feat: add SunEnergyXT 500 Series battery storage module

packages/modules/devices/sunenergyxt/sunenergyxt/SunEnergyXT500_bat_test.py

@@ -0,0 +1,200 @@
+#!/usr/bin/env python3
+"""Unit tests for SunEnergyXT 500 Series battery module."""
+import pytest
+import requests_mock as req_mock
+
+from unittest.mock import MagicMock, patch
+from modules.devices.sunenergyxt.sunenergyxt.bat import SunEnergyXTBat
+from modules.devices.sunenergyxt.sunenergyxt.config import (
+    SunEnergyXT,
+    SunEnergyXTBatSetup,
+    SunEnergyXTConfiguration,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+BASE_URL = "http://192.168.1.100"
+
+
+def _make_bat() -> SunEnergyXTBat:
+    """Create a SunEnergyXTBat instance with mocked stores."""
+    device_config = SunEnergyXT(
+        configuration=SunEnergyXTConfiguration(ip_address="192.168.1.100")
+    )
+    component_config = SunEnergyXTBatSetup()
+    bat = SunEnergyXTBat(component_config, device_config=device_config)
+
+    # Mock internal helpers so no real HA infrastructure is needed
+    bat.sim_counter = MagicMock()
+    bat.sim_counter.sim_count.return_value = (1000.0, 500.0)
+    bat.store = MagicMock()
+    bat.fault_state = MagicMock()
+    bat._base_url = BASE_URL
+    bat._gs_max = 800  # Fallback-Wert wie in initialize()
+
+    return bat
+
+
+# ---------------------------------------------------------------------------
+# update() – Parsing SC / PB / IS
+# ---------------------------------------------------------------------------
+
+class TestUpdate:
+    def test_update_parses_soc_and_power(self, requests_mock):
+        """update() liest SC (SoC) und PB (Batteriepower) korrekt aus."""
+        bat = _make_bat()
+        requests_mock.get(
+            f"{BASE_URL}/read",
+            json={"state": {"reported": {"SC": 75, "PB": -500, "IS": 800}}}
+        )
+
+        bat.update()
+
+        bat.store.set.assert_called_once()
+        bat_state = bat.store.set.call_args[0][0]
+        assert bat_state.soc == 75
+        assert bat_state.power == -500.0
+
+    def test_update_sets_gs_max_from_is(self, requests_mock):
+        """update() setzt self._gs_max dynamisch aus IS."""
+        bat = _make_bat()
+        requests_mock.get(
+            f"{BASE_URL}/read",
+            json={"state": {"reported": {"SC": 50, "PB": 0, "IS": 2400}}}
+        )
+
+        bat.update()
+
+        assert bat._gs_max == 2400
+
+    def test_update_keeps_fallback_if_is_zero(self, requests_mock):
+        """update() behält Fallback-_gs_max wenn IS=0 geliefert wird."""
+        bat = _make_bat()
+        requests_mock.get(
+            f"{BASE_URL}/read",
+            json={"state": {"reported": {"SC": 50, "PB": 0, "IS": 0}}}
+        )
+
+        bat.update()
+
+        assert bat._gs_max == 800  # Fallback bleibt
+
+    def test_update_handles_flat_json(self, requests_mock):
+        """update() akzeptiert auch flaches JSON ohne 'state'/'reported'."""
+        bat = _make_bat()
+        requests_mock.get(
+            f"{BASE_URL}/read",
+            json={"SC": 42, "PB": 300, "IS": 800}
+        )
+
+        bat.update()
+
+        bat_state = bat.store.set.call_args[0][0]
+        assert bat_state.soc == 42
+        assert bat_state.power == 300.0
+
+    def test_update_uses_simcount(self, requests_mock):
+        """update() ruft sim_counter.sim_count() auf und übergibt Werte."""
+        bat = _make_bat()
+        bat.sim_counter.sim_count.return_value = (2000.0, 1000.0)
+        requests_mock.get(
+            f"{BASE_URL}/read",
+            json={"state": {"reported": {"SC": 80, "PB": -1000, "IS": 800}}}
+        )
+
+        bat.update()
+
+        bat.sim_counter.sim_count.assert_called_once_with(-1000.0)
+        bat_state = bat.store.set.call_args[0][0]
+        assert bat_state.imported == 2000.0
+        assert bat_state.exported == 1000.0
+
+
+# ---------------------------------------------------------------------------
+# set_power_limit() – MM/GS Steuerung
+# ---------------------------------------------------------------------------
+
+class TestSetPowerLimit:
+    def test_none_sets_automatic_mode(self, requests_mock):
+        """power_limit=None → MM=1, GS=0 (Self-Consumption)."""
+        bat = _make_bat()
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(None)
+
+        assert requests_mock.last_request.json() == {"state": {"MM": 1, "GS": 0}}
+
+    def test_zero_stops_discharge(self, requests_mock):
+        """power_limit=0 → MM=0, GS=0 (Entladung gesperrt)."""
+        bat = _make_bat()
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(0)
+
+        assert requests_mock.last_request.json() == {"state": {"MM": 0, "GS": 0}}
+
+    def test_positive_discharges(self, requests_mock):
+        """power_limit>0 → MM=0, GS=+p (Entladen)."""
+        bat = _make_bat()
+        bat._gs_max = 800
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(500)
+
+        assert requests_mock.last_request.json() == {"state": {"MM": 0, "GS": 500}}
+
+    def test_negative_charges(self, requests_mock):
+        """power_limit<0 → MM=0, GS=-p (Laden)."""
+        bat = _make_bat()
+        bat._gs_max = 800
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(-600)
+
+        assert requests_mock.last_request.json() == {"state": {"MM": 0, "GS": -600}}
+
+    def test_discharge_capped_at_gs_max(self, requests_mock):
+        """Entladeleistung wird auf _gs_max (IS) begrenzt."""
+        bat = _make_bat()
+        bat._gs_max = 800
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(9999)  # Weit über Limit
+
+        payload = requests_mock.last_request.json()
+        assert payload["state"]["GS"] == 800
+
+    def test_charge_capped_at_gs_max(self, requests_mock):
+        """Ladeleistung wird auf _gs_max (IS) begrenzt."""
+        bat = _make_bat()
+        bat._gs_max = 2400
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(-9999)  # Weit über Limit
+
+        payload = requests_mock.last_request.json()
+        assert payload["state"]["GS"] == -2400
+
+    def test_gs_max_reflects_pro_model(self, requests_mock):
+        """Nach update() mit IS=2400 (Pro) wird 2400W als Limit genutzt."""
+        bat = _make_bat()
+        bat._gs_max = 2400  # Wie nach update() mit IS=2400
+        requests_mock.post(f"{BASE_URL}/write", json={})
+
+        bat.set_power_limit(-2400)
+
+        payload = requests_mock.last_request.json()
+        assert payload["state"]["GS"] == -2400
+
+
+# ---------------------------------------------------------------------------
+# power_limit_controllable()
+# ---------------------------------------------------------------------------
+
+def test_power_limit_controllable():
+    """power_limit_controllable() muss True zurückgeben."""
+    bat = _make_bat()
+    assert bat.power_limit_controllable() is True

packages/modules/devices/sunenergyxt/sunenergyxt/bat.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+"""SunEnergyXT 500 Series – openWB Batteriespeicher-Modul."""
+import logging
+from typing import Any, Optional
+
+from modules.common import req
+from modules.common.abstract_device import AbstractBat
+from modules.common.component_state import BatState
+from modules.common.component_type import ComponentDescriptor
+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.devices.sunenergyxt.sunenergyxt.config import SunEnergyXT, SunEnergyXTBatSetup
+
+log = logging.getLogger(__name__)
+
+# Fallback-Limit bis IS vom Gerät gelesen wurde (SunEnergyXT 500, 1 Modul)
+_GS_MAX_FALLBACK: int = 800
+
+
+class SunEnergyXTBat(AbstractBat):
+    def __init__(self, component_config: SunEnergyXTBatSetup, **kwargs: Any) -> None:
+        self.component_config = component_config
+        self.kwargs = kwargs
+
+    def initialize(self) -> None:
+        self.device_config: SunEnergyXT = self.kwargs['device_config']
+        self.sim_counter = SimCounter(self.device_config.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._base_url = f"http://{self.device_config.configuration.ip_address}"
+        # Wird beim ersten update() aus IS (Max. Inverterleistung) gesetzt.
+        # IS berücksichtigt automatisch Modell (500 / 500 Pro) und Anzahl Module (BN).
+        self._gs_max: int = _GS_MAX_FALLBACK
+
+    def _read(self) -> dict:
+        url = f"{self._base_url}/read"
+        return req.get_http_session().get(url, timeout=5).json()
+
+    def _write(self, **kwargs) -> None:
+        url = f"{self._base_url}/write"
+        payload = {"state": kwargs}
+        resp = req.get_http_session().post(url, json=payload, timeout=5)
+        log.debug("SunEnergyXT write %s → %s", kwargs, resp.text)
+
+    def update(self) -> None:
+        data = self._read()
+        reported = data.get("state", {}).get("reported", data)
+
+        soc = int(float(reported.get("SC", 0)))
+        power = float(reported.get("PB", 0))
+
+        # IS = max. Inverterleistung: hängt von Modell (500/Pro) und Modulanzahl (BN) ab.
+        # Wird als dynamisches GS-Limit verwendet.
+        is_value = int(float(reported.get("IS", _GS_MAX_FALLBACK)))
+        if is_value > 0:
+            self._gs_max = is_value
+
+        imported, exported = self.sim_counter.sim_count(power)
+
+        bat_state = BatState(
+            power=power,
+            soc=soc,
+            imported=imported,
+            exported=exported,
+        )
+        self.store.set(bat_state)
+        log.debug("SunEnergyXT: SoC=%d%%, PB=%.0fW, IS=%dW (gs_max)", soc, power, self._gs_max)
+
+    def set_power_limit(self, power_limit: Optional[int]) -> None:
+        if power_limit is None:
+            log.debug("SunEnergyXT: Automatik (MM=1, GS=0)")
+            self._write(MM=1, GS=0)
+        elif power_limit == 0:
+            log.debug("SunEnergyXT: Entladung gesperrt (MM=0, GS=0)")
+            self._write(MM=0, GS=0)
+        elif power_limit > 0:
+            p = int(min(power_limit, self._gs_max))
+            log.debug("SunEnergyXT: Entladen mit %dW (gs_max=%dW)", p, self._gs_max)
+            self._write(MM=0, GS=p)
+        else:
+            p = int(min(abs(power_limit), self._gs_max))
+            log.debug("SunEnergyXT: Laden mit %dW (gs_max=%dW)", p, self._gs_max)
+            self._write(MM=0, GS=-p)
+
+    def power_limit_controllable(self) -> bool:
+        return True
+
+
+component_descriptor = ComponentDescriptor(configuration_factory=SunEnergyXTBatSetup)

packages/modules/devices/sunenergyxt/sunenergyxt/config.py

@@ -0,0 +1,36 @@
+from typing import Optional
+from modules.common.component_setup import ComponentSetup
+from ..vendor import vendor_descriptor
+
+
+class SunEnergyXTConfiguration:
+    def __init__(self,
+                 ip_address: Optional[str] = None):
+        self.ip_address = ip_address
+
+
+class SunEnergyXT:
+    def __init__(self,
+                 name: str = "SunEnergyXT 500 Series",
+                 type: str = "sunenergyxt",
+                 id: int = 0,
+                 configuration: SunEnergyXTConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.vendor = vendor_descriptor.configuration_factory().type
+        self.id = id
+        self.configuration = configuration or SunEnergyXTConfiguration()
+
+
+class SunEnergyXTBatConfiguration:
+    def __init__(self):
+        pass
+
+
+class SunEnergyXTBatSetup(ComponentSetup[SunEnergyXTBatConfiguration]):
+    def __init__(self,
+                 name: str = "SunEnergyXT Speicher",
+                 type: str = "bat",
+                 id: int = 0,
+                 configuration: SunEnergyXTBatConfiguration = None) -> None:
+        super().__init__(name, type, id, configuration or SunEnergyXTBatConfiguration())

packages/modules/devices/sunenergyxt/sunenergyxt/device.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+from typing import Iterable
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.component_context import SingleComponentUpdateContext
+from modules.common.configurable_device import ComponentFactoryByType, ConfigurableDevice, MultiComponentUpdater
+from modules.devices.sunenergyxt.sunenergyxt.bat import SunEnergyXTBat
+from modules.devices.sunenergyxt.sunenergyxt.config import SunEnergyXT, SunEnergyXTBatSetup
+
+
+def create_device(device_config: SunEnergyXT):
+    def create_bat_component(component_config: SunEnergyXTBatSetup):
+        return SunEnergyXTBat(component_config, device_config=device_config)
+
+    def update_components(components: Iterable[SunEnergyXTBat]):
+        for component in components:
+            with SingleComponentUpdateContext(component.fault_state):
+                component.update()
+
+    return ConfigurableDevice(
+        device_config=device_config,
+        component_factory=ComponentFactoryByType(
+            bat=create_bat_component,
+        ),
+        component_updater=MultiComponentUpdater(update_components)
+    )
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=SunEnergyXT)

packages/modules/devices/sunenergyxt/vendor.py

@@ -0,0 +1,11 @@
+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 = "SunEnergyXT"
+        self.group = VendorGroup.VENDORS.value
+
+vendor_descriptor = DeviceDescriptor(configuration_factory=Vendor)