BatterX: add bat control

packages/modules/devices/batterx/batterx/bat.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
-from typing import Dict, TypedDict, Any
+import logging
+from typing import Dict, TypedDict, Any, Optional
 
 from modules.devices.batterx.batterx.config import BatterXBatSetup
 from modules.common.abstract_device import AbstractBat
@@ -8,10 +9,14 @@
 from modules.common.fault_state import ComponentInfo, FaultState
 from modules.common.simcount import SimCounter
 from modules.common.store import get_bat_value_store
+from modules.common import req
+
+log = logging.getLogger(__name__)
 
 
 class KwargsDict(TypedDict):
     device_id: int
+    ip_address: str
 
 
 class BatterXBat(AbstractBat):
@@ -21,9 +26,11 @@ def __init__(self, component_config: BatterXBatSetup, **kwargs: Any) -> None:
 
     def initialize(self) -> None:
         self.__device_id: int = self.kwargs['device_id']
+        self.__ip_address: str = self.kwargs['ip_address']
         self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="speicher")
         self.store = get_bat_value_store(self.component_config.id)
         self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
+        self.last_mode = 'Undefined'
 
     def update(self, resp: Dict) -> None:
         power = resp["1121"]["1"]
@@ -37,5 +44,57 @@ def update(self, resp: Dict) -> None:
         )
         self.store.set(bat_state)
 
+    def set_power_limit(self, power_limit: Optional[int]) -> None:
+        log.debug(f'last_mode: {self.last_mode}')
+
+        if power_limit is None:
+            # Kein Powerlimit gefordert, externe Steuerung deaktivieren
+            log.debug("Keine Batteriesteuerung gefordert, deaktiviere externe Steuerung.")
+            if self.last_mode is not None:
+                # Battery Charge AC - OFF
+                req.get_http_session().get(
+                    f"http://{self.__ip_address}/api.php?set=command&type=20738&text1=3&text2=0",
+                    timeout=5
+                )
+                # Battery Discharging - ON
+                req.get_http_session().get(
+                    f"http://{self.__ip_address}/api.php?set=command&type=20738&text1=4&text2=1",
+                    timeout=5
+                )
+                self.last_mode = None
+        elif power_limit <= 0:
+            # BatterX kann Entladung nur komplett sperren
+            log.debug("Aktive Batteriesteuerung angestoßen. Setze Entladesperre.")
+            if self.last_mode != 'stop':
+                # Battery Charge AC - OFF
+                req.get_http_session().get(
+                    f"http://{self.__ip_address}/api.php?set=command&type=20738&text1=3&text2=0",
+                    timeout=5
+                )
+                # Battery Discharging - OFF
+                req.get_http_session().get(
+                    f"http://{self.__ip_address}/api.php?set=command&type=20738&text1=4&text2=0",
+                    timeout=5
+                )
+                self.last_mode = 'stop'
+        else:
+            # Aktive Ladung
+            log.debug("Aktive Batteriesteuerung angestoßen. Setze aktive Ladung.")
+            if self.last_mode != 'charge':
+                # Battery Charge AC - ON
+                req.get_http_session().get(
+                    f"http://{self.__ip_address}/api.php?set=command&type=20738&text1=3&text2=1",
+                    timeout=5
+                )
+                # Battery Discharging - OFF
+                req.get_http_session().get(
+                    f"http://{self.__ip_address}/api.php?set=command&type=20738&text1=4&text2=0",
+                    timeout=5
+                )
+                self.last_mode = 'charge'
+
+    def power_limit_controllable(self) -> bool:
+        return True
+
 
 component_descriptor = ComponentDescriptor(configuration_factory=BatterXBatSetup)

packages/modules/devices/batterx/batterx/device.py

@@ -23,7 +23,9 @@
 
 def create_device(device_config: BatterX):
     def create_bat_component(component_config: BatterXBatSetup):
-        return bat.BatterXBat(component_config=component_config, device_id=device_config.id)
+        return bat.BatterXBat(component_config=component_config,
+                              device_id=device_config.id,
+                              ip_address=device_config.configuration.ip_address)
 
     def create_counter_component(component_config: BatterXCounterSetup):
         return counter.BatterXCounter(component_config=component_config, device_id=device_config.id)