diff --git a/AGENTS.md b/AGENTS.md index 01a9b68..ba475a3 100644 --- a/AGENTS.md +++ b/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. diff --git a/tesla_fleet_api/__init__.py b/tesla_fleet_api/__init__.py index bd146c8..f4fa1e0 100644 --- a/tesla_fleet_api/__init__.py +++ b/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", diff --git a/tesla_fleet_api/funnel.py b/tesla_fleet_api/funnel.py index 2014c59..81d41dd 100644 --- a/tesla_fleet_api/funnel.py +++ b/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 + ) diff --git a/tests/test_funnel_stream.py b/tests/test_funnel_stream.py new file mode 100644 index 0000000..b4a3414 --- /dev/null +++ b/tests/test_funnel_stream.py @@ -0,0 +1,294 @@ +"""Tests for TeslemetryStreamPublisher: signal-update translation, and no way to fetch. + +Every update here is a literal dictionary written in the test. The publisher is +given no client, session, or callable, so a value it produces can only have come +from that literal - which is what makes the "cannot request" claim checkable +rather than asserted. +""" + +from __future__ import annotations + +import inspect +from typing import Any +from unittest import TestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.funnel import ( + BleBroadcastPublisher, + FieldPath, + ObservationFunnel, + TeslemetryStreamPublisher, + Value, +) +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_protocol.command.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) +from tesla_protocol.command.vcsec_pb2 import ( + FromVCSECMessage, + VehicleLockState_E, + VehicleStatus, +) + +# A trimmed but structurally real Teslemetry stream ``data`` push. +UPDATE: dict[str, Any] = { + "Locked": True, + "ChargePortDoorOpen": True, + "DoorState": { + "DriverFront": False, + "TrunkFront": False, + "TrunkRear": False, + }, + "Soc": 72, +} + + +class _Clock: + def __init__(self, now: float = 0.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + +def _translate(data: dict[str, Any]) -> dict[FieldPath, Value]: + publisher = TeslemetryStreamPublisher(clock=_Clock()) + return {o.path: o.value for o in publisher.publish_update(data)} + + +class TestSignalTranslation(TestCase): + def test_maps_exactly_the_three_audited_signals(self) -> None: + self.assertEqual( + _translate(UPDATE), + { + FieldPath.LOCKED: True, + FieldPath.CHARGE_PORT_DOOR_OPEN: True, + FieldPath.DOOR_STATE_TRUNK_FRONT: False, + }, + ) + + def test_absent_signals_emit_no_observation(self) -> None: + self.assertEqual(_translate({"Soc": 72}), {}) + + def test_a_null_signal_is_an_explicit_unavailable_reading(self) -> None: + self.assertEqual( + _translate( + { + "Locked": None, + "ChargePortDoorOpen": None, + "DoorState": None, + } + ), + { + FieldPath.LOCKED: None, + FieldPath.CHARGE_PORT_DOOR_OPEN: None, + FieldPath.DOOR_STATE_TRUNK_FRONT: None, + }, + ) + + def test_string_encoded_booleans_are_coerced(self) -> None: + """Some vehicles stream 'true'/'false' strings instead of JSON booleans.""" + self.assertEqual( + _translate( + { + "Locked": "true", + "ChargePortDoorOpen": "false", + "DoorState": {"TrunkFront": "true"}, + } + ), + { + FieldPath.LOCKED: True, + FieldPath.CHARGE_PORT_DOOR_OPEN: False, + FieldPath.DOOR_STATE_TRUNK_FRONT: True, + }, + ) + + def test_a_non_boolean_locked_emits_no_observation(self) -> None: + self.assertEqual(_translate({"Locked": "unlocked"}), {}) + + def test_a_partial_door_state_only_reports_trunk_front_when_present(self) -> None: + self.assertEqual( + _translate({"DoorState": {"DriverFront": True}}), + {}, + ) + + def test_a_malformed_door_state_is_ignored_rather_than_guessed(self) -> None: + self.assertEqual(_translate({"DoorState": "open"}), {}) + + def test_unaudited_signals_are_never_routed(self) -> None: + self.assertEqual(_translate({"Soc": 72, "ChargerVoltage": 240.0}), {}) + + +class TestSuppliedUpdateFunnelling(TestCase): + def test_a_supplied_update_reaches_listeners(self) -> None: + funnel = ObservationFunnel() + publisher = TeslemetryStreamPublisher(clock=_Clock()) + funnel.attach(publisher) + + seen: dict[FieldPath, list[Value]] = {path: [] for path in FieldPath} + for path in FieldPath: + funnel.listen(path, seen[path].append) + + publisher.publish_update(UPDATE, observed_at=1.0) + + self.assertEqual( + seen, + { + FieldPath.LOCKED: [True], + FieldPath.CHARGE_PORT_DOOR_OPEN: [True], + FieldPath.DOOR_STATE_TRUNK_FRONT: [False], + }, + ) + + def test_a_repeated_update_is_not_re_dispatched(self) -> None: + funnel = ObservationFunnel() + publisher = TeslemetryStreamPublisher(clock=_Clock()) + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + + publisher.publish_update(UPDATE, observed_at=1.0) + publisher.publish_update(UPDATE, observed_at=2.0) + + self.assertEqual(seen, [True]) + + def test_a_partial_update_leaves_other_fields_untouched(self) -> None: + """A stream push carrying only one changed signal doesn't blank the rest.""" + funnel = ObservationFunnel() + publisher = TeslemetryStreamPublisher(clock=_Clock()) + funnel.attach(publisher) + + seen: dict[FieldPath, list[Value]] = {path: [] for path in FieldPath} + for path in FieldPath: + funnel.listen(path, seen[path].append) + + publisher.publish_update(UPDATE, observed_at=1.0) + publisher.publish_update({"Locked": False}, observed_at=2.0) + + self.assertEqual(seen[FieldPath.LOCKED], [True, False]) + self.assertEqual(seen[FieldPath.CHARGE_PORT_DOOR_OPEN], [True]) + self.assertEqual(seen[FieldPath.DOOR_STATE_TRUNK_FRONT], [False]) + + def test_activation_subscribes_a_passive_source_to_nothing(self) -> None: + funnel = ObservationFunnel() + publisher = TeslemetryStreamPublisher(clock=_Clock()) + funnel.attach(publisher) + before = dict(vars(publisher)) + + release = funnel.listen(FieldPath.LOCKED, lambda _: None) + release() + + self.assertEqual(vars(publisher), before) + + def test_publishing_before_attach_translates_but_reaches_no_listener(self) -> None: + funnel = ObservationFunnel() + publisher = TeslemetryStreamPublisher(clock=_Clock()) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + observations = publisher.publish_update(UPDATE, observed_at=1.0) + + self.assertEqual(len(observations), 3) + self.assertEqual(seen, []) + + +class TestPublisherCannotRequestData(TestCase): + """The publisher's only data source is the mapping handed to it.""" + + def test_it_exposes_no_coroutine_and_no_awaitable_member(self) -> None: + publisher = TeslemetryStreamPublisher(clock=_Clock()) + for name, member in inspect.getmembers(publisher): + self.assertFalse( + inspect.iscoroutinefunction(member), + msg=f"{name} is a coroutine function", + ) + self.assertFalse(inspect.isawaitable(member), msg=f"{name} is awaitable") + + def test_it_holds_no_client_session_or_fetch_callable(self) -> None: + publisher = TeslemetryStreamPublisher(clock=_Clock()) + self.assertEqual(set(vars(publisher)), {"_clock", "_sink"}) + + self.assertEqual( + list(inspect.signature(publisher._clock).parameters), # type: ignore[attr-defined] + [], + ) + + def test_it_yields_nothing_when_no_update_is_supplied(self) -> None: + clock = _Clock() + funnel = ObservationFunnel() + publisher = TeslemetryStreamPublisher(clock=clock) + funnel.attach(publisher) + + seen: list[Value] = [] + for path in FieldPath: + funnel.listen(path, seen.append) + + clock.now = 10_000.0 + self.assertEqual(seen, []) + for path in FieldPath: + self.assertIsNone(funnel.value(path)) + + def test_an_object_that_could_fetch_is_never_called(self) -> None: + """An update-shaped mapping whose lookups are counted proves the reads.""" + calls: list[str] = [] + + class _Tripwire(dict[str, Any]): + def __getitem__(self, key: str) -> Any: + calls.append(key) + return super().__getitem__(key) + + def fetch(self) -> None: # pragma: no cover - must never be reached + raise AssertionError("the publisher invoked a fetch callable") + + publisher = TeslemetryStreamPublisher(clock=_Clock()) + publisher.publish_update(_Tripwire(UPDATE)) + self.assertEqual(calls, ["Locked", "ChargePortDoorOpen", "DoorState"]) + + +class TestRegressionWalkthrough(TestCase): + """Bluetooth and streaming feed the same funnel; neither blanks the other. + + The captain's direction is that streaming is the primary source of truth + and Bluetooth is opportunistic - here both publishers are attached to one + funnel and each field keeps a value whichever source is producing, with + no ranking between them. + """ + + def test_streaming_and_bluetooth_both_reach_the_same_listeners(self) -> None: + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + vehicle: VehicleBluetooth[Any] = VehicleBluetooth(parent, "5YJXCAE43LF123456") + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle.connect = AsyncMock() # type: ignore[method-assign] + vehicle.client = MagicMock() + vehicle.client.is_connected = True + vehicle.client.write_gatt_char = AsyncMock() + + funnel = ObservationFunnel() + funnel.attach(BleBroadcastPublisher(vehicle, clock=_Clock(100.0))) + stream_publisher = TeslemetryStreamPublisher(clock=_Clock(0.0)) + funnel.attach(stream_publisher) + + seen: dict[FieldPath, list[Value]] = {path: [] for path in FieldPath} + for path in FieldPath: + funnel.listen(path, seen[path].append) + + stream_publisher.publish_update({"Locked": True}, observed_at=1.0) + vehicle._on_message( + RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY), + protobuf_message_as_bytes=FromVCSECMessage( + vehicleStatus=VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED + ) + ).SerializeToString(), + ) + ) + + self.assertEqual(seen[FieldPath.LOCKED], [True, False]) + self.assertIs(funnel.value(FieldPath.LOCKED), False)