Qcells active battery control

packages/modules/devices/qcells/qcells/bat.py

@@ -1,7 +1,9 @@
 #!/usr/bin/env python3
 import logging
 from typing import TypedDict, Any, Optional
+from pymodbus.constants import Endian
 
+from control import data
 from modules.common.abstract_device import AbstractBat
 from modules.common.component_state import BatState
 from modules.common.component_type import ComponentDescriptor
@@ -14,22 +16,19 @@
 
 log = logging.getLogger(__name__)
 
-# Solax/QCells Mode 8 Remote Control Registers (Holding Registers)
-# Speichersteuerung via "Individual Setting - Duration Mode"
-# Unterstuetzte Hardware: QCells Q.VOLT HYB-G3-3P (Solax Gen4),
-# Solax Gen4/Gen5/Gen6 Hybrid und AC Wechselrichter.
-REMOTE_CONTROL_MODE_REG = 0xA0           # U16: 0=Disabled, 8=Individual Duration
-REMOTE_CONTROL_SET_TYPE_REG = 0xA1       # U16: 1=Set
-REMOTE_CONTROL_PV_LIMIT_REG = 0xA2       # U32: PV Power Limit in Watt (keine Begrenzung = 30000)
-REMOTE_CONTROL_PUSH_POWER_REG = 0xA4     # S32: Battery Push Power (+Entladung, -Ladung)
-REMOTE_CONTROL_DURATION_REG = 0xA6       # U16: Dauer in Sekunden
-REMOTE_CONTROL_TIMEOUT_REG = 0xA7        # U16: Timeout in Sekunden
-
-MODE_8_INDIVIDUAL_DURATION = 8
+# Solax/QCells Mode 1 Remote Control Registers (Holding Registers)
+# Speichersteuerung via Active Power Sollwert.
+REMOTE_CONTROL_MODE_REG = 0x7C           # U16: 0=Disabled, 1=Enabled Power Control
+REMOTE_CONTROL_SET_TYPE_REG = 0x7D       # U16: 1=Set
+REMOTE_CONTROL_ACTIVE_POWER_REG = 0x7E   # S32: Active Power Sollwert in Watt
+REMOTE_CONTROL_DURATION_REG = 0x82       # U16: Dauer in Sekunden
+REMOTE_CONTROL_TIMEOUT_REG = 0x88        # U16: Timeout in Sekunden
+
+MODE_1_DISABLED = 0
+MODE_1_ENABLED_POWER_CONTROL = 1
 SET_TYPE_SET = 1
-PV_LIMIT_NO_CURTAILMENT = 30000
-REMOTE_CONTROL_DURATION = 300
-REMOTE_CONTROL_TIMEOUT = 300
+MIN_REMOTE_CONTROL_DURATION = 20
+MIN_REMOTE_CONTROL_TIMEOUT = 60
 
 
 class KwargsDict(TypedDict):
@@ -77,40 +76,75 @@ def set_power_limit(self, power_limit: Optional[int]) -> None:
             if self.last_mode is not None:
                 with self.client:
                     self.client.write_register(
-                        REMOTE_CONTROL_MODE_REG, 0, data_type=ModbusDataType.UINT_16, unit=unit)
+                        REMOTE_CONTROL_MODE_REG, MODE_1_DISABLED,
+                        data_type=ModbusDataType.UINT_16, unit=unit)
                 self.last_mode = None
         else:
