Consumer: Add Kermi, LG Therma V, Luxtronik

packages/modules/consumers/kermi/kermi/config.py

@@ -0,0 +1,31 @@
+from typing import Optional, Tuple
+
+from control.consumer.consumer_data import ConsumerUsage
+from helpermodules.auto_str import auto_str
+from modules.common.consumer_setup import ConsumerSetup
+from ..vendor import vendor_descriptor
+
+
+@auto_str
+class KermiConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502,
+                 modbus_id: int = 40):
+        self.ip_address = ip_address
+        self.port = port
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class Kermi(ConsumerSetup[KermiConfiguration]):
+    def __init__(self,
+                 name: str = "Kermi Wärmepumpe (x-center/x-change)",
+                 type: str = "kermi",
+                 id: int = 0,
+                 configuration: Optional[KermiConfiguration] = None,
+                 usage: Tuple[ConsumerUsage, ...] = (ConsumerUsage.SUSPENDABLE_TUNABLE,
+                                                     ConsumerUsage.METER_ONLY),
+                 **kwargs) -> None:
+        super().__init__(name, type, id, vendor=vendor_descriptor.configuration_factory(
+        ).type, configuration=configuration or KermiConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/kermi/kermi/consumer.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+import logging
+from typing import Optional
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_state import ConsumerState
+from modules.common.component_type import ComponentType
+from modules.common.configurable_consumer import ConfigurableConsumer, SetLimitData
+from modules.common.modbus import ModbusDataType, ModbusTcpClient_
+from modules.common.simcount._simcounter import SimCounterConsumer
+from modules.consumers.kermi.kermi.config import Kermi
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: Kermi):
+    client: Optional[ModbusTcpClient_] = None
+    sim_counter: Optional[SimCounterConsumer] = None
+
+    def initializer():
+        nonlocal client, sim_counter
+        client = ModbusTcpClient_(config.configuration.ip_address, config.configuration.port)
+        sim_counter = SimCounterConsumer(config.id, ComponentType.CONSUMER)
+
+    def update() -> ConsumerState:
+        # Reg 108: aktuelle Leistungsaufnahme, Auflösung 100W
+        power = client.read_holding_registers(
+            108, ModbusDataType.INT_16, unit=config.configuration.modbus_id) * 100
+        imported, exported = sim_counter.sim_count(power)
+
+        return ConsumerState(
+            power=power,
+            imported=imported,
+            exported=exported
+        )
+
+    def set_power_limit(power_limit: Optional[float], data: SetLimitData) -> None:
+        # Reg 301: PV-Überschussleistung, Auflösung 0,1W (raw = W * 10)
+        value = int(round((0 if power_limit is None else max(power_limit, 0)) * 10))
+        client.write_register(301, value, data_type=ModbusDataType.INT_16, unit=config.configuration.modbus_id)
+
+    return ConfigurableConsumer(consumer_config=config,
+                                initializer=initializer,
+                                update=update,
+                                set_power_limit=set_power_limit)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Kermi)

packages/modules/consumers/kermi/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 = "Kermi"
+        self.group = VendorGroup.VENDORS.value
+
+
+vendor_descriptor = DeviceDescriptor(configuration_factory=Vendor)

packages/modules/consumers/lg_therma/lg_therma/config.py

@@ -0,0 +1,33 @@
+from typing import Optional, Tuple
+
+from control.consumer.consumer_data import ConsumerUsage
+from helpermodules.auto_str import auto_str
+from modules.common.consumer_setup import ConsumerSetup
+from ..vendor import vendor_descriptor
+
+
+@auto_str
+class LgThermaConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502,
+                 modbus_id: int = 33):
+        self.ip_address = ip_address
+        self.port = port
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class LgTherma(ConsumerSetup[LgThermaConfiguration]):
+    def __init__(self,
+                 name: str = "LG Therma V Wärmepumpe",
+                 type: str = "lg_therma",
+                 id: int = 0,
+                 configuration: Optional[LgThermaConfiguration] = None,
+                 # UNVERIFIZIERT: keine offizielle LG-Dokumentation für dieses Register gefunden.
+                 # PI-485-Protokoll selbst ist offiziell dokumentiert.
+                 usage: Tuple[ConsumerUsage, ...] = (ConsumerUsage.SUSPENDABLE_ONOFF,
+                                                     ConsumerUsage.METER_ONLY),
+                 **kwargs) -> None:
+        super().__init__(name, type, id, vendor=vendor_descriptor.configuration_factory(
+        ).type, configuration=configuration or LgThermaConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/lg_therma/lg_therma/consumer.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+import logging
+from typing import Optional
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_state import ConsumerState
+from modules.common.component_type import ComponentType
+from modules.common.configurable_consumer import ConfigurableConsumer
+from modules.common.modbus import ModbusDataType, ModbusTcpClient_
+from modules.common.simcount._simcounter import SimCounterConsumer
+from modules.consumers.lg_therma.lg_therma.config import LgTherma
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: LgTherma):
+    client: Optional[ModbusTcpClient_] = None
+    sim_counter: Optional[SimCounterConsumer] = None
+
+    def initializer():
+        nonlocal client, sim_counter
+        client = ModbusTcpClient_(config.configuration.ip_address, config.configuration.port)
+        sim_counter = SimCounterConsumer(config.id, ComponentType.CONSUMER)
+
+    def switch_on() -> None:
+        # Reg 9: Energy State, 3 = Ein-Empfehlung (Boost)
+        client.write_register(9, 3, data_type=ModbusDataType.UINT_16, unit=config.configuration.modbus_id)
+
+    def switch_off() -> None:
+        # Reg 9: Energy State, 2 = Normalbetrieb
+        client.write_register(9, 2, data_type=ModbusDataType.UINT_16, unit=config.configuration.modbus_id)
+
+    def update() -> ConsumerState:
+        # Reg 35: elektrische Leistungsaufnahme [W]
+        power = client.read_input_registers(35, ModbusDataType.UINT_16, unit=config.configuration.modbus_id)
+        imported, exported = sim_counter.sim_count(power)
+
+        return ConsumerState(
+            power=power,
+            imported=imported,
+            exported=exported
+        )
+
+    return ConfigurableConsumer(consumer_config=config,
+                                initializer=initializer,
+                                update=update,
+                                switch_on=switch_on,
+                                switch_off=switch_off)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=LgTherma)

