Merge pull request #45 from KLBa/add_tab_intp

andlaus · web-flow · commit a0f3f910c6c1 · 2022-06-08T16:15:05.000+02:00
Add `TAB-INTP` compu method
diff --git a/odxtools/compumethods/readcompumethod.py b/odxtools/compumethods/readcompumethod.py
@@ -168,7 +168,16 @@ def read_compu_method_from_odx(et_element, internal_type: DataType, physical_typ
         return ScaleLinearCompuMethod(linear_methods)
 
     elif compu_category == "TAB-INTP":
-        return TabIntpCompuMethod(internal_type=internal_type, physical_type=physical_type)
+
+        scales = et_element.findall(
+            "COMPU-INTERNAL-TO-PHYS/COMPU-SCALES/COMPU-SCALE")
+        internal = [scale.findtext("LOWER-LIMIT") for scale in scales]
+        physical = [scale.findtext("COMPU-CONST/V") for scale in scales]
+
+        internal = [internal_type.from_string(x) for x in internal]
+        physical = [physical_type.from_string(x) for x in physical]
+
+        return TabIntpCompuMethod(internal_type=internal_type, physical_type=physical_type, internal_points=internal, physical_points=physical)
 
     # TODO: Implement other categories (never instantiate CompuMethod)
     logger.warning(
diff --git a/odxtools/compumethods/tabintpcompumethod.py b/odxtools/compumethods/tabintpcompumethod.py
@@ -2,28 +2,148 @@
 # Copyright (c) 2022 MBition GmbH
 
 
+from typing import List, Tuple, Union
+
+from ..exceptions import EncodeError, DecodeError
 from ..globals import logger
+from ..odxtypes import DataType
 
 from .compumethodbase import CompuMethod
+from .limit import IntervalType, Limit
 
 
 class TabIntpCompuMethod(CompuMethod):
+    """
+    A compu method of type Tab Interpolated is used for linear interpolation.
+
+    A `TabIntpCompuMethod` is defined by a set of points. Each point is an (internal, physical) value pair.
+    When converting from internal to physical or vice-versa, the result is linearly interpolated.
+
+    The function defined by a `TabIntpCompuMethod` is similar to the one of a `ScaleLinearCompuMethod` with the following differences:
+
+    * `TabIntpCompuMethod`s are always continuous whereas `ScaleLinearCompuMethod` might have jumps
+    * `TabIntpCompuMethod`s are always invertible: Even if the linear interpolation is not monotonic, the first matching interval is taken.
+
+    Refer to ASAM MCD-2 D (ODX) Specification, section 7.3.6.6.8 for details.
+
+    Examples
+    --------
+
+    Create a TabIntpCompuMethod defined by the points (0, -1), (10, 1), (30, 2)::
+
+        method = TabIntpCompuMethod(
+            internal_type=DataType.A_UINT32,
+            physical_type=DataType.A_UINT32,
+            internal_points=[0, 10, 30],
+            physical_points=[-1, 1, 2]
+        )
+
+    Note that the points are given as two lists. The equivalent odx definition is::
+
+        <COMPU-METHOD>
+            <CATEGORY>TAB-INTP</CATEGORY>
+            <COMPU-INTERNAL-TO-PHYS>
+                <COMPU-SCALES>
+                    <COMPU-SCALE>
+                        <LOWER-LIMIT INTERVAL-TYPE = "CLOSED">0</LOWER-LIMIT>
+                        <COMPU-CONST>
+                            <V>-1</V>
+                        </COMPU-CONST>
+                    </COMPU-SCALE>
+                    <COMPU-SCALE>
+                        <LOWER-LIMIT INTERVAL-TYPE = "CLOSED">10</LOWER-LIMIT>
+                        <COMPU-CONST>
+                            <V>1</V>
+                        </COMPU-CONST>
+                    </COMPU-SCALE>
+                    <COMPU-SCALE>
+                        <LOWER-LIMIT INTERVAL-TYPE = "CLOSED">30</LOWER-LIMIT>
+                        <COMPU-CONST>
+                            <V>2</V>
+                        </COMPU-CONST>
+                    </COMPU-SCALE>
+                </COMPU-SCALES>
+            </COMPU-INTERNAL-TO-PHYS>
+        </COMPU-METHOD>
+
+    """
+
     def __init__(self,
-                 internal_type,
-                 physical_type):
+                 internal_type: Union[DataType, str],
+                 physical_type: Union[DataType, str],
+                 internal_points: List[Union[float, int]],
+                 physical_points: List[Union[float, int]]):
         super().__init__(internal_type, physical_type, "TAB-INTP")
-        logger.debug("Created table interpolation compu method!")
-        logger.warning(
-            "TODO: Implement table interpolation compu method properly!")
 
-    def convert_physical_to_internal(self, physical_value):
-        return self.internal_type.make_from(physical_value)
+        self.internal_points = internal_points
+        self.physical_points = physical_points
+
+        self._physical_lower_limit = Limit(
+            min(physical_points), IntervalType.CLOSED)
+        self._physical_upper_limit = Limit(
+            max(physical_points), IntervalType.CLOSED)
+
+        logger.debug("Created compu method of type tab interpolated !")
+        self._assert_validity()
+
+    @property
+    def physical_lower_limit(self) -> Limit:
+        return self._physical_lower_limit
+
+    @property
+    def physical_upper_limit(self) -> Limit:
+        return self._physical_upper_limit
+
+    def _assert_validity(self) -> None:
+        assert len(self.internal_points) == len(self.physical_points)
+
+        assert self.internal_type in [DataType.A_INT32, DataType.A_UINT32,
+                                      DataType.A_FLOAT32, DataType.A_FLOAT64], \
+            ("Internal data type of tab-intp compumethod must be one of"
+             " [DataType.A_INT32, DataType.A_UINT32, DataType.A_FLOAT32, DataType.A_FLOAT64]")
+        assert self.physical_type in [DataType.A_INT32, DataType.A_UINT32,
+                                      DataType.A_FLOAT32, DataType.A_FLOAT64], \
+            ("Physical data type of tab-intp compumethod must be one of"
+             " [DataType.A_INT32, DataType.A_UINT32, DataType.A_FLOAT32, DataType.A_FLOAT64]")
+
+    def _piecewise_linear_interpolate(self,
+                                      x: Union[int, float],
+                                      points: List[Tuple[Union[int, float], Union[int, float]]]) \
+            -> Union[float, None]:
+        for ((x0, y0), (x1, y1)) in zip(points[:-1], points[1:]):
+            if x0 <= x and x <= x1:
+                return y0 + (x - x0) * (y1 - y0) / (x1 - x0)
+
+        return None
+
+    def convert_physical_to_internal(self, physical_value: Union[int, float]) -> Union[int, float]:
+        reference_points = list(zip(
+            self.physical_points, self.internal_points))
+        result = self._piecewise_linear_interpolate(
+            physical_value, reference_points)
+
+        if result is None:
+            raise EncodeError(f"Internal value {physical_value} must be inside the range"
+                              f" [{min(self.physical_points)}, {max(self.physical_points)}]")
+        res = self.internal_type.make_from(result)
+        assert isinstance(res, (int, float))
+        return res
+
+    def convert_internal_to_physical(self, internal_value: Union[int, float]) -> Union[int, float]:
+        reference_points = list(zip(
+            self.internal_points, self.physical_points))
+        result = self._piecewise_linear_interpolate(
+            internal_value, reference_points)
 
-    def convert_internal_to_physical(self, internal_value):
-        return self.physical_type.make_from(internal_value)
+        if result is None:
+            raise DecodeError(f"Internal value {internal_value} must be inside the range"
+                              f" [{min(self.internal_points)}, {max(self.internal_points)}]")
+        res = self.physical_type.make_from(result)
+        assert isinstance(res, (int, float))
+        return res
 
-    def is_valid_physical_value(self, physical_value):
-        return True
+    def is_valid_physical_value(self, physical_value: Union[int, float]) -> bool:
+        return min(self.physical_points) <= physical_value and physical_value <= max(self.physical_points)
 
-    def is_valid_internal_value(self, internal_value):
-        return True
+    def is_valid_internal_value(self, internal_value: Union[int, float]) -> bool:
+        return min(self.internal_points) <= internal_value and internal_value <= max(self.internal_points)
diff --git a/odxtools/pdx_stub/macros/printDOP.tpl b/odxtools/pdx_stub/macros/printDOP.tpl
@@ -137,6 +137,19 @@
   {%- endfor %}
 		</COMPU-SCALES>
  </COMPU-INTERNAL-TO-PHYS>
+ {%- elif cm.category == "TAB-INTP" %}
+ <COMPU-INTERNAL-TO-PHYS>
+  <COMPU-SCALES>
+  {%- for idx in range( cm.internal_points | length ) %}
+   <COMPU-SCALE>
+    <LOWER-LIMIT INTERVAL-TYPE="CLOSED">{{ cm.internal_points[idx] }}</LOWER-LIMIT>
+    <COMPU-CONST>
+     <V>{{ cm.physical_points[idx] }}</V>
+    </COMPU-CONST>
+   </COMPU-SCALE>
+  {%- endfor %}
+  </COMPU-SCALES>
+ </COMPU-INTERNAL-TO-PHYS>
  {%- endif %}
 </COMPU-METHOD>
 {%- endmacro -%}
diff --git a/tests/test_compu_methods.py b/tests/test_compu_methods.py
@@ -1,13 +1,22 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2022 MBition GmbH
 
+import inspect
+import os
 import unittest
+from xml.etree import ElementTree
 
-from odxtools.compumethods import Limit, LinearCompuMethod, IntervalType
+import jinja2
+import odxtools
+
+from odxtools.compumethods import Limit, LinearCompuMethod, IntervalType, TabIntpCompuMethod
+from odxtools.compumethods.readcompumethod import read_compu_method_from_odx
+from odxtools.exceptions import DecodeError, EncodeError
+from odxtools.odxtypes import DataType
 
 
 class TestLinearCompuMethod(unittest.TestCase):
-    def test_linear_compu_method(self):
+    def test_linear_compu_method_type_int_int(self):
         compu_method = LinearCompuMethod(1, 3, "A_INT32", "A_INT32")
 
         self.assertEqual(compu_method.convert_internal_to_physical(4), 13)
@@ -90,5 +99,117 @@ def test_linear_compu_method_physical_limits(self):
         self.assertFalse(compu_method.is_valid_physical_value(-13))
 
 
+class TestTabIntpCompuMethod(unittest.TestCase):
+    def setUp(self) -> None:
+        """Prepares the jinja environment and the sample tab-intp compumethod"""
+
+        def _get_jinja_environment():
+            __module_filname = inspect.getsourcefile(odxtools)
+            assert isinstance(__module_filname, str)
+            stub_dir = os.path.sep.join([os.path.dirname(__module_filname),
+                                        "pdx_stub"])
+
+            jinja_env = jinja2.Environment(
+                loader=jinja2.FileSystemLoader(stub_dir))
+
+            # allows to put XML attributes on a separate line while it is
+            # collapsed with the previous line in the rendering
+            jinja_env.filters["odxtools_collapse_xml_attribute"] = lambda x: " " + \
+                x.strip() if x.strip() else ""
+            return jinja_env
+
+        self.jinja_env = _get_jinja_environment()
+
+        self.compumethod = TabIntpCompuMethod(DataType.A_INT32,
+                                              DataType.A_FLOAT32,
+                                              internal_points=[0, 10, 30],
+                                              physical_points=[-1, 1, 2]
+                                              )
+
+        self.compumethod_odx = f"""
+        <COMPU-METHOD>
+            <CATEGORY>TAB-INTP</CATEGORY>
+            <COMPU-INTERNAL-TO-PHYS>
+                <COMPU-SCALES>
+                    <COMPU-SCALE>
+                        <LOWER-LIMIT INTERVAL-TYPE="CLOSED">{self.compumethod.internal_points[0]}</LOWER-LIMIT>
+                        <COMPU-CONST>
+                            <V>{self.compumethod.physical_points[0]}</V>
+                        </COMPU-CONST>
+                    </COMPU-SCALE>
+                    <COMPU-SCALE>
+                        <LOWER-LIMIT INTERVAL-TYPE="CLOSED">{self.compumethod.internal_points[1]}</LOWER-LIMIT>
+                        <COMPU-CONST>
+                            <V>{self.compumethod.physical_points[1]}</V>
+                        </COMPU-CONST>
+                    </COMPU-SCALE>
+                    <COMPU-SCALE>
+                        <LOWER-LIMIT INTERVAL-TYPE="CLOSED">{self.compumethod.internal_points[2]}</LOWER-LIMIT>
+                        <COMPU-CONST>
+                            <V>{self.compumethod.physical_points[2]}</V>
+                        </COMPU-CONST>
+                    </COMPU-SCALE>
+                </COMPU-SCALES>
+            </COMPU-INTERNAL-TO-PHYS>
+        </COMPU-METHOD>
+        """
+
+    def test_tabintp_convert_type_int_float(self):
+        method = self.compumethod
+
+        for internal, physical in [
+            (0, -1),
+            (2, -0.6),
+            (3, -0.4),
+            (5, 0),
+            (10, 1),
+            (20, 1.5),
+            (25, 1.75),
+            (30, 2)
+        ]:
+            self.assertTrue(method.is_valid_internal_value(internal))
+            self.assertTrue(method.is_valid_physical_value(physical))
+            self.assertEqual(method.convert_internal_to_physical(internal),
+                             physical)
+            self.assertEqual(method.convert_physical_to_internal(physical),
+                             internal)
+
+        self.assertRaises(DecodeError,
+                          method.convert_internal_to_physical, -2)
+        self.assertRaises(DecodeError,
+                          method.convert_internal_to_physical, 31)
+        self.assertRaises(EncodeError,
+                          method.convert_physical_to_internal, -2)
+        self.assertRaises(EncodeError,
+                          method.convert_physical_to_internal, 2.1)
+
+    def test_read_odx(self):
+        expected = self.compumethod
+
+        et_element = ElementTree.fromstring(self.compumethod_odx)
+        actual = read_compu_method_from_odx(et_element,
+                                            expected.internal_type,
+                                            expected.physical_type)
+        self.assertIsInstance(actual, TabIntpCompuMethod)
+        self.assertEqual(expected.physical_type, actual.physical_type)
+        self.assertEqual(expected.internal_type, actual.internal_type)
+        self.assertEqual(expected.internal_points, actual.internal_points)
+        self.assertEqual(expected.physical_points, actual.physical_points)
+
+    def test_write_odx(self):
+        dlc_tpl = self.jinja_env.get_template("macros/printDOP.tpl")
+        module = dlc_tpl.make_module()
+
+        out = module.printCompuMethod(self.compumethod)
+
+        expected_odx = self.compumethod_odx
+
+        # We ignore spaces
+        def remove_spaces(string):
+            return "".join(string.split())
+
+        self.assertEqual(remove_spaces(out), remove_spaces(expected_odx))
+
+
 if __name__ == '__main__':
     unittest.main()
