Consumer: Add Tasmota

packages/modules/consumers/tasmota/tasmota/config.py

@@ -0,0 +1,27 @@
+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 TasmotaConfiguration:
+    def __init__(self, ip_address: Optional[str] = None):
+        self.ip_address = ip_address
+
+
+@auto_str
+class Tasmota(ConsumerSetup[TasmotaConfiguration]):
+    def __init__(self,
+                 name: str = "Tasmota",
+                 type: str = "tasmota",
+                 id: int = 0,
+                 configuration: Optional[TasmotaConfiguration] = None,
+                 usage: Tuple[ConsumerUsage, ...] = (ConsumerUsage.CONTINUOUS,
+                                                     ConsumerUsage.SUSPENDABLE_ONOFF,
+                                                     ConsumerUsage.METER_ONLY),
+                 **kwargs) -> None:
+        super().__init__(name, type, id, vendor=vendor_descriptor.configuration_factory(
+        ).type, configuration=configuration or TasmotaConfiguration(), usage=usage, **kwargs)

packages/modules/consumers/tasmota/tasmota/consumer.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+import logging
+from typing import Optional
+
+from modules.common import req
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_state import ConsumerState
+from modules.common.configurable_consumer import ConfigurableConsumer
+from modules.consumers.tasmota.tasmota.config import Tasmota
+from modules.devices.tasmota.tasmota.parser import TasmotaParser
+
+log = logging.getLogger(__name__)
+
+
+def create_consumer(config: Tasmota):
+    session: Optional[req.CustomSession] = None
+
+    def initializer():
+        nonlocal session
+        session = req.get_http_session()
+
+    def error_handler() -> None:
+        initializer()
+
+    def update() -> ConsumerState:
+        # Status 0 liefert Relais- und Energiemesswerte in einer Antwort.
+        url = f"http://{config.configuration.ip_address}/cm?cmnd=Status%200"
+        response = session.get(url, timeout=5).json()
+        parser = TasmotaParser(response)
+        return ConsumerState(
+            power=parser.power,
+            state=parser.state,
+            voltages=parser.voltages,
+            currents=parser.currents,
+            powers=parser.powers,
+        )
+
+    def switch_on() -> None:
+        session.get(f"http://{config.configuration.ip_address}/cm?cmnd=Power%20on", timeout=5)
+
+    def switch_off() -> None:
+        session.get(f"http://{config.configuration.ip_address}/cm?cmnd=Power%20off", timeout=5)
+
+    return ConfigurableConsumer(consumer_config=config,
+                                initializer=initializer,
+                                error_handler=error_handler,
+                                update=update,
+                                switch_on=switch_on,
+                                switch_off=switch_off)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=Tasmota)

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