From 3fbad17ae8a5f25a9caa39a8bfecfa748395db02 Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 10:15:32 +1000 Subject: [PATCH 1/2] feat(energy): add typed live_status/site_info listeners for energy sites Mirrors the vehicle listener idiom: get_energysite(site_id) returns a TeslemetryStreamEnergySite with listen_LiveStatus/listen_SiteInfo, unwrapping the flat live_status/site_info envelope (full documents, not field deltas) and delivering snapshot-then-live the same way vehicle State does. --- AGENTS.md | 1 + CLAUDE.md | 1 + README.md | 50 ++++++++ teslemetry_stream/__init__.py | 2 + teslemetry_stream/const.py | 5 + teslemetry_stream/energysite.py | 55 +++++++++ teslemetry_stream/stream.py | 14 +++ tests/test_energysite_events.py | 196 ++++++++++++++++++++++++++++++++ 8 files changed, 324 insertions(+) create mode 120000 CLAUDE.md create mode 100644 teslemetry_stream/energysite.py create mode 100644 tests/test_energysite_events.py diff --git a/AGENTS.md b/AGENTS.md index 4df7a88..59ec65f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -9,6 +9,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `Signal` in `const.py` tracks ; the config route rejects names it does not know with `fst_err_validation`. Fields the API has retired are not rejected - it accepts the request and names them in a top-level `ignoredFields` list - so the library can lag the published list without breaking. - 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. ## Maintaining this file diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..47dc3e3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/README.md b/README.md index 1f474f1..4da4bbf 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,42 @@ async def main(): await stream.close() ``` +## Energy Site Streaming + +Energy sites stream two events, `live_status` and `site_info`. Unlike vehicle +signals these are delivered as full documents rather than field deltas, and +there is nothing to enable - subscribed sites are polled automatically. On +connect you get an initial snapshot event for each, the same way vehicle +`State` is delivered on connect. + +```python +async def main(): + async with aiohttp.ClientSession() as session: + async with TeslemetryStream( + access_token="", + session=session, + ) as stream: + + site = stream.get_energysite("") + + def live_status_callback(live_status): + print(f"Battery Power: {live_status.get('battery_power')}") + + def site_info_callback(site_info): + print(f"Site Name: {site_info.get('site_name')}") + + remove_live_status_listener = site.listen_LiveStatus(live_status_callback) + remove_site_info_listener = site.listen_SiteInfo(site_info_callback) + + print("Running") + await asyncio.sleep(60) + remove_live_status_listener() + remove_site_info_listener() +``` + +> **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. + ## Public Methods in TeslemetryStream Class ### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False)` @@ -122,6 +158,9 @@ Initialize the TeslemetryStream client. ### `get_vehicle(vin: str) -> TeslemetryStreamVehicle` Create a vehicle object to manage config and create listeners. +### `get_energysite(site_id: str | int) -> TeslemetryStreamEnergySite` +Create an energy site object to create listeners for `live_status` and `site_info`. + ### `connected -> bool` Return if connected. @@ -197,3 +236,14 @@ Listen for vehicle error events. The callback receives a list of dictionaries co The `TeslemetryStreamVehicle` class contains a `listen_*` methods for each telemetry signal. These methods allow you to listen to specific signals and handle their data in a type-safe manner. A full list of fields and metadata can be found at [api.teslemetry.com/fields.json](https://api.teslemetry.com/fields.json) + +## Public Methods in TeslemetryStreamEnergySite Class + +### `__init__(stream: TeslemetryStream, site_id: str)` +Initialize the TeslemetryStreamEnergySite instance. + +### `listen_LiveStatus(callback: Callable[[dict], None]) -> Callable[[],None]` +Listen for energy site live status events. The callback receives the full `live_status` document. + +### `listen_SiteInfo(callback: Callable[[dict], None]) -> Callable[[],None]` +Listen for energy site info events. The callback receives the full `site_info` document. diff --git a/teslemetry_stream/__init__.py b/teslemetry_stream/__init__.py index bc72d04..cbf3dbd 100644 --- a/teslemetry_stream/__init__.py +++ b/teslemetry_stream/__init__.py @@ -1,5 +1,6 @@ from .stream import TeslemetryStream from .vehicle import TeslemetryStreamVehicle +from .energysite import TeslemetryStreamEnergySite from .exception import ( TeslemetryStreamError, TeslemetryStreamConnectionError, @@ -11,6 +12,7 @@ __all__ = [ "TeslemetryStream", "TeslemetryStreamVehicle", + "TeslemetryStreamEnergySite", "TeslemetryStreamError", "TeslemetryStreamConnectionError", "TeslemetryStreamVehicleNotConfigured", diff --git a/teslemetry_stream/const.py b/teslemetry_stream/const.py index 95d67ca..b22531d 100644 --- a/teslemetry_stream/const.py +++ b/teslemetry_stream/const.py @@ -24,6 +24,11 @@ class Key(StrEnum): STATE = "state" STATUS = "status" NETWORK_INTERFACE = "networkInterface" + SITE_ID = "site_id" + LIVE_STATUS = "live_status" + SITE_INFO = "site_info" + IS_CACHE = "isCache" + CREATED_AT = "createdAt" class Signal(StrEnum): diff --git a/teslemetry_stream/energysite.py b/teslemetry_stream/energysite.py new file mode 100644 index 0000000..18d851d --- /dev/null +++ b/teslemetry_stream/energysite.py @@ -0,0 +1,55 @@ +"""Energy site class for handling streaming live_status and site_info updates.""" + +from __future__ import annotations +from typing import TYPE_CHECKING, Any, Callable + +from .const import Key + +if TYPE_CHECKING: + from .stream import TeslemetryStream +else: + TeslemetryStream = None + + +class TeslemetryStreamEnergySite: + """Handle streaming energy site updates. + + Unlike vehicle signals, energy `live_status` and `site_info` events are + full documents rather than field deltas, and there is no per-field + config to enable - the server auto-polls subscribed sites - so listeners + here just filter and unwrap the matching topic. + """ + + def __init__(self, stream: TeslemetryStream, site_id: str): + self.stream = stream + self.site_id = str(site_id) + + def listen_LiveStatus( + self, callback: Callable[[dict[str, Any]], None] + ) -> Callable[[], None]: + """Listen for energy site live status. + + The callback receives the full live_status document. On connect (and + whenever a snapshot exists), an initial event is delivered with + `isCache` set, matching the same snapshot-then-live semantics as + vehicle state. + """ + return self.stream.async_add_listener( + lambda x: callback(x[Key.LIVE_STATUS]), + {Key.SITE_ID: self.site_id, Key.LIVE_STATUS: None}, + ) + + def listen_SiteInfo( + self, callback: Callable[[dict[str, Any]], None] + ) -> Callable[[], None]: + """Listen for energy site info. + + The callback receives the full site_info document. On connect (and + whenever a snapshot exists), an initial event is delivered with + `isCache` set, matching the same snapshot-then-live semantics as + vehicle state. + """ + return self.stream.async_add_listener( + lambda x: callback(x[Key.SITE_INFO]), + {Key.SITE_ID: self.site_id, Key.SITE_INFO: None}, + ) diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 4032125..c70ef35 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -9,6 +9,7 @@ from .exception import TeslemetryStreamEnded from .vehicle import TeslemetryStreamVehicle +from .energysite import TeslemetryStreamEnergySite LOGGER = logging.getLogger(__package__) @@ -53,6 +54,7 @@ def __init__( self.manual = manual self.retries: int = 0 self.vehicles: dict[str, TeslemetryStreamVehicle] = {} + self.energysites: dict[str, TeslemetryStreamEnergySite] = {} self.fields: dict[str, Any] = {} if self.vin: @@ -80,6 +82,18 @@ def get_vehicle(self, vin: str) -> TeslemetryStreamVehicle: self.vehicles[vin] = TeslemetryStreamVehicle(self, vin) return self.vehicles[vin] + def get_energysite(self, site_id: str | int) -> TeslemetryStreamEnergySite: + """ + Create an energy site stream. + + :param site_id: Numeric energy site ID. + :return: TeslemetryStreamEnergySite instance. + """ + site_id = str(site_id) + if site_id not in self.energysites: + self.energysites[site_id] = TeslemetryStreamEnergySite(self, site_id) + return self.energysites[site_id] + @property def connected(self) -> bool: """ diff --git a/tests/test_energysite_events.py b/tests/test_energysite_events.py new file mode 100644 index 0000000..f42ff5b --- /dev/null +++ b/tests/test_energysite_events.py @@ -0,0 +1,196 @@ +"""Checks energy site listener filtering against PR 310'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). +""" +from __future__ import annotations + +from typing import Any + +from teslemetry_stream.stream import TeslemetryStream, recursive_match + +SITE_A = "12345" +SITE_B = "67890" + +LIVE_STATUS_SNAPSHOT: dict[str, Any] = { + "createdAt": "2026-07-28T10:15:30.000Z", + "site_id": SITE_A, + "isCache": True, + "live_status": { + "battery_power": 1200, + "load_power": 900, + "grid_power": -300, + "solar_power": 2400, + "percentage_charged": 82.5, + }, +} + +LIVE_STATUS_LIVE: dict[str, Any] = { + "createdAt": "2026-07-28T10:16:00.000Z", + "site_id": SITE_A, + "live_status": { + "battery_power": 1100, + "load_power": 950, + "grid_power": -150, + "solar_power": 2200, + "percentage_charged": 82.6, + }, +} + +SITE_INFO_SNAPSHOT: dict[str, Any] = { + "createdAt": "2026-07-28T10:15:30.000Z", + "site_id": SITE_A, + "isCache": True, + "site_info": { + "site_name": "Home", + "backup_reserve_percent": 20, + "default_real_mode": "self_consumption", + }, +} + +OTHER_SITE_LIVE_STATUS: dict[str, Any] = { + "createdAt": "2026-07-28T10:16:00.000Z", + "site_id": SITE_B, + "live_status": {"battery_power": 0}, +} + +CREDITS_EVENT: dict[str, Any] = { + "credits": {"type": "snapshot", "cost": 0, "name": "snapshot", "balance": 5, "quota": {}}, + "createdAt": "2026-07-28T10:16:00.000Z", + "isCache": True, +} + + +def make_stream() -> TeslemetryStream: + """Build a stream that never actually connects.""" + return TeslemetryStream(session=None, access_token="test-token", manual=True) # type: ignore[arg-type] + + +def dispatch(stream: TeslemetryStream, event: dict[str, Any]) -> None: + """Replicate stream.listen()'s per-event dispatch without a live connection.""" + for listener, filters in list(stream._listeners.values()): + if recursive_match(filters, event): + listener(event) + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<64} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +def main() -> None: + results = [] + + # listen_LiveStatus receives the unwrapped live_status document, snapshot and live alike. + stream = make_stream() + site = stream.get_energysite(SITE_A) + received: list[dict[str, Any]] = [] + site.listen_LiveStatus(received.append) + dispatch(stream, LIVE_STATUS_SNAPSHOT) + dispatch(stream, LIVE_STATUS_LIVE) + results.append( + check( + "listen_LiveStatus unwraps the live_status document", + received == [LIVE_STATUS_SNAPSHOT["live_status"], LIVE_STATUS_LIVE["live_status"]], + f"got {received}", + ) + ) + + # listen_SiteInfo receives the unwrapped site_info document. + stream = make_stream() + site = stream.get_energysite(SITE_A) + received = [] + site.listen_SiteInfo(received.append) + dispatch(stream, SITE_INFO_SNAPSHOT) + results.append( + check( + "listen_SiteInfo unwraps the site_info document", + received == [SITE_INFO_SNAPSHOT["site_info"]], + f"got {received}", + ) + ) + + # A live_status event for another site is not delivered. + stream = make_stream() + site = stream.get_energysite(SITE_A) + received = [] + site.listen_LiveStatus(received.append) + dispatch(stream, OTHER_SITE_LIVE_STATUS) + results.append( + check( + "a different site's live_status is filtered out", + received == [], + f"got {received}", + ) + ) + + # A site_info listener does not receive live_status events for the same site. + stream = make_stream() + site = stream.get_energysite(SITE_A) + received = [] + site.listen_SiteInfo(received.append) + dispatch(stream, LIVE_STATUS_SNAPSHOT) + results.append( + check( + "listen_SiteInfo ignores live_status 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) + live_received: list[dict[str, Any]] = [] + info_received: list[dict[str, Any]] = [] + site.listen_LiveStatus(live_received.append) + site.listen_SiteInfo(info_received.append) + dispatch(stream, CREDITS_EVENT) + results.append( + check( + "credits events are not delivered to energy site listeners", + live_received == [] and info_received == [], + f"live={live_received} info={info_received}", + ) + ) + + # Removing a listener stops further delivery. + stream = make_stream() + site = stream.get_energysite(SITE_A) + received = [] + remove = site.listen_LiveStatus(received.append) + dispatch(stream, LIVE_STATUS_SNAPSHOT) + remove() + dispatch(stream, LIVE_STATUS_LIVE) + results.append( + check( + "removing a listener stops further delivery", + received == [LIVE_STATUS_SNAPSHOT["live_status"]], + f"got {received}", + ) + ) + + # get_energysite is idempotent per id, mirroring get_vehicle. + stream = make_stream() + results.append( + check( + "get_energysite returns the same instance for the same id", + stream.get_energysite(SITE_A) is stream.get_energysite(SITE_A), + ) + ) + results.append( + check( + "get_energysite normalizes int and str ids to the same instance", + stream.get_energysite(12345) is stream.get_energysite("12345"), + ) + ) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + main() From 2790c391c9c150a6a66f1e2429f8403a6ec00cb5 Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 10:23:04 +1000 Subject: [PATCH 2/2] docs(energy): fix example to match the library's real instantiation pattern TeslemetryStream has no __aenter__/__aexit__, so `async with TeslemetryStream(...)` raises a TypeError - switch to direct instantiation with an explicit close(), matching the working custom-listener example already in the README. --- README.md | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 4da4bbf..0ab5f07 100644 --- a/README.md +++ b/README.md @@ -125,26 +125,27 @@ connect you get an initial snapshot event for each, the same way vehicle ```python async def main(): async with aiohttp.ClientSession() as session: - async with TeslemetryStream( + stream = TeslemetryStream( access_token="", session=session, - ) as stream: + ) - site = stream.get_energysite("") + site = stream.get_energysite("") - def live_status_callback(live_status): - print(f"Battery Power: {live_status.get('battery_power')}") + def live_status_callback(live_status): + print(f"Battery Power: {live_status.get('battery_power')}") - def site_info_callback(site_info): - print(f"Site Name: {site_info.get('site_name')}") + def site_info_callback(site_info): + print(f"Site Name: {site_info.get('site_name')}") - remove_live_status_listener = site.listen_LiveStatus(live_status_callback) - remove_site_info_listener = site.listen_SiteInfo(site_info_callback) + remove_live_status_listener = site.listen_LiveStatus(live_status_callback) + remove_site_info_listener = site.listen_SiteInfo(site_info_callback) - print("Running") - await asyncio.sleep(60) - remove_live_status_listener() - remove_site_info_listener() + print("Running") + await asyncio.sleep(60) + remove_live_status_listener() + remove_site_info_listener() + stream.close() ``` > **Note:** Energy site streaming ships flag-gated behind [Teslemetry/api#310](https://github.com/Teslemetry/api/pull/310).