Merge pull request #129 from gerw/common_base_class

Common base class for Parameters,Calculations,Visibilities

luxtronik/calculations.py

@@ -1,6 +1,8 @@
 """Parse luxtronik calculations."""
 import logging
 
+from luxtronik.data_vector import DataVector
+
 from luxtronik.datatypes import (
     BivalenceLevel,
     Bool,
@@ -33,14 +35,16 @@
     Voltage,
 )
 
-LOGGER = logging.getLogger("Luxtronik.Calculations")
-
 
-class Calculations:
+class Calculations(DataVector):
     """Class that holds all calculations."""
 
+    logger = logging.getLogger("Luxtronik.Calculations")
+    name = "Calculation"
+
     def __init__(self):
-        self._calculations = {
+        super().__init__()
+        self._data = {
             0: Unknown("Unknown_Calculation_0"),
             1: Unknown("Unknown_Calculation_1"),
             2: Unknown("Unknown_Calculation_2"),
@@ -302,40 +306,3 @@ def __init__(self):
             258: MajorMinorVersion("RBE_Version"),
             259: Unknown("Unknown_Calculation_259"),
         }
-
-    def __iter__(self):
-        return iter(self._calculations.items())
-
-    def parse(self, raw_data):
-        """Parse raw calculations data."""
-        for index, data in enumerate(raw_data):
-            calculation = self._calculations.get(index, False)
-            if calculation is not False:
-                calculation.raw = data
-            else:
-                # LOGGER.warning("Calculation '%d' not in list of calculations", index)
-                calculation = Unknown(f"Unknown_Calculation_{index}")
-                calculation.raw = data
-                self._calculations[index] = calculation
-
-    def _lookup(self, target):
-        """Lookup calculation by either id or name."""
-        # Get calculation by id
-        if isinstance(target, int):
-            return self._calculations.get(target, None)
-        # Get calculation by name
-        if isinstance(target, str):
-            try:
-                target = int(target)
-                return self._calculations.get(target, None)
-            except ValueError:
-                for _, calculation in self._calculations.items():
-                    if calculation.name == target:
-                        return calculation
-        LOGGER.warning("Calculation '%s' not found", target)
-        return None
-
-    def get(self, target):
-        """Get calculation by id or name."""
-        calculation = self._lookup(target)
-        return calculation

luxtronik/data_vector.py

@@ -0,0 +1,67 @@
+"""Provides a base class for parameters, calculations, visibilities."""
+import logging
+
+from luxtronik.datatypes import Unknown
+
+
+class DataVector:
+    """Class that holds a vector of data entries."""
+
+    logger = logging.getLogger("Luxtronik.DataVector")
+    name = "DataVector"
+
+    def __init__(self):
+        """Initialize DataVector class."""
+        self._data = {}
+
+    def __iter__(self):
+        """Iterator for the data entries."""
+        return iter(self._data.items())
+
+    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)
+                entry = Unknown(f"Unknown_{self.name}_{index}")
+                entry.raw = data
+                self._data[index] = entry
+
+    def _lookup(self, target, with_index=False):
+        """
+        Lookup an entry
+
+        "target" could either be its id or its name.
+
+        In case "with_index" is set, also the index is returned.
+        """
+        if isinstance(target, str):
+            try:
+                # Try to get entry by id
+                target_index = int(target)
+            except ValueError:
+                # Get entry by name
+                target_index = None
+                for index, entry in self._data.items():
+                    if entry.name == target:
+                        target_index = index
+        elif isinstance(target, int):
+            # Get entry by id
+            target_index = target
+        else:
+            target_index = None
+
+        target_entry = self._data.get(target_index, None)
+        if target_entry is None:
+            self.logger.warning("entry '%s' not found", target)
+        if with_index:
+            return target_index, target_entry
+        return target_entry
+
+    def get(self, target):
+        """Get entry by id or name."""
+        entry = self._lookup(target)
+        return entry

luxtronik/parameters.py

@@ -3,6 +3,8 @@
 """Parse luxtronik parameters."""
 import logging
 
+from luxtronik.data_vector import DataVector
+
 from luxtronik.datatypes import (
     AccessLevel,
     Bool,
@@ -22,18 +24,20 @@
     VentilationMode,
 )
 
-LOGGER = logging.getLogger("Luxtronik.Parameters")
-
 
-class Parameters:
+class Parameters(DataVector):
     """Class that holds all parameters."""
 
+    logger = logging.getLogger("Luxtronik.Parameters")
+    name = "Parameter"
+
     def __init__(self, safe=True):
         """Initialize parameters class."""
+        super().__init__()
         self.safe = safe
         self.queue = {}
 
