feat(energy): add typed energy_totals listener for calendar_history refreshes

Teslemetry/api PR 316 adds server-side polling of energy calendar history,
publishing a compact energy_totals notification (id/product_type/topic/
url/createdAt/isCache/totals) instead of the full document. Adds
listen_EnergyTotals on TeslemetryStreamEnergySite plus the typed
EnergyHistoryTotals dataclass to expose the cumulative totals.

Author: Bre77

AGENTS.md

@@ -10,6 +10,7 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - Config responses are shaped inconsistently: success is flat, `{"updated_vehicles": n}` plus `ignoredFields` when some were dropped, while errors are wrapped, `{"response": null, "error": ...}`. Do not look for `updated_vehicles` under `response`; that lookup silently never matches.
 - `update_config` funnels every caller through one per-vehicle single-flight flush (`TeslemetryStreamVehicle._flush`): the first caller starts it, later callers merge into the same pending config and await it rather than starting their own PATCH. This exists because a batch of listeners scheduled at once (e.g. HA integration setup) must produce one PATCH, not one per listener - see `tests/test_batch_retry_storm.py`. A body-shaped error (`{"error": ...}`) is terminal for that batch: it is not replayed, but the pending config is kept for the next explicit `update_config` call. A transport-level failure (`aiohttp.ClientError`/timeout) gets one bounded retry inside the same flush. `tests/test_config_update.py` covers the response-shape handling.
 - Energy site events (`teslemetry_stream/energysite.py`) are shaped differently from vehicle signals: `live_status`/`site_info` are flat top-level envelopes (`{createdAt, site_id, isCache?, live_status|site_info}`), not nested under `data`, and the payload is a full opaque document rather than a field delta - there is no per-field config to enable, the server auto-polls subscribed sites. Contract source: Teslemetry/api PR 310 (`src/routes/sse/index.ts`, `liveStatusSchema.ts`, `siteInfoSchema.ts`), flag-gated server-side as of this writing - `tests/test_energysite_events.py` fixtures mirror that PR's schemas.
+- `energy_totals` (Teslemetry/api PR 316) is shaped differently again: the site id rides the `id` field, not `site_id`, alongside `product_type: "energy_site"` and `topic: "energy_totals"` - filter on those three keys, not `site_id`. It carries a compact cumulative `totals` object (`EnergyHistoryTotals` in `const.py`) instead of a document, fires only when the server's periodic `calendar_history` poll detects a change (silence is not staleness), and has no snapshot-on-connect delivery. The `url` field is the canonical REST path to GET the full time series.
 
 ## Maintaining this file
 

README.md

