Qcells, Solax active battery control

packages/modules/conftest.py

@@ -31,6 +31,7 @@
 sys.modules['pymodbus.constants'] = module
 
 module = type(sys)('pymodbus.payload')
+module.BinaryPayloadBuilder = Mock()
 module.BinaryPayloadDecoder = Mock()
 sys.modules['pymodbus.payload'] = module
 

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

@@ -1,15 +1,38 @@
 #!/usr/bin/env python3
-from typing import TypedDict, Any
+import logging
+from typing import Any, Optional, TypedDict
+
+from pymodbus.constants import Endian
+from pymodbus.payload import BinaryPayloadBuilder
 
 from modules.common.abstract_device import AbstractBat
 from modules.common.component_state import BatState
-from modules.common.component_type import ComponentDescriptor
+from modules.common.component_type import ComponentDescriptor, ComponentType
 from modules.common.fault_state import ComponentInfo, FaultState
 from modules.common.modbus import ModbusDataType, ModbusTcpClient_
 from modules.common.store import get_bat_value_store
-from modules.devices.qcells.qcells.config import QCellsBatSetup
 from modules.common.utils.peak_filter import PeakFilter
-from modules.common.component_type import ComponentType
+from modules.devices.qcells.qcells.config import QCellsBatSetup
+
+log = logging.getLogger(__name__)
+
+# Solax/QCells Remote Control Registers (Holding Registers)
+REMOTE_CONTROL_MODE_REG = 0x7C
+REMOTE_CONTROL_SET_TYPE_REG = 0x7D
+REMOTE_CONTROL_ACTIVE_POWER_REG = 0x7E
+REMOTE_CONTROL_REACTIVE_POWER_REG = 0x80
+REMOTE_CONTROL_DURATION_REG = 0x82
+REMOTE_CONTROL_TARGET_SOC_REG = 0x83
+REMOTE_CONTROL_TARGET_ENERGY_REG = 0x84
+REMOTE_CONTROL_TARGET_POWER_REG = 0x86
+REMOTE_CONTROL_TIMEOUT_REG = 0x88
+REMOTE_CONTROL_PUSH_POWER_MODE4_REG = 0x89
+
+MODE_DISABLED = 0
+MODE_4_PUSH_POWER = 4
+SET_TYPE_SET = 1
+MODE_4_TIMEOUT_DISABLED = 0
+MODE4_BLOCK_REG_COUNT = 15
 
 
 class KwargsDict(TypedDict):
@@ -23,28 +46,94 @@ def __init__(self, component_config: QCellsBatSetup, **kwargs: Any) -> None:
         self.kwargs: KwargsDict = kwargs
 
     def initialize(self) -> None:
-        self.__modbus_id: int = self.kwargs['modbus_id']
-        self.client: ModbusTcpClient_ = self.kwargs['client']
+        self.__modbus_id: int = self.kwargs["modbus_id"]
+        self.client: ModbusTcpClient_ = self.kwargs["client"]
         self.store = get_bat_value_store(self.component_config.id)
         self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
         self.peak_filter = PeakFilter(ComponentType.BAT, self.component_config.id, self.fault_state)
+        self.last_mode: Optional[str] = "Undefined"
 
     def update(self) -> None:
         power = self.client.read_input_registers(0x0016, ModbusDataType.INT_16, unit=self.__modbus_id)
         soc = self.client.read_input_registers(0x001C, ModbusDataType.UINT_16, unit=self.__modbus_id)
-        imported = self.client.read_input_registers(
-            0x0021, ModbusDataType.UINT_16, unit=self.__modbus_id) * 100
-        exported = self.client.read_input_registers(
-            0x001D, ModbusDataType.UINT_16, unit=self.__modbus_id) * 100
+        imported = self.client.read_input_registers(0x0021, ModbusDataType.UINT_16, unit=self.__modbus_id) * 100
+        exported = self.client.read_input_registers(0x001D, ModbusDataType.UINT_16, unit=self.__modbus_id) * 100
 
         imported, exported = self.peak_filter.check_values(power, imported, exported)
         bat_state = BatState(
             power=power,
             soc=soc,
             imported=imported,
-            exported=exported
+            exported=exported,
         )
         self.store.set(bat_state)
 
