feat: add parsed battery storage data with power integration

src/aiosolaredge/__init__.py

@@ -2,6 +2,7 @@
 
 __version__ = "1.0.2"
 
+from .models import BatteryStorageData, StorageData, integrate_power
 from .solaredge import SolarEdge
 
-__all__ = ["SolarEdge"]
+__all__ = ["BatteryStorageData", "SolarEdge", "StorageData", "integrate_power"]

src/aiosolaredge/models.py

@@ -0,0 +1,142 @@
+"""Models and parsing for the SolarEdge Monitoring API storage data."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from itertools import pairwise
+from typing import Any
+
+_TELEMETRY_DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S"
+
+
+def _parse_timestamp(value: Any) -> datetime | None:
+    """Parse a storageData telemetry timestamp, returning None when invalid."""
+    if not isinstance(value, str):
+        return None
+    try:
+        return datetime.strptime(value, _TELEMETRY_DATETIME_FORMAT)
+    except ValueError:
+        return None
+
+
+def integrate_power(telemetries: list[dict[str, Any]]) -> tuple[float, float]:
+    """
+    Integrate battery power telemetry into charged and discharged energy.
+
+    SolarEdge frequently reports ``lifeTimeEnergyCharged`` and
+    ``lifeTimeEnergyDischarged`` as 0 in the ``storageData`` response, even
+    while the battery is actively charging or discharging. The ``power``
+    telemetry is reliably populated though, so the energy (Wh) is derived by
+    trapezoidal integration of the power (W) over time. Positive power means
+    the battery is charging, negative power means it is discharging.
+
+    :param telemetries: The ``telemetries`` list of a single battery, ordered
+        chronologically. Each entry is expected to have a ``timeStamp``
+        (``"%Y-%m-%d %H:%M:%S"``) and a ``power`` value in Watts.
+    :return: A ``(charge_energy, discharge_energy)`` tuple in Wh.
+    """
+    charge_energy = 0.0
+    discharge_energy = 0.0
+    for previous, current in pairwise(telemetries):
+        previous_time = _parse_timestamp(previous.get("timeStamp"))
+        current_time = _parse_timestamp(current.get("timeStamp"))
+        if previous_time is None or current_time is None:
+            continue
+        interval = (current_time - previous_time).total_seconds() / 3600
+        if interval <= 0:
+            continue
+        # Trapezoidal integration of power (W) over the interval (h) -> Wh.
+        average_power = (
+            (previous.get("power") or 0.0) + (current.get("power") or 0.0)
+        ) / 2
+        energy = average_power * interval
+        if energy >= 0:
+            charge_energy += energy
+        else:
+            discharge_energy -= energy
+    return charge_energy, discharge_energy
+
+
+@dataclass(slots=True)
+class BatteryStorageData:
+    """
+    Parsed storage data for a single battery.
+
+    The ``charge_energy`` and ``discharge_energy`` values are derived by
+    integrating the ``power`` telemetry rather than read from
+    ``lifeTimeEnergyCharged`` / ``lifeTimeEnergyDischarged``, because SolarEdge
+    frequently reports those lifetime counters as 0 in the ``storageData``
+    response even while the battery is actively cycling.
+    """
+
+    serial_number: str
+    model_number: str | None
+    nameplate: float | None
+    state_of_charge: float | None
+    power: float | None
+    charge_energy: float
+    discharge_energy: float
+
+    @classmethod
+    def from_dict(cls, battery: dict[str, Any]) -> BatteryStorageData | None:
+        """
+        Create a ``BatteryStorageData`` from a raw ``storageData`` battery.
+
+        :param battery: A single entry from ``storageData.batteries``.
+        :return: The parsed battery, or ``None`` when the battery has no serial
+            number or no telemetry to derive values from.
+        """
+        serial_number = battery.get("serialNumber")
+        if not serial_number:
+            return None
+        telemetries = battery.get("telemetries") or []
+        if not telemetries:
+            return None
+        latest = telemetries[-1]
+        charge_energy, discharge_energy = integrate_power(telemetries)
+        return cls(
+            serial_number=serial_number,
+            model_number=battery.get("modelNumber"),
+            nameplate=battery.get("nameplate"),
+            state_of_charge=latest.get("batteryPercentageState"),
+            power=latest.get("power"),
+            charge_energy=charge_energy,
+            discharge_energy=discharge_energy,
+        )
+
+
+@dataclass(slots=True)
+class StorageData:
+    """Parsed storage data for all batteries at a SolarEdge site."""
+
+    batteries: list[BatteryStorageData] = field(default_factory=list)
+
+    @property
+    def total_charge_energy(self) -> float:
+        """Total charged energy across all batteries in Wh."""
+        return sum(battery.charge_energy for battery in self.batteries)
+
+    @property
+    def total_discharge_energy(self) -> float:
+        """Total discharged energy across all batteries in Wh."""
+        return sum(battery.discharge_energy for battery in self.batteries)
+
+    @classmethod
+    def from_response(cls, response: dict[str, Any]) -> StorageData:
+        """
+        Create ``StorageData`` from a raw ``storageData`` API response.
+
+        :param response: The JSON returned by the ``storageData`` endpoint.
+        :raises KeyError: if the response is missing ``storageData`` or its
+            ``batteries`` list.
+        :return: The parsed storage data; batteries without a serial number or
+            telemetry are skipped.
+        """
+        batteries = response["storageData"]["batteries"]
+        parsed = [
+            parsed_battery
+            for battery in batteries
+            if (parsed_battery := BatteryStorageData.from_dict(battery)) is not None
+        ]
+        return cls(batteries=parsed)

src/aiosolaredge/solaredge.py

@@ -7,6 +7,8 @@
 import aiohttp
 import yarl
 
+from .models import StorageData
+
 _BASE_URL = yarl.URL("https://monitoringapi.solaredge.com")
 _DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S"
 
@@ -126,6 +128,33 @@ async def get_storage_data(
             params["serials"] = ",".join(serials)
         return await self._get_json(url, params=params)
 
+    async def get_parsed_storage_data(
+        self,
+        site_id: int | str,
+        start_time: datetime,
+        end_time: datetime,
+        serials: Iterable[str] = (),
+    ) -> StorageData:
+        """
+        Get parsed battery storage data from the SolarEdge site.
+
+        Convenience wrapper around :meth:`get_storage_data` that parses the raw
+        response into :class:`~aiosolaredge.models.StorageData`. The charge and
+        discharge energy are derived by integrating the ``power`` telemetry,
+        because SolarEdge frequently reports the lifetime energy counters as 0
+        even while the battery is cycling. Limited to a one-week period.
+
+        :param site_id: The site ID.
+        :param start_time: The start time.
+        :param end_time: The end time.
+        :param serials: Optional battery serial numbers to filter by.
+        :return: The parsed storage data of the SolarEdge system.
+        :raises KeyError: if the response is missing ``storageData`` or its
+            ``batteries`` list.
+        """
+        response = await self.get_storage_data(site_id, start_time, end_time, serials)
+        return StorageData.from_response(response)
+
     async def get_current_power_flow(self, site_id: int | str) -> dict[str, Any]:
         """
         Get current power flow of the SolarEdge system.

