Hinzufügen des Victron 3P75CT Moduls

packages/modules/common/modbus.py

@@ -11,7 +11,7 @@
 from typing import Any, Callable, Iterable, Optional, Union, overload, List
 
 import pymodbus
-from pymodbus.client.sync import ModbusTcpClient, ModbusSerialClient
+from pymodbus.client.sync import ModbusTcpClient, ModbusUdpClient, ModbusSerialClient
 from pymodbus.constants import Endian
 from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder
 from pymodbus.transaction import ModbusSocketFramer
@@ -50,7 +50,7 @@ def __init__(self, bits: int, encoding_method: str, decoding_method: str):
 
 class ModbusClient:
     def __init__(self,
-                 delegate: Union[ModbusSerialClient, ModbusTcpClient],
+                 delegate: Union[ModbusSerialClient, ModbusTcpClient, ModbusUdpClient],
                  address: str, port: int = 502,
                  sleep_after_connect: Optional[int] = 0):
         self._delegate = delegate
@@ -308,6 +308,19 @@ def __init__(self,
         super().__init__(ModbusTcpClient(host, port, framer, **kwargs), address, port, sleep_after_connect)
 
 
+class ModbusUdpClient_(ModbusClient):
+    def __init__(self,
+                 address: str,
+                 port: int = 502,
+                 sleep_after_connect: Optional[int] = 0,
+                 **kwargs):
+        parsed_url = parse_url(address)
+        host = parsed_url.host
+        if parsed_url.port is not None:
+            port = parsed_url.port
+        super().__init__(ModbusUdpClient(host, port, **kwargs), address, port, sleep_after_connect)
+
+
 class ModbusSerialClient_(ModbusClient):
     def __init__(self,
                  port: int,

packages/modules/conftest.py

@@ -23,6 +23,7 @@
 module = type(sys)('pymodbus.client.sync')
 module.ModbusSerialClient = Mock()
 module.ModbusTcpClient = Mock()
+module.ModbusUdpClient = Mock()
 sys.modules['pymodbus.client.sync'] = module
 
 module = type(sys)('pymodbus.constants')

packages/modules/devices/victron/victron_3p75ct/config.py

@@ -0,0 +1,37 @@
+from typing import Optional
+
+from modules.common.component_setup import ComponentSetup
+from ..vendor import vendor_descriptor
+
+
+class VictronConfiguration:
+    def __init__(self, ip_address: Optional[str] = None, port: int = 502):
+        self.ip_address = ip_address
+        self.port = port
+
+
+class Victron:
+    def __init__(self,
+                 name: str = "Victron 3P75CT",
+                 type: str = "victron_3p75ct",
+                 id: int = 0,
+                 configuration: VictronConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.vendor = vendor_descriptor.configuration_factory().type
+        self.id = id
+        self.configuration = configuration or VictronConfiguration()
+
+
+class VictronCounterConfiguration:
+    def __init__(self):
+        pass
+
+
+class VictronCounterSetup(ComponentSetup[VictronCounterConfiguration]):
+    def __init__(self,
+                 name: str = "Victron Zähler",
+                 type: str = "counter",
+                 id: int = 0,
+                 configuration: VictronCounterConfiguration = None) -> None:
+        super().__init__(name, type, id, configuration or VictronCounterConfiguration())

packages/modules/devices/victron/victron_3p75ct/counter.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+from typing import TypedDict, Any
+
+from modules.common import modbus
+from modules.common.abstract_device import AbstractCounter
+from modules.common.component_state import CounterState
+from modules.common.component_type import ComponentDescriptor
+from modules.common.fault_state import ComponentInfo, FaultState
+from modules.common.modbus import ModbusDataType
+from modules.common.store import get_counter_value_store
+from modules.devices.victron.victron_3p75ct.config import VictronCounterSetup
+
+
+class KwargsDict(TypedDict):
+    device_id: int
+    client: modbus.ModbusUdpClient_
+
+
+class VictronCounter(AbstractCounter):
+    def __init__(self, component_config: VictronCounterSetup, **kwargs: Any) -> None:
+        self.component_config = component_config
+        self.kwargs: KwargsDict = kwargs
+
+    def initialize(self) -> None:
+        self.__udp_client = self.kwargs['client']
+        self.store = get_counter_value_store(self.component_config.id)
+        self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
+
+    def update(self):
+        unit = 1  # Modbus ID ist immer 1 da Standalone Produkt
+        with self.__udp_client:
+            powers = [
+                self.__udp_client.read_holding_registers(reg, ModbusDataType.INT_32, unit=unit) / 1
+                for reg in [0x3082, 0x3086, 0x308A]]
+            currents = [
+                self.__udp_client.read_holding_registers(reg, ModbusDataType.INT_16, unit=unit) / 100
+                for reg in [0x3041, 0x3049, 0x3051]]
+            voltages = [
+                self.__udp_client.read_holding_registers(reg, ModbusDataType.INT_16, unit=unit) / 100
+                for reg in [0x3040, 0x3048, 0x3050]]
+            power_factors = [
+                self.__udp_client.read_holding_registers(reg, ModbusDataType.INT_16, unit=unit) / 1000
+                for reg in [0x3047, 0x304F, 0x3057]]
+
+            power = sum(powers)
+            frequency = self.__udp_client.read_holding_registers(0x3032, ModbusDataType.UINT_16, unit=unit) / 100
+            imported = self.__udp_client.read_holding_registers(0x3034, ModbusDataType.UINT_32, unit=unit) * 10
+            exported = self.__udp_client.read_holding_registers(0x3036, ModbusDataType.UINT_32, unit=unit) * 10
+
+        counter_state = CounterState(
+            voltages=voltages,
+            currents=currents,
+            powers=powers,
+            power_factors=power_factors,
+            imported=imported,
+            exported=exported,
+            power=power,
+            frequency=frequency
+        )
+
+        self.store.set(counter_state)
+
+
+component_descriptor = ComponentDescriptor(configuration_factory=VictronCounterSetup)

packages/modules/devices/victron/victron_3p75ct/device.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+import logging
+from typing import Iterable
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_context import SingleComponentUpdateContext
+from modules.common.configurable_device import ComponentFactoryByType, ConfigurableDevice, MultiComponentUpdater
+from modules.common.modbus import ModbusUdpClient_
+from modules.devices.victron.victron_3p75ct.config import Victron, VictronCounterSetup
+from modules.devices.victron.victron_3p75ct.counter import VictronCounter
+
+log = logging.getLogger(__name__)
+
+
+def create_device(device_config: Victron):
+    client = None
+
+    def create_counter_component(component_config: VictronCounterSetup):
+        nonlocal client
+        return VictronCounter(component_config, device_id=device_config.id, client=client)
+
+    def update_components(components: Iterable[VictronCounter]):
+        nonlocal client
+        with client:
+            for component in components:
+                with SingleComponentUpdateContext(component.fault_state):
+                    component.update()
+
+    def initializer():
+        nonlocal client
+        client = ModbusUdpClient_(device_config.configuration.ip_address, device_config.configuration.port, timeout=30)
+
+    return ConfigurableDevice(
+        device_config=device_config,
+        initializer=initializer,
+        component_factory=ComponentFactoryByType(
+            counter=create_counter_component
+        ),
+        component_updater=MultiComponentUpdater(update_components)
+    )
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Victron)