diff --git a/packages/dataclass_utils/factories.py b/packages/dataclass_utils/factories.py index acb55a2fa8..1043fe375d 100644 --- a/packages/dataclass_utils/factories.py +++ b/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 diff --git a/packages/helpermodules/subdata.py b/packages/helpermodules/subdata.py index 4c59046940..884ad3e0f4 100644 --- a/packages/helpermodules/subdata.py +++ b/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) diff --git a/packages/modules/common/io_setup.py b/packages/modules/common/io_setup.py index 5ae5792e12..f17d7c416b 100644 --- a/packages/modules/common/io_setup.py +++ b/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 diff --git a/packages/modules/common/modbus.py b/packages/modules/common/modbus.py index 5de47a04aa..e3602a8c6a 100644 --- a/packages/modules/common/modbus.py +++ b/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, diff --git a/packages/modules/io_devices/dimm_kit/api.py b/packages/modules/io_devices/dimm_kit/api.py index a94883d812..8d03fb6129 100644 --- a/packages/modules/io_devices/dimm_kit/api.py +++ b/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,16 +42,20 @@ 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, @@ -59,15 +63,19 @@ def read(): ) 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) diff --git a/packages/modules/io_devices/dimm_kit/config.py b/packages/modules/io_devices/dimm_kit/config.py index d2e888932a..a960e88e41 100644 --- a/packages/modules/io_devices/dimm_kit/config.py +++ b/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: