From b0e45af9fb51df87a786369b965d0b30916e7c05 Mon Sep 17 00:00:00 2001 From: it-rec Date: Sat, 6 Jun 2026 13:08:34 +0200 Subject: [PATCH] feat: add parsed battery storage data with power integration Add a StorageData/BatteryStorageData model layer and a SolarEdge.get_parsed_storage_data() convenience method that parse the raw storageData response into typed objects. SolarEdge frequently reports lifeTimeEnergyCharged/lifeTimeEnergyDischarged as 0 in the storageData response even while the battery is actively cycling, so the charge and discharge energy are derived by trapezoidal integration of the reliably-populated power telemetry instead. This keeps the device-specific parsing in the library rather than in downstream consumers such as Home Assistant (see home-assistant/core#169964). --- src/aiosolaredge/__init__.py | 3 +- src/aiosolaredge/models.py | 142 ++++++++++++++++++++++++++++++++ src/aiosolaredge/solaredge.py | 29 +++++++ tests/test_init.py | 28 +++++++ tests/test_models.py | 149 ++++++++++++++++++++++++++++++++++ 5 files changed, 350 insertions(+), 1 deletion(-) create mode 100644 src/aiosolaredge/models.py create mode 100644 tests/test_models.py diff --git a/src/aiosolaredge/__init__.py b/src/aiosolaredge/__init__.py index 5d32894..693c982 100644 --- a/src/aiosolaredge/__init__.py +++ b/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"] diff --git a/src/aiosolaredge/models.py b/src/aiosolaredge/models.py new file mode 100644 index 0000000..36b02f1 --- /dev/null +++ b/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) diff --git a/src/aiosolaredge/solaredge.py b/src/aiosolaredge/solaredge.py index 2d8d9bd..d478bf2 100644 --- a/src/aiosolaredge/solaredge.py +++ b/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. diff --git a/tests/test_init.py b/tests/test_init.py index 9eeba45..a11be4f 100644 --- a/tests/test_init.py +++ b/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() diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..ec666db --- /dev/null +++ b/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": {}})