Add the shi module __init__ file

Author: Guzz-T

luxtronik/shi/__init__.py

@@ -0,0 +1,124 @@
+"""
+Python module for controlling a Luxtronik heat pump controller
+via the smart home interface. Powered by Guzz-T.
+"""
+
+from luxtronik.datatypes import FullVersion, MajorMinorVersion
+from luxtronik.shi.constants import (
+    LUXTRONIK_DEFAULT_MODBUS_PORT,
+    LUXTRONIK_DEFAULT_MODBUS_TIMEOUT,
+    LUXTRONIK_LATEST_SHI_VERSION,
+)
+from luxtronik.shi.common import LOGGER, parse_version
+from luxtronik.shi.inputs import INPUTS_DEFINITIONS
+from luxtronik.shi.modbus import LuxtronikModbusTcpInterface
+from luxtronik.shi.interface import LuxtronikSmartHomeInterface
+
+VERSION_DETECT = "detect"
+VERSION_LATEST = "latest"
+
+
+###############################################################################
+# Helper methods
+###############################################################################
+
+def get_version_definitions(definitions):
+    """
+    Retrieve all definitions that represent version fields.
+
+    Args:
+        definitions (LuxtronikDefinitionsList): List of definitions
+
+    Returns:
+        list[LuxtronikDefinition]: List of definitions whose data_type
+        is either FullVersion or MajorMinorVersion.
+    """
+    version_definitions = []
+    for d in definitions:
+        if d.data_type in (FullVersion, MajorMinorVersion):
+            version_definitions.append(d)
+    return version_definitions
+
+def determine_version(interface):
+    """
+    Determine the version of the luxtronik controller.
+
+    This is a little bit ugly! The controller version is required
+    to locate the version field. As workaround, probe each known
+    version field until one yields a valid read and a parsable version.
+    This approach works as long as the version-field has not changed.
+
+    Args:
+        interface (LuxtronikModbusTcpInterface):
+            Simple read/write interface to read out the version.
+
+    Returns:
+        tuple[int] | None: The version of the controller on success,
+            or None if no version could be determined.
+    """
+    definitions = get_version_definitions(INPUTS_DEFINITIONS)
+    for definition in definitions:
+        data = interface.read_inputs(definition.addr, definition.count)
+        if data is not None:
+            field = definition.create_field()
+            field.raw = data
+            parsed = parse_version(field.value)
+            if parsed is not None:
+                return parsed
+    LOGGER.warning("It was not possible to determine the controller version. " \
+        + "Switch to trial-and-error mode.")
+    return None
+
+
+###############################################################################
+# Factory methods
+###############################################################################
+
+def create_modbus_tcp(
+    host,
+    port=LUXTRONIK_DEFAULT_MODBUS_PORT,
+    timeout=LUXTRONIK_DEFAULT_MODBUS_TIMEOUT,
+    version=VERSION_DETECT
+):
+    """
+    Create a LuxtronikSmartHomeInterface using a Modbus TCP connection.
+
+    The function constructs a Modbus TCP low-level interface and resolves the
+    controller version according to the supplied `version` argument:
+      - If `version` equals VERSION_DETECT, attempt to determine the version.
+      - If `version` equals VERSION_LATEST, use LUXTRONIK_LATEST_SHI_VERSION as version.
+      - If `version` is a string, parse it into a version tuple.
+      - If `version` is None, the interface is initialized in trial-and-error mode.
+      - Otherwise assume `version` is already a parsed version tuple.
+
+    Args:
+        host (str): Hostname or IP address of the Luxtronik controller.
+        port (int): TCP port for the Modbus connection.
+        timeout (float): Timeout in seconds for the Modbus connection.
+        version (tuple[int] | str | None): Version used to initialize the interface.
+            If VERSION_DETECT is passed, the function will attempt to determine the version.
+            If a str is passed, the string will be parsed into a version tuple.
+            If None is passed, trial-and-error mode is activated.
+
+    Returns:
+        LuxtronikSmartHomeInterface:
+            Initialized interface instance bound to the Modbus TCP connection.
+    """
+    modbus_interface = LuxtronikModbusTcpInterface(host, port, timeout)
+
+    resolved_version = version
+    if resolved_version == VERSION_DETECT:
+        # return None in case of an error -> trial-and-error mode
+        resolved_version = determine_version(modbus_interface)
+    elif isinstance(resolved_version, str):
+        if resolved_version.lower() == VERSION_LATEST:
+            resolved_version = LUXTRONIK_LATEST_SHI_VERSION
+        else:
+            # return None in case of an error -> trial-and-error mode
+            resolved_version = parse_version(resolved_version)
+    else:
+        resolved_version = parse_version(resolved_version)
+
+    LOGGER.info(f"Create smart-home-interface via modbus-TCP on {host}:{port}"
+        + f" for version {resolved_version}")
+    return LuxtronikSmartHomeInterface(modbus_interface, resolved_version)

tests/shi/test_interface.py

@@ -1,4 +1,5 @@
 import pytest
+from unittest.mock import patch
 
 from luxtronik.datatypes import Base, Unknown
 
@@ -24,6 +25,7 @@
     LuxtronikSmartHomeData,
     LuxtronikSmartHomeInterface,
 )
+from luxtronik.shi import create_modbus_tcp
 
 ###############################################################################
 # Fake modbus client
@@ -33,14 +35,23 @@ class FakeInterface:
     telegram_list = []
     result = True
 
+    def __init__(self, host="", port="", timeout=0):
+        pass
+
+    def _get_data(self, addr, count):
+        return [addr - 10000 + i for i in range(count)]
+
+    def read_inputs(self, addr, count):
+        return self._get_data(addr, count) if self.result else None
+
     def send(self, telegrams):
         if not isinstance(telegrams, list):
             telegrams = [telegrams]
         FakeInterface.telegram_list = telegrams
 
         for t in telegrams:
             if isinstance(t, LuxtronikSmartHomeReadTelegram):
-                t.data = [t.addr - 10000 + i for i in range(t.count)]
+                t.data = self._get_data(t.addr, t.count)
         return self.result
 
 
@@ -73,6 +84,7 @@ def test_empty(self):
         assert data.inputs.version == (1, 2, 0, 0)
 
 
+@patch("luxtronik.shi.LuxtronikModbusTcpInterface", FakeInterface)
 class TestLuxtronikSmartHomeInterface:
 
     @classmethod
@@ -1246,3 +1258,26 @@ def test_trial_and_error_mode(self):
         assert FakeInterface.telegram_list[7].addr == offset + 4
         assert FakeInterface.telegram_list[7].count == 1
         assert FakeInterface.telegram_list[7].data == [16]
+
+    def test_create_modbus(self):
+        interface = create_modbus_tcp('host', version=None)
+        assert interface.version is None
+
+        interface = create_modbus_tcp('host', version=1)
+        assert interface.version is None
+
+        interface = create_modbus_tcp('host', version="1.2.3")
+        assert interface.version == (1, 2, 3, 0)
+
+        interface = create_modbus_tcp('host', version="latest")
+        assert interface.version == LUXTRONIK_LATEST_SHI_VERSION
+
+        interface = create_modbus_tcp('host')
+        assert interface.version == (400, 401, 402, 0)
+
+        FakeInterface.result = False
+
+        interface = create_modbus_tcp('host')
+        assert interface.version is None
+
+        FakeInterface.result = True