feat(sse): support topics allowlist and tariff_content_v2 topic

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
 

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="<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="<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.

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",
 ]

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.

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]:

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,13 +38,18 @@ 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")
 
         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