+    def set_power_limit(self, power_limit: Optional[int]) -> None:
+        unit = self.__modbus_id
+        log.debug(f"QCells set_power_limit: power_limit={power_limit}, last_mode={self.last_mode}")
+
+        if power_limit is None:
+            log.debug("Keine Batteriesteuerung, Selbstregelung durch Wechselrichter")
+            if self.last_mode is not None:
+                with self.client:
+                    self.client.write_register(
+                        REMOTE_CONTROL_MODE_REG,
+                        MODE_DISABLED,
+                        data_type=ModbusDataType.UINT_16,
+                        unit=unit,
+                    )
+                self.last_mode = None
+            return
+
+        if power_limit < 0:
+            self.last_mode = "discharge"
+        elif power_limit > 0:
+            self.last_mode = "charge"
+        else:
+            self.last_mode = "stop"
+
+        push_power = self._get_mode4_push_power(int(power_limit))
+        self._write_mode4(push_power, unit)
+
+    def _get_mode4_push_power(self, power_limit: int) -> int:
+        # openWB power_limit semantics:
+        # <0 discharge, 0 stop, >0 charge
+        # Mode 4 push_power semantics:
+        # >0 discharge, 0 stop, <0 charge
+        push_power = int(power_limit * -1)
+        log.debug(f"QCells Mode4 target: power_limit={power_limit}W -> push_power={push_power}W")
+        return push_power
+
+    def _write_mode4(self, push_power: int, unit: int) -> None:
+        log.debug(
+            (
+                f"QCells Mode4 write: mode={MODE_4_PUSH_POWER}, set_type={SET_TYPE_SET}, "
+                f"timeout={MODE_4_TIMEOUT_DISABLED}s, push_power={push_power}W"
+            )
+        )
+        builder = BinaryPayloadBuilder(byteorder=Endian.Big, wordorder=Endian.Little)
+        builder.add_16bit_uint(MODE_4_PUSH_POWER)
+        builder.add_16bit_uint(SET_TYPE_SET)
+        builder.add_32bit_int(0)
+        builder.add_32bit_int(0)
+        builder.add_16bit_uint(0)
+        builder.add_16bit_uint(0)
+        builder.add_32bit_uint(0)
+        builder.add_32bit_int(0)
+        builder.add_16bit_uint(MODE_4_TIMEOUT_DISABLED)
+        builder.add_32bit_int(push_power)
+        payload = builder.to_registers()
+        if len(payload) != MODE4_BLOCK_REG_COUNT:
+            raise RuntimeError(
+                f"Unexpected mode4 payload size {len(payload)}, expected {MODE4_BLOCK_REG_COUNT}"
+            )
+
+        with self.client:
+            # data_type=None with list payload writes a contiguous FC16 block.
+            self.client.write_register(REMOTE_CONTROL_MODE_REG, payload, unit=unit)
+
+    def power_limit_controllable(self) -> bool:
+        return True
+
 
 component_descriptor = ComponentDescriptor(configuration_factory=QCellsBatSetup)

packages/modules/devices/qcells/qcells/bat_test.py

@@ -0,0 +1,25 @@
+from types import SimpleNamespace
+
+from modules.devices.qcells.qcells import bat
+from modules.devices.qcells.qcells.config import QCellsBatSetup
+
+
+def _create_qcells_bat() -> bat.QCellsBat:
+    return bat.QCellsBat(QCellsBatSetup(), modbus_id=1, client=SimpleNamespace())
+
+
+def test_get_mode4_push_power_stop_is_zero() -> None:
+    qcells_bat = _create_qcells_bat()
+    assert qcells_bat._get_mode4_push_power(0) == 0
+
+
+def test_get_mode4_push_power_discharge_is_positive() -> None:
+    qcells_bat = _create_qcells_bat()
+    # openWB discharge limit is negative -> mode4 push power must be positive
+    assert qcells_bat._get_mode4_push_power(-700) == 700
+
+
+def test_get_mode4_push_power_charge_is_negative() -> None:
+    qcells_bat = _create_qcells_bat()
+    # openWB charge limit is positive -> mode4 push power must be negative
+    assert qcells_bat._get_mode4_push_power(1000) == -1000

