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.

Author: Bre77

AGENTS.md

@@ -9,6 +9,7 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `Signal` in `const.py` tracks <https://api.teslemetry.com/fields.json>; 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
 

CLAUDE.md

@@ -0,0 +1 @@
+AGENTS.md

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="<token>",
+            session=session,
+        ) as stream:
+
+            site = stream.get_energysite("<site_id>")
+
+            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.

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

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

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},
+        )

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

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()