From 5dfd8a9846ebc20d4d1eaf9dbcc44c94dd4a7166 Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 15:46:53 +1000 Subject: [PATCH] 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. --- AGENTS.md | 1 + README.md | 16 ++++++ teslemetry_stream/const.py | 55 ++++++++++++++++++++ teslemetry_stream/energysite.py | 23 ++++++++- tests/test_energysite_events.py | 89 ++++++++++++++++++++++++++++++++- 5 files changed, 182 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 59ec65f..95997be 100644 --- a/AGENTS.md +++ b/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 diff --git a/README.md b/README.md index 0ab5f07..49dddc3 100644 --- a/README.md +++ b/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. diff --git a/teslemetry_stream/const.py b/teslemetry_stream/const.py index b22531d..59c873e 100644 --- a/teslemetry_stream/const.py +++ b/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""" diff --git a/teslemetry_stream/energysite.py b/teslemetry_stream/energysite.py index 18d851d..c65e223 100644 --- a/teslemetry_stream/energysite.py +++ b/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, + }, + ) diff --git a/tests/test_energysite_events.py b/tests/test_energysite_events.py index f42ff5b..a8b84aa 100644 --- a/tests/test_energysite_events.py +++ b/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)