diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cb14a16..cd9a816b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index fc61112c..eb5a76c0 100755 --- a/README.md +++ b/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: diff --git a/luxtronik/__main__.py b/luxtronik/__main__.py index 51746286..7280a494 100755 --- a/luxtronik/__main__.py +++ b/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: diff --git a/luxtronik/cfi/interface.py b/luxtronik/cfi/interface.py index e7ebf6e9..28700cc8 100644 --- a/luxtronik/cfi/interface.py +++ b/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) diff --git a/luxtronik/scripts/__init__.py b/luxtronik/scripts/__init__.py index 86e259bf..6a020617 100644 --- a/luxtronik/scripts/__init__.py +++ b/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}") diff --git a/luxtronik/scripts/debug_cfi.py b/luxtronik/scripts/debug_cfi.py new file mode 100644 index 00000000..bab87133 --- /dev/null +++ b/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() diff --git a/luxtronik/scripts/debug_shi.py b/luxtronik/scripts/debug_shi.py new file mode 100644 index 00000000..7fa851a3 --- /dev/null +++ b/luxtronik/scripts/debug_shi.py @@ -0,0 +1,63 @@ +#! /usr/bin/env python3 + +# pylint: disable=invalid-name +""" +Script to debug the smart-home interface of a Luxtronik controller +""" +import logging + +from luxtronik.shi import create_modbus_tcp, LUXTRONIK_DEFAULT_MODBUS_PORT +from luxtronik.scripts import create_default_args_parser, get_port_from_ip_string + +logging.basicConfig(level=logging.DEBUG) + +SUPPORTED_CMDS = ["write-holding", "read-holding", "read-input"] + + +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_shi(): + parser = create_default_args_parser( + "Debug commands for the smart-home interface of a Luxtronik controller", + LUXTRONIK_DEFAULT_MODBUS_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_MODBUS_PORT) + + client = create_modbus_tcp(args.ip, port) + + if args.cmd == SUPPORTED_CMDS[0]: + print(f"Write {args.value} to SHI holding {args.target} of {args.ip}:{port}") + value = try_text_to_number(args.value) + client.write_holding(args.target, value) + + elif args.cmd == SUPPORTED_CMDS[1]: + print(f"Read holding from SHI {args.target} of {args.ip}:{port}") + holding = client.read_holding(args.target) + print(f"{repr(holding)}") + + elif args.cmd == SUPPORTED_CMDS[2]: + print(f"Read input from SHI {args.target} of {args.ip}:{port}") + input = client.read_input(args.target) + print(f"{repr(input)}") + + else: + print(f"Cmd '{args.cmd}' not supported. Use on of {SUPPORTED_CMDS}") + + +if __name__ == "__main__": + debug_shi()