Add device und component modules for Solakon One battery

packages/modules/devices/solakon/solakon_one/bat.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+import logging
+from typing import TypedDict, Any
+from modules.common.abstract_device import AbstractBat
+from modules.common.component_state import BatState
+from modules.common.component_type import ComponentDescriptor
+from modules.common.fault_state import ComponentInfo, FaultState
+from modules.common.modbus import ModbusDataType, ModbusTcpClient_
+from modules.common.store import get_bat_value_store
+from modules.devices.solakon.solakon_one.config import SolakonOneBatSetup
+
+log = logging.getLogger(__name__)
+
+
+class KwargsDict(TypedDict):
+    client: ModbusTcpClient_
+
+
+class SolakonOneBat(AbstractBat):
+    def __init__(self, component_config: SolakonOneBatSetup, **kwargs: Any) -> None:
+        self.component_config = component_config
+        self.kwargs: KwargsDict = kwargs
+
+    def initialize(self) -> None:
+        self.client: ModbusTcpClient_ = self.kwargs['client']
+        self.store = get_bat_value_store(self.component_config.id)
+        self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
+
+    def update(self) -> None:
+        unit = self.component_config.configuration.modbus_id
+
+        # AC Leistung am Stecker, Batterie aus dem Netz aufladen hat positive Werte,
+        # Leistung aus der Batterie und/oder aus PV ins Netz abgeben hat negative Werte
+        power = self.client.read_holding_registers(39134, ModbusDataType.INT_32, unit=unit) * -1
+        # SoC Ladezustand der Batterie in %
+        soc = self.client.read_holding_registers(39424, ModbusDataType.INT_16, unit=unit)
+        # gesamte DC Ladung der Batterie in Wh
+        imported = self.client.read_holding_registers(39605, ModbusDataType.UINT_32, unit=unit) * 10
+        # gesamte DC Entladung der Batterie in Wh
+        exported = self.client.read_holding_registers(39609, ModbusDataType.UINT_32, unit=unit) * 10
+
+        bat_state = BatState(
+            power=power,
+            soc=soc,
+            imported=imported,
+            exported=exported
+        )
+        self.store.set(bat_state)
+
+
+component_descriptor = ComponentDescriptor(configuration_factory=SolakonOneBatSetup)

packages/modules/devices/solakon/solakon_one/config.py