packages/modules/devices/solax/solax/bat.py

@@ -1,5 +1,9 @@
 #!/usr/bin/env python3
-from typing import Any, TypedDict
+import logging
+from typing import Any, Optional, TypedDict
+
+from pymodbus.constants import Endian
+from pymodbus.payload import BinaryPayloadBuilder
 
 from modules.common import modbus
 from modules.common.abstract_device import AbstractBat
@@ -10,9 +14,20 @@
 from modules.common.simcount import SimCounter
 from modules.common.store import get_bat_value_store
 from modules.devices.solax.solax.config import SolaxBatSetup, Solax
+from modules.devices.solax.solax.version import SolaxVersion
 from modules.common.utils.peak_filter import PeakFilter
 from modules.common.component_type import ComponentType
 
+log = logging.getLogger(__name__)
+
+# Solax Remote Control Registers (Holding Registers)
+REMOTE_CONTROL_MODE_REG = 0x7C
+MODE_DISABLED = 0
+MODE_4_PUSH_POWER = 4
+SET_TYPE_SET = 1
+MODE_4_TIMEOUT_DISABLED = 0
+MODE4_BLOCK_REG_COUNT = 15
+
 
 class KwargsDict(TypedDict):
     client: modbus.ModbusTcpClient_
@@ -31,11 +46,12 @@ def initialize(self) -> None:
         self.store = get_bat_value_store(self.component_config.id)
         self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
         self.peak_filter = PeakFilter(ComponentType.BAT, self.component_config.id, self.fault_state)
+        self.last_mode: Optional[str] = 'Undefined'
 
     def update(self) -> None:
         unit = self.device_config.configuration.modbus_id
 
-        # kein Speicher für Versionen G2 und G4
+        # Basiswerte aus dem Batterie-Registersatz lesen
         power = self.__tcp_client.read_input_registers(0x0016, ModbusDataType.INT_16, unit=unit)
         soc = self.__tcp_client.read_input_registers(0x001C, ModbusDataType.UINT_16, unit=unit)
         self.peak_filter.check_values(power)
@@ -48,5 +64,81 @@ def update(self) -> None:
         )
         self.store.set(bat_state)
 