tests/test_init.py

@@ -109,4 +109,32 @@ async def test_simple_requests() -> None:
             datetime.datetime.now(),
             serials=["SN1", "SN2"],
         ) == {"storageData": {"batteryCount": 1, "batteries": []}}
+
+        parsed_pattern = re.compile(
+            r"^https://monitoringapi\.solaredge\.com/site/123/storageData"
+        )
+        mocked.get(
+            parsed_pattern,
+            payload={
+                "storageData": {
+                    "batteryCount": 1,
+                    "batteries": [
+                        {
+                            "serialNumber": "SN1",
+                            "telemetries": [
+                                {"timeStamp": "2026-06-05 00:00:00", "power": 0},
+                                {"timeStamp": "2026-06-05 01:00:00", "power": 2000},
+                            ],
+                        }
+                    ],
+                }
+            },
+        )
+        storage = await solar_edge.get_parsed_storage_data(
+            123, datetime.datetime.now(), datetime.datetime.now()
+        )
+        assert len(storage.batteries) == 1
+        assert storage.batteries[0].serial_number == "SN1"
+        assert storage.total_charge_energy == 1000.0
+        assert storage.total_discharge_energy == 0.0
         await solar_edge.close()

tests/test_models.py

@@ -0,0 +1,149 @@
+import pytest
+
+from aiosolaredge import BatteryStorageData, StorageData, integrate_power
+
+
+def _telemetry(timestamp, power):
+    return {"timeStamp": timestamp, "power": power}
+
+
+@pytest.mark.parametrize(
+    ("telemetries", "expected"),
+    [
+        # Not enough data points to integrate anything.
+        ([], (0.0, 0.0)),
+        ([_telemetry("2026-06-05 00:00:00", 1000)], (0.0, 0.0)),
+        # Constant 2000 W charge over two hourly intervals -> 2000 Wh charged.
+        (
+            [
+                _telemetry("2026-06-05 00:00:00", 0),
+                _telemetry("2026-06-05 01:00:00", 2000),
+                _telemetry("2026-06-05 02:00:00", 0),
+            ],
+            (2000.0, 0.0),
+        ),
+        # Discharge (negative power) is accumulated as a positive number.
+        (
+            [
+                _telemetry("2026-06-05 00:00:00", 0),
+                _telemetry("2026-06-05 01:00:00", -1000),
+            ],
+            (0.0, 500.0),
+        ),
+        # Mixed charge then discharge.
+        (
+            [
+                _telemetry("2026-06-05 00:00:00", 1000),
+                _telemetry("2026-06-05 01:00:00", 1000),
+                _telemetry("2026-06-05 02:00:00", -1000),
+                _telemetry("2026-06-05 03:00:00", -1000),
+            ],
+            (1000.0, 1000.0),
+        ),
+        # Unparsable / missing timestamps are skipped.
+        (
+            [
+                _telemetry("not-a-timestamp", 1000),
+                _telemetry("2026-06-05 01:00:00", 2000),
+                _telemetry(None, 2000),
+            ],
+            (0.0, 0.0),
+        ),
+        # Non-monotonic timestamps (interval <= 0) are skipped.
+        (
+            [
+                _telemetry("2026-06-05 02:00:00", 2000),
+                _telemetry("2026-06-05 01:00:00", 2000),
+            ],
+            (0.0, 0.0),
+        ),
+        # Missing power is treated as 0 W.
+        (
+            [
+                _telemetry("2026-06-05 00:00:00", None),
+                _telemetry("2026-06-05 01:00:00", 2000),
+            ],
+            (1000.0, 0.0),
+        ),
+    ],
+)
+def test_integrate_power(telemetries, expected):
+    assert integrate_power(telemetries) == expected
+
+
+def test_battery_from_dict():
+    battery = {
+        "serialNumber": "SN1",
+        "modelNumber": "BAT-1",
+        "nameplate": 10000,
+        "telemetries": [
+            _telemetry("2026-06-05 00:00:00", 0) | {"batteryPercentageState": 50},
+            _telemetry("2026-06-05 01:00:00", 2000) | {"batteryPercentageState": 70},
+        ],
+    }
+    parsed = BatteryStorageData.from_dict(battery)
+    assert parsed is not None
+    assert parsed.serial_number == "SN1"
+    assert parsed.model_number == "BAT-1"
+    assert parsed.nameplate == 10000
+    assert parsed.state_of_charge == 70
+    assert parsed.power == 2000
+    assert parsed.charge_energy == 1000.0
+    assert parsed.discharge_energy == 0.0
+
+
+def test_battery_from_dict_without_serial():
+    assert BatteryStorageData.from_dict({"telemetries": []}) is None
+
+
+def test_battery_from_dict_without_telemetry():
+    assert BatteryStorageData.from_dict({"serialNumber": "SN1"}) is None
+    assert (
+        BatteryStorageData.from_dict({"serialNumber": "SN1", "telemetries": []}) is None
+    )
+
+
+def test_storage_data_from_response_totals():
+    response = {
+        "storageData": {
+            "batteryCount": 2,
+            "batteries": [
+                {
+                    "serialNumber": "SN1",
+                    "telemetries": [
+                        _telemetry("2026-06-05 00:00:00", 0),
+                        _telemetry("2026-06-05 01:00:00", 2000),
+                    ],
+                },
+                {
+                    "serialNumber": "SN2",
+                    "telemetries": [
+                        _telemetry("2026-06-05 00:00:00", 0),
+                        _telemetry("2026-06-05 01:00:00", -1000),
+                    ],
+                },
+                # Skipped: no serial number.
+                {"telemetries": []},
+            ],
+        }
+    }
+    storage = StorageData.from_response(response)
+    assert len(storage.batteries) == 2
+    assert storage.total_charge_energy == 1000.0
+    assert storage.total_discharge_energy == 500.0
+
+
+def test_storage_data_from_response_empty():
+    storage = StorageData.from_response(
+        {"storageData": {"batteryCount": 0, "batteries": []}}
+    )
+    assert storage.batteries == []
+    assert storage.total_charge_energy == 0.0
+    assert storage.total_discharge_energy == 0.0
+
+
+def test_storage_data_from_response_missing_keys():
+    with pytest.raises(KeyError):
+        StorageData.from_response({})
+    with pytest.raises(KeyError):
+        StorageData.from_response({"storageData": {}})