@@ -0,0 +1,58 @@
+from typing import Optional
+
+from helpermodules.auto_str import auto_str
+from modules.common.component_setup import ComponentSetup
+from ..vendor import vendor_descriptor
+
+
+class SolakonOneConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502):
+        self.ip_address = ip_address
+        self.port = port
+
+
+class SolakonOne:
+    def __init__(self,
+                 name: str = "Solakon One",
+                 type: str = "solakon_one",
+                 id: int = 0,
+                 configuration: SolakonOneConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.vendor = vendor_descriptor.configuration_factory().type
+        self.id = id
+        self.configuration = configuration or SolakonOneConfiguration()
+
+
+@auto_str
+class SolakonOneBatConfiguration:
+    def __init__(self, modbus_id: int = 1):
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class SolakonOneBatSetup(ComponentSetup[SolakonOneBatConfiguration]):
+    def __init__(self,
+                 name: str = "Solakon One Speicher",
+                 type: str = "bat",
+                 id: int = 0,
+                 configuration: SolakonOneBatConfiguration = None) -> None:
+        super().__init__(name, type, id, configuration or SolakonOneBatConfiguration())
+
+
+@auto_str
+class SolakonOneInverterConfiguration:
+    def __init__(self, modbus_id: int = 1):
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class SolakonOneInverterSetup(ComponentSetup[SolakonOneInverterConfiguration]):
+    def __init__(self,
+                 name: str = "Solakon One Wechselrichter",
+                 type: str = "inverter",
+                 id: int = 0,
+                 configuration: SolakonOneInverterConfiguration = None) -> None:
+        super().__init__(name, type, id, configuration or SolakonOneInverterConfiguration())

packages/modules/devices/solakon/solakon_one/device.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+import logging
+from typing import Iterable, Union
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_context import SingleComponentUpdateContext
+from modules.common.configurable_device import ConfigurableDevice, ComponentFactoryByType, MultiComponentUpdater
+from modules.common.modbus import ModbusTcpClient_
+from modules.devices.solakon.solakon_one.bat import SolakonOneBat
+from modules.devices.solakon.solakon_one.inverter import SolakonOneInverter
+from modules.devices.solakon.solakon_one.config import SolakonOne, SolakonOneBatSetup, SolakonOneInverterSetup
+
+log = logging.getLogger(__name__)
+
+
+def create_device(device_config: SolakonOne):
+    client = None
+
+    def create_bat_component(component_config: SolakonOneBatSetup):
+        nonlocal client
+        return SolakonOneBat(component_config=component_config, client=client)
+
+    def create_inverter_component(component_config: SolakonOneInverterSetup):
+        nonlocal client
+        return SolakonOneInverter(component_config=component_config, client=client)
+
+    def update_components(components: Iterable[Union[SolakonOneBat, SolakonOneInverter]]):
+        nonlocal client
+        with client:
+            for component in components:
+                with SingleComponentUpdateContext(component.fault_state):
+                    component.update()
+
+    def initializer():
+        nonlocal client
+        client = ModbusTcpClient_(device_config.configuration.ip_address, device_config.configuration.port)
+
+    return ConfigurableDevice(
+        device_config=device_config,
+        initializer=initializer,
+        component_factory=ComponentFactoryByType(
+            bat=create_bat_component,
+            inverter=create_inverter_component,
+        ),
+        component_updater=MultiComponentUpdater(update_components)
+    )
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=SolakonOne)

packages/modules/devices/solakon/solakon_one/inverter.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+from typing import TypedDict, Any
+
+from modules.common.abstract_device import AbstractInverter
+from modules.common.component_state import InverterState
+from modules.common.component_type import ComponentDescriptor
+from modules.common.fault_state import ComponentInfo, FaultState
+from modules.common.modbus import ModbusDataType, ModbusTcpClient_
+from modules.common.store import get_inverter_value_store
+from modules.devices.solakon.solakon_one.config import SolakonOneInverterSetup
+
+
+class KwargsDict(TypedDict):
+    client: ModbusTcpClient_
+
+
+class SolakonOneInverter(AbstractInverter):
+    def __init__(self, component_config: SolakonOneInverterSetup, **kwargs: Any) -> None:
+        self.component_config = component_config
+        self.kwargs: KwargsDict = kwargs
+
+    def initialize(self) -> None:
+        self.client: ModbusTcpClient_ = self.kwargs['client']
+        self.store = get_inverter_value_store(self.component_config.id)
+        self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
+
+    def update(self) -> None:
+        unit = self.component_config.configuration.modbus_id
+        # Gesamte DC PV Leistung aller vier MPPT in W
+        power = self.client.read_holding_registers(39118, ModbusDataType.INT_32, unit=unit)
+        # Gesamte DC PV Produktion in Wh
+        exported = self.client.read_holding_registers(39601, ModbusDataType.UINT_32, unit=unit) * 10
+
+        inverter_state = InverterState(
+            power=power,
+            exported=exported
+        )
+        self.store.set(inverter_state)
+
+
+component_descriptor = ComponentDescriptor(configuration_factory=SolakonOneInverterSetup)

packages/modules/devices/solakon/vendor.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.devices.vendors import VendorGroup
+
+
+class Vendor:
+    def __init__(self):
+        self.type = Path(__file__).parent.name
+        self.vendor = "Solakon"
+        self.group = VendorGroup.VENDORS.value
+
+
+vendor_descriptor = DeviceDescriptor(configuration_factory=Vendor)