Try/logger

luxtronik/cfi/constants.py

@@ -26,7 +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: Final = 1
 
 # The data from the config interface are transmitted in 32-bit chunks.
 LUXTRONIK_CFI_REGISTER_BIT_SIZE: Final = 32

luxtronik/cfi/interface.py

@@ -76,7 +76,7 @@ def _with_lock_and_connect(self, func, *args, **kwargs):
                 ret_val = None
                 with socket.create_connection((self._host, self._port)) as sock:
                     self._socket = sock
-                    LOGGER.info("Connected to Luxtronik heat pump %s:%s", self._host, self._port)
+                    LOGGER.info("Connected to CFI of Luxtronik heat pump %s:%s", self._host, self._port)
                     ret_val = func(*args, **kwargs)
             except socket.gaierror as e:
                 LOGGER.error("Failed to connect to Luxtronik heat pump %s:%s. %s.",
@@ -166,6 +166,7 @@ def _write(self, parameters):
         if not isinstance(parameters, Parameters):
             LOGGER.error("Only parameters are writable!")
             return
+        count = 0
         for definition, field in parameters.items():
             if field.write_pending:
                 field.write_pending = False
@@ -178,12 +179,14 @@ def _write(self, parameters):
                         value,
                     )
                     continue
-                LOGGER.info("%s: Parameter '%d' set to '%s'", self._host, definition.index, value)
+                LOGGER.debug("%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)
+                count += 1
+        LOGGER.info("%s: Write %d parameters", self._host, count)
         # Give the heatpump a short time to handle the value changes/calculations:
         time.sleep(WAIT_TIME_AFTER_PARAMETER_WRITE)
 

luxtronik/constants.py

@@ -20,4 +20,4 @@
 LUXTRONIK_32BIT_FUNCTION_NOT_AVAILABLE: Final = 0x7FFFFFFF
 
 # If True, preserve the last set field value on clear and assign `None` to raw
-LUXTRONIK_PRESERVE_LAST_VALUE = True
+LUXTRONIK_PRESERVE_LAST_VALUE: Final = True

luxtronik/definitions/__init__.py

@@ -69,6 +69,8 @@ def __init__(self, data_dict, type_name, offset, data_type=""):
               that have been checked for correctness using pytest.
               This eliminates the need for type tests here.
         """
+        self.report_successor = True
+        self.report_outdated_name = True
         try:
             data_dict = self.DEFAULT_DATA | data_dict
             index = int(data_dict["index"])
@@ -194,11 +196,7 @@ def names(self):
 
     @property
     def successor(self):
-        # Clear the successor after first use,
-        # not to generate a lot of warnings
-        s = self._successor
-        self._successor = None
-        return s
+        return self._successor
 
     @property
     def name(self):
@@ -301,12 +299,12 @@ def get(self, name_or_idx, default=None):
         if d is None:
             LOGGER.debug(f"Definition for '{name_or_idx}' not found")
         else:
-            # The successor is returned only once for each definition,
-            # not to generate a lot of warnings
-            successor = d.successor
-            if successor is not None:
-                LOGGER.warning(f"Definition for '{name_or_idx}' is outdated and will " \
-                    + f"be removed soon! Please use '{successor}' instead.")
+            # Report a successor only once per definition instance
+            # to avoid generating too many warnings
+            if d.successor is not None and d.report_successor:
+                d.report_successor = False
+                LOGGER.warning(f"'{d.type_name}' definition for '{name_or_idx}' is outdated and will " \
+                    + f"be removed soon! Please use '{d.successor}' instead.")
         return d if d is not None else default
 
     def _get(self, name_or_idx):
@@ -362,8 +360,16 @@ def _get_definition_by_name(self, name):
             If multiple definitions added for the same name, the last added takes precedence.
         """
         definition = self._name_dict.get(name.lower(), None)
-        if definition is not None and definition.valid and name.lower() != definition.name.lower():
-            LOGGER.warning(f"'{name}' is outdated! Use '{definition.name}' instead.")
+
+        # Report an outdated name only once per definition instance
+        # to avoid generating too many warnings
+
+        if definition is not None and definition.valid \
+                and name.lower() != definition.name.lower() \
+                and definition.report_outdated_name:
+            definition.report_outdated_name = False
+            LOGGER.warning(f"'{definition.type_name}' name '{name}' is outdated! " \
+                + f"Use '{definition.name}' instead.")
         return definition
 
 

luxtronik/shi/interface.py

@@ -506,6 +506,25 @@ def _send_and_integrate(self, blocks_list):
         # Send all telegrams. The retrieved data is returned within the telegrams
         telegrams = [data[1] for data in telegrams_data]
         success = self._interface.send(telegrams)
+        # Report send/received data
+        count = {'hr': 0, 'hw': 0, 'ir': 0, 'u': 0}
+        for t in telegrams:
+            if isinstance(t, LuxtronikSmartHomeReadHoldingsTelegram):
+                count['hr'] += t.count
+            elif isinstance(t, LuxtronikSmartHomeReadInputsTelegram):
+                count['ir'] += t.count
+            elif isinstance(t, LuxtronikSmartHomeWriteHoldingsTelegram):
+                count['hw'] += t.count
+            else:
+                count['u'] += t.count
+        if count['hr'] > 0:
+            LOGGER.info(f"{self._interface._host}: Read {count['hr']} holdings")
+        if count['ir'] > 0:
+            LOGGER.info(f"{self._interface._host}: Read {count['ir']} inputs")
+        if count['hw'] > 0:
+            LOGGER.info(f"{self._interface._host}: Write {count['hw']} holdings")
+        if count['u'] > 0:
+            LOGGER.info(f"{self._interface._host}: Write {count['u']} unknowns?")
         # Transfer the data from the telegrams into the fields
         success &= self._integrate_data(telegrams_data)
         return success

luxtronik/shi/modbus.py

@@ -53,6 +53,8 @@ def __init__(
         self._lock = get_host_lock(host)
 
         # Create the Modbus client (connection is not opened/closed automatically)
+        self._host = host
+        self._port = port
         self._client = ModbusClient(
             host=host,
             port=port,
@@ -86,6 +88,8 @@ def _connect(self):
                 + f"{self._client.last_error_as_txt}")
             self._client.close()
             return False
+        else:
+            LOGGER.info(f"Connected to SHI of Luxtronik heat pump {self._host}:{self._port}")
 
         return True
 
@@ -295,7 +299,6 @@ def send(self, telegrams):
 
         # Exit the function if no operation is necessary
         if total_count <= 0:
-            LOGGER.warning("No data requested/provided. Abort operation.")
             return False
 
         # Acquire lock, connect and read/write data. Disconnect afterwards.

tests/fake/fake_modbus.py

@@ -6,6 +6,8 @@ class FakeModbus:
     result = True
 
     def __init__(self, host="", port="", timeout=0):
+        self._host = host
+        self._port = port
         self._connected = False
         self._blocking = False
 

tests/test_definitions.py

@@ -38,7 +38,6 @@ def test_init(self):
         assert definition.names == names
         assert definition.name == names[0]
         assert definition.successor == 'abcd'
-        assert definition.successor is None
         assert definition.data_type == 'INT16'
         assert definition.valid
         assert definition