Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
1 change: 1 addition & 0 deletions CLAUDE.md
51 changes: 51 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,43 @@ 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:
stream = TeslemetryStream(
access_token="<token>",
session=session,
)

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()
stream.close()
```

> **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)`
Expand All @@ -122,6 +159,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.

Expand Down Expand Up @@ -197,3 +237,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.
2 changes: 2 additions & 0 deletions teslemetry_stream/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from .stream import TeslemetryStream
from .vehicle import TeslemetryStreamVehicle
from .energysite import TeslemetryStreamEnergySite
from .exception import (
TeslemetryStreamError,
TeslemetryStreamConnectionError,
Expand All @@ -11,6 +12,7 @@
__all__ = [
"TeslemetryStream",
"TeslemetryStreamVehicle",
"TeslemetryStreamEnergySite",
"TeslemetryStreamError",
"TeslemetryStreamConnectionError",
"TeslemetryStreamVehicleNotConfigured",
Expand Down
5 changes: 5 additions & 0 deletions teslemetry_stream/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
55 changes: 55 additions & 0 deletions teslemetry_stream/energysite.py
Original file line number Diff line number Diff line change
@@ -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},
)
14 changes: 14 additions & 0 deletions teslemetry_stream/stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from .exception import TeslemetryStreamEnded
from .vehicle import TeslemetryStreamVehicle
from .energysite import TeslemetryStreamEnergySite

LOGGER = logging.getLogger(__package__)

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
"""
Expand Down
196 changes: 196 additions & 0 deletions tests/test_energysite_events.py
Original file line number Diff line number Diff line change
@@ -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()
Loading