Consumer: Add M-TEC, Weishaupt, Xtherma

packages/modules/consumers/mtec/mtec/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 MtecConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502,
+                 modbus_id: int = 1):
+        self.ip_address = ip_address
+        self.port = port
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class Mtec(ConsumerSetup[MtecConfiguration]):
+    def __init__(self,
+                 name: str = "M-TEC Wärmepumpe",
+                 type: str = "mtec",
+                 id: int = 0,
+                 configuration: Optional[MtecConfiguration] = None,
+                 # Am Wärmepumpenregler muss "Modul Type" auf "Virtual" und "WMZ Modus" auf
+                 # "Analog" gestellt werden, siehe M-TEC-Anleitung zur PV-Überschussnutzung.
+                 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 MtecConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/mtec/mtec/consumer.py

@@ -0,0 +1,47 @@
+#!/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.mtec.mtec.config import Mtec
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: Mtec):
+    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 707: elektrische Leistungsaufnahme [W]
+        power = client.read_holding_registers(707, ModbusDataType.INT_16, unit=config.configuration.modbus_id)
+        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 1000: PV-Überschussleistung [W]
+        value = 0 if power_limit is None else max(round(power_limit), 0)
+        client.write_register(1000, 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=Mtec)

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

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

packages/modules/consumers/weishaupt/weishaupt_wwp/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 WeishauptConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502,
+                 modbus_id: int = 1):
+        self.ip_address = ip_address
+        self.port = port
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class Weishaupt(ConsumerSetup[WeishauptConfiguration]):
+    def __init__(self,
+                 name: str = "Weishaupt Wärmepumpe (AEROBLOCK/BIBLOCK/SPLITBLOCK/GEOBLOCK)",
+                 type: str = "weishaupt_wwp",
+                 id: int = 0,
+                 configuration: Optional[WeishauptConfiguration] = None,
+                 # kein SUSPENDABLE_ONOFF: die SG-Ready-Eingänge sind auf diesen Geräten laut
+                 # Weishaupt-Datenpunktliste (Modbus TCP WWP) nur lesbar, die Leistungsvorgabe
+                 # SollwertPV übersteuert SG-Ready ohnehin.
+                 usage: Tuple[ConsumerUsage, ...] = (ConsumerUsage.SUSPENDABLE_TUNABLE,),
+                 **kwargs) -> None:
+        super().__init__(name, type, id, vendor=vendor_descriptor.configuration_factory(
+        ).type, configuration=configuration or WeishauptConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/weishaupt/weishaupt_wwp/consumer.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+import logging
+from typing import Optional
+
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.configurable_consumer import ConfigurableConsumer, SetLimitData
+from modules.common.modbus import ModbusDataType, ModbusTcpClient_
+from modules.consumers.weishaupt.weishaupt_wwp.config import Weishaupt
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: Weishaupt):
+    client: Optional[ModbusTcpClient_] = None
+
+    def initializer():
+        nonlocal client
+        client = ModbusTcpClient_(config.configuration.ip_address, config.configuration.port)
+
+    def set_power_limit(power_limit: Optional[float], data: SetLimitData) -> None:
+        # Reg 40002: SollwertPV [W], 0 gibt die Regelung an die Wärmepumpe zurück (0..65535,
+        # daher clamp statt negativer Werte).
+        value = 0 if power_limit is None else min(max(round(power_limit), 0), 65535)
+        client.write_register(40002, value, data_type=ModbusDataType.UINT_16, unit=config.configuration.modbus_id)
+
+    # Keine update()-Funktion: die Datenpunktliste (83807301) enthält keine elektrische
+    # Leistungsaufnahme, nur Leistungsanforderung in % (Reg 33103). Wer den Verbrauch erfassen
+    # will, braucht eine separate Leistungsmessung (zB Zwischenzähler/Smart-Plug).
+    return ConfigurableConsumer(consumer_config=config,
+                                initializer=initializer,
+                                set_power_limit=set_power_limit)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Weishaupt)

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

packages/modules/consumers/xtherma/xtherma/config.py

@@ -0,0 +1,32 @@
+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 XthermaConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None,
+                 port: int = 502,
+                 modbus_id: int = 1):
+        self.ip_address = ip_address
+        self.port = port
+        self.modbus_id = modbus_id
+
+
+@auto_str
+class Xtherma(ConsumerSetup[XthermaConfiguration]):
+    def __init__(self,
+                 name: str = "Xtherma Wärmepumpe",
+                 type: str = "xtherma",
+                 id: int = 0,
+                 configuration: Optional[XthermaConfiguration] = None,
+                 usage: Tuple[ConsumerUsage, ...] = (ConsumerUsage.SUSPENDABLE_TUNABLE,
+                                                     ConsumerUsage.SUSPENDABLE_ONOFF,
+                                                     ConsumerUsage.METER_ONLY),
+                 **kwargs) -> None:
+        super().__init__(name, type, id, vendor=vendor_descriptor.configuration_factory(
+        ).type, configuration=configuration or XthermaConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/xtherma/xtherma/consumer.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+import logging
+from typing import Optional
+
+from helpermodules import timecheck
+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.xtherma.xtherma.config import Xtherma
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: Xtherma):
+    client: Optional[ModbusTcpClient_] = None
+    sim_counter: Optional[SimCounterConsumer] = None
+    last_write: Optional[float] = None
+    last_sgready_write: Optional[float] = 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 177 (in_total): elektrische Leistungsaufnahme, Auflösung 10W
+        power = client.read_holding_registers(
+            177, ModbusDataType.UINT_16, unit=config.configuration.modbus_id) * 10
+        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 70/71 dürfen laut Xtherma-Protokoll höchstens 1x/min beschrieben werden.
+        nonlocal last_write
+        if last_write is not None and timecheck.check_timestamp(last_write, 60):
+            return
+        last_write = timecheck.create_timestamp()
+        unit = config.configuration.modbus_id
+        if power_limit is None:
+            client.write_register(70, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+        else:
+            # Reg 71: Überschuss [W], 0..100000
+            client.write_register(
+                71, min(max(round(power_limit), 0), 100000), data_type=ModbusDataType.UINT_16, unit=unit)
+            client.write_register(70, 1, data_type=ModbusDataType.UINT_16, unit=unit)
+
+    def switch_on() -> None:
+        # Reg 60: SG-Ready-Modus, darf laut Xtherma-Protokoll höchstens 1x/30min beschrieben
+        # werden. 1 = Normalbetrieb.
+        nonlocal last_sgready_write
+        if last_sgready_write is not None and timecheck.check_timestamp(last_sgready_write, 1800):
+            return
+        last_sgready_write = timecheck.create_timestamp()
+        client.write_register(60, 1, data_type=ModbusDataType.UINT_16, unit=config.configuration.modbus_id)
+
+    def switch_off() -> None:
+        # Reg 60: SG-Ready-Modus, 2 = Sperre.
+        nonlocal last_sgready_write
+        if last_sgready_write is not None and timecheck.check_timestamp(last_sgready_write, 1800):
+            return
+        last_sgready_write = timecheck.create_timestamp()
+        client.write_register(60, 2, data_type=ModbusDataType.UINT_16, unit=config.configuration.modbus_id)
+
+    return ConfigurableConsumer(consumer_config=config,
+                                initializer=initializer,
+                                update=update,
+                                set_power_limit=set_power_limit,
+                                switch_on=switch_on,
+                                switch_off=switch_off)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Xtherma)