dimm kit: modbus client context manager, type hints

packages/dataclass_utils/factories.py

@@ -1,11 +1,11 @@
-from typing import Dict, List
+from typing import Any, Dict, List
 
 
-def empty_dict_factory() -> Dict:
+def empty_dict_factory() -> Dict[str, Any]:
     return {}
 
 
-def empty_list_factory() -> List:
+def empty_list_factory() -> List[Any]:
     return []
 
 
@@ -17,7 +17,7 @@ def voltages_list_factory() -> List[float]:
     return [230.0]*3
 
 
-def empty_io_pattern_boolean_factory():
+def empty_io_pattern_boolean_factory() -> List[Dict[str, Any]]:
     return [
         {
             "value": True,  # dimmen
@@ -30,7 +30,7 @@ def empty_io_pattern_boolean_factory():
     ]
 
 
-def empty_io_pattern_stepwise_factory():
+def empty_io_pattern_stepwise_factory() -> List[Dict[str, Any]]:
     return [
         {
             "value": 1.0,  # keine Begrenzung

packages/helpermodules/subdata.py

@@ -1051,7 +1051,10 @@ def process_system_topic(self, client: mqtt.Client, var: dict, msg: mqtt.MQTTMes
                     dev = importlib.import_module(f".io_devices.{io_config['type']}.api",
                                                   "modules")
                     config = dataclass_from_dict(dev.device_descriptor.configuration_factory, io_config)
-                    var["io"+index] = dev.create_io(config)
+                    if (self.event_subdata_initialized.is_set() is False or
+                            "io"+index not in var or
+                            io_config != asdict(var["io"+index].config)):
+                        var["io"+index] = dev.create_io(config)
             elif re.search("^.+/io/[0-9]+/set/manual/analog_output", msg.topic) is not None:
                 index = get_index(msg.topic)
                 self.set_json_payload(var["io"+index].set_manual["analog_output"], msg)

packages/modules/common/io_setup.py

@@ -1,5 +1,5 @@
 
-from typing import Dict, Generic, Optional, TypeVar
+from typing import Dict, Generic, Optional, TypeVar, Union
 
 from dataclass_utils.factories import empty_dict_factory
 
@@ -13,8 +13,8 @@ def __init__(self,
                  type: str,
                  id: int,
                  configuration: T,
-                 input: Optional[Dict[str, Dict[int, float]]] = None,
-                 output: Optional[Dict[str, Dict[int, float]]] = None) -> None:
+                 input: Optional[Dict[str, Union[Dict[str, bool], Dict[str, float], Dict[str, None]]]] = None,
+                 output: Optional[Dict[str, Union[Dict[str, bool], Dict[str, float], Dict[str, None]]]] = None) -> None:
         self.name = name
         self.type = type
         self.id = id

packages/modules/common/modbus.py

@@ -8,7 +8,8 @@
 import struct
 from enum import Enum
 import time
-from typing import Any, Callable, Iterable, Optional, Union, overload, List
+from types import TracebackType
+from typing import Any, Callable, Dict, Iterable, Literal, Optional, Type, Union, overload, List
 
 import pymodbus
 from pymodbus.client.sync import ModbusTcpClient, ModbusUdpClient, ModbusSerialClient
@@ -52,7 +53,7 @@ class ModbusClient:
     def __init__(self,
                  delegate: Union[ModbusSerialClient, ModbusTcpClient, ModbusUdpClient],
                  address: str, port: int = 502,
-                 sleep_after_connect: Optional[int] = 0):
+                 sleep_after_connect: int = 0):
         self._delegate = delegate
         self.address = address
         self.port = port
@@ -67,7 +68,10 @@ def __enter__(self):
             raise e
         return self
 
-    def __exit__(self, exc_type, exc_value, exc_traceback):
+    def __exit__(self,
+                 exc_type: Optional[Type[BaseException]],
+                 exc_value: Optional[BaseException],
+                 exc_traceback: Optional[TracebackType]):
         self._delegate.__exit__(exc_type, exc_value, exc_traceback)
 
     def connect(self) -> None:
@@ -84,12 +88,12 @@ def close(self) -> None:
     def is_socket_open(self) -> bool:
         return self._delegate.is_socket_open()
 
-    def __read_registers(self, read_register_method: Callable,
+    def __read_registers(self, read_register_method: Callable[[int, int], Any],
                          address: int,
                          types: Union[Iterable[ModbusDataType], ModbusDataType],
-                         byteorder: Endian = Endian.Big,
-                         wordorder: Endian = Endian.Big,
-                         **kwargs):
+                         byteorder: str = Endian.Big,
+                         wordorder: str = Endian.Big,
+                         **kwargs: Any):
         if self.is_socket_open() is False:
             self.connect()
         try:
@@ -123,40 +127,40 @@ def divide_rounding_up(numerator: int, denominator: int):
             raise Exception(__name__+" "+str(type(e))+" " + str(e)) from e
 
     @overload
-    def read_holding_registers(self, address: int, types: Iterable[ModbusDataType], byteorder: Endian = Endian.Big,
-                               wordorder: Endian = Endian.Big, **kwargs) -> List[Number]:
+    def read_holding_registers(self, address: int, types: Iterable[ModbusDataType], byteorder: str = Endian.Big,
+                               wordorder: str = Endian.Big, **kwargs: Any) -> List[Number]:
         pass
 
     @overload
-    def read_holding_registers(self, address: int, types: ModbusDataType, byteorder: Endian = Endian.Big,
-                               wordorder: Endian = Endian.Big, **kwargs) -> Number:
+    def read_holding_registers(self, address: int, types: ModbusDataType, byteorder: str = Endian.Big,
+                               wordorder: str = Endian.Big, **kwargs: Any) -> Number:
         pass
 
     def read_holding_registers(self, address: int,
                                types: Union[Iterable[ModbusDataType], ModbusDataType],
-                               byteorder: Endian = Endian.Big,
-                               wordorder: Endian = Endian.Big,
-                               **kwargs):
+                               byteorder: str = Endian.Big,
+                               wordorder: str = Endian.Big,
+                               **kwargs: Any):
         return self.__read_registers(
             self._delegate.read_holding_registers, address, types, byteorder, wordorder, **kwargs
         )
 
     @overload
-    def read_input_registers(self, address: int, types: Iterable[ModbusDataType], byteorder: Endian = Endian.Big,
-                             wordorder: Endian = Endian.Big,
-                             **kwargs) -> List[Number]:
+    def read_input_registers(self, address: int, types: Iterable[ModbusDataType], byteorder: str = Endian.Big,
+                             wordorder: str = Endian.Big,
+                             **kwargs: Any) -> List[Number]:
         pass
 
     @overload
-    def read_input_registers(self, address: int, types: ModbusDataType, byteorder: Endian = Endian.Big,
-                             wordorder: Endian = Endian.Big, **kwargs) -> Number:
+    def read_input_registers(self, address: int, types: ModbusDataType, byteorder: str = Endian.Big,
+                             wordorder: str = Endian.Big, **kwargs: Any) -> Number:
         pass
 
     def read_input_registers(self, address: int,
                              types: Union[Iterable[ModbusDataType], ModbusDataType],
-                             byteorder: Endian = Endian.Big,
-                             wordorder: Endian = Endian.Big,
-                             **kwargs):
+                             byteorder: str = Endian.Big,
+                             wordorder: str = Endian.Big,
+                             **kwargs: Any):
         return self.__read_registers(self._delegate.read_input_registers,
                                      address,
                                      types,
@@ -165,16 +169,14 @@ def read_input_registers(self, address: int,
                                      **kwargs)
 
     @overload
-    def read_coils(self, address: int, types: Iterable[ModbusDataType], byteorder: Endian = Endian.Big,
-                   wordorder: Endian = Endian.Big,
-                   **kwargs) -> List[bool]:
+    def read_coils(self, address: int, count: Literal[1] = 1, **kwargs: Any) -> bool:
         pass
 
     @overload
-    def read_coils(self, address: int, count: int, **kwargs) -> bool:
+    def read_coils(self, address: int, count: int, **kwargs: Any) -> Union[bool, List[bool]]:
         pass
 
-    def read_coils(self, address: int, count: int, **kwargs):
+    def read_coils(self, address: int, count: int = 1, **kwargs: Any) -> Union[bool, List[bool]]:
         try:
             response = self._delegate.read_coils(address, count, **kwargs)
             if response.isError():
@@ -190,8 +192,8 @@ def read_coils(self, address: int, count: int, **kwargs):
     def _build_binary_payload(self,
                               value: Union[int, float],
                               data_type: ModbusDataType,
-                              byteorder: Endian = Endian.Big,
-                              wordorder: Endian = Endian.Big) -> list:
+                              byteorder: str = Endian.Big,
+                              wordorder: str = Endian.Big) -> List[int]:
         builder = BinaryPayloadBuilder(byteorder=byteorder, wordorder=wordorder)
         if data_type == ModbusDataType.FLOAT_16:
             # FLOAT_16 (IEEE 754 Half-Precision) manuelle Konvertierung