-            log.debug("Aktive Batteriesteuerung aktiv")
-            if self.last_mode != 'limited':
-                self.last_mode = 'limited'
-
-            # Berechne power value: 0 = stop, != 0 = multipliziere mit -1
-            power_value = 0 if power_limit == 0 else int(power_limit) * -1
-
-            self._write_mode8(power_value, unit=unit)
-
-    def _write_mode8(self, power_value: int, unit: int) -> None:
-        """Schreibt die Mode 8 Remote Control Register (0xA0-0xA7)."""
+            if power_limit < 0:
+                self.last_mode = 'discharge'
+            elif power_limit > 0:
+                self.last_mode = 'charge'
+            else:
+                self.last_mode = 'stop'
+
+            ap_target = self._get_active_power_target(int(power_limit))
+            self._write_mode1(ap_target, unit=unit)
+
+    def _get_active_power_target(self, power_limit: int) -> int:
+        # openWB-Werte verwenden (nicht WR-Berechnungen):
+        # power_limit < 0 = Entladen, 0 = Stop, > 0 = Laden
+        home_consumption = int(data.data.counter_all_data.data.set.home_consumption)
+        cp_power = int(data.data.cp_all_data.data.get.power)
+        house_load = max(0, home_consumption + cp_power)
+        pv_generation = max(0, int(data.data.pv_all_data.data.get.power * -1))
+        ap_target = power_limit + house_load - pv_generation
+
+        try:
+            evu_counter = data.data.counter_all_data.get_evu_counter()
+            import_limit = int(evu_counter.data.config.max_total_power)
+        except Exception:
+            import_limit = 0
+
+        if import_limit > 0:
+            ap_target = min(ap_target, import_limit)
+
+        log.debug((
+            f"QCells Mode1 target: power_limit={power_limit}W, home_consumption={home_consumption}W, "
+            f"cp_power={cp_power}W, house_load={house_load}W, "
+            f"pv_generation={pv_generation}W, import_limit={import_limit}W -> ap_target={ap_target}W"
+        ))
+        return int(ap_target)
+
+    def _write_mode1(self, ap_target: int, unit: int) -> None:
+        """Schreibt die Mode 1 Remote Control Register (0x7C-0x88)."""
+        duration, timeout = self._get_mode1_timing()
         with self.client:
             self.client.write_register(
-                REMOTE_CONTROL_MODE_REG, MODE_8_INDIVIDUAL_DURATION,
+                REMOTE_CONTROL_MODE_REG, MODE_1_ENABLED_POWER_CONTROL,
                 data_type=ModbusDataType.UINT_16, unit=unit)
             self.client.write_register(
                 REMOTE_CONTROL_SET_TYPE_REG, SET_TYPE_SET,
                 data_type=ModbusDataType.UINT_16, unit=unit)
             self.client.write_register(
-                REMOTE_CONTROL_PV_LIMIT_REG, PV_LIMIT_NO_CURTAILMENT,
-                data_type=ModbusDataType.UINT_32, unit=unit)
+                REMOTE_CONTROL_ACTIVE_POWER_REG, ap_target,
+                data_type=ModbusDataType.INT_32, wordorder=Endian.Little, unit=unit)
             self.client.write_register(
-                REMOTE_CONTROL_PUSH_POWER_REG, power_value,
-                data_type=ModbusDataType.INT_32, unit=unit)
-            self.client.write_register(
-                REMOTE_CONTROL_DURATION_REG, REMOTE_CONTROL_DURATION,
+                REMOTE_CONTROL_DURATION_REG, duration,
                 data_type=ModbusDataType.UINT_16, unit=unit)
             self.client.write_register(
-                REMOTE_CONTROL_TIMEOUT_REG, REMOTE_CONTROL_TIMEOUT,
+                REMOTE_CONTROL_TIMEOUT_REG, timeout,
                 data_type=ModbusDataType.UINT_16, unit=unit)
 
+    def _get_mode1_timing(self) -> tuple[int, int]:
+        try:
+            control_interval = int(data.data.general_data.data.control_interval)
+        except Exception:
+            control_interval = 10
+
+        duration = max(MIN_REMOTE_CONTROL_DURATION, control_interval * 2)
+        timeout = max(MIN_REMOTE_CONTROL_TIMEOUT, control_interval * 3)
+        return duration, timeout
+
     def power_limit_controllable(self) -> bool:
         return True