Try/all

README.md

@@ -256,8 +256,8 @@ print(parameters.get("ID_Ba_Hz_akt").options()) # returns a list of possible val
 
 # Now we increase the heating controller target temperature by 2 Kelvin
 heating_offset = l.holdings.get(2)    # Get an object for the offset
-heating_offset.value = 2.0            # Set the desired value
-l.holdings["heating_mode"] = "Offset" # Set the value to activate the offset mode
+heating_offset.value = 2.0            # Queue the desired value by setting the field's value
+l.holdings["heating_mode"] = "Offset" # Queue the value to activate the offset mode
 l.write()                             # Write down the values to the heatpump
 ```
 

luxtronik/cfi/calculations.py

@@ -4,10 +4,15 @@
 from typing import Final
 
 from luxtronik.definitions import LuxtronikDefinitionsList
-from luxtronik.definitions.calculations import CALCULATIONS_DEFINITIONS_LIST, CALCULATIONS_OFFSET
+from luxtronik.definitions.calculations import (
+    CALCULATIONS_DEFINITIONS_LIST,
+    CALCULATIONS_OFFSET,
+    CALCULATIONS_DEFAULT_DATA_TYPE,
+    CALCULATIONS_OUTDATED,
+)
 
 from luxtronik.cfi.constants import CALCULATIONS_FIELD_NAME
-from luxtronik.data_vector import DataVector
+from luxtronik.cfi.vector import DataVectorConfig
 from luxtronik.datatypes import Base
 
 
@@ -16,32 +21,20 @@
 CALCULATIONS_DEFINITIONS: Final = LuxtronikDefinitionsList(
     CALCULATIONS_DEFINITIONS_LIST,
     CALCULATIONS_FIELD_NAME,
-    CALCULATIONS_OFFSET
+    CALCULATIONS_OFFSET,
+    CALCULATIONS_DEFAULT_DATA_TYPE
 )
 
-class Calculations(DataVector):
+class Calculations(DataVectorConfig):
     """Class that holds all calculations."""
 
-    logger = LOGGER
     name = CALCULATIONS_FIELD_NAME
     definitions = CALCULATIONS_DEFINITIONS
-
-    _obsolete = {
-        "ID_WEB_SoftStand": "get_firmware_version()"
-    }
-
-    def __init__(self):
-        super().__init__()
-        for d in CALCULATIONS_DEFINITIONS:
-            self._data.add(d, d.create_field())
-
-    @property
-    def calculations(self):
-        return self._data
+    _outdated = CALCULATIONS_OUTDATED
 
     def get_firmware_version(self):
         """Get the firmware version as string."""
-        return "".join([super(Calculations, self).get(i).value for i in range(81, 91)])
+        return "".join([str(super(Calculations, self).get(i).value) for i in range(81, 91)])
 
     def _get_firmware_version(self):
         """Get the firmware version as string like in previous versions."""
@@ -50,7 +43,7 @@ def _get_firmware_version(self):
     def get(self, target):
         """Treats certain names specially. For all others, the function of the base class is called."""
         if target == "ID_WEB_SoftStand":
-            self.logger.debug("The name 'ID_WEB_SoftStand' is obsolete! Use 'get_firmware_version()' instead.")
+            LOGGER.debug("The name 'ID_WEB_SoftStand' is obsolete! Use 'get_firmware_version()' instead.")
             entry = Base("ID_WEB_SoftStand")
             entry.raw = self._get_firmware_version()
             return entry

luxtronik/cfi/constants.py

@@ -26,4 +26,7 @@
 
 # Wait time (in seconds) after writing parameters to give controller
 # some time to re-calculate values, etc.
-WAIT_TIME_AFTER_PARAMETER_WRITE = 1
+WAIT_TIME_AFTER_PARAMETER_WRITE = 1
+
+# The data from the config interface are transmitted in 32-bit chunks.
+LUXTRONIK_CFI_REGISTER_BIT_SIZE: Final = 32

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
@@ -161,23 +162,25 @@ def _write_and_read(self, parameters, data):
         return self._read(data)
 
     def _write(self, parameters):
-        for index, value in parameters.queue.items():
-            if not isinstance(index, int) or not isinstance(value, int):
-                LOGGER.warning(
-                    "%s: Parameter id '%s' or value '%s' invalid!",
-                    self._host,
-                    index,
-                    value,
-                )
-                continue
-            LOGGER.info("%s: Parameter '%d' set to '%s'", self._host, index, value)
-            self._send_ints(LUXTRONIK_PARAMETERS_WRITE, index, value)
-            cmd = self._read_int()
-            LOGGER.debug("%s: Command %s", self._host, cmd)
-            val = self._read_int()
-            LOGGER.debug("%s: Value %s", self._host, val)
-        # Flush queue after writing all values
-        parameters.queue = {}
+        for definition, field in parameters.items():
+            if field.write_pending:
+                value = field.raw
+                if not isinstance(definition.index, int) or not isinstance(value, int):
+                    LOGGER.warning(
+                        "%s: Parameter id '%s' or value '%s' invalid!",
+                        self._host,
+                        definition.index,
+                        value,
+                    )
+                    field.write_pending = False
+                    continue
+                LOGGER.info("%s: Parameter '%d' set to '%s'", self._host, definition.index, value)
+                self._send_ints(LUXTRONIK_PARAMETERS_WRITE, definition.index, value)
+                cmd = self._read_int()
+                LOGGER.debug("%s: Command %s", self._host, cmd)
+                val = self._read_int()
+                LOGGER.debug("%s: Value %s", self._host, val)
+            field.write_pending = False
         # Give the heatpump a short time to handle the value changes/calculations:
         time.sleep(WAIT_TIME_AFTER_PARAMETER_WRITE)
 
@@ -191,7 +194,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 +209,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 +222,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 +259,38 @@ 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:
+            raw_data (list[int]): List of raw register values.
+                The raw data must start at register index 0.
+            num_bits (int): Number of bits per register.
+        """
+        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.items():
+            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
+                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/cfi/parameters.py

@@ -4,50 +4,29 @@
 from typing import Final
 
 from luxtronik.definitions import LuxtronikDefinitionsList
-from luxtronik.definitions.parameters import PARAMETERS_DEFINITIONS_LIST, PARAMETERS_OFFSET
+from luxtronik.definitions.parameters import (
+    PARAMETERS_DEFINITIONS_LIST,
+    PARAMETERS_OFFSET,
+    PARAMETERS_DEFAULT_DATA_TYPE,
+    PARAMETERS_OUTDATED,
+)
 
 from luxtronik.cfi.constants import PARAMETERS_FIELD_NAME
-from luxtronik.data_vector import DataVector
+from luxtronik.cfi.vector import DataVectorConfig
 
 
 LOGGER = logging.getLogger(__name__)
 
 PARAMETERS_DEFINITIONS: Final = LuxtronikDefinitionsList(
     PARAMETERS_DEFINITIONS_LIST,
     PARAMETERS_FIELD_NAME,
-    PARAMETERS_OFFSET
+    PARAMETERS_OFFSET,
+    PARAMETERS_DEFAULT_DATA_TYPE
 )
 
-class Parameters(DataVector):
+class Parameters(DataVectorConfig):
     """Class that holds all parameters."""
 
-    logger = LOGGER
     name = PARAMETERS_FIELD_NAME
     definitions = PARAMETERS_DEFINITIONS
-
-    def __init__(self, safe=True):
-        """Initialize parameters class."""
-        super().__init__()
-        self.safe = safe
-        self.queue = {}
-        for d in PARAMETERS_DEFINITIONS:
-            self._data.add(d, d.create_field())
-
-    @property
-    def parameters(self):
-        return self._data
-
-    def set(self, target, value):
-        """Set parameter to new value."""
-        index, parameter = self._lookup(target, with_index=True)
-        if index is not None:
-            if parameter.writeable or not self.safe:
-                raw = parameter.to_heatpump(value)
-                if isinstance(raw, int):
-                    self.queue[index] = raw
-                else:
-                    self.logger.error("Value '%s' for Parameter '%s' not valid!", value, parameter.name)
-            else:
-                self.logger.warning("Parameter '%s' not safe for writing!", parameter.name)
-        else:
-            self.logger.warning("Parameter '%s' not found", target)
+    _outdated = PARAMETERS_OUTDATED

luxtronik/cfi/vector.py

@@ -0,0 +1,83 @@
+
+import logging
+
+from luxtronik.data_vector import DataVector
+
+
+LOGGER = logging.getLogger(__name__)
+
+###############################################################################
+# Configuration interface data-vector
+###############################################################################
+
+class DataVectorConfig(DataVector):
+    """Specialized DataVector for Luxtronik configuration fields."""
+
+    def _init_instance(self, safe):
+        """Re-usable method to initialize all instance variables."""
+        super()._init_instance(safe)
+
+    def __init__(self, safe=True):
+        """
+        Initialize the data-vector instance.
+        Creates field objects for definitions and stores them in the data vector.
+
+        Args:
+            safe (bool): If true, prevent fields marked as
+                not secure from being written to.
+        """
+        self._init_instance(safe)
+
+        # Add all available fields
+        for d in self.definitions:
+            self._data.add(d, d.create_field())
+
+    @classmethod
+    def empty(cls, safe=True):
+        """
+        Initialize the data-vector instance without any fields.
+
+        Args:
+            safe (bool): If true, prevent fields marked as
+                not secure from being written to.
+        """
+        obj = cls.__new__(cls) # this don't call __init__()
+        obj._init_instance(safe)
+        return obj
+
+    def add(self, def_field_name_or_idx, alias=None):
+        """
+        Adds an additional field to this data vector.
+        Mainly used for data vectors created via `empty()`
+        to read/write individual fields. Existing fields will not be overwritten.
+
+        Args:
+            def_field_name_or_idx (LuxtronikDefinition | Base | str | int):
+                Field to add. Either by definition, name or index, or the field itself.
+            alias (Hashable | None): Alias, which can be used to access the field again.
+
+        Returns:
+            Base | None: The added field object if this could be added or
+                the existing field, otherwise None. In case a field
+
+        Note:
+            It is not possible to add fields which are not defined.
+            To add custom fields, add them to the used `LuxtronikDefinitionsList`
+            (`cls.definitions`) first.
+            If multiple fields added for the same index/name, the last added takes precedence.
+        """
+        # Look-up the related definition
+        definition, field = self._get_definition(def_field_name_or_idx, True)
+        if definition is None:
+            return None
+
+        # Check if the field already exists
+        existing_field = self._data.get(definition, None)
+        if existing_field is not None:
+            return existing_field
+
+        # Add a (new) field
+        if field is None:
+            field = definition.create_field()
+        self._data.add_sorted(definition, field, alias)
+        return field

luxtronik/cfi/visibilities.py

@@ -4,32 +4,29 @@
 from typing import Final
 
 from luxtronik.definitions import LuxtronikDefinitionsList
-from luxtronik.definitions.visibilities import VISIBILITIES_DEFINITIONS_LIST, VISIBILITIES_OFFSET
+from luxtronik.definitions.visibilities import (
+    VISIBILITIES_DEFINITIONS_LIST,
+    VISIBILITIES_OFFSET,
+    VISIBILITIES_DEFAULT_DATA_TYPE,
+    VISIBILITIES_OUTDATED,
+)
 
 from luxtronik.cfi.constants import VISIBILITIES_FIELD_NAME
-from luxtronik.data_vector import DataVector
+from luxtronik.cfi.vector import DataVectorConfig
 
 
 LOGGER = logging.getLogger(__name__)
 
 VISIBILITIES_DEFINITIONS: Final = LuxtronikDefinitionsList(
     VISIBILITIES_DEFINITIONS_LIST,
     VISIBILITIES_FIELD_NAME,
-    VISIBILITIES_OFFSET
+    VISIBILITIES_OFFSET,
+    VISIBILITIES_DEFAULT_DATA_TYPE,
 )
 
-class Visibilities(DataVector):
+class Visibilities(DataVectorConfig):
     """Class that holds all visibilities."""
 
-    logger = LOGGER
     name = VISIBILITIES_FIELD_NAME
     definitions = VISIBILITIES_DEFINITIONS
-
-    def __init__(self):
-        super().__init__()
-        for d in VISIBILITIES_DEFINITIONS:
-            self._data.add(d, d.create_field())
-
-    @property
-    def visibilities(self):
-        return self._data
+    _outdated = VISIBILITIES_OUTDATED