From 67afbbcb130f057686a908c1faba2659facb1930 Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 18:19:30 +1000 Subject: [PATCH 1/5] feat(sse): support topics allowlist and tariff_content_v2 topic Adds Pythonic client support for api PR 318 (tariff_content_v2 split out of site_info) and PR 319 (topics allowlist query param), fully additive - existing listener contracts are unchanged. - TeslemetryStream(topics=...) sends the exact comma-separated SSE topics allowlist; omitting it preserves legacy-all behavior forever. SseTopic plus SSE_VEHICLE_TOPICS/SSE_ENERGY_TOPICS/SSE_ALL_TOPICS presets expand client-side to exact wire names. - listen_TariffContentV2 exposes the V2 tariff document verbatim, with a null body surfaced as an explicit removal signal (None). - listen_SiteInfo keeps working against the now-slim site_info shape. - listen_ComposedSiteInfo merges the latest site_info with the last known tariff piece into a whole-document view. Co-Authored-By: Claude Sonnet 5 --- AGENTS.md | 2 + README.md | 68 ++++++++- teslemetry_stream/__init__.py | 15 +- teslemetry_stream/const.py | 71 +++++++++ teslemetry_stream/energysite.py | 64 +++++++- teslemetry_stream/stream.py | 10 +- tests/test_sse_topics.py | 263 ++++++++++++++++++++++++++++++++ 7 files changed, 483 insertions(+), 10 deletions(-) create mode 100644 tests/test_sse_topics.py diff --git a/AGENTS.md b/AGENTS.md index 95997be..dfd5245 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,6 +11,8 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `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. +- `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". `listen_ComposedSiteInfo` merges the latest `site_info` with the last known tariff piece for callers that want the whole Tesla-shaped document back together. Both halves share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. +- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets. Omitting `topics` is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, `listen_ComposedSiteInfo`, and the `topics` param's URL construction. ## Maintaining this file diff --git a/README.md b/README.md index 49dddc3..93be9b4 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,34 @@ 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. +`site_info` never carries `tariff_content`/`tariff_content_v2` - the site's V2 +tariff is its own `tariff_content_v2` event, published only when it changes. +Like `site_info`, event silence means no change, never staleness; freshness +always lives in a REST call, never in event cadence. A `None` body is an +explicit removal signal (the site's V2 tariff was cleared), not "no data yet". + +```python + def tariff_callback(tariff_content_v2): + if tariff_content_v2 is None: + print("V2 tariff removed") + else: + print(f"Tariff code: {tariff_content_v2.get('code')}") + + remove_tariff_listener = site.listen_TariffContentV2(tariff_callback) +``` + +If you want the whole Tesla-shaped document back together, `listen_ComposedSiteInfo` +merges the latest `site_info` with the last known `tariff_content_v2` piece so +you don't have to hand-assemble it from the two separate events: + +```python + def composed_callback(site_info): + print(f"Site Name: {site_info.get('site_name')}") + print(f"Tariff: {site_info.get('tariff_content_v2')}") + + remove_composed_listener = site.listen_ComposedSiteInfo(composed_callback) +``` + 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` @@ -164,10 +192,38 @@ change, never a stale value. remove_energy_totals_listener = site.listen_EnergyTotals(energy_totals_callback) ``` +## SSE Topic Selection + +By default a connection receives every applicable event (legacy-all +behavior, unchanged forever). Pass `topics` to `TeslemetryStream` to +subscribe to only the SSE wire events you need - an exact allowlist, +comma-joined onto the connection's `topics` query parameter: + +```python +from teslemetry_stream import TeslemetryStream, SseTopic, SSE_ENERGY_TOPICS + +stream = TeslemetryStream( + access_token="", + session=session, + topics=[SseTopic.LIVE_STATUS, SseTopic.SITE_INFO], +) + +# Or use a preset that expands client-side to every topic in a group: +stream = TeslemetryStream( + access_token="", + session=session, + topics=SSE_ENERGY_TOPICS, +) +``` + +`SseTopic` is the closed set of exact wire names the server recognizes; +`SSE_VEHICLE_TOPICS`, `SSE_ENERGY_TOPICS`, and `SSE_ALL_TOPICS` are +convenience presets that expand to those exact names client-side. + ## Public Methods in TeslemetryStream Class -### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False)` -Initialize the TeslemetryStream client. +### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, topics: Iterable[str] | None = None)` +Initialize the TeslemetryStream client. `topics` is an optional exact SSE wire event allowlist (see `SseTopic`); omitting it preserves legacy-all behavior. ### `get_vehicle(vin: str) -> TeslemetryStreamVehicle` Create a vehicle object to manage config and create listeners. @@ -260,7 +316,13 @@ Initialize the TeslemetryStreamEnergySite instance. 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. +Listen for energy site info events. The callback receives the `site_info` document. This document never carries `tariff_content`/`tariff_content_v2` - use `listen_TariffContentV2` for the V2 tariff. + +### `listen_TariffContentV2(callback: Callable[[dict | None], None]) -> Callable[[],None]` +Listen for the site's V2 tariff document. The callback receives the `tariff_content_v2` document verbatim, or `None` when the server sends an explicit removal signal. Published only when it changes. + +### `listen_ComposedSiteInfo(callback: Callable[[dict], None]) -> Callable[[],None]` +Listen for a whole-document view merging the latest `site_info` with the last known `tariff_content_v2` piece under a `tariff_content_v2` key, so consumers don't have to hand-assemble it from the two separate events. ### `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/__init__.py b/teslemetry_stream/__init__.py index cbf3dbd..462b671 100644 --- a/teslemetry_stream/__init__.py +++ b/teslemetry_stream/__init__.py @@ -7,7 +7,14 @@ TeslemetryStreamVehicleNotConfigured, TeslemetryStreamEnded ) -from .const import Signal, Alert +from .const import ( + Signal, + Alert, + SseTopic, + SSE_VEHICLE_TOPICS, + SSE_ENERGY_TOPICS, + SSE_ALL_TOPICS, +) __all__ = [ "TeslemetryStream", @@ -18,5 +25,9 @@ "TeslemetryStreamVehicleNotConfigured", "TeslemetryStreamEnded", "Signal", - "Alert" + "Alert", + "SseTopic", + "SSE_VEHICLE_TOPICS", + "SSE_ENERGY_TOPICS", + "SSE_ALL_TOPICS", ] diff --git a/teslemetry_stream/const.py b/teslemetry_stream/const.py index 59c873e..1e86a0c 100644 --- a/teslemetry_stream/const.py +++ b/teslemetry_stream/const.py @@ -34,6 +34,7 @@ class Key(StrEnum): TOPIC = "topic" URL = "url" TOTALS = "totals" + TARIFF_CONTENT_V2 = "tariff_content_v2" class Signal(StrEnum): @@ -328,6 +329,76 @@ class RefreshTopic(StrEnum): ENERGY_TOTALS = "energy_totals" +class SseTopic(StrEnum): + """Exact SSE wire event names selectable via `TeslemetryStream(topics=...)`. + + Mirrors the api's closed `SSE_TOPICS` set (`src/lib/sseTopics.ts`) - a + name here must match the server's allowlist exactly, since the server + validates `topics` and 400s on anything it does not recognize. + """ + + STATE = "state" + DATA = "data" + ALERTS = "alerts" + ERRORS = "errors" + CONNECTIVITY = "connectivity" + VEHICLE_DATA = "vehicle_data" + CONFIG = "config" + LIVE_STATUS = "live_status" + SITE_INFO = "site_info" + TARIFF_CONTENT_V2 = "tariff_content_v2" + ENERGY_TOTALS = "energy_totals" + CREDITS = "credits" + + +# Vehicle topics with a connect-time cache snapshot. +SSE_VEHICLE_SNAPSHOT_TOPICS: tuple[SseTopic, ...] = ( + SseTopic.STATE, + SseTopic.DATA, + SseTopic.ALERTS, + SseTopic.ERRORS, + SseTopic.CONNECTIVITY, + SseTopic.VEHICLE_DATA, +) + +# Vehicle topics that are live-only - never part of a connect-time snapshot. +SSE_VEHICLE_LIVE_ONLY_TOPICS: tuple[SseTopic, ...] = (SseTopic.CONFIG,) + +# Energy site topics with a connect-time cache snapshot. +SSE_ENERGY_SNAPSHOT_TOPICS: tuple[SseTopic, ...] = ( + SseTopic.LIVE_STATUS, + SseTopic.SITE_INFO, + SseTopic.TARIFF_CONTENT_V2, +) + +# Energy site topics that are live-only - never part of a connect-time snapshot. +SSE_ENERGY_LIVE_ONLY_TOPICS: tuple[SseTopic, ...] = (SseTopic.ENERGY_TOTALS,) + +# Account-wide topics with a connect-time cache snapshot. +SSE_ACCOUNT_SNAPSHOT_TOPICS: tuple[SseTopic, ...] = (SseTopic.CREDITS,) + +#: Convenience preset - every vehicle topic. Expands client-side to exact +#: wire names; passing this to `TeslemetryStream(topics=...)` is equivalent +#: to legacy-all for a vehicle connection, minus energy/account topics. +SSE_VEHICLE_TOPICS: tuple[SseTopic, ...] = ( + *SSE_VEHICLE_SNAPSHOT_TOPICS, + *SSE_VEHICLE_LIVE_ONLY_TOPICS, +) + +#: Convenience preset - every energy site topic. +SSE_ENERGY_TOPICS: tuple[SseTopic, ...] = ( + *SSE_ENERGY_SNAPSHOT_TOPICS, + *SSE_ENERGY_LIVE_ONLY_TOPICS, +) + +#: Convenience preset - every known topic, equivalent to omitting `topics`. +SSE_ALL_TOPICS: tuple[SseTopic, ...] = ( + *SSE_VEHICLE_TOPICS, + *SSE_ENERGY_TOPICS, + *SSE_ACCOUNT_SNAPSHOT_TOPICS, +) + + @dataclass class EnergyHistoryTotals: """Cumulative per-type totals from a refreshed energy_totals document. diff --git a/teslemetry_stream/energysite.py b/teslemetry_stream/energysite.py index c65e223..5968929 100644 --- a/teslemetry_stream/energysite.py +++ b/teslemetry_stream/energysite.py @@ -44,16 +44,72 @@ def listen_SiteInfo( ) -> 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. + The callback receives the site_info document. This document no + longer carries `tariff_content`/`tariff_content_v2` - subscribe to + `listen_TariffContentV2` for the V2 tariff, or use the REST + site_info endpoint for the full Tesla-shaped document including + both tariffs. 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}, ) + def listen_TariffContentV2( + self, callback: Callable[[dict[str, Any] | None], None] + ) -> Callable[[], None]: + """Listen for the site's V2 tariff document. + + The callback receives the `tariff_content_v2` document verbatim, or + `None` when the server sends an explicit removal signal (the + site's V2 tariff was cleared). Published only when it changes - + silence means no change, never staleness, matching `listen_SiteInfo`. + """ + return self.stream.async_add_listener( + lambda x: callback(x[Key.TARIFF_CONTENT_V2]), + {Key.SITE_ID: self.site_id, Key.TARIFF_CONTENT_V2: None}, + ) + + def listen_ComposedSiteInfo( + self, callback: Callable[[dict[str, Any]], None] + ) -> Callable[[], None]: + """Listen for a whole-document view combining site_info and tariff. + + Merges the latest slim `site_info` with the last known + `tariff_content_v2` piece under a `tariff_content_v2` key, so + consumers get the same shape the REST site_info endpoint returns + without hand-assembling it from two separate listeners. Fires + whenever either half updates; nothing is emitted until the first + `site_info` document has arrived. `tariff_content_v2` is `None` + until a value has been received, and again after an explicit + removal. + """ + state: dict[str, Any] = {"site_info": None, "tariff_content_v2": None} + + def emit() -> None: + if state["site_info"] is None: + return + callback({**state["site_info"], Key.TARIFF_CONTENT_V2: state["tariff_content_v2"]}) + + def on_site_info(site_info: dict[str, Any]) -> None: + state["site_info"] = site_info + emit() + + def on_tariff(tariff_content_v2: dict[str, Any] | None) -> None: + state["tariff_content_v2"] = tariff_content_v2 + emit() + + remove_site_info = self.listen_SiteInfo(on_site_info) + remove_tariff = self.listen_TariffContentV2(on_tariff) + + def remove_listener() -> None: + remove_site_info() + remove_tariff() + + return remove_listener + def listen_EnergyTotals( self, callback: Callable[[EnergyHistoryTotals], None] ) -> Callable[[], None]: diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index c70ef35..c367dd0 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -3,7 +3,7 @@ import json import logging from datetime import datetime, timezone -from typing import Any, Awaitable, Callable, cast +from typing import Any, Awaitable, Callable, Iterable, cast import aiohttp @@ -27,6 +27,7 @@ def __init__( vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, + topics: Iterable[str] | None = None, ): """ Initialize the TeslemetryStream client. @@ -37,6 +38,10 @@ def __init__( :param vin: Vehicle Identification Number. :param parse_timestamp: Whether to parse timestamps. :param manual: Whether to start listening manually. + :param topics: Exact SSE wire event names (see `SseTopic` and its + presets in `const.py`) to subscribe to. Omitting this preserves + legacy-all behavior: every applicable event is delivered + unfiltered, forever. """ if server and not server.endswith(".teslemetry.com"): raise ValueError("Server must be on the teslemetry.com domain") @@ -44,6 +49,7 @@ def __init__( self.active: bool = False self.server = server self.vin = vin + self.topics = list(topics) if topics is not None else None self._listeners: dict[ Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None] ] = {} @@ -214,9 +220,11 @@ async def connect(self) -> None: if self.vin: url += f"/{self.vin}" headers = await self.headers() + params = {"topics": ",".join(self.topics)} if self.topics else None self._response = await self._session.get( url, headers=headers, + params=params, raise_for_status=True, timeout=aiohttp.ClientTimeout( connect=5, sock_connect=5, sock_read=30, total=None diff --git a/tests/test_sse_topics.py b/tests/test_sse_topics.py new file mode 100644 index 0000000..20a3550 --- /dev/null +++ b/tests/test_sse_topics.py @@ -0,0 +1,263 @@ +"""Checks the api PR 318/319 SSE protocol additions: the `topics` allowlist +query parameter, the `tariff_content_v2` topic (including null-as-removal), +and the composed site_info+tariff helper. + +Fixtures mirror the authoritative schemas in Teslemetry/api: +- `src/lib/sseTopics.ts` for the exact wire topic names and `topics` being + an additive, omission-preserves-legacy-behavior query parameter. +- `src/routes/sse/tariffContentV2Schema.ts` for the tariff_content_v2 event + shape (`createdAt`/`site_id`/`isCache?`/`tariff_content_v2`, the last + nullable to signal removal). +- `src/routes/sse/siteInfoSchema.ts` for the now-slim `site_info` event + (never carries `tariff_content`/`tariff_content_v2`). +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from teslemetry_stream.const import SseTopic +from teslemetry_stream.stream import TeslemetryStream, recursive_match + +SITE_A = "12345" + +SITE_INFO_SNAPSHOT: dict[str, Any] = { + "createdAt": "2026-07-29T10:15:30.000Z", + "site_id": SITE_A, + "isCache": True, + "site_info": { + "site_name": "Home", + "backup_reserve_percent": 20, + "default_real_mode": "self_consumption", + }, +} + +SITE_INFO_UPDATE: dict[str, Any] = { + "createdAt": "2026-07-29T10:16:00.000Z", + "site_id": SITE_A, + "site_info": { + "site_name": "Home", + "backup_reserve_percent": 25, + "default_real_mode": "self_consumption", + }, +} + +TARIFF_SNAPSHOT: dict[str, Any] = { + "createdAt": "2026-07-29T10:15:30.000Z", + "site_id": SITE_A, + "isCache": True, + "tariff_content_v2": {"code": "PLAN-1", "utility": "Acme Power"}, +} + +TARIFF_REMOVED: dict[str, Any] = { + "createdAt": "2026-07-29T10:17:00.000Z", + "site_id": SITE_A, + "tariff_content_v2": None, +} + + +class FakeResponse: + """Minimal stand-in for the aiohttp response `connect()` awaits.""" + + def __init__(self) -> None: + self.url = "https://fake.teslemetry.com/sse" + self.status = 200 + + +class FakeSession: + """Captures the kwargs `connect()` passes to `session.get`.""" + + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.calls.append({"url": url, **kwargs}) + return FakeResponse() + + +def make_stream(topics: Any = None) -> tuple[TeslemetryStream, FakeSession]: + session = FakeSession() + stream = TeslemetryStream( + session=session, # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + manual=True, + topics=topics, + ) + return stream, session + + +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 = [] + + # Omitting topics preserves legacy behavior: no topics param sent at all. + stream, session = make_stream(topics=None) + asyncio.run(stream.connect()) + results.append( + check( + "omitted topics sends no topics query param", + session.calls[0]["params"] is None, + f"got {session.calls[0]['params']}", + ) + ) + + # Passing topics builds the exact comma-separated allowlist param. + stream, session = make_stream(topics=[SseTopic.LIVE_STATUS, SseTopic.SITE_INFO]) + asyncio.run(stream.connect()) + results.append( + check( + "topics builds a comma-separated params entry", + session.calls[0]["params"] == {"topics": "live_status,site_info"}, + f"got {session.calls[0]['params']}", + ) + ) + + # Plain strings work too, not just SseTopic members. + stream, session = make_stream(topics=["state", "config"]) + asyncio.run(stream.connect()) + results.append( + check( + "plain string topics are accepted", + session.calls[0]["params"] == {"topics": "state,config"}, + f"got {session.calls[0]['params']}", + ) + ) + + # listen_TariffContentV2 receives the tariff document verbatim. + stream, _ = make_stream() + site = stream.get_energysite(SITE_A) + received: list[dict[str, Any] | None] = [] + site.listen_TariffContentV2(received.append) + dispatch(stream, TARIFF_SNAPSHOT) + results.append( + check( + "listen_TariffContentV2 receives the tariff document", + received == [TARIFF_SNAPSHOT["tariff_content_v2"]], + f"got {received}", + ) + ) + + # A null tariff_content_v2 body is delivered as None - the removal signal. + stream, _ = make_stream() + site = stream.get_energysite(SITE_A) + received = [] + site.listen_TariffContentV2(received.append) + dispatch(stream, TARIFF_SNAPSHOT) + dispatch(stream, TARIFF_REMOVED) + results.append( + check( + "a null tariff_content_v2 body surfaces as None (removal)", + received == [TARIFF_SNAPSHOT["tariff_content_v2"], None], + f"got {received}", + ) + ) + + # listen_SiteInfo keeps working against the slim shape and ignores tariff events. + stream, _ = make_stream() + site = stream.get_energysite(SITE_A) + received = [] + site.listen_SiteInfo(received.append) + dispatch(stream, SITE_INFO_SNAPSHOT) + dispatch(stream, TARIFF_SNAPSHOT) + results.append( + check( + "listen_SiteInfo ignores tariff_content_v2 events", + received == [SITE_INFO_SNAPSHOT["site_info"]], + f"got {received}", + ) + ) + + # listen_ComposedSiteInfo merges the latest site_info with the last tariff piece. + stream, _ = make_stream() + site = stream.get_energysite(SITE_A) + composed: list[dict[str, Any]] = [] + site.listen_ComposedSiteInfo(composed.append) + dispatch(stream, SITE_INFO_SNAPSHOT) + results.append( + check( + "composed view emits site_info alone with tariff_content_v2=None first", + composed == [{**SITE_INFO_SNAPSHOT["site_info"], "tariff_content_v2": None}], + f"got {composed}", + ) + ) + + dispatch(stream, TARIFF_SNAPSHOT) + results.append( + check( + "composed view merges in the tariff once it arrives", + composed[-1] + == {**SITE_INFO_SNAPSHOT["site_info"], "tariff_content_v2": TARIFF_SNAPSHOT["tariff_content_v2"]}, + f"got {composed[-1]}", + ) + ) + + dispatch(stream, SITE_INFO_UPDATE) + results.append( + check( + "composed view keeps the last tariff when only site_info updates", + composed[-1] + == {**SITE_INFO_UPDATE["site_info"], "tariff_content_v2": TARIFF_SNAPSHOT["tariff_content_v2"]}, + f"got {composed[-1]}", + ) + ) + + dispatch(stream, TARIFF_REMOVED) + results.append( + check( + "composed view reflects an explicit tariff removal as None", + composed[-1] == {**SITE_INFO_UPDATE["site_info"], "tariff_content_v2": None}, + f"got {composed[-1]}", + ) + ) + + # Nothing is emitted before the first site_info document arrives. + stream, _ = make_stream() + site = stream.get_energysite(SITE_A) + composed = [] + site.listen_ComposedSiteInfo(composed.append) + dispatch(stream, TARIFF_SNAPSHOT) + results.append( + check( + "composed view withholds emission until site_info has arrived", + composed == [], + f"got {composed}", + ) + ) + + # Removing the composed listener removes both underlying listeners. + stream, _ = make_stream() + site = stream.get_energysite(SITE_A) + composed = [] + remove = site.listen_ComposedSiteInfo(composed.append) + dispatch(stream, SITE_INFO_SNAPSHOT) + remove() + dispatch(stream, TARIFF_SNAPSHOT) + dispatch(stream, SITE_INFO_UPDATE) + results.append( + check( + "removing the composed listener stops delivery from either half", + len(composed) == 1, + f"got {composed}", + ) + ) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + main() From f69819440e479b001e97fcbeb90529795d1cbdcb Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 18:34:15 +1000 Subject: [PATCH 2/5] fix(sse): reject an explicitly empty topics iterable Per PR review: a truthiness check on self.topics treated an empty iterable the same as None, silently falling back to legacy-all instead of subscribing to nothing. Distinguish the two at construction time and raise ValueError on empty, mirroring the server's own 400. Co-Authored-By: Claude Sonnet 5 --- AGENTS.md | 2 +- teslemetry_stream/stream.py | 18 ++++++++++++++---- tests/test_sse_topics.py | 13 +++++++++++++ 3 files changed, 28 insertions(+), 5 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index dfd5245..9bbb03b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,7 +12,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - 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. - `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". `listen_ComposedSiteInfo` merges the latest `site_info` with the last known tariff piece for callers that want the whole Tesla-shaped document back together. Both halves share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. -- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets. Omitting `topics` is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, `listen_ComposedSiteInfo`, and the `topics` param's URL construction. +- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, `listen_ComposedSiteInfo`, the `topics` param's URL construction, and the empty-iterable rejection. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index c367dd0..f6f047e 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -39,9 +39,11 @@ def __init__( :param parse_timestamp: Whether to parse timestamps. :param manual: Whether to start listening manually. :param topics: Exact SSE wire event names (see `SseTopic` and its - presets in `const.py`) to subscribe to. Omitting this preserves - legacy-all behavior: every applicable event is delivered - unfiltered, forever. + presets in `const.py`) to subscribe to. Omitting this (`None`) + preserves legacy-all behavior: every applicable event is + delivered unfiltered, forever. An explicitly empty iterable is + rejected - it means "no topics", not "all topics", mirroring + the server's own 400 on an empty `topics` value. """ if server and not server.endswith(".teslemetry.com"): raise ValueError("Server must be on the teslemetry.com domain") @@ -49,7 +51,15 @@ def __init__( self.active: bool = False self.server = server self.vin = vin - self.topics = list(topics) if topics is not None else None + self.topics: list[str] | None + if topics is not None: + self.topics = list(topics) + if not self.topics: + raise ValueError( + "topics must not be empty - omit it (None) for legacy-all behavior" + ) + else: + self.topics = None self._listeners: dict[ Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None] ] = {} diff --git a/tests/test_sse_topics.py b/tests/test_sse_topics.py index 20a3550..9bcd140 100644 --- a/tests/test_sse_topics.py +++ b/tests/test_sse_topics.py @@ -135,6 +135,19 @@ def main() -> None: ) ) + # An explicitly empty topics iterable is rejected, not treated as legacy-all. + raised = False + try: + make_stream(topics=[]) + except ValueError: + raised = True + results.append( + check( + "an empty topics iterable raises ValueError instead of silently enabling legacy-all", + raised, + ) + ) + # listen_TariffContentV2 receives the tariff document verbatim. stream, _ = make_stream() site = stream.get_energysite(SITE_A) From f656202c5fac20ce40a1583d6f7fc18d2f1adfee Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 18:44:31 +1000 Subject: [PATCH 3/5] docs(sse): scope listen_ComposedSiteInfo honestly to streamed data only Per PR review: the docstring/README claimed the composed view produces "the whole REST-shaped document" / "whole Tesla-shaped document," but the stream never carries the legacy V1 tariff_content - it has no SSE topic and stays REST-only by design. Reword to describe the helper as composing only what the stream itself carries (slim site_info + V2 tariff), and point consumers needing V1 at the REST endpoint. No behavior change. Co-Authored-By: Claude Sonnet 5 --- AGENTS.md | 2 +- README.md | 9 ++++++--- teslemetry_stream/energysite.py | 17 ++++++++++------- 3 files changed, 17 insertions(+), 11 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 9bbb03b..7307b54 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,7 +11,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `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. -- `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". `listen_ComposedSiteInfo` merges the latest `site_info` with the last known tariff piece for callers that want the whole Tesla-shaped document back together. Both halves share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. +- `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". `listen_ComposedSiteInfo` merges the latest `site_info` with the last known tariff piece for callers that want the two streamed halves back together - it only ever carries what the stream carries, never the legacy V1 `tariff_content` (no SSE topic, REST-only by design), so it is not a substitute for the REST site_info endpoint. Both halves share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. - `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, `listen_ComposedSiteInfo`, the `topics` param's URL construction, and the empty-iterable rejection. ## Maintaining this file diff --git a/README.md b/README.md index 93be9b4..ab1514e 100644 --- a/README.md +++ b/README.md @@ -167,9 +167,12 @@ explicit removal signal (the site's V2 tariff was cleared), not "no data yet". remove_tariff_listener = site.listen_TariffContentV2(tariff_callback) ``` -If you want the whole Tesla-shaped document back together, `listen_ComposedSiteInfo` +If you want the two streamed halves back together, `listen_ComposedSiteInfo` merges the latest `site_info` with the last known `tariff_content_v2` piece so -you don't have to hand-assemble it from the two separate events: +you don't have to hand-assemble it from the two separate events. This only +ever carries what the stream carries - it never restores the legacy V1 +`tariff_content`, which has no SSE topic and stays REST-only by design; fetch +the REST site_info endpoint directly if you need V1: ```python def composed_callback(site_info): @@ -322,7 +325,7 @@ Listen for energy site info events. The callback receives the `site_info` docume Listen for the site's V2 tariff document. The callback receives the `tariff_content_v2` document verbatim, or `None` when the server sends an explicit removal signal. Published only when it changes. ### `listen_ComposedSiteInfo(callback: Callable[[dict], None]) -> Callable[[],None]` -Listen for a whole-document view merging the latest `site_info` with the last known `tariff_content_v2` piece under a `tariff_content_v2` key, so consumers don't have to hand-assemble it from the two separate events. +Listen for a view merging the latest `site_info` with the last known `tariff_content_v2` piece under a `tariff_content_v2` key, so consumers don't have to hand-assemble it from the two separate events. Carries only what the stream carries - it never restores the legacy V1 `tariff_content`, which stays REST-only; use the REST site_info endpoint for that. ### `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/energysite.py b/teslemetry_stream/energysite.py index 5968929..c106c23 100644 --- a/teslemetry_stream/energysite.py +++ b/teslemetry_stream/energysite.py @@ -75,16 +75,19 @@ def listen_TariffContentV2( def listen_ComposedSiteInfo( self, callback: Callable[[dict[str, Any]], None] ) -> Callable[[], None]: - """Listen for a whole-document view combining site_info and tariff. + """Listen for a view composing the two streamed site_info pieces. Merges the latest slim `site_info` with the last known `tariff_content_v2` piece under a `tariff_content_v2` key, so - consumers get the same shape the REST site_info endpoint returns - without hand-assembling it from two separate listeners. Fires - whenever either half updates; nothing is emitted until the first - `site_info` document has arrived. `tariff_content_v2` is `None` - until a value has been received, and again after an explicit - removal. + consumers don't have to hand-assemble the two separate listeners + themselves. This only ever carries what the stream itself carries - + slim `site_info` plus the V2 tariff - never the legacy V1 + `tariff_content`, which has no SSE topic and stays REST-only by + design; a consumer needing V1 must fetch the REST site_info + endpoint directly. Fires whenever either half updates; nothing is + emitted until the first `site_info` document has arrived. + `tariff_content_v2` is `None` until a value has been received, and + again after an explicit removal. """ state: dict[str, Any] = {"site_info": None, "tariff_content_v2": None} From da1130e6c0d5f583e78bd75f1388263be329c6d3 Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 18:51:06 +1000 Subject: [PATCH 4/5] refactor(sse): drop listen_ComposedSiteInfo, stop encoding snapshot/live-only in topic presets Per captain review: - Drop listen_ComposedSiteInfo entirely. It could never actually produce the whole REST-shaped site_info document (legacy V1 tariff_content has no SSE topic and stays REST-only by design), so the promise it made was structurally unkeepable - not just a docs wording problem. A consumer wanting both tariffs together should use the REST site_info endpoint. - Collapse SSE_VEHICLE_SNAPSHOT_TOPICS/SSE_VEHICLE_LIVE_ONLY_TOPICS and their energy counterparts into flat SSE_VEHICLE_TOPICS/ SSE_ENERGY_TOPICS/SSE_ALL_TOPICS presets. Whether a topic happens to have a connect-time snapshot is server-side behavior the library has no reason to hardcode - the presets only need to know which topics apply to which product kind to expand client-side into exact names. Co-Authored-By: Claude Sonnet 5 --- AGENTS.md | 4 +- README.md | 18 ------- teslemetry_stream/const.py | 38 +++++--------- teslemetry_stream/energysite.py | 41 --------------- tests/test_sse_topics.py | 88 +-------------------------------- 5 files changed, 15 insertions(+), 174 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7307b54..50d139f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,8 +11,8 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `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. -- `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". `listen_ComposedSiteInfo` merges the latest `site_info` with the last known tariff piece for callers that want the two streamed halves back together - it only ever carries what the stream carries, never the legacy V1 `tariff_content` (no SSE topic, REST-only by design), so it is not a substitute for the REST site_info endpoint. Both halves share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. -- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, `listen_ComposedSiteInfo`, the `topics` param's URL construction, and the empty-iterable rejection. +- `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design), so it can't actually promise the whole REST-shaped document; a consumer wanting both tariffs together should use the REST site_info endpoint. +- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, and the empty-iterable rejection. ## Maintaining this file diff --git a/README.md b/README.md index ab1514e..da72f8f 100644 --- a/README.md +++ b/README.md @@ -167,21 +167,6 @@ explicit removal signal (the site's V2 tariff was cleared), not "no data yet". remove_tariff_listener = site.listen_TariffContentV2(tariff_callback) ``` -If you want the two streamed halves back together, `listen_ComposedSiteInfo` -merges the latest `site_info` with the last known `tariff_content_v2` piece so -you don't have to hand-assemble it from the two separate events. This only -ever carries what the stream carries - it never restores the legacy V1 -`tariff_content`, which has no SSE topic and stays REST-only by design; fetch -the REST site_info endpoint directly if you need V1: - -```python - def composed_callback(site_info): - print(f"Site Name: {site_info.get('site_name')}") - print(f"Tariff: {site_info.get('tariff_content_v2')}") - - remove_composed_listener = site.listen_ComposedSiteInfo(composed_callback) -``` - 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` @@ -324,8 +309,5 @@ Listen for energy site info events. The callback receives the `site_info` docume ### `listen_TariffContentV2(callback: Callable[[dict | None], None]) -> Callable[[],None]` Listen for the site's V2 tariff document. The callback receives the `tariff_content_v2` document verbatim, or `None` when the server sends an explicit removal signal. Published only when it changes. -### `listen_ComposedSiteInfo(callback: Callable[[dict], None]) -> Callable[[],None]` -Listen for a view merging the latest `site_info` with the last known `tariff_content_v2` piece under a `tariff_content_v2` key, so consumers don't have to hand-assemble it from the two separate events. Carries only what the stream carries - it never restores the legacy V1 `tariff_content`, which stays REST-only; use the REST site_info endpoint for that. - ### `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 1e86a0c..15f0653 100644 --- a/teslemetry_stream/const.py +++ b/teslemetry_stream/const.py @@ -351,51 +351,35 @@ class SseTopic(StrEnum): CREDITS = "credits" -# Vehicle topics with a connect-time cache snapshot. -SSE_VEHICLE_SNAPSHOT_TOPICS: tuple[SseTopic, ...] = ( +#: Convenience preset - every vehicle topic. Expands client-side to exact +#: wire names; passing this to `TeslemetryStream(topics=...)` is equivalent +#: to legacy-all for a vehicle connection, minus energy/account topics. +SSE_VEHICLE_TOPICS: tuple[SseTopic, ...] = ( SseTopic.STATE, SseTopic.DATA, SseTopic.ALERTS, SseTopic.ERRORS, SseTopic.CONNECTIVITY, SseTopic.VEHICLE_DATA, + SseTopic.CONFIG, ) -# Vehicle topics that are live-only - never part of a connect-time snapshot. -SSE_VEHICLE_LIVE_ONLY_TOPICS: tuple[SseTopic, ...] = (SseTopic.CONFIG,) - -# Energy site topics with a connect-time cache snapshot. -SSE_ENERGY_SNAPSHOT_TOPICS: tuple[SseTopic, ...] = ( +#: Convenience preset - every energy site topic. +SSE_ENERGY_TOPICS: tuple[SseTopic, ...] = ( SseTopic.LIVE_STATUS, SseTopic.SITE_INFO, SseTopic.TARIFF_CONTENT_V2, + SseTopic.ENERGY_TOTALS, ) -# Energy site topics that are live-only - never part of a connect-time snapshot. -SSE_ENERGY_LIVE_ONLY_TOPICS: tuple[SseTopic, ...] = (SseTopic.ENERGY_TOTALS,) - -# Account-wide topics with a connect-time cache snapshot. -SSE_ACCOUNT_SNAPSHOT_TOPICS: tuple[SseTopic, ...] = (SseTopic.CREDITS,) - -#: Convenience preset - every vehicle topic. Expands client-side to exact -#: wire names; passing this to `TeslemetryStream(topics=...)` is equivalent -#: to legacy-all for a vehicle connection, minus energy/account topics. -SSE_VEHICLE_TOPICS: tuple[SseTopic, ...] = ( - *SSE_VEHICLE_SNAPSHOT_TOPICS, - *SSE_VEHICLE_LIVE_ONLY_TOPICS, -) - -#: Convenience preset - every energy site topic. -SSE_ENERGY_TOPICS: tuple[SseTopic, ...] = ( - *SSE_ENERGY_SNAPSHOT_TOPICS, - *SSE_ENERGY_LIVE_ONLY_TOPICS, -) +#: Convenience preset - every account-wide topic. +SSE_ACCOUNT_TOPICS: tuple[SseTopic, ...] = (SseTopic.CREDITS,) #: Convenience preset - every known topic, equivalent to omitting `topics`. SSE_ALL_TOPICS: tuple[SseTopic, ...] = ( *SSE_VEHICLE_TOPICS, *SSE_ENERGY_TOPICS, - *SSE_ACCOUNT_SNAPSHOT_TOPICS, + *SSE_ACCOUNT_TOPICS, ) diff --git a/teslemetry_stream/energysite.py b/teslemetry_stream/energysite.py index c106c23..49adf6d 100644 --- a/teslemetry_stream/energysite.py +++ b/teslemetry_stream/energysite.py @@ -72,47 +72,6 @@ def listen_TariffContentV2( {Key.SITE_ID: self.site_id, Key.TARIFF_CONTENT_V2: None}, ) - def listen_ComposedSiteInfo( - self, callback: Callable[[dict[str, Any]], None] - ) -> Callable[[], None]: - """Listen for a view composing the two streamed site_info pieces. - - Merges the latest slim `site_info` with the last known - `tariff_content_v2` piece under a `tariff_content_v2` key, so - consumers don't have to hand-assemble the two separate listeners - themselves. This only ever carries what the stream itself carries - - slim `site_info` plus the V2 tariff - never the legacy V1 - `tariff_content`, which has no SSE topic and stays REST-only by - design; a consumer needing V1 must fetch the REST site_info - endpoint directly. Fires whenever either half updates; nothing is - emitted until the first `site_info` document has arrived. - `tariff_content_v2` is `None` until a value has been received, and - again after an explicit removal. - """ - state: dict[str, Any] = {"site_info": None, "tariff_content_v2": None} - - def emit() -> None: - if state["site_info"] is None: - return - callback({**state["site_info"], Key.TARIFF_CONTENT_V2: state["tariff_content_v2"]}) - - def on_site_info(site_info: dict[str, Any]) -> None: - state["site_info"] = site_info - emit() - - def on_tariff(tariff_content_v2: dict[str, Any] | None) -> None: - state["tariff_content_v2"] = tariff_content_v2 - emit() - - remove_site_info = self.listen_SiteInfo(on_site_info) - remove_tariff = self.listen_TariffContentV2(on_tariff) - - def remove_listener() -> None: - remove_site_info() - remove_tariff() - - return remove_listener - def listen_EnergyTotals( self, callback: Callable[[EnergyHistoryTotals], None] ) -> Callable[[], None]: diff --git a/tests/test_sse_topics.py b/tests/test_sse_topics.py index 9bcd140..c8aa3be 100644 --- a/tests/test_sse_topics.py +++ b/tests/test_sse_topics.py @@ -1,6 +1,6 @@ """Checks the api PR 318/319 SSE protocol additions: the `topics` allowlist -query parameter, the `tariff_content_v2` topic (including null-as-removal), -and the composed site_info+tariff helper. +query parameter and the `tariff_content_v2` topic (including +null-as-removal). Fixtures mirror the authoritative schemas in Teslemetry/api: - `src/lib/sseTopics.ts` for the exact wire topic names and `topics` being @@ -32,16 +32,6 @@ }, } -SITE_INFO_UPDATE: dict[str, Any] = { - "createdAt": "2026-07-29T10:16:00.000Z", - "site_id": SITE_A, - "site_info": { - "site_name": "Home", - "backup_reserve_percent": 25, - "default_real_mode": "self_consumption", - }, -} - TARIFF_SNAPSHOT: dict[str, Any] = { "createdAt": "2026-07-29T10:15:30.000Z", "site_id": SITE_A, @@ -192,80 +182,6 @@ def main() -> None: ) ) - # listen_ComposedSiteInfo merges the latest site_info with the last tariff piece. - stream, _ = make_stream() - site = stream.get_energysite(SITE_A) - composed: list[dict[str, Any]] = [] - site.listen_ComposedSiteInfo(composed.append) - dispatch(stream, SITE_INFO_SNAPSHOT) - results.append( - check( - "composed view emits site_info alone with tariff_content_v2=None first", - composed == [{**SITE_INFO_SNAPSHOT["site_info"], "tariff_content_v2": None}], - f"got {composed}", - ) - ) - - dispatch(stream, TARIFF_SNAPSHOT) - results.append( - check( - "composed view merges in the tariff once it arrives", - composed[-1] - == {**SITE_INFO_SNAPSHOT["site_info"], "tariff_content_v2": TARIFF_SNAPSHOT["tariff_content_v2"]}, - f"got {composed[-1]}", - ) - ) - - dispatch(stream, SITE_INFO_UPDATE) - results.append( - check( - "composed view keeps the last tariff when only site_info updates", - composed[-1] - == {**SITE_INFO_UPDATE["site_info"], "tariff_content_v2": TARIFF_SNAPSHOT["tariff_content_v2"]}, - f"got {composed[-1]}", - ) - ) - - dispatch(stream, TARIFF_REMOVED) - results.append( - check( - "composed view reflects an explicit tariff removal as None", - composed[-1] == {**SITE_INFO_UPDATE["site_info"], "tariff_content_v2": None}, - f"got {composed[-1]}", - ) - ) - - # Nothing is emitted before the first site_info document arrives. - stream, _ = make_stream() - site = stream.get_energysite(SITE_A) - composed = [] - site.listen_ComposedSiteInfo(composed.append) - dispatch(stream, TARIFF_SNAPSHOT) - results.append( - check( - "composed view withholds emission until site_info has arrived", - composed == [], - f"got {composed}", - ) - ) - - # Removing the composed listener removes both underlying listeners. - stream, _ = make_stream() - site = stream.get_energysite(SITE_A) - composed = [] - remove = site.listen_ComposedSiteInfo(composed.append) - dispatch(stream, SITE_INFO_SNAPSHOT) - remove() - dispatch(stream, TARIFF_SNAPSHOT) - dispatch(stream, SITE_INFO_UPDATE) - results.append( - check( - "removing the composed listener stops delivery from either half", - len(composed) == 1, - f"got {composed}", - ) - ) - print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") if not all(results): From 1b070c4d1fb2540ee5807146ebabd57c7c5ebb48 Mon Sep 17 00:00:00 2001 From: Brett Date: Wed, 29 Jul 2026 18:54:34 +1000 Subject: [PATCH 5/5] fix(sse): treat a bare topics string as one topic, not characters Per PR review: topics="state" (or a bare SseTopic member) satisfies Iterable[str], so list(topics) split it into ['s','t','a','t','e'], sent as a garbled topics query param the server would reject. Accept str | Iterable[str] | None and wrap a lone string as a single-element list before validating emptiness. Co-Authored-By: Claude Sonnet 5 --- AGENTS.md | 2 +- README.md | 4 ++-- teslemetry_stream/stream.py | 17 ++++++++++------- tests/test_sse_topics.py | 22 ++++++++++++++++++++++ 4 files changed, 35 insertions(+), 10 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 50d139f..df7e49c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,7 +12,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - 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. - `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design), so it can't actually promise the whole REST-shaped document; a consumer wanting both tariffs together should use the REST site_info endpoint. -- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, and the empty-iterable rejection. +- `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes (must stay in sync with the api's `SSE_TOPICS`), and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered - and existing callers that never pass it are unaffected. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. ## Maintaining this file diff --git a/README.md b/README.md index da72f8f..24aed6c 100644 --- a/README.md +++ b/README.md @@ -210,8 +210,8 @@ convenience presets that expand to those exact names client-side. ## Public Methods in TeslemetryStream Class -### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, topics: Iterable[str] | None = None)` -Initialize the TeslemetryStream client. `topics` is an optional exact SSE wire event allowlist (see `SseTopic`); omitting it preserves legacy-all behavior. +### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, topics: str | Iterable[str] | None = None)` +Initialize the TeslemetryStream client. `topics` is an optional exact SSE wire event allowlist (see `SseTopic`) - a single topic or an iterable of them; omitting it preserves legacy-all behavior. ### `get_vehicle(vin: str) -> TeslemetryStreamVehicle` Create a vehicle object to manage config and create listeners. diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index f6f047e..48aec65 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -27,7 +27,7 @@ def __init__( vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, - topics: Iterable[str] | None = None, + topics: str | Iterable[str] | None = None, ): """ Initialize the TeslemetryStream client. @@ -39,11 +39,12 @@ def __init__( :param parse_timestamp: Whether to parse timestamps. :param manual: Whether to start listening manually. :param topics: Exact SSE wire event names (see `SseTopic` and its - presets in `const.py`) to subscribe to. Omitting this (`None`) - preserves legacy-all behavior: every applicable event is - delivered unfiltered, forever. An explicitly empty iterable is - rejected - it means "no topics", not "all topics", mirroring - the server's own 400 on an empty `topics` value. + presets in `const.py`) to subscribe to - a single topic or an + iterable of them. Omitting this (`None`) preserves legacy-all + behavior: every applicable event is delivered unfiltered, + forever. An explicitly empty iterable is rejected - it means + "no topics", not "all topics", mirroring the server's own 400 + on an empty `topics` value. """ if server and not server.endswith(".teslemetry.com"): raise ValueError("Server must be on the teslemetry.com domain") @@ -53,7 +54,9 @@ def __init__( self.vin = vin self.topics: list[str] | None if topics is not None: - self.topics = list(topics) + # A bare str (or SseTopic, itself a str) is iterable character-by-character - + # wrap it as a single topic rather than silently splitting it into letters. + self.topics = [topics] if isinstance(topics, str) else list(topics) if not self.topics: raise ValueError( "topics must not be empty - omit it (None) for legacy-all behavior" diff --git a/tests/test_sse_topics.py b/tests/test_sse_topics.py index c8aa3be..1d898da 100644 --- a/tests/test_sse_topics.py +++ b/tests/test_sse_topics.py @@ -138,6 +138,28 @@ def main() -> None: ) ) + # A bare string is one topic, not an iterable of characters. + stream, session = make_stream(topics="state") + asyncio.run(stream.connect()) + results.append( + check( + "a bare str topics value is treated as a single topic, not character-split", + session.calls[0]["params"] == {"topics": "state"}, + f"got {session.calls[0]['params']}", + ) + ) + + # A bare SseTopic member (itself a str) is likewise one topic. + stream, session = make_stream(topics=SseTopic.LIVE_STATUS) + asyncio.run(stream.connect()) + results.append( + check( + "a bare SseTopic topics value is treated as a single topic, not character-split", + session.calls[0]["params"] == {"topics": "live_status"}, + f"got {session.calls[0]['params']}", + ) + ) + # listen_TariffContentV2 receives the tariff document verbatim. stream, _ = make_stream() site = stream.get_energysite(SITE_A)