-        self._parameters = {
+        self._data = {
             0: Unknown("ID_Transfert_LuxNet"),
             1: Celsius("ID_Einst_WK_akt", True),
             2: Celsius("ID_Einst_BWS_akt", True),
@@ -1162,61 +1166,15 @@ def __init__(self, safe=True):
             1125: Unknown("Unknown_Parameter_1125"),
         }
 
-    def __iter__(self):
-        return iter(self._parameters.items())
-
-    def parse(self, raw_data):
-        """Parse raw parameter data."""
-        for index, data in enumerate(raw_data):
-            parameter = self._parameters.get(index, False)
-            if parameter is not False:
-                parameter.raw = data
-            else:
-                # LOGGER.warning("Parameter '%d' not in list of parameters", index)
-                parameter = Unknown(f"Unknown_Parameter_{index}")
-                parameter.raw = data
-                self._parameters[index] = parameter
-
-    def _lookup(self, target, with_index=False):
-        # pylint: disable=too-many-return-statements,fixme
-        # TODO Evaluate whether logic can be re-arranged to get rid of the
-        # pylint error regarding too many return statements.
-        """Lookup parameter by either id or name."""
-        # Get parameter by id
-        if isinstance(target, int):
-            if with_index:
-                return target, self._parameters.get(target, None)
-            return self._parameters.get(target, None)
-        # Get parameter by name
-        if isinstance(target, str):
-            try:
-                target = int(target)
-                if with_index:
-                    return target, self._parameters.get(target, None)
-                return self._parameters.get(target, None)
-            except ValueError:
-                for index, parameter in self._parameters.items():
-                    if parameter.name == target:
-                        if with_index:
-                            return index, parameter
-                        return parameter
-        LOGGER.warning("Parameter '%s' not found", target)
-        if with_index:
-            return None, None
-        return None
-
-    def get(self, target):
-        """Get parameter by id or name."""
-        parameter = self._lookup(target)
-        return parameter
-
     def set(self, target, value):
         """Set parameter to new value."""
         index, parameter = self._lookup(target, with_index=True)
-        if index:
+        if index is not None:
             if parameter.writeable or not self.safe:
                 self.queue[index] = parameter.to_heatpump(value)
             else:
-                LOGGER.warning("Parameter '%s' not safe for writing!", parameter.name)
+                self.logger.warning(
+                    "Parameter '%s' not safe for writing!", parameter.name
+                )
         else:
-            LOGGER.warning("Parameter '%s' not found", target)
+            self.logger.warning("Parameter '%s' not found", target)

luxtronik/visibilities.py

@@ -1,16 +1,20 @@
 """Parse luxtronik visibilities."""
 import logging
 
-from luxtronik.datatypes import Unknown
+from luxtronik.data_vector import DataVector
 
-LOGGER = logging.getLogger("Luxtronik.Visibilities")
+from luxtronik.datatypes import Unknown
 
 
-class Visibilities:
+class Visibilities(DataVector):
     """Class that holds all visibilities."""
 
+    logger = logging.getLogger("Luxtronik.Visibilities")
+    name = "Visibility"
+
     def __init__(self):
-        self._visibilities = {
+        super().__init__()
+        self._data = {
             0: Unknown("ID_Visi_NieAnzeigen"),
             1: Unknown("ID_Visi_ImmerAnzeigen"),
             2: Unknown("ID_Visi_Heizung"),
@@ -367,40 +371,3 @@ def __init__(self):
             353: Unknown("Unknown_Visibility_353"),
             354: Unknown("Unknown_Visibility_354"),
         }
-
-    def __iter__(self):
-        return iter(self._visibilities.items())
-
-    def parse(self, raw_data):
-        """Parse raw visibility data."""
-        for index, data in enumerate(raw_data):
-            visibility = self._visibilities.get(index, False)
-            if visibility is not False:
-                visibility.raw = data
-            else:
-                # LOGGER.warning("Visibility '%d' not in list of visibilities", index)
-                visibility = Unknown(f"Unknown_Parameter_{index}")
-                visibility.raw = data
-                self._visibilities[index] = visibility
-
-    def _lookup(self, target):
-        """Lookup visibility by either id or name."""
-        # Get visibility by id
-        if isinstance(target, int):
-            return self._visibilities.get(target, None)
-        # Get visibility by name
-        if isinstance(target, str):
-            try:
-                target = int(target)
-                return self._visibilities.get(target, None)
-            except ValueError:
-                for _, visibility in self._visibilities.items():
-                    if visibility.name == target:
-                        return visibility
-        LOGGER.warning("Visibility '%s' not found", target)
-        return None
-
-    def get(self, target):
-        """Get visibility by id or name."""
-        visibility = self._lookup(target)
-        return visibility

tests/test_calculations.py

@@ -0,0 +1,14 @@
+"""Test suite for parameters module"""
+
+# pylint: disable=too-few-public-methods
+
+from luxtronik.calculations import Calculations
+
+
+class TestCalculations:
+    """Test suite for Calculations"""
+
+    def test_init(self):
+        """Test cases for initialization"""
+        calculations = Calculations()
+        assert calculations.name == "Calculation"