packages/modules/consumers/lg_therma/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 = "LG"
+        self.group = VendorGroup.VENDORS.value
+
+
+vendor_descriptor = DeviceDescriptor(configuration_factory=Vendor)

packages/modules/consumers/luxtronik/luxtronik/config.py

@@ -0,0 +1,36 @@
+from typing import Optional, Tuple
+
+from control.consumer.consumer_data import ConsumerUsage
+from helpermodules.auto_str import auto_str
+from modules.common.consumer_setup import ConsumerSetup
+from ..vendor import vendor_descriptor
+
+
+@auto_str
+class LuxtronikConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502,
+                 modbus_id: int = 1,
+                 # Anhebung von Heiz- und Warmwassersolltemperatur im Überschussbetrieb [K] -
+                 # abhängig vom Heizsystem (Fußbodenheizung/Heizkörper, Puffergröße), daher
+                 # bewusst konfigurierbar statt fest verdrahtet.
+                 boost_offset: float = 5.0):
+        self.ip_address = ip_address
+        self.port = port
+        self.modbus_id = modbus_id
+        self.boost_offset = boost_offset
+
+
+@auto_str
+class Luxtronik(ConsumerSetup[LuxtronikConfiguration]):
+    def __init__(self,
+                 name: str = "Luxtronik 2.1 Wärmepumpe (Alpha Innotec/Buderus Logamatic HMC/Novelan/Roth/Wolf u.a.)",
+                 type: str = "luxtronik",
+                 id: int = 0,
+                 configuration: Optional[LuxtronikConfiguration] = None,
+                 usage: Tuple[ConsumerUsage, ...] = (ConsumerUsage.SUSPENDABLE_ONOFF,
+                                                     ConsumerUsage.METER_ONLY),
+                 **kwargs) -> None:
+        super().__init__(name, type, id, vendor=vendor_descriptor.configuration_factory(
+        ).type, configuration=configuration or LuxtronikConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/luxtronik/luxtronik/consumer.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+import logging
+from typing import Optional
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_state import ConsumerState
+from modules.common.component_type import ComponentType
+from modules.common.configurable_consumer import ConfigurableConsumer
+from modules.common.modbus import ModbusDataType, ModbusTcpClient_
+from modules.common.simcount._simcounter import SimCounterConsumer
+from modules.consumers.luxtronik.luxtronik.config import Luxtronik
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: Luxtronik):
+    client: Optional[ModbusTcpClient_] = None
+    sim_counter: Optional[SimCounterConsumer] = None
+
+    def initializer():
+        nonlocal client, sim_counter
+        client = ModbusTcpClient_(config.configuration.ip_address, config.configuration.port)
+        sim_counter = SimCounterConsumer(config.id, ComponentType.CONSUMER)
+
+    def switch_on() -> None:
+        unit = config.configuration.modbus_id
+        offset = round(config.configuration.boost_offset * 10)
+        # Reg 10040: LPC-Modus 0=No-Limit. Reg 10000/10005: Heiz./WW-Modus 2=Offset,
+        # Reg 10002/10007: Offset [0.1K].
+        client.write_register(10040, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+        client.write_register(10000, 2, data_type=ModbusDataType.UINT_16, unit=unit)
+        client.write_register(10002, offset, data_type=ModbusDataType.INT_16, unit=unit)
+        client.write_register(10005, 2, data_type=ModbusDataType.UINT_16, unit=unit)
+        client.write_register(10007, offset, data_type=ModbusDataType.INT_16, unit=unit)
+
+    def switch_off() -> None:
+        unit = config.configuration.modbus_id
+        client.write_register(10040, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+        client.write_register(10000, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+        client.write_register(10005, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+
+    def update() -> ConsumerState:
+        # Reg 10301: el. Leistungsaufnahme, Auflösung 100W (kW x0.1)
+        power = client.read_input_registers(
+            10301, ModbusDataType.UINT_16, unit=config.configuration.modbus_id) * 100
+        imported, exported = sim_counter.sim_count(power)
+
+        return ConsumerState(
+            power=power,
+            imported=imported,
+            exported=exported
+        )
+
+    return ConfigurableConsumer(consumer_config=config,
+                                initializer=initializer,
+                                update=update,
+                                switch_on=switch_on,
+                                switch_off=switch_off)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Luxtronik)

packages/modules/consumers/luxtronik/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 = "Luxtronik"
+        self.group = VendorGroup.VENDORS.value
+
+
+vendor_descriptor = DeviceDescriptor(configuration_factory=Vendor)