@@ -151,6 +151,19 @@ async def main():
 > **Note:** Energy site streaming ships flag-gated behind [Teslemetry/api#310](https://github.com/Teslemetry/api/pull/310).
 > Until that server feature is enabled, `listen_LiveStatus`/`listen_SiteInfo` will simply never fire.
 
+A third event, `energy_totals`, fires when the server's periodic
+`calendar_history` poll detects the day's history actually changed. It never
+carries the full time series - just cumulative per-type totals and a `url`
+to re-fetch the full document via your own REST client. Silence means no
+change, never a stale value.
+
+```python
+        def energy_totals_callback(totals):
+            print(f"Total home usage: {totals.total_home_usage}")
+
+        remove_energy_totals_listener = site.listen_EnergyTotals(energy_totals_callback)
+```
+
 ## Public Methods in TeslemetryStream Class
 
 ### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False)`
@@ -248,3 +261,6 @@ Listen for energy site live status events. The callback receives the full `live_
 
 ### `listen_SiteInfo(callback: Callable[[dict], None]) -> Callable[[],None]`
 Listen for energy site info events. The callback receives the full `site_info` document.
+
+### `listen_EnergyTotals(callback: Callable[[EnergyHistoryTotals], None]) -> Callable[[],None]`
+Listen for `energy_totals` refresh notifications. The callback receives an `EnergyHistoryTotals` dataclass of cumulative per-type totals - never the full time series. Fires only when the server's periodic history poll detects a change; a consumer wanting the full series should GET the underlying event's `url` via their own REST client.

teslemetry_stream/const.py

@@ -29,6 +29,11 @@ class Key(StrEnum):
     SITE_INFO = "site_info"
     IS_CACHE = "isCache"
     CREATED_AT = "createdAt"
+    ID = "id"
+    PRODUCT_TYPE = "product_type"
+    TOPIC = "topic"
+    URL = "url"
+    TOTALS = "totals"
 
 
 class Signal(StrEnum):
@@ -311,6 +316,56 @@ class Status(StrEnum):
     DISCONNECTED = "DISCONNECTED"
 
 
+class ProductType(StrEnum):
+    """Product types carried by the uniform refresh-notification schema."""
+
+    ENERGY_SITE = "energy_site"
+
+
+class RefreshTopic(StrEnum):
+    """Topics carried by the uniform refresh-notification schema."""
+
+    ENERGY_TOTALS = "energy_totals"
+
+
+@dataclass
+class EnergyHistoryTotals:
+    """Cumulative per-type totals from a refreshed energy_totals document.
+
+    Mirrors the api's `ENERGY_HISTORY_TOTAL_FIELDS` (same names, same
+    order): each field sums that quantity across the polled day's
+    time_series, and stays `None` rather than 0 when the field never
+    appeared in any period.
+    """
+
+    solar_energy_exported: float | None
+    generator_energy_exported: float | None
+    grid_energy_imported: float | None
+    grid_services_energy_imported: float | None
+    grid_services_energy_exported: float | None
+    grid_energy_exported_from_solar: float | None
+    grid_energy_exported_from_generator: float | None
+    grid_energy_exported_from_battery: float | None
+    battery_energy_exported: float | None
+    battery_energy_imported_from_grid: float | None
+    battery_energy_imported_from_solar: float | None
+    battery_energy_imported_from_generator: float | None
+    consumer_energy_imported_from_grid: float | None
+    consumer_energy_imported_from_solar: float | None
+    consumer_energy_imported_from_battery: float | None
+    consumer_energy_imported_from_generator: float | None
+    total_home_usage: float | None
+    total_battery_charge: float | None
+    total_battery_discharge: float | None
+    total_solar_generation: float | None
+    total_grid_energy_exported: float | None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, float | None]) -> "EnergyHistoryTotals":
+        """Build from the event's `totals` dict."""
+        return cls(**{field: data.get(field) for field in cls.__dataclass_fields__})
+
+
 @dataclass
 class TeslaLocation:
     """Location data"""

teslemetry_stream/energysite.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 from typing import TYPE_CHECKING, Any, Callable
 
-from .const import Key
+from .const import EnergyHistoryTotals, Key, ProductType, RefreshTopic
 
 if TYPE_CHECKING:
     from .stream import TeslemetryStream
@@ -53,3 +53,24 @@ def listen_SiteInfo(
             lambda x: callback(x[Key.SITE_INFO]),
             {Key.SITE_ID: self.site_id, Key.SITE_INFO: None},
         )
+
+    def listen_EnergyTotals(
+        self, callback: Callable[[EnergyHistoryTotals], None]
+    ) -> Callable[[], None]:
+        """Listen for energy_totals refresh notifications.
+
+        Unlike live_status/site_info, this event carries no full document -
+        just cumulative totals and the event fires only when the server's
+        5-minute poll actually detects a change. Silence means no change,
+        never staleness; there is no snapshot-on-connect delivery. A
+        consumer wanting the full time series must GET the event's `url`
+        via their own REST client - this listener only exposes the totals.
+        """
+        return self.stream.async_add_listener(
+            lambda x: callback(EnergyHistoryTotals.from_dict(x[Key.TOTALS])),
+            {
+                Key.ID: self.site_id,
+                Key.PRODUCT_TYPE: ProductType.ENERGY_SITE,
+                Key.TOPIC: RefreshTopic.ENERGY_TOTALS,
+            },
+        )

tests/test_energysite_events.py

@@ -1,13 +1,19 @@
-"""Checks energy site listener filtering against PR 310's SSE event shapes.
+"""Checks energy site listener filtering against PR 310/316's SSE event shapes.
 
 Fixtures mirror the `liveStatusSchema`/`siteInfoSchema` from Teslemetry/api
 PR 310: a flat envelope of `createdAt`, `site_id`, optional `isCache`, and
 the full document under `live_status`/`site_info` (opaque, not a delta).
+`energy_totals` fixtures mirror PR 316's notification schema (renamed
+from `calendar_history_refreshed` to `energy_totals` post-merge): the
+uniform notification shape
+(`id`/`product_type`/`topic`/`url`/`createdAt`/`isCache`) plus a compact
+`totals` object - no `site_id` key, the site id rides `id` instead.
 """
 from __future__ import annotations
 
 from typing import Any
 
+from teslemetry_stream.const import EnergyHistoryTotals
 from teslemetry_stream.stream import TeslemetryStream, recursive_match
 
 SITE_A = "12345"
@@ -55,6 +61,45 @@
     "live_status": {"battery_power": 0},
 }
 
+ENERGY_TOTALS_FIXTURE: dict[str, float | None] = {
+    "solar_energy_exported": 12.3,
+    "generator_energy_exported": None,
+    "grid_energy_imported": 4.5,
+    "grid_services_energy_imported": None,
+    "grid_services_energy_exported": None,
+    "grid_energy_exported_from_solar": None,
+    "grid_energy_exported_from_generator": None,
+    "grid_energy_exported_from_battery": None,
+    "battery_energy_exported": 1.1,
+    "battery_energy_imported_from_grid": None,
+    "battery_energy_imported_from_solar": None,
+    "battery_energy_imported_from_generator": None,
+    "consumer_energy_imported_from_grid": None,
+    "consumer_energy_imported_from_solar": None,
+    "consumer_energy_imported_from_battery": None,
+    "consumer_energy_imported_from_generator": None,
+    "total_home_usage": 20.0,
+    "total_battery_charge": None,
+    "total_battery_discharge": None,
+    "total_solar_generation": 12.3,
+    "total_grid_energy_exported": None,
+}
+
+ENERGY_TOTALS_EVENT: dict[str, Any] = {
+    "id": SITE_A,
+    "product_type": "energy_site",
+    "topic": "energy_totals",
+    "url": f"/api/1/energy_sites/{SITE_A}/calendar_history?kind=energy&period=day",
+    "createdAt": "2026-07-29T10:16:00.000Z",
+    "isCache": False,
+    "totals": ENERGY_TOTALS_FIXTURE,
+}
+
+OTHER_SITE_ENERGY_TOTALS: dict[str, Any] = {
+    **ENERGY_TOTALS_EVENT,
+    "id": SITE_B,
+}
+
 CREDITS_EVENT: dict[str, Any] = {
     "credits": {"type": "snapshot", "cost": 0, "name": "snapshot", "balance": 5, "quota": {}},
     "createdAt": "2026-07-28T10:16:00.000Z",
@@ -139,6 +184,48 @@ def main() -> None:
         )
     )
 
+    # listen_EnergyTotals receives the parsed totals dataclass.
+    stream = make_stream()
+    site = stream.get_energysite(SITE_A)
+    totals_received: list[EnergyHistoryTotals] = []
+    site.listen_EnergyTotals(totals_received.append)
+    dispatch(stream, ENERGY_TOTALS_EVENT)
+    results.append(
+        check(
+            "listen_EnergyTotals parses the totals dict",
+            totals_received == [EnergyHistoryTotals(**ENERGY_TOTALS_FIXTURE)],
+            f"got {totals_received}",
+        )
+    )
+
+    # An energy_totals event for another site is not delivered.
+    stream = make_stream()
+    site = stream.get_energysite(SITE_A)
+    totals_received = []
+    site.listen_EnergyTotals(totals_received.append)
+    dispatch(stream, OTHER_SITE_ENERGY_TOTALS)
+    results.append(
+        check(
+            "a different site's energy_totals is filtered out",
+            totals_received == [],
+            f"got {totals_received}",
+        )
+    )
+
+    # listen_LiveStatus does not receive energy_totals events.
+    stream = make_stream()
+    site = stream.get_energysite(SITE_A)
+    received = []
+    site.listen_LiveStatus(received.append)
+    dispatch(stream, ENERGY_TOTALS_EVENT)
+    results.append(
+        check(
+            "listen_LiveStatus ignores energy_totals events",
+            received == [],
+            f"got {received}",
+        )
+    )
+
     # Unrelated account-wide events (e.g. credits) are not delivered to energy listeners.
     stream = make_stream()
     site = stream.get_energysite(SITE_A)