feat: accept externally sourced events via TeslemetryStream.ingest

AGENTS.md

@@ -22,7 +22,8 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record).
 - `TeslemetryStream` has no `__aenter__`/`__aexit__` - do not reintroduce `async with TeslemetryStream(...)` in the README or examples. Connection lifecycle is entirely listener-driven: `async_add_listener` connects on the first (public) listener and disconnects on the last one removed; `connect()`/`close()`/`listen()` exist for callers who want to manage the connection themselves instead.
 - `__anext__` treats a `aiohttp.ClientResponseError` with `status` 401 or 403 as terminal, not transient: it sets `active = False` and raises `TeslemetryStreamAuthenticationError` (chaining the original error) rather than retrying, since a rejected token can never succeed on retry. Every other `aiohttp.ClientError` (including other response statuses) keeps the pre-existing backoff-and-reconnect behavior. `tests/test_auth_failure.py` covers both the 401/403 surfacing and that a genuine transient `ClientError` still retries and reconnects.
-- `TeslemetryStream` owns exactly one `_listen_task`: `async_add_listener`'s zero-to-one transition only creates it when absent/done, and `listen()` itself checks `asyncio.current_task()` against `self._listen_task` - a second concurrent `listen()` call joins the owner via `await existing_task` instead of racing it for the connection. `connect()` serializes the actual GET behind `self._connect_lock` and re-checks `self.active` after acquiring both the lock and the response, discarding a response that arrived after a stop/supersede rather than publishing it. Internal reconnect paths (EOF, `ClientError`, unexpected exceptions in `__anext__`) call `_close_response()`, which only clears the response and notifies connection listeners - they must not call `close()`, which additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally, so task cancellation (however triggered) still releases the connection. `listen()` dispatches over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch.
+- `TeslemetryStream` owns exactly one `_listen_task`: `async_add_listener`'s zero-to-one transition only creates it when absent/done, and `listen()` itself checks `asyncio.current_task()` against `self._listen_task` - a second concurrent `listen()` call joins the owner via `await existing_task` instead of racing it for the connection. `connect()` serializes the actual GET behind `self._connect_lock` and re-checks `self.active` after acquiring both the lock and the response, discarding a response that arrived after a stop/supersede rather than publishing it. Internal reconnect paths (EOF, `ClientError`, unexpected exceptions in `__anext__`) call `_close_response()`, which only clears the response and notifies connection listeners - they must not call `close()`, which additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally, so task cancellation (however triggered) still releases the connection. `listen()` dispatches through `_dispatch()` (shared with `ingest()`), over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch.
+- `TeslemetryStream.ingest()` (and `TeslemetryStreamVehicle.ingest()`, the same call with the VIN filled in) is the ingestion point for an observation the library did not read off its own SSE connection - a Bluetooth broadcast, today. It builds the native wire event (`vin`/`data`/`createdAt`) plus an open-ended `metadata` dict (`Metadata.SOURCE`/`Metadata.RAW` in `const.py`) and hands it to `_dispatch`, the single fan-out `listen()` also uses - so native events pass through untouched and a consumer's existing `listen_*` callbacks receive both sources with no translation and no second subscription. Dispatch is arrival-ordered and the stream holds no per-field value: nothing is deduplicated, reordered, or dropped, and there is deliberately no source ranking, precedence, or preferred-source field - which report to believe is the consumer's decision, made on `metadata`. Ingesting neither requires nor opens a connection. Neither library depends on the other: the BLE-side shim that shapes a broadcast into this format lives in `tesla-fleet-api` and the consumer wires the two, following the `aiopowerwall`/`EnergySiteRouter` duck-typing precedent. `tests/test_external_ingest.py` covers the native-event regression, the two sources being indistinguishable apart from metadata, and the no-dedup/no-ranking contract.
 
 ## Maintaining this file
 

README.md

@@ -6,6 +6,7 @@ This is an asynchronous Python 3 library that connects to the Teslemetry Stream
 - Listen to various telemetry signals from Tesla vehicles
 - Handle signals using typed listen methods
 - Write custom listeners for multiple signals
+- Ingest observations from other sources into the same listeners
 
 ## Installation
 
@@ -208,6 +209,48 @@ stream = TeslemetryStream(
 `SSE_VEHICLE_TOPICS`, `SSE_ENERGY_TOPICS`, and `SSE_ALL_TOPICS` are
 convenience presets that expand to those exact names client-side.
 
+## Ingesting Externally Sourced Events
+
+Some observations arrive from somewhere other than the SSE connection - a
+Bluetooth broadcast read locally, for instance. `ingest` accepts one and
+delivers it to the listeners you already registered, in the same wire format
+the connection itself sends, so no consumer needs a second subscription or a
+translation step:
+
+```python
+from teslemetry_stream import Metadata, Signal
+
+vehicle = stream.get_vehicle("<vin>")
+vehicle.listen_Locked(print)
+
+vehicle.ingest(
+    {Signal.LOCKED: False},
+    {
+        Metadata.SOURCE: "bluetooth",
+        Metadata.RAW: "VEHICLELOCKSTATE_SELECTIVE_UNLOCKED",
+    },
+)
+```
+
+`metadata` records where the observation came from and what its untranslated
+wire value was. It is carried on the event and never acted on: `source` keeps
+provenance visible for debugging or for a decision the consumer makes, and
+`raw` keeps the fidelity a translation discards (any unlocked state reads as
+unlocked, but which one is still worth having). It is a plain dict, so a
+source can add keys of its own without a format break; `Metadata` names the
+two every source is expected to speak.
+
+Ingesting holds no client and opens no connection, so an observation is
+delivered whether or not the stream is connected.
+
+**Ordering and deduplication.** Every event - native or ingested - is
+dispatched in arrival order, exactly as given. The stream keeps no per-field
+value and compares nothing against what came before, so if two sources report
+the same field, listeners are called twice: once per report, later report
+last, neither dropped. There is deliberately no source ranking, precedence,
+or preferred source. Which report to believe is the consumer's decision, and
+`metadata` is what it decides on.
+
 ## 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: str | Iterable[str] | None = None)`
@@ -249,6 +292,9 @@ Add listener for data updates.
 ### `listen(self)`
 Listen to the telemetry stream.
 
+### `ingest(data: dict, vin: str | None = None, metadata: dict | None = None, created_at: str | None = None) -> dict`
+Deliver an externally sourced observation to this stream's listeners, in the same wire format the connection sends. `vin` defaults to the stream's own. Returns the event as dispatched. See [Ingesting Externally Sourced Events](#ingesting-externally-sourced-events).
+
 ### `listen_Credits(callback: Callable[[CreditsEvent], None]) -> Callable[[], None]`
 Add listener for credit events.
 
@@ -272,6 +318,9 @@ Replace Fleet Telemetry configuration for the vehicle.
 ### `config(self) -> dict`
 Return current configuration for the vehicle.
 
+### `ingest(data: dict, metadata: dict | None = None, created_at: str | None = None) -> dict`
+Ingest an externally sourced observation for this vehicle - `TeslemetryStream.ingest` with the VIN filled in.
+
 ### `listen_State(callback: Callable[[bool], None]) -> Callable[[],None]`
 Listen for vehicle online state polling. The callback receives a boolean value representing whether the vehicle is online.
 

teslemetry_stream/__init__.py

@@ -3,6 +3,7 @@
     SSE_ENERGY_TOPICS,
     SSE_VEHICLE_TOPICS,
     Alert,
+    Metadata,
     Signal,
     SseTopic,
 )
@@ -22,6 +23,7 @@
     "SSE_ENERGY_TOPICS",
     "SSE_VEHICLE_TOPICS",
     "Alert",
+    "Metadata",
     "Signal",
     "SseTopic",
     "TeslemetryStream",

teslemetry_stream/const.py

@@ -37,6 +37,22 @@ class Key(StrEnum):
     URL = "url"
     TOTALS = "totals"
     TARIFF_CONTENT_V2 = "tariff_content_v2"
+    METADATA = "metadata"
+
+
+class Metadata(StrEnum):
+    """Conventional keys inside an ingested event's `metadata` dict.
+
+    The dict is open ended by design, so a new key can be added without a
+    format break; these are the ones every source is expected to speak.
+    """
+
+    #: Where the observation came from, e.g. "bluetooth". Provenance is
+    #: recorded so it stays visible, never so the library can rank sources.
+    SOURCE = "source"
+    #: The wire value before any translation, kept because collapsing it to
+    #: the streamed type (e.g. any-unlocked-is-unlocked) discards fidelity.
+    RAW = "raw"
 
 
 class Signal(StrEnum):

teslemetry_stream/stream.py

@@ -9,7 +9,7 @@
 
 import aiohttp
 
-from .const import CreditsEvent
+from .const import CreditsEvent, Key
 from .energysite import TeslemetryStreamEnergySite
 from .exception import TeslemetryStreamAuthenticationError, TeslemetryStreamEnded
 from .vehicle import TeslemetryStreamVehicle
@@ -339,12 +339,7 @@ async def __anext__(self) -> dict[str, Any]:
                     if field == "data":
                         data = json.loads(value)
                         if self.parse_timestamp:
-                            main, _, ns = data["createdAt"].partition(".")
-                            data["timestamp"] = int(
-                                datetime.strptime(main, "%Y-%m-%dT%H:%M:%S")
-                                .replace(tzinfo=timezone.utc)
-                                .timestamp()
-                            ) * 1000 + int(ns[:3])
+                            data["timestamp"] = _parse_created_at(data["createdAt"])
                         return cast(dict[str, Any], data)
                 raise TeslemetryStreamEnded()
             except StopAsyncIteration as e:
@@ -452,26 +447,99 @@ async def listen(self) -> None:
         try:
             async for event in self:
                 if event:
-                    # A snapshot, not a live view - a callback that creates a
-                    # vehicle (get_vehicle) or otherwise adds a listener
-                    # mid-dispatch must not mutate _listeners while this is
-                    # iterating it, which would raise RuntimeError and kill
-                    # the loop. Internal (bookkeeping) listeners go first, so
-                    # one can cache from the pristine event before any public
-                    # callback gets a chance to mutate it in place.
-                    ordered = sorted(self._listeners.values(), key=lambda item: not item[2])
-                    for listener, filters, _internal in ordered:
-                        if recursive_match(filters, event):
-                            try:
-                                listener(event)
-                            except Exception as error:
-                                LOGGER.error("Uncaught error in listener: %s", error)
+                    self._dispatch(event)
         finally:
             self._close_response()
             if self._listen_task is current_task:
                 self._listen_task = None
         LOGGER.debug("Listen has finished")
 
+    def _dispatch(self, event: dict[str, Any]) -> None:
+        """
+        Fan one event out to every listener whose filters match it.
+
+        Shared by the SSE reader and by `ingest`, so an externally sourced
+        event reaches consumers by the same path a native one does.
+
+        :param event: Event to dispatch.
+        """
+        # A snapshot, not a live view - a callback that creates a vehicle
+        # (get_vehicle) or otherwise adds a listener mid-dispatch must not
+        # mutate _listeners while this is iterating it, which would raise
+        # RuntimeError and kill the loop. Internal (bookkeeping) listeners go
+        # first, so one can cache from the pristine event before any public
+        # callback gets a chance to mutate it in place.
+        ordered = sorted(self._listeners.values(), key=lambda item: not item[2])
+        for listener, filters, _internal in ordered:
+            if recursive_match(filters, event):
+                try:
+                    listener(event)
+                except Exception as error:
+                    LOGGER.error("Uncaught error in listener: %s", error)
+
+    def ingest(
+        self,
+        data: dict[str, Any],
+        vin: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        created_at: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Feed an externally sourced observation into this stream's listeners.
+
+        The event is built in the same wire format the SSE connection
+        delivers - `{"vin": ..., "data": {...}, "createdAt": ...}` - and goes
+        out through the same dispatch, so a consumer's existing `listen_*`
+        callbacks receive it with no translation and no separate
+        subscription. Nothing here connects, reads, or holds a client: the
+        observation is an argument, and ingesting one works whether or not
+        the SSE connection is up.
+
+        Every ingested event is dispatched, in arrival order, exactly as it
+        was given. The stream keeps no per-field value and does not compare
+        an event against what came before, so two sources reporting the same
+        field produce two dispatches - the later one simply arrives later,
+        the way repeated native events already do. There is deliberately no
+        source ranking, precedence, or deduplication: which source to
+        believe is the consumer's call, and `metadata` is what it decides on.
+
+        :param data: Signal payload keyed by signal name, e.g.
+            `{"Locked": True}` or `{"DoorState": {"TrunkFront": False}}`.
+        :param vin: Vehicle Identification Number. Defaults to the stream's
+            own `vin` for a single-vehicle client.
+        :param metadata: Provenance carried alongside the event and never
+            acted on - see `Metadata` for the conventional keys (`source`,
+            `raw`). A dict so it can grow without a format break.
+        :param created_at: Observation time in the stream's own
+            `%Y-%m-%dT%H:%M:%S.%fZ` format. Defaults to now.
+        :return: The event as dispatched.
+        :raises ValueError: If no VIN is available.
+        :raises TypeError: If `data` or `metadata` is not a dict.
+        """
+        vin = vin or self.vin
+        if not vin:
+            raise ValueError("ingest requires a vin, either its own or the stream's")
+        if not isinstance(data, dict):
+            raise TypeError("data must be a dict keyed by signal name")
+        if metadata is not None and not isinstance(metadata, dict):
+            raise TypeError("metadata must be a dict")
+
+        event: dict[str, Any] = {
+            # Plain string keys, exactly as a decoded SSE event carries them.
+            Key.VIN.value: vin,
+            # Copied, not aliased - a caller reusing its own payload dict
+            # must not retroactively change an event already dispatched.
+            Key.DATA.value: dict(data),
+            Key.CREATED_AT.value: created_at or _now(),
+            Key.METADATA.value: dict(metadata) if metadata else {},
+        }
+        if self.parse_timestamp:
+            # Before dispatch, so a malformed created_at raises to the caller
+            # instead of half-delivering an event.
+            event["timestamp"] = _parse_created_at(event[Key.CREATED_AT])
+        self._dispatch(event)
+        return event
+
     def listen_Credits(
         self, callback: Callable[[CreditsEvent], None]
     ) -> Callable[[], None]:
@@ -527,3 +595,23 @@ def recursive_match(dict1: dict[str, Any] | None, dict2: dict[str, Any]) -> bool
                 return False
     # No differences found
     return True
+
+
+def _parse_created_at(created_at: str) -> int:
+    """
+    Convert a stream `createdAt` string to epoch milliseconds.
+
+    :param created_at: Timestamp as sent on the wire.
+    :return: Milliseconds since the epoch.
+    """
+    main, _, ns = created_at.partition(".")
+    return int(
+        datetime.strptime(main, "%Y-%m-%dT%H:%M:%S")
+        .replace(tzinfo=timezone.utc)
+        .timestamp()
+    ) * 1000 + int(ns[:3])
+
+
+def _now() -> str:
+    """Current time in the same format the stream sends `createdAt` in."""
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"