Add per-charger energy statistics feed for Energy Dashboard

README.md

@@ -55,12 +55,18 @@ Confirmed to work with Zaptec products
   [here](https://www.home-assistant.io/common-tasks/general/#defining-a-custom-polling-interval),
   will have unexpected effects. If the automatic polling is turned off, not all
   the data in the integration will update properly.
-* Using the _Energy Meter_ entity as an input to the Energy Dashboard will give values that are delayed by 1 hour
-  in the graphs (see [issue 162](https://github.com/custom-components/zaptec/issues/162) for details).
-  There is a plan to solve this in [issue 300](https://github.com/custom-components/zaptec/issues/300), but until that is implemented,
-  a workaround is to use the more frequently updated _Session total charge_ entity instead. This reduces the delay-issue,
-  but has a separate drawback where a restart of Home Assistant during a charging session can give a fake spike in the logged
-  consumption that needs to be manually edited using "Adjust sum" in the Statistics tab of the Developer tools dashboard.
+* Using the _Energy Meter_ entity directly as an input to the Energy Dashboard will still give values delayed by up to an hour
+  in the graphs, since that entity reflects live (polling-delayed) state (see [issue 162](https://github.com/custom-components/zaptec/issues/162)).
+  Each tracked charger now also gets a separate, invisible statistics feed (backdated hourly from Zaptec's charge history)
+  that appears in the Energy Dashboard's device picker
+  as "<charger name> Energy" - use that entry instead of _Energy Meter_ or _Session total charge_ for accurate,
+  correctly-timed consumption graphs.
+  Because it is built from completed charge sessions and imported hourly, the most recent hour or two can lag -
+  an active session's energy only appears once that session ends and the next import runs - so it is meant for
+  accurate historical graphs rather than real-time monitoring (the _Energy Meter_ sensor covers the live view).
+  This feed requires Home Assistant's [Recorder](https://www.home-assistant.io/integrations/recorder/)
+  integration (enabled by default). If you have disabled the recorder, the "<charger name> Energy" entry
+  simply isn't created - the rest of the integration continues to work as normal.
 
 ## Features missing from the API
 

custom_components/zaptec/__init__.py

@@ -4,6 +4,7 @@
 
 import logging
 
+from homeassistant.components.recorder import DOMAIN as RECORDER_DOMAIN
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
 from homeassistant.core import HomeAssistant
@@ -25,9 +26,11 @@
 from .coordinator import ZaptecUpdateCoordinator, ZaptecUpdateOptions
 from .manager import ZaptecConfigEntry, ZaptecManager
 from .services import async_setup_services, async_unload_services
+from .statistics import ZaptecStatisticsCoordinator
 from .zaptec import (
     RETRYABLE_HTTP_STATUSES,
     AuthenticationError,
+    Charger,
     Installation,
     RequestConnectionError,
     RequestError,
@@ -185,9 +188,26 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             ),
         )
 
+    # One statistics coordinator per tracked charger, backdating hourly energy
+    # into the Energy Dashboard. Recorder is an after_dependency, so
+    # if it's disabled we skip only this feed, not the whole integration.
+    if RECORDER_DOMAIN in hass.config.components:
+        for deviceid in tracked_devices:
+            zaptec_obj = zaptec[deviceid]
+            if isinstance(zaptec_obj, Charger):
+                manager.statistics_coordinators[deviceid] = ZaptecStatisticsCoordinator(
+                    hass, entry=entry, charger=zaptec_obj
+                )
+    else:
+        _LOGGER.debug("Recorder not enabled; skipping energy-statistics import")
+
     # Initialize the coordinators
     for co in manager.all_coordinators:
         await co.async_config_entry_first_refresh()
+    # async_refresh (not async_config_entry_first_refresh): a failure on this
+    # secondary, Owner-only feed must not abort setup of the whole config entry.
+    for co in manager.statistics_coordinators.values():
+        await co.async_refresh()
 
     # Done setting up, change back to not log all updates. Having this enabled
     # will create a lot of debug log output.

custom_components/zaptec/const.py

@@ -51,3 +51,9 @@
     "three_to_one_phase_switch_current",
     "total_charge_power_session",
 }
+
+ZAPTEC_STATISTICS_POLL_INTERVAL = 60 * 60
+"""Interval in seconds between imports of archived charge sessions into HA statistics."""
+
+ZAPTEC_STATISTICS_BACKFILL_DAYS = 730
+"""How far back (in days) to backfill energy statistics on first import."""

custom_components/zaptec/manager.py

@@ -18,6 +18,7 @@
 from .const import DOMAIN, KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, MANUFACTURER
 from .coordinator import ZaptecUpdateCoordinator
 from .entity import KeyUnavailableError, ZaptecBaseEntity
+from .statistics import ZaptecStatisticsCoordinator
 from .zaptec import Charger, Installation, Zaptec, ZaptecBase
 
 _LOGGER = logging.getLogger(__name__)
@@ -53,6 +54,9 @@ class ZaptecManager:
     device_coordinators: dict[str, ZaptecUpdateCoordinator]
     """Coordinators for the devices, both installation and chargers."""
 
+    statistics_coordinators: dict[str, ZaptecStatisticsCoordinator]
+    """Coordinators that backdate hourly energy statistics, one per tracked charger."""
+
     streams: list[tuple[asyncio.Task, Installation]]
     """List of active streams for the installations."""
 
@@ -72,6 +76,7 @@ def __init__(
         self.tracked_devices = tracked_devices or set()
         self.name_prefix = name_prefix
         self.device_coordinators = {}
+        self.statistics_coordinators = {}
         self.streams = []
 
     @property

custom_components/zaptec/manifest.json

@@ -1,12 +1,14 @@
 {
   "domain": "zaptec",
   "name": "Zaptec EV charger",
+  "after_dependencies": [
+    "recorder"
+  ],
   "codeowners": [
     "@sveinse",
     "@hellowlol"
   ],
   "config_flow": true,
-  "dependencies": [],
   "documentation": "https://github.com/custom-components/zaptec",
   "integration_type": "hub",
   "iot_class": "cloud_polling",

custom_components/zaptec/statistics.py

@@ -0,0 +1,224 @@
+"""Import Zaptec charge history into Home Assistant's long-term statistics."""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import datetime, timedelta
+from http import HTTPStatus
+import logging
+from typing import TYPE_CHECKING, Any
+
+from homeassistant.components.recorder.db_schema import StatisticsMeta
+from homeassistant.components.recorder.models import (
+    StatisticData,
+    StatisticMeanType,
+    StatisticMetaData,
+)
+from homeassistant.components.recorder.statistics import (
+    async_add_external_statistics,
+    get_last_statistics,
+)
+from homeassistant.const import UnitOfEnergy
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.recorder import get_instance
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.util import dt as dt_util
+from homeassistant.util.unit_conversion import EnergyConverter
+
+from .const import DOMAIN, ZAPTEC_STATISTICS_BACKFILL_DAYS, ZAPTEC_STATISTICS_POLL_INTERVAL
+from .zaptec import Charger, RequestError, ZaptecApiError
+
+if TYPE_CHECKING:
+    from .manager import ZaptecConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+RESUME_MARGIN = timedelta(hours=26)
+"""Lookback added to the fetch window on resume - not a session-length limit.
+
+The API filters by session *end* time, which can lag a session's final meter
+timestamp slightly; querying from `last_start - RESUME_MARGIN` keeps such a
+session in the window. No-double-count is guaranteed by bucket_sessions_hourly's
+`after` filter regardless - this only bounds how far back we re-scan."""
+
+_SUPPORTS_UNIT_CLASS = hasattr(StatisticsMeta, "unit_class")
+"""HA core gained the statistics `unit_class` field around 2026.4 - feature-detect
+it rather than pin a version."""
+
+
+_HOUR_BOUNDARY_TOLERANCE = timedelta(seconds=5)
+"""Snap tolerance before flooring, so a scheduled on-the-hour report landing a
+few seconds either side of the boundary still buckets into the intended hour.
+
+5s is safely between the two relevant scales: an order of magnitude above the
+sub-second jitter observed on real on-the-hour reports (so it reliably catches
+them), yet far below the minutes-scale gap between distinct reports (meter
+interval is 30 min or hourly), so it can never merge two different reports."""
+
+
+def _floor_hour(value: datetime) -> datetime:
+    """Floor a datetime to the start of its UTC hour.
+
+    Snaps up first if `value` is within `_HOUR_BOUNDARY_TOLERANCE` of the next
+    hour, so a report timestamped a few seconds early/late buckets correctly.
+    """
+    value = dt_util.as_utc(value)
+    floor = value.replace(minute=0, second=0, microsecond=0)
+    if value - floor >= timedelta(hours=1) - _HOUR_BOUNDARY_TOLERANCE:
+        floor += timedelta(hours=1)
+    return floor
+
+
+def bucket_sessions_hourly(
+    sessions: list[dict[str, Any]],
+    *,
+    after: datetime | None,
+    running_sum: float,
+) -> list[StatisticData]:
+    """Convert archived charge sessions into hourly external-statistics points.
+
+    Each `energyDetails` point's `energy` is the incremental delta since the
+    previous point (not a cumulative total; verified against the session's
+    OCMF-signed meter readings). Reports arrive roughly hourly while metering
+    but are skipped while idle, so gaps can span many hours. A delta is bucketed
+    to the hour immediately *before* its own timestamp - where the energy was
+    actually drawn, confirmed against the charger's power sensor - except a
+    session's final, irregular point, which is not walked back past the previous
+    point's hour. For back-to-back hourly reports both rules agree. A delta
+    straddling an hour boundary lands wholly in one hour rather than being split,
+    but this still fixes the live sensor's hour lag.
+
+    Sessions without `energyDetails` (pre-3.2 firmware) fall back to a single
+    point at `endDateTime` with the session's total `energy`; `voided`/`aborted`
+    sessions are skipped (no meaningful energy).
+
+    `sessions` must be oldest-first (guaranteed by the API). `after` is the start
+    of the last hour already imported; points are skipped when their *floored
+    hour* is <= `after`, not by raw timestamp - otherwise a mid-hour point like
+    11:10 would be re-added to the already-stored 11:00 bucket and compound on
+    every poll. `running_sum` is the kWh imported so far; returned points chain
+    onto it so `sum` stays monotonic.
+    """
+    hourly_deltas: dict[datetime, float] = defaultdict(float)
+
+    for session in sessions:
+        if session.get("voided") or session.get("aborted"):
+            continue
+
+        details = session.get("energyDetails") or []
+        if not details:
+            end = session.get("endDateTime")
+            energy = session.get("energy") or 0.0
+            if end and energy:
+                details = [{"timestamp": end, "energy": energy}]
+
+        prev_timestamp: datetime | None = None
+        for point in details:
+            timestamp = dt_util.parse_datetime(point["timestamp"])
+            if timestamp is None:
+                continue
+            delta = point["energy"]
+            if prev_timestamp is None:
+                hour = _floor_hour(timestamp)
+            else:
+                hour = max(
+                    _floor_hour(timestamp) - timedelta(hours=1), _floor_hour(prev_timestamp)
+                )
+            prev_timestamp = timestamp
+            if after is not None and hour <= after:
+                continue
+            hourly_deltas[hour] += delta
+
+    statistics: list[StatisticData] = []
+    for hour in sorted(hourly_deltas):
+        running_sum += hourly_deltas[hour]
+        statistics.append(StatisticData(start=hour, state=hourly_deltas[hour], sum=running_sum))
+    return statistics
+
+
+class ZaptecStatisticsCoordinator(DataUpdateCoordinator[None]):
+    """Imports one charger's archived sessions into HA long-term statistics.
+
+    Independent of the live-state coordinators (coordinator.py), so a failure
+    here (e.g. a non-Owner 403) degrades only the Energy Dashboard feed,
+    not the whole integration.
+    """
+
+    config_entry: ZaptecConfigEntry
+
+    def __init__(
+        self, hass: HomeAssistant, *, entry: ZaptecConfigEntry, charger: Charger
+    ) -> None:
+        """Initialize the statistics coordinator for one charger."""
+        self.charger = charger
+        self.statistic_id = f"{DOMAIN}:energy_{charger.id.replace('-', '')}"
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=entry,
+            name=f"{DOMAIN}-statistics-{charger.qual_id}",
+            update_interval=timedelta(seconds=ZAPTEC_STATISTICS_POLL_INTERVAL),
+        )
+
+    async def _async_update_data(self) -> None:
+        """Fetch new archived sessions and import them as external statistics."""
+        last_stats = await get_instance(self.hass).async_add_executor_job(
+            get_last_statistics,
+            self.hass,
+            1,
+            self.statistic_id,
+            True,  # noqa: FBT003
+            {"sum"},
+        )
+        rows = last_stats.get(self.statistic_id) if last_stats else None
+        last = rows[0] if rows else None
+        if last is not None and (start_ts := last.get("start")) is not None:
+            last_start = dt_util.utc_from_timestamp(start_ts)
+            running_sum = last.get("sum") or 0.0
+        else:
+            last_start = dt_util.utcnow() - timedelta(days=ZAPTEC_STATISTICS_BACKFILL_DAYS)
+            running_sum = 0.0
+
+        sessions: list[dict[str, Any]] = []
+        cursor: str | None = None
+        try:
+            while True:
+                page = await self.charger.get_archived_sessions(
+                    from_time=last_start - RESUME_MARGIN,
+                    to_time=dt_util.utcnow(),
+                    cursor=cursor,
+                )
+                sessions.extend(page.get("sessions") or [])
+                if not page.get("hasMore"):
+                    break
+                cursor = page.get("cursor")
+        except RequestError as err:
+            if err.error_code == HTTPStatus.FORBIDDEN:
+                # Owner-only endpoint: warn rather than fail the coordinator
+                # every poll for non-Owner accounts.
+                _LOGGER.warning(
+                    "No permission to read charge history for %s (requires Owner role), "
+                    "skipping energy statistics import",
+                    self.charger.qual_id,
+                )
+                return
+            raise UpdateFailed(err) from err
+        except ZaptecApiError as err:
+            raise UpdateFailed(err) from err
+
+        statistics = bucket_sessions_hourly(sessions, after=last_start, running_sum=running_sum)
+        if not statistics:
+            return
+
+        metadata_kwargs: dict[str, Any] = {
+            "mean_type": StatisticMeanType.NONE,
+            "has_sum": True,
+            "name": f"{self.charger.name} Energy",
+            "source": DOMAIN,
+            "statistic_id": self.statistic_id,
+            "unit_of_measurement": UnitOfEnergy.KILO_WATT_HOUR,
+        }
+        if _SUPPORTS_UNIT_CLASS:
+            metadata_kwargs["unit_class"] = EnergyConverter.UNIT_CLASS
+        metadata = StatisticMetaData(**metadata_kwargs)
+        async_add_external_statistics(self.hass, metadata, statistics)