Extend the CLI to be able to read and write single values from and to the heatpump

CHANGELOG.md

@@ -14,7 +14,8 @@ This changelog follows the "Keep a Changelog" format and Semantic Versioning.
 - Support smart-home-interface. Luxtronik firmware v3.90.1 or higher
 is required for this. See README for further information. [#190]
 - Add a command-line-interface (CLI) with the following commands:
-`dump`, `dump-cfi`, `dump.shi`, `changes`, `watch-cfi`, `watch-shi`, `discover`
+`dump`, `dump-cfi`, `dump.shi`, `changes`, `watch-cfi`, `watch-shi`,
+`debug-cfi`, `debug-shi`, `discover`
 - Provide an automatically generated documentation for the data fields. [#189]
 
 ### Changed

README.md

@@ -363,6 +363,33 @@ calc: Number: 10    Name: ID_WEB_Temperatur_TVL
 calc: Number: 13    Name: ID_WEB_Temperatur_TRL_ext                                    Value: 26.2 -> 26.1
 ```
 
+#### Debug luxtronik interface
+
+Similar to the other features, some interface commands are available via the following:
+
+```sh
+luxtronik debug-cfi 192.168.178.123 read-calc ID_WEB_Temperatur_TA
+
+# or analog for Modbus TCP register (port is optional)
+luxtronik debug-shi 192.168.178.123:502 read-input 0
+```
+
+or call the script that comes with the python package:
+
+```python
+PYTHONPATH=. ./luxtronik/scripts/debug_cfi.py 192.168.178.123:8889 write-param ID_Ba_Bw_akt Party
+
+# or from the base folder of luxtronik
+python -m luxtronik debug-shi 192.168.178.123 write-holding heating_setpoint 30.5
+```
+
+Currently implemented:
+
+- CFI: `write-param`, `read-param`, `read-calc`, `read-visi`
+- SHI: `write-holding`, `read-holding`, `read-input`
+
+For more information, you'll need to look at the source code.
+
 ### WRITING VALUES TO HEAT PUMP
 
 The following example writes data to the heat pump:

luxtronik/__main__.py

@@ -15,6 +15,12 @@
 from luxtronik.scripts.watch_shi import (
     watch_shi,
 )  # pylint: disable=unused-import # noqa: F401
+from luxtronik.scripts.debug_cfi import (
+    debug_cfi,
+)  # pylint: disable=unused-import # noqa: F401
+from luxtronik.scripts.debug_shi import (
+    debug_shi,
+)  # pylint: disable=unused-import # noqa: F401
 
 
 def discover():
@@ -37,6 +43,8 @@ def main() -> int:
         changes    Watch all config interface value changes of the Luxtronik controller
         watch-cfi  Watch all config interface value changes of the Luxtronik controller
         watch-shi  Watch all smart home interface value changes of the Luxtronik controller
+        debug-cfi  Debug interface for testing the config interface of the Luxtronik controller
+        debug-shi  Debug interface for testing the smart-home interface of the Luxtronik controller
         discover   Discover Luxtronik controllers on the network (via magic packet) and output results
         """,
     )
@@ -51,6 +59,8 @@ def main() -> int:
         "changes": watch_cfi,
         "watch-cfi": watch_cfi,
         "watch-shi": watch_shi,
+        "debug-cfi": debug_cfi,
+        "debug-shi": debug_shi,
         "discover": discover,
     }
     if args.command not in commands:

luxtronik/cfi/interface.py

@@ -7,6 +7,8 @@
 
 from luxtronik.collections import integrate_data
 from luxtronik.common import get_host_lock
+from luxtronik.datatypes import Base
+from luxtronik.definitions import LuxtronikDefinition
 from luxtronik.cfi.constants import (
     LUXTRONIK_DEFAULT_PORT,
     LUXTRONIK_PARAMETERS_WRITE,
@@ -133,6 +135,10 @@ def read_visibilities(self, visibilities=None):
             visibilities = Visibilities()
         return self._with_lock_and_connect(self._read_visibilities, visibilities)
 
+    def write_parameter(self, def_field_name_or_idx, value=None, safe=True):
+        """Calls `_write_parameter` with the lock-and-connect decorator."""
+        self._with_lock_and_connect(self._write_parameter, def_field_name_or_idx, value, safe)
+
     def write(self, parameters):
         """
         Write all set parameters to the heat pump.
@@ -169,27 +175,69 @@ def _write(self, parameters):
         count = 0
         for definition, field in parameters.items():
             if field.write_pending:
-                field.write_pending = False
-                value = field.raw
-                if not isinstance(definition.index, int) or not field.check_for_write(parameters.safe):
-                    LOGGER.warning(
-                        "%s: Parameter id '%s' or value '%s' invalid!",
-                        self._host,
-                        definition.index,
-                        value,
-                    )
-                    continue
-                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
+                if self._do_write_field(LUXTRONIK_PARAMETERS_WRITE, \
+                        definition.index, field, parameters.safe):
+                    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)
 
+    def _do_write_field(self, cmd, index, field, safe=True):
+        """
+        Write a single field to the Luxtronik controller.
+
+        This method checks whether the field can be safely written,
+        resets its write-pending flag, validates the raw value, and
+        delegates the actual write operation to _do_write_raw().
+
+        Args:
+            cmd (int): Command specifying the type of write operation.
+            index (int): Index of the field to write.
+            field (Field): Field object containing the raw value and metadata.
+            safe (bool): If True, perform safety checks before writing.
+
+        Returns:
+            bool: True if the write operation was executed, otherwise False.
+        """
+        # Reset the write_pending flag
+        field.write_pending = False
+        value = field.raw
+        if not field.check_for_write(safe):
+            LOGGER.warning(f"{self._host} - Write: Parameter id '{index}'" \
+                + f" or value '{value}' invalid!")
+            return False
+        return self._do_write_raw(cmd, index, value)
+
+    def _do_write_raw(self, cmd, index, value):
+        """
+        Write a single raw value to the Luxtronik controller.
+
+        Args:
+            cmd (int): Command specifying the type of data to write.
+                Currently only LUXTRONIK_PARAMETERS_WRITE is supported.
+            index (int): Index of the field to write.
+            value (int): Value to write.
+
+        Returns:
+            bool: True if the write operation was executed, otherwise False.
+        """
+        if not isinstance(cmd, int):
+            LOGGER.error(f"{self._host} - Write: Command '{cmd}' invalid! Must be an integer.")
+            return False
+        if not isinstance(index, int):
+            LOGGER.error(f"{self._host} - Write: Index '{index}' invalid! Must be an integer.")
+            return False
+        if not isinstance(value, int):
+            LOGGER.error(f"{self._host} - Write: Value '{value}' invalid! Must be an integer.")
+            return False
+        LOGGER.debug(f"{self._host} - Write: Index '{index}' set to '{value}'")
+        self._send_ints(cmd, index, value)
+        command = self._read_int()
+        LOGGER.debug(f"{self._host} - Write: Command {command}")
+        idx = self._read_int()
+        LOGGER.debug(f"{self._host} - Write: Index {idx}")
+        return (command == cmd) and (index == idx)
+
     def _read_parameters(self, parameters):
         data = []
         self._send_ints(LUXTRONIK_PARAMETERS_READ, 0)
@@ -231,6 +279,55 @@ def _read_visibilities(self, visibilities):
         self._parse(visibilities, data)
         return visibilities
 
+    def _write_parameter(self, def_field_name_or_idx, value=None, safe=True):
+        """
+        Write a single parameter to the Luxtronik controller. Primarily used for debug purposes.
+
+        Args:
+            def_field_name_or_idx (LuxtronikDefinition | Base | str | int):
+                Target to write to. May be a definition, a field, a name, or an index.
+                Unknown fields can be written, but no safety checks are possible.
+            value (int): Value to write. Overrides the field's value if a field is provided.
+            safe (bool): If True, perform safety checks before non-raw writing.
+
+        Returns:
+            bool: True if the write operation was executed, otherwise False.
+        """
+        if isinstance(def_field_name_or_idx, LuxtronikDefinition):
+            # input parameter is a definition
+            definition = def_field_name_or_idx
+            field = definition.create_field()
+            field.value = value
+            self._do_write_field(LUXTRONIK_PARAMETERS_WRITE, definition.index, field, safe)
+        elif isinstance(def_field_name_or_idx, Base):
+            # input parameter is a field
+            field = def_field_name_or_idx
+            definition = Parameters.definitions.get(field.name)
+            if value is not None:
+                field.value = value
+            self._do_write_field(LUXTRONIK_PARAMETERS_WRITE, definition.index, field, safe)
+        else:
+            # input parameter is either a name or an index
+            definition = Parameters.definitions.get(def_field_name_or_idx)
+            if definition is not None:
+                field = definition.create_field()
+                field.value = value
+                self._do_write_field(LUXTRONIK_PARAMETERS_WRITE, definition.index, field, safe)
+            else:
+                # check for an integer string
+                try:
+                    def_field_name_or_idx = int(def_field_name_or_idx)
+                except ValueError:
+                    pass
+                if isinstance(def_field_name_or_idx, int):
+                    index = def_field_name_or_idx
+                    self._do_write_raw(LUXTRONIK_PARAMETERS_WRITE, index, value)
+                else:
+                    LOGGER.warning(f"{self._host} - Write: Target '{def_field_name_or_idx}' invalid!")
+
+        # Give the heatpump a short time to handle the value changes/calculations:
+        time.sleep(WAIT_TIME_AFTER_PARAMETER_WRITE)
+
     def _send_ints(self, *ints):
         "Low-level helper to send a tuple of ints"
         data = struct.pack(">" + "i" * len(ints), *ints)

luxtronik/scripts/__init__.py

@@ -3,18 +3,26 @@
 import argparse
 
 
-def create_default_args_parser(func_desc, default_port):
+def create_default_args_parser(func_desc, default_port, with_port=True):
     parser = argparse.ArgumentParser(description=func_desc)
-    parser.add_argument("ip", help="IP address of Luxtronik controller to connect to")
-    parser.add_argument(
-        "port",
-        nargs="?",
-        type=int,
-        default=default_port,
-        help="Port to use to connect to Luxtronik controller",
-    )
+    parser.add_argument("ip", help="IP address of Luxtronik controller to connect to. Add :port to specify a custom port to use.")
+    if with_port:
+        parser.add_argument(
+            "port",
+            nargs="?",
+            type=int,
+            default=default_port,
+            help="Port to use to connect to Luxtronik controller",
+        )
     return parser
 
+def get_port_from_ip_string(args, default_port):
+    vals = args.ip.split(":")
+    if len(vals) > 1:
+        return vals[1]
+    else:
+        return default_port
+
 def print_dump_header(caption):
     print("=" * 130)
     print(f"{' ' + caption + ' ': ^120}")

luxtronik/scripts/debug_cfi.py

@@ -0,0 +1,71 @@
+#! /usr/bin/env python3
+
+# pylint: disable=invalid-name
+"""
+Script to debug the config interface of a Luxtronik controller
+"""
+import logging
+
+from luxtronik import LuxtronikSocketInterface, LUXTRONIK_DEFAULT_PORT
+from luxtronik.scripts import create_default_args_parser, get_port_from_ip_string
+
+logging.basicConfig(level=logging.DEBUG)
+
+SUPPORTED_CMDS = ["write-param", "read-param", "read-calc", "read-visi"]
+
+
+def try_text_to_number(text):
+    value = text
+    try:
+        value = float(value)
+    except ValueError:
+        pass
+    try:
+        value = int(value)
+    except ValueError:
+        pass
+    return value
+
+def debug_cfi():
+    parser = create_default_args_parser(
+        "Debug commands for the config interface of a Luxtronik controller",
+        LUXTRONIK_DEFAULT_PORT,
+        False
+    )
+    parser.add_argument("cmd", type=str, help="Command to execute. Currently supported: {SUPPORTED_CMDS}.")
+    parser.add_argument("target", help="The target to be read or written. May be a index or name.")
+    parser.add_argument("value", nargs="?", default=None, help="Value to be written")
+    args = parser.parse_args()
+    port = get_port_from_ip_string(args, LUXTRONIK_DEFAULT_PORT)
+
+    client = LuxtronikSocketInterface(args.ip, port)
+
+    if args.cmd == SUPPORTED_CMDS[0]:
+        print(f"Write {args.value} to CFI parameter {args.target} of {args.ip}:{port}")
+        value = try_text_to_number(args.value)
+        client.write_parameter(args.target, value)
+
+    elif args.cmd == SUPPORTED_CMDS[1]:
+        print(f"Read parameter from CFI {args.target} of {args.ip}:{port}")
+        parameters = client.read_parameters()
+        field = parameters[args.target]
+        print(f"{repr(field)}")
+
+    elif args.cmd == SUPPORTED_CMDS[2]:
+        print(f"Read calculation from CFI {args.target} of {args.ip}:{port}")
+        calculations = client.read_calculations()
+        field = calculations[args.target]
+        print(f"{repr(field)}")
+
+    elif args.cmd == SUPPORTED_CMDS[3]:
+        print(f"Read visibilities from CFI {args.target} of {args.ip}:{port}")
+        visibilities = client.read_visibilities()
+        field = visibilities[args.target]
+        print(f"{repr(field)}")
+
+    else:
+        print(f"Cmd '{args.cmd}' not supported. Use on of {SUPPORTED_CMDS}")
+
+
+if __name__ == "__main__":
+    debug_cfi()