@@ -205,7 +207,7 @@ def _build_binary_payload(self,
         return builder.to_registers()
 
     def write_register(self, address: int, value: Union[int, float], data_type: Optional[ModbusDataType] = None,
-                       byteorder: Endian = Endian.Big, wordorder: Endian = Endian.Big, **kwargs):
+                       byteorder: str = Endian.Big, wordorder: str = Endian.Big, **kwargs: Any):
         if data_type is not None:
             if data_type.bits > 16 or data_type in [ModbusDataType.FLOAT_16,
                                                     ModbusDataType.FLOAT_32,
@@ -219,15 +221,20 @@ def write_register(self, address: int, value: Union[int, float], data_type: Opti
             # Fallback für bestehenden Code ohne data_type
             self._delegate.write_registers(address, value, **kwargs)
 
-    def write_single_coil(self, address: int, value: Any, **kwargs):
+    def write_single_coil(self, address: int, value: bool, **kwargs: Any):
         self._delegate.write_coil(address, value, **kwargs)
 
+    def write_coils(self, address: int, value: List[bool], **kwargs: Any):
+        self._delegate.write_coils(address, value, **kwargs)
+
     def __read_bulk(self,
-                    read_register_method: Callable,
+                    read_register_method: Callable[[int, int], Any],
                     start_address: int,
                     count: int,
                     mapping: list[tuple[int, Union[ModbusDataType, Iterable[ModbusDataType]]]],
-                    byteorder: Endian = Endian.Big, wordorder: Endian = Endian.Big, **kwargs):
+                    byteorder: str = Endian.Big,
+                    wordorder: str = Endian.Big,
+                    **kwargs: Any) -> Dict[int, Union[Number, list[Number]]]:
         """
         Liest einen Registerbereich und gibt ein dict mit reg als Key und dekodiertem Wert als Value zurück.
         mapping: Liste von Tupeln (reg, ModbusDataType)
@@ -267,9 +274,9 @@ def read_input_registers_bulk(self,
                                   start_address: int,
                                   count: int,
                                   mapping: list[tuple[int, Union[ModbusDataType, Iterable[ModbusDataType]]]],
-                                  byteorder: Endian = Endian.Big,
-                                  wordorder: Endian = Endian.Big,
-                                  **kwargs):
+                                  byteorder: str = Endian.Big,
+                                  wordorder: str = Endian.Big,
+                                  **kwargs: Any):
         return self.__read_bulk(self._delegate.read_input_registers,
                                 start_address,
                                 count,
@@ -282,9 +289,9 @@ def read_holding_registers_bulk(self,
                                     start_address: int,
                                     count: int,
                                     mapping: list[tuple[int, Union[ModbusDataType, Iterable[ModbusDataType]]]],
-                                    byteorder: Endian = Endian.Big,
-                                    wordorder: Endian = Endian.Big,
-                                    **kwargs):
+                                    byteorder: str = Endian.Big,
+                                    wordorder: str = Endian.Big,
+                                    **kwargs: Any):
         return self.__read_bulk(self._delegate.read_holding_registers,
                                 start_address,
                                 count,
@@ -298,9 +305,9 @@ class ModbusTcpClient_(ModbusClient):
     def __init__(self,
                  address: str,
                  port: int = 502,
-                 sleep_after_connect: Optional[int] = 0,
+                 sleep_after_connect: int = 0,
                  framer: type[ModbusSocketFramer] = ModbusSocketFramer,
-                 **kwargs):
+                 **kwargs: Any):
         parsed_url = parse_url(address)
         host = parsed_url.host
         if parsed_url.port is not None:
@@ -316,8 +323,8 @@ class ModbusUdpClient_(ModbusClient):
     def __init__(self,
                  address: str,
                  port: int = 502,
-                 sleep_after_connect: Optional[int] = 0,
-                 **kwargs):
+                 sleep_after_connect: int = 0,
+                 **kwargs: Any):
         parsed_url = parse_url(address)
         host = parsed_url.host
         if parsed_url.port is not None:
@@ -328,8 +335,8 @@ def __init__(self,
 class ModbusSerialClient_(ModbusClient):
     def __init__(self,
                  port: int,
-                 sleep_after_connect: Optional[int] = 0,
-                 **kwargs):
+                 sleep_after_connect: int = 0,
+                 **kwargs: Any):
         super().__init__(ModbusSerialClient(method="rtu",
                                             port=port,
                                             baudrate=9600,

packages/modules/io_devices/dimm_kit/api.py

@@ -19,7 +19,7 @@
 
 def create_io(config: IoLan):
     version = False
-    client = None
+    client: ModbusTcpClient_
 
     def read():
         nonlocal version
@@ -42,32 +42,40 @@ def read():
                 raise Exception("Die IP-Adresse ist nicht erreichbar. Bitte überprüfe die Einstellungen.")
         # analog inputs are configured as 0-5V (AI1-AI4) and 0-25mA (AI5-AI8) as default
         # the values are reported as integers in range of 0-1024
-        time.sleep(0.1)
-        analog_read = client.read_input_registers(0x00, [ModbusDataType.UINT_16]*8, unit=config.configuration.modbus_id)
-        analog_input = {getattr(AnalogInputMapping, f'AI{pin+1}').name: analog_read[pin] * 5 for pin in range(8)}
-        time.sleep(0.1)
-        digital_input_read = client.read_coils(0x00, 8, unit=config.configuration.modbus_id)
-        digital_input = {getattr(DigitalInputMapping, f'DI{pin+1}').name: digital_input_read[pin] for pin in range(8)}
-        time.sleep(0.1)
-        digital_output_read = client.read_coils(0x10, 8, unit=config.configuration.modbus_id)
-        digital_output = {
-            getattr(DigitalOutputMapping, f'DO{pin+1}').name: digital_output_read[pin] for pin in range(8)}
+        with client:
+            time.sleep(0.1)
+            analog_read = client.read_input_registers(
+                0x00, [ModbusDataType.UINT_16]*8, unit=config.configuration.modbus_id)
+            analog_input = {getattr(AnalogInputMapping,
+                                    f'AI{pin+1}').name: analog_read[pin] * 5 for pin in range(8)}
+            time.sleep(0.1)
+            digital_input_output_read = client.read_coils(0x00, 24, unit=config.configuration.modbus_id)
+            digital_input = {
+                getattr(DigitalInputMapping,
+                        f'DI{pin+1}').name: digital_input_output_read[pin] for pin in range(8)}
+            digital_output = {
+                getattr(DigitalOutputMapping,
+                        f'DO{pin-15}').name: digital_input_output_read[pin] for pin in range(16, 24)}
         return IoState(
             analog_input=analog_input,
             digital_input=digital_input,
             digital_output=digital_output
         )
 
     def write(analog_output: Optional[Dict[str, int]], digital_output: Optional[Dict[str, bool]]) -> None:
-        for i, value in digital_output.items():
-            client.write_single_coil(DigitalOutputMapping[i].value, 1 if value is True else 0,
-                                     unit=config.configuration.modbus_id)
+        with client:
+            if digital_output is not None:
+                client.write_coils(DigitalOutputMapping[list(digital_output.keys())[0]].value,
+                                   list(digital_output.values()),
+                                   unit=config.configuration.modbus_id)
 
     def initializer():
         nonlocal client
         client = ModbusTcpClient_(config.configuration.host, config.configuration.port)
-        for output, value in config.output["digital"].items():
-            client.write_single_coil(DigitalOutputMapping[output].value, value, unit=config.configuration.modbus_id)
+        with client:
+            client.write_coils(DigitalOutputMapping[list(config.output["digital"].keys())[0]].value,
+                               list(config.output["digital"].values()),
+                               unit=config.configuration.modbus_id)
     return ConfigurableIo(config=config, component_reader=read, component_writer=write, initializer=initializer)
 
 

packages/modules/io_devices/dimm_kit/config.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 from helpermodules.auto_str import auto_str
 from modules.common.io_setup import IoDeviceSetup
@@ -45,12 +45,12 @@ def __init__(self, host: Optional[str] = None, port: int = 8899, modbus_id: int
         self.modbus_id = modbus_id
 
 
-def init_input():
+def init_input() -> Dict[str, Union[Dict[str, bool], Dict[str, float], Dict[str, None]]]:
     return {"analog": {pin.name: None for pin in AnalogInputMapping},
             "digital": {pin.name: False for pin in DigitalInputMapping}}
 
 
-def init_output():
+def init_output() -> Dict[str, Union[Dict[str, bool], Dict[str, float], Dict[str, None]]]:
     return {"analog": {},
             "digital": {pin.name: False for pin in DigitalOutputMapping}}
 
@@ -61,9 +61,9 @@ def __init__(self,
                  name: str = "openWB Dimm- & Control-Kit",
                  type: str = "dimm_kit",
                  id: int = 0,
-                 configuration: IoLanConfiguration = None,
-                 input: Dict[str, Dict[int, float]] = None,
-                 output: Dict[str, Dict[int, float]] = None) -> None:
+                 configuration: Optional[IoLanConfiguration] = None,
+                 input: Optional[Dict[str, Union[Dict[str, bool], Dict[str, float], Dict[str, None]]]] = None,
+                 output: Optional[Dict[str, Union[Dict[str, bool], Dict[str, float], Dict[str, None]]]] = None) -> None:
         if input is None:
             input = init_input()
         if output is None: