feat(funnel): add Teslemetry stream publisher

AGENTS.md

@@ -69,7 +69,7 @@ Commands (vehicle/commands.py) - protobuf-based signed command implementation (A
 
 `Router` (`router/base.py`) is an entity-agnostic composition wrapper (not part of the inheritance chain) that chains an ordered list of two-or-more backends sharing a common method surface and dispatches each method call down the chain with automatic per-command failover: it tries the first backend that has the method and, on any exception except `BluetoothUnconfirmedCommand`, retries the same call on the next backend that has it, returning the first success (raising the last error only if every applicable backend fails, `AttributeError` only if none has the method). Non-callable attributes resolve to the first backend that has them. Constructor: `Router(primary, secondary, *more_backends, health=None)`. The health check (`bool` | sync callable | async callable returning `bool`; omitted = attempt primary, fail over on exception with no probe) gates **only the primary**; the rest of the chain is reached purely through per-command failover — there is deliberately no per-backend health matrix. Double-execution caveat: a non-idempotent command that fails mid-flight can be re-run on the next backend, except for `BluetoothUnconfirmedCommand`, which propagates without replay.
 
-`ObservationFunnel` (`funnel.py`) is the **read** side, a separate mechanism from the command `Router` and not in its inheritance chain. It is a **funnel, not a selector**: every attached publisher feeds the same per-field listeners, so a field bound to one source survives that source dropping. There is deliberately no source health, availability, grace window, failback delay, priority, stickiness or per-field selection anywhere in it — **unavailability is a value a source reports** (a null/SNA reading), never something the funnel infers from a link dropping; inferring it would be the funnel asserting data it does not have. The only arbitration is `publish()` ignoring an observation older than the last one for that field and not re-dispatching an unchanged value; both are hard-coded, not configurable. It is **entirely synchronous and can never originate a request**: no `async def`/`await`, no polling loop, no request callable, no scheduling task — `tests/test_funnel.py::TestFunnelCannotOriginateWork` locks that in against the module's own AST, so keep the module synchronous rather than adding a fetch path. Polling belongs entirely to an external consumer, which may gate its own schedule on `listen_demand(paths, cb)` (a read-only observer over the activation counts) and feed results back through `VehicleDataResultPublisher.publish_result(dict)` — that publisher holds no client, session, or callable able to obtain one. Publishers push into the funnel (which is itself the `ObservationSink`) via `publish(Observation)`; `observed_at` values must come from one monotonic clock shared by every publisher on a funnel. `value(path)` returns the last observed value, its `None` meaning either never observed or reported unavailable. Fields are deliberately three (`Locked`, `ChargePortDoorOpen`, `DoorState.TrunkFront`); translations are positive allowlists, and an unmapped VCSEC enum or absent JSON leaf emits no observation rather than a guess, while an explicit JSON null emits an unavailable value. `BleBroadcastPublisher` reuses the existing `VehicleBluetooth` `listen_vehicle_lock_state`/`listen_charge_port`/`listen_front_trunk` seams and never connects, reads, or commands; because `VEHICLELOCKSTATE_UNLOCKED` is 0 with no proto3 presence, every VCSEC status broadcast reports a lock state and the funnel deduplicates the repeats. Any unlocked VCSEC lock state, including `INTERNAL_LOCKED`→locked and `SELECTIVE_UNLOCKED`→unlocked, maps to a boolean per the "any unlocked is unlocked" ruling; closure `UNKNOWN`/`FAILED_UNLATCH` remain unmapped pending live-frame validation.
+`ObservationFunnel` (`funnel.py`) is the **read** side, a separate mechanism from the command `Router` and not in its inheritance chain. It is a **funnel, not a selector**: every attached publisher feeds the same per-field listeners, so a field bound to one source survives that source dropping. There is deliberately no source health, availability, grace window, failback delay, priority, stickiness or per-field selection anywhere in it — **unavailability is a value a source reports** (a null/SNA reading), never something the funnel infers from a link dropping; inferring it would be the funnel asserting data it does not have. The only arbitration is `publish()` ignoring an observation older than the last one for that field and not re-dispatching an unchanged value; both are hard-coded, not configurable. It is **entirely synchronous and can never originate a request**: no `async def`/`await`, no polling loop, no request callable, no scheduling task — `tests/test_funnel.py::TestFunnelCannotOriginateWork` locks that in against the module's own AST, so keep the module synchronous rather than adding a fetch path. Polling belongs entirely to an external consumer, which may gate its own schedule on `listen_demand(paths, cb)` (a read-only observer over the activation counts) and feed results back through `VehicleDataResultPublisher.publish_result(dict)` — that publisher holds no client, session, or callable able to obtain one. Publishers push into the funnel (which is itself the `ObservationSink`) via `publish(Observation)`; `observed_at` values must come from one monotonic clock shared by every publisher on a funnel. `value(path)` returns the last observed value, its `None` meaning either never observed or reported unavailable. Fields are deliberately three (`Locked`, `ChargePortDoorOpen`, `DoorState.TrunkFront`); translations are positive allowlists, and an unmapped VCSEC enum or absent JSON leaf emits no observation rather than a guess, while an explicit JSON null emits an unavailable value. `BleBroadcastPublisher` reuses the existing `VehicleBluetooth` `listen_vehicle_lock_state`/`listen_charge_port`/`listen_front_trunk` seams and never connects, reads, or commands; because `VEHICLELOCKSTATE_UNLOCKED` is 0 with no proto3 presence, every VCSEC status broadcast reports a lock state and the funnel deduplicates the repeats. Any unlocked VCSEC lock state, including `INTERNAL_LOCKED`→locked and `SELECTIVE_UNLOCKED`→unlocked, maps to a boolean per the "any unlocked is unlocked" ruling; closure `UNKNOWN`/`FAILED_UNLATCH` remain unmapped pending live-frame validation. `TeslemetryStreamPublisher` is the intended primary source (Bluetooth is opportunistic) and follows the same caller-supplied-payload shape as `VehicleDataResultPublisher` rather than depending on the separate `teslemetry-stream` package: `publish_update(data)` takes one stream push's `data` mapping, keyed by signal name (`Locked`, `ChargePortDoorOpen`, `DoorState` with a nested `TrunkFront`) — the same string keys `teslemetry-stream`'s own `Signal` `StrEnum` values equal, so a caller's dict matches whether or not it's keyed with that enum. It coerces the `"true"`/`"false"` wire strings some vehicles stream in place of JSON booleans. The funnel has no source ranking between Bluetooth and stream publishers by design — see the `ObservationFunnel` description above.
 
 `VehicleRouter` and `EnergySiteRouter` (`router/vehicle.py`, `router/energysite.py`) are thin entity-specific `Router` subclasses. `VehicleRouter(bluetooth_primary, teslemetry_secondary)` pairs a `VehicleBluetooth` primary with a cloud (`TeslemetryVehicle`) secondary; `EnergySiteRouter(local_energysite, teslemetry_energysite)` pairs a duck-typed local `EnergySite`-shaped object (e.g. aiopowerwall's `PowerwallEnergySite`, no dependency added) with a cloud `TeslemetryEnergySite` fallback. Both re-export from `router/__init__.py` (`tesla_fleet_api.router.Router` etc.) and from `tesla/__init__.py` (`tesla_fleet_api.tesla.Router`) for backward compatibility. They have no factory on the `Vehicles`/`EnergySites` collections. This repo owns the RSA keypair lifecycle and cloud registration (`Tesla.get_rsa_private_key`, `EnergySite.add_authorized_client`) that aiopowerwall's local signed transport depends on but does not implement itself; see `docs/energy_local_control.md` for the end-to-end pairing + `EnergySiteRouter` composition flow. The cloud-only `set_island_mode`/`go_off_grid`/`reconnect_grid` (`tesla/energysite.py`) can only send an unsigned `grpc_command`, which gateways can acknowledge without actuating the contactor — rather than ship that as a silent no-op, they unconditionally raise `SignedCommandRequired` (`exceptions.py`); only the signed local path via `add_authorized_client` + `EnergySiteRouter` actually actuates, and a success response from that transport still doesn't prove the contactor moved — verify state after the call.
 

tesla_fleet_api/__init__.py

@@ -11,6 +11,7 @@
     ObservationFunnel,
     ObservationSink,
     Publisher,
+    TeslemetryStreamPublisher,
     VehicleDataResultPublisher,
 )
 from tesla_fleet_api.tariff import (
@@ -47,6 +48,7 @@
     "TeslaFleetOAuth",
     "Teslemetry",
     "TeslemetryClientRegistration",
+    "TeslemetryStreamPublisher",
     "Tessie",
     "VehicleDataResultPublisher",
     "firmware_at_least",

tesla_fleet_api/funnel.py

@@ -456,3 +456,88 @@ def _translate(
         # boolean meaning.
         elif (code := _int_code(front_trunk)) in (0, 1):
             yield Observation(FieldPath.DOOR_STATE_TRUNK_FRONT, code == 1, observed_at)
+
+
+# -- Supplied Teslemetry stream publisher -----------------------------------
+
+
+def _coerce_stream_bool(value: object) -> object:
+    """``"true"``/``"false"`` become real booleans; anything else passes through.
+
+    Some vehicles stream a boolean signal as that literal string rather than a
+    JSON boolean even when the stream prefers typed values.
+    """
+    if value == "true":
+        return True
+    if value == "false":
+        return False
+    return value
+
+
+class TeslemetryStreamPublisher:
+    """Translates a caller-supplied Teslemetry stream signal update into observations.
+
+    The update is an argument, keyed by signal name the way a Teslemetry
+    stream push does (``Locked``, ``ChargePortDoorOpen``, ``DoorState`` with a
+    nested ``TrunkFront``). This class holds no stream client, connection, or
+    callable able to obtain one, so nothing here can originate a subscription
+    or a connection - the same shape ``VehicleDataResultPublisher`` uses for a
+    supplied ``vehicle_data`` result, kept here rather than depending on the
+    separate stream client package for a type this library does not need.
+    """
+
+    def __init__(self, *, clock: Clock = time.monotonic) -> None:
+        self._clock = clock
+        self._sink: ObservationSink | None = None
+
+    @property
+    def paths(self) -> frozenset[FieldPath]:
+        return frozenset(FieldPath)
+
+    def attach(self, sink: ObservationSink) -> Unsubscribe:
+        self._sink = sink
+
+        def detach() -> None:
+            self._sink = None
+
+        return detach
+
+    def request(self, paths: frozenset[FieldPath]) -> None:
+        """Passive source: activation subscribes to nothing."""
+
+    def release(self, paths: frozenset[FieldPath]) -> None:
+        """Passive source: activation subscribes to nothing."""
+
+    def publish_update(
+        self, data: Mapping[str, Any], *, observed_at: float | None = None
+    ) -> tuple[Observation, ...]:
+        """Translate a supplied stream signal update and feed it to the sink."""
+        at = self._clock() if observed_at is None else observed_at
+        observations = tuple(self._translate(data, at))
+        sink = self._sink
+        if sink is not None:
+            for observation in observations:
+                sink.publish(observation)
+        return observations
+
+    def _translate(
+        self, data: Mapping[str, Any], observed_at: float
+    ) -> Iterator[Observation]:
+        locked = _coerce_stream_bool(_leaf(data, "Locked"))
+        if locked is None or isinstance(locked, bool):
+            yield Observation(FieldPath.LOCKED, locked, observed_at)
+
+        charge_port = _coerce_stream_bool(_leaf(data, "ChargePortDoorOpen"))
+        if charge_port is None or isinstance(charge_port, bool):
+            yield Observation(FieldPath.CHARGE_PORT_DOOR_OPEN, charge_port, observed_at)
+
+        door_state = _leaf(data, "DoorState")
+        if door_state is None:
+            yield Observation(FieldPath.DOOR_STATE_TRUNK_FRONT, None, observed_at)
+        elif isinstance(door_state, Mapping):
+            door_section: Mapping[str, Any] = door_state  # pyright: ignore[reportUnknownVariableType]
+            trunk_front = _coerce_stream_bool(_leaf(door_section, "TrunkFront"))
+            if trunk_front is None or isinstance(trunk_front, bool):
+                yield Observation(
+                    FieldPath.DOOR_STATE_TRUNK_FRONT, trunk_front, observed_at
+                )