+    def set_power_limit(self, power_limit: Optional[int]) -> None:
+        if self.power_limit_controllable() is False:
+            log.debug("SolaX set_power_limit: aktive Speichersteuerung für diese Version nicht unterstützt")
+            return
+
+        unit = self.device_config.configuration.modbus_id
+        log.debug(f"SolaX set_power_limit: power_limit={power_limit}, last_mode={self.last_mode}")
+
+        if power_limit is None:
+            log.debug("Keine Batteriesteuerung, Selbstregelung durch Wechselrichter")
+            if self.last_mode is not None:
+                with self.__tcp_client:
+                    self.__tcp_client.write_register(
+                        REMOTE_CONTROL_MODE_REG,
+                        MODE_DISABLED,
+                        data_type=ModbusDataType.UINT_16,
+                        unit=unit,
+                    )
+                self.last_mode = None
+            return
+
+        if power_limit < 0:
+            self.last_mode = 'discharge'
+        elif power_limit > 0:
+            self.last_mode = 'charge'
+        else:
+            self.last_mode = 'stop'
+
+        push_power = self._get_mode4_push_power(int(power_limit))
+        self._write_mode4(push_power, unit)
+
+    def _get_mode4_push_power(self, power_limit: int) -> int:
+        # openWB power_limit semantics:
+        # <0 discharge, 0 stop, >0 charge
+        # Mode 4 push_power semantics:
+        # >0 discharge, 0 stop, <0 charge
+        push_power = int(power_limit * -1)
+        log.debug(f"SolaX Mode4 target: power_limit={power_limit}W -> push_power={push_power}W")
+        return push_power
+
+    def _write_mode4(self, push_power: int, unit: int) -> None:
+        log.debug(
+            (
+                f"SolaX Mode4 write: mode={MODE_4_PUSH_POWER}, set_type={SET_TYPE_SET}, "
+                f"timeout={MODE_4_TIMEOUT_DISABLED}s, push_power={push_power}W"
+            )
+        )
+        builder = BinaryPayloadBuilder(byteorder=Endian.Big, wordorder=Endian.Little)
+        builder.add_16bit_uint(MODE_4_PUSH_POWER)
+        builder.add_16bit_uint(SET_TYPE_SET)
+        builder.add_32bit_int(0)
+        builder.add_32bit_int(0)
+        builder.add_16bit_uint(0)
+        builder.add_16bit_uint(0)
+        builder.add_32bit_uint(0)
+        builder.add_32bit_int(0)
+        builder.add_16bit_uint(MODE_4_TIMEOUT_DISABLED)
+        builder.add_32bit_int(push_power)
+        payload = builder.to_registers()
+        if len(payload) != MODE4_BLOCK_REG_COUNT:
+            raise RuntimeError(
+                f"Unexpected mode4 payload size {len(payload)}, expected {MODE4_BLOCK_REG_COUNT}"
+            )
+
+        with self.__tcp_client:
+            self.__tcp_client.write_register(REMOTE_CONTROL_MODE_REG, payload, unit=unit)
+
+    def power_limit_controllable(self) -> bool:
+        device_config = getattr(self, 'device_config', self.kwargs.get('device_config'))
+        if device_config is None:
+            return False
+        try:
+            return SolaxVersion(device_config.configuration.version) == SolaxVersion.G3
+        except ValueError:
+            return False
+
 
 component_descriptor = ComponentDescriptor(configuration_factory=SolaxBatSetup)

packages/modules/devices/solax/solax/bat_test.py

@@ -0,0 +1,41 @@
+from types import SimpleNamespace
+
+from modules.devices.solax.solax import bat
+from modules.devices.solax.solax.config import Solax, SolaxBatSetup, SolaxConfiguration
+from modules.devices.solax.solax.version import SolaxVersion
+
+
+def _create_solax_bat(version: SolaxVersion) -> bat.SolaxBat:
+    config = SolaxConfiguration(version=version)
+    device_config = Solax(configuration=config)
+    return bat.SolaxBat(SolaxBatSetup(), device_config=device_config, client=SimpleNamespace())
+
+
+def test_get_mode4_push_power_stop_is_zero() -> None:
+    solax_bat = _create_solax_bat(SolaxVersion.G3)
+    assert solax_bat._get_mode4_push_power(0) == 0
+
+
+def test_get_mode4_push_power_discharge_is_positive() -> None:
+    solax_bat = _create_solax_bat(SolaxVersion.G3)
+    assert solax_bat._get_mode4_push_power(-700) == 700
+
+
+def test_get_mode4_push_power_charge_is_negative() -> None:
+    solax_bat = _create_solax_bat(SolaxVersion.G3)
+    assert solax_bat._get_mode4_push_power(1000) == -1000
+
+
+def test_power_limit_controllable_true_for_g3() -> None:
+    solax_bat = _create_solax_bat(SolaxVersion.G3)
+    assert solax_bat.power_limit_controllable() is True
+
+
+def test_power_limit_controllable_false_for_g2() -> None:
+    solax_bat = _create_solax_bat(SolaxVersion.G2)
+    assert solax_bat.power_limit_controllable() is False
+
+
+def test_power_limit_controllable_false_for_g4() -> None:
+    solax_bat = _create_solax_bat(SolaxVersion.G4)
+    assert solax_bat.power_limit_controllable() is False