Move parse() into the cfi-interface

luxtronik/cfi/interface.py

@@ -15,6 +15,7 @@
     LUXTRONIK_SOCKET_READ_SIZE_INTEGER,
     LUXTRONIK_SOCKET_READ_SIZE_CHAR,
     WAIT_TIME_AFTER_PARAMETER_WRITE,
+    LUXTRONIK_CFI_REGISTER_BIT_SIZE,
 )
 from luxtronik.cfi.calculations import Calculations
 from luxtronik.cfi.parameters import Parameters
@@ -191,7 +192,7 @@ def _read_parameters(self, parameters):
         for _ in range(0, length):
             data.append(self._read_int())
         LOGGER.info("%s: Read %d parameters", self._host, length)
-        parameters.parse(data)
+        self._parse(parameters, data)
         return parameters
 
     def _read_calculations(self, calculations):
@@ -206,7 +207,7 @@ def _read_calculations(self, calculations):
         for _ in range(0, length):
             data.append(self._read_int())
         LOGGER.info("%s: Read %d calculations", self._host, length)
-        calculations.parse(data)
+        self._parse(calculations, data)
         return calculations
 
     def _read_visibilities(self, visibilities):
@@ -219,7 +220,7 @@ def _read_visibilities(self, visibilities):
         for _ in range(0, length):
             data.append(self._read_char())
         LOGGER.info("%s: Read %d visibilities", self._host, length)
-        visibilities.parse(data)
+        self._parse(visibilities, data)
         return visibilities
 
     def _send_ints(self, *ints):
@@ -256,4 +257,40 @@ def _read_int(self):
     def _read_char(self):
         "Low-level helper to receive a signed int"
         reading = self._read_bytes(LUXTRONIK_SOCKET_READ_SIZE_CHAR)
-        return struct.unpack(">b", reading)[0]
+        return struct.unpack(">b", reading)[0]
+
+    def _parse(self, data_vector, raw_data):
+        """
+        Parse raw data into the corresponding fields.
+
+        Args:
+            data_vector (DataVector): Data vector in which
+                the raw data is to be integrated.
+            raw_data (list[int]): List of raw register values.
+                The raw data must start at register index 0.
+        """
+        raw_len = len(raw_data)
+        # Prepare a list of undefined indices
+        undefined = {i for i in range(0, raw_len)}
+
+        # integrate the data into the fields
+        for pair in data_vector.data.pairs():
+            definition, field = pair
+            # skip this field if there are not enough data
+            next_idx = definition.index + definition.count
+            if next_idx > raw_len:
+                # not enough registers
+                field.raw = None
+                continue
+            # remove all used indices from the list of undefined indices
+            for index in range(definition.index, next_idx):
+                undefined.discard(index)
+            pair.integrate_data(raw_data, LUXTRONIK_CFI_REGISTER_BIT_SIZE)
+
+        # create an unknown field for additional data
+        for index in undefined:
+            # LOGGER.warning(f"Entry '%d' not in list of {self.name}", index)
+            definition = data_vector.definitions.create_unknown_definition(index)
+            field = definition.create_field()
+            field.raw = raw_data[index]
+            data_vector.data.add_sorted(definition, field)

luxtronik/data_vector.py

@@ -8,7 +8,6 @@
 )
 
 from luxtronik.collections import LuxtronikFieldsDictionary
-from luxtronik.definitions import LuxtronikDefinition
 
 
 LOGGER = logging.getLogger(__name__)
@@ -38,19 +37,6 @@ def __iter__(self):
     def data(self):
         return self._data
 
-    def parse(self, raw_data):
-        """Parse raw data."""
-        for index, data in enumerate(raw_data):
-            entry = self._data.get(index, None)
-            if entry is not None:
-                entry.raw = data
-            else:
-                # self.logger.warning(f"Entry '%d' not in list of {self.name}", index)
-                definition = LuxtronikDefinition.unknown(index, self.name, 0)
-                field = definition.create_field()
-                field.raw = data
-                self._data.add_sorted(definition, field)
-
     def _name_lookup(self, name):
         """
         Try to find the index using the given field name.

luxtronik/shi/vector.py

@@ -2,12 +2,12 @@
 import logging
 
 from luxtronik.common import version_in_range
-from luxtronik.collections import integrate_data, LuxtronikFieldsDictionary
+from luxtronik.collections import LuxtronikFieldsDictionary
 from luxtronik.data_vector import DataVector
 from luxtronik.datatypes import Base, Unknown
 from luxtronik.definitions import LuxtronikDefinition
 
-from luxtronik.shi.constants import LUXTRONIK_LATEST_SHI_VERSION, LUXTRONIK_SHI_REGISTER_BIT_SIZE
+from luxtronik.shi.constants import LUXTRONIK_LATEST_SHI_VERSION
 from luxtronik.shi.contiguous import ContiguousDataBlockList
 
 
@@ -292,20 +292,6 @@ def update_read_blocks(self):
 
 # Data and access methods #####################################################
 
-    def parse(self, raw_data):
-        """
-        Parse raw data into the corresponding fields.
-
-        Args:
-            raw_data (list[int]): List of raw register values.
-                The raw data must start at register index 0.
-        """
-        raw_len = len(raw_data)
-        for definition, field in self._data.pairs():
-            if definition.index + definition.count >= raw_len:
-                continue
-            integrate_data(definition, field, raw_data, LUXTRONIK_SHI_REGISTER_BIT_SIZE)
-
     def get(self, def_name_or_idx, default=None):
         """
         Retrieve a field by definition, name or register index.

tests/cfi/test_cfi_interface.py

@@ -0,0 +1,52 @@
+
+from luxtronik import (
+  Parameters,
+  Calculations,
+  Visibilities,
+  LuxtronikSocketInterface,
+)
+
+
+class TestLuxtronikSocketInterface:
+
+    def test_parse(self):
+        lux = LuxtronikSocketInterface('host')
+        parameters = Parameters()
+        calculations = Calculations()
+        visibilities = Visibilities()
+
+        n = 2000
+        t = list(range(0, n + 1))
+
+        lux._parse(parameters, t)
+        p = parameters.get(n)
+        assert p.name == f"unknown_parameter_{n}"
+        assert p.raw == n
+
+        lux._parse(calculations, t)
+        c = calculations.get(n)
+        assert c.name == f"unknown_calculation_{n}"
+        assert c.raw == n
+
+        lux._parse(visibilities, t)
+        v = visibilities.get(n)
+        assert v.name == f"unknown_visibility_{n}"
+        assert v.raw == n
+
+        n = 10
+        t = list(range(0, n + 1))
+
+        lux._parse(parameters, t)
+        for definition, field in parameters.data.pairs():
+            if definition.index > n:
+                assert field.raw is None
+
+        lux._parse(calculations, t)
+        for definition, field in calculations.data.pairs():
+            if definition.index > n:
+                assert field.raw is None
+
+        lux._parse(visibilities, t)
+        for definition, field in visibilities.data.pairs():
+            if definition.index > n:
+                assert field.raw is None

tests/cfi/test_cfi_parameters.py

@@ -66,19 +66,6 @@ def test__lookup(self):
         j = 0.0
         assert parameters._lookup(j) is None
 
-    def test_parse(self):
-        """Test cases for _parse"""
-        parameters = Parameters()
-
-        n = 2000
-        t = list(range(0, n + 1))
-        parameters.parse(t)
-
-        p = parameters.get(n)
-
-        assert p.name == f"unknown_parameter_{n}"
-        assert p.raw == n
-
     def test___iter__(self):
         """Test cases for __iter__"""
         parameters = Parameters()

tests/shi/test_shi_vector.py

@@ -309,33 +309,6 @@ def test_set(self):
         assert field_9.value == 6
         assert field_9.write_pending
 
-    def test_parse(self):
-        data_vector = DataVectorTest(parse_version("1.1.2"))
-        field_5 = data_vector[5]
-        field_9 = data_vector[9]
-        field_9a = data_vector['field_9a']
-
-        # not enough data
-        data = [1]
-        data_vector.parse(data)
-        assert field_5.value is None
-        assert field_9.value is None
-        assert field_9a.value is None
-
-        # data only for field 5
-        data = [1, 2, 3, 4, 5, 6, 7]
-        data_vector.parse(data)
-        assert field_5.value == 6
-        assert field_9.value is None
-        assert field_9a.value is None
-
-        # data for all fields
-        data = [9, 8, 7, 6, 5, 4, 3, 2, 1, 0, -1, -2]
-        data_vector.parse(data)
-        assert field_5.value == 4
-        assert field_9.value == [0, -1]
-        assert field_9a.value == 0
-
     def test_alias(self):
         TEST_DEFINITIONS.register_alias('field_9a', 10)
         data_vector = DataVectorTest(parse_version("1.1.2"))

tests/test_socket_interaction.py

@@ -3,6 +3,7 @@
 import unittest.mock as mock
 
 from luxtronik import Luxtronik, LuxtronikSocketInterface, Parameters, Calculations, Visibilities
+from luxtronik.collections import integrate_data
 from tests.fake import (
     fake_create_connection,
     fake_parameter_value,
@@ -16,6 +17,7 @@
 @mock.patch("socket.create_connection", fake_create_connection)
 @mock.patch("luxtronik.LuxtronikModbusTcpInterface", FakeModbus)
 class TestSocketInteraction:
+
     def check_luxtronik_data(self, lux, check_for_true=True):
         cp = self.check_data_vector(lux.parameters)
         cc = self.check_data_vector(lux.calculations)
@@ -32,8 +34,12 @@ def check_data_vector(self, data_vector):
             fct = fake_calculation_value
         elif type(data_vector) is Visibilities:
             fct = fake_visibility_value
-        for idx, entry in data_vector:
-            if entry.raw != fct(idx):
+        for d, f in data_vector.data.pairs():
+            # get raw data
+            raw = [fct(idx) for idx in range(d.index, d.index + d.count)]
+            temp_field = d.create_field()
+            integrate_data(d, temp_field, raw, 32, 0)
+            if f.raw != temp_field.raw:
                 return False
         return True