From e59e23799c6f07440ed4d275dde0db5d7137345b Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 14:31:26 +1000 Subject: [PATCH 1/3] feat(vehicle): add typed BLE broadcast listeners for vehicle-state VCSEC status broadcasts were only consumed by the one-shot per-command confirmation ladder (PR 68). Add a persistent listener API so callers can receive vehicle state passively instead of polling: typed listen_ methods for every VehicleStatus leaf field (lock state, sleep status, presence, all 8 closures, tonneau percent), plus an untyped listen_broadcast catch-all for payloads VehicleStatus decoding doesn't cover. Listeners fan out from the existing _on_message notification path - no second subscription - and coexist with the confirmation ladder's watcher. --- AGENTS.md | 1 + docs/bluetooth_vehicles.md | 60 +++ tesla_fleet_api/tesla/vehicle/bluetooth.py | 14 +- tesla_fleet_api/tesla/vehicle/broadcast.py | 209 ++++++++++ tests/test_ble_broadcast_listeners.py | 446 +++++++++++++++++++++ 5 files changed, 728 insertions(+), 2 deletions(-) create mode 100644 tesla_fleet_api/tesla/vehicle/broadcast.py create mode 100644 tests/test_ble_broadcast_listeners.py diff --git a/AGENTS.md b/AGENTS.md index c3cfbf7..ad9f591 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -129,6 +129,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te - **BLE mutating-command timeout is inconclusive - never assume "the write didn't land"**: this corrects an earlier version of this note, which observed `adjust_volume` write timeouts on one rig leaving the car unmutated and concluded a write timeout means the write never landed. Later live testing disproved that as a general rule: `door_unlock` and `door_lock` each raised a timeout yet both physically executed - a VCSEC state read after each confirmed the lock state had flipped. Reads (state readers, `wake_up()`'s effect) came back reliably all night; only mutating VCSEC/RKE actions showed this false-negative pattern, most likely because the vehicle doesn't reliably return an ack `_send()` observes within the timeout, not because the write failed. Treat `BluetoothUnconfirmedCommand` (a `BluetoothTimeout` subclass) from any mutating BLE command as **inconclusive, not failure**: snapshot state before acting, then verify the outcome with a follow-up state read whenever a mutation times out. Never blind-retry a non-idempotent command (toggles like `media_toggle_playback`, volume steps, schedule add/remove) on timeout alone - see the retry-double-execution entry below for why the library's own retry has the same exposure. The `adjust_volume`/`TimeoutAPIError` root cause from the original observation is still unexplained; it just isn't evidence that timed-out writes never land. VCSEC actuations now use the shorter `_actuation_timeout` when their terminal ack is lost. - **The BLE mutating-command confirmation ladder is one `confirmation` enum + one `raise_unconfirmed` bool**: `VehicleBluetooth.confirmation` (`"optimistic" | "ack" | "verify"`, default `"ack"`; threaded through `Vehicles`/`VehiclesBluetooth.create*`) picks how many of write → ack-or-broadcast wait → state-read confirmation run; `raise_unconfirmed` (default `False`) picks what happens when the ladder still can't tell. `"optimistic"` short-circuits `_sendVehicleSecurity`/`_sendInfotainment` (`bluetooth.py`) to `_send_optimistic()`, which signs and writes but never waits for any reply - a provably pre-submission write failure still raises `BluetoothTransportError` unconditionally, but a submitted-then-ambiguous write (see the write-delivery-certainty entry below) follows `raise_unconfirmed` like every other rung instead of being consulted as "nothing else." `"verify"` adds a post-timeout state-read rung: on an unresolved ack/broadcast wait, `_resolve_timeout()` reads the mapped prover state (`_vcsec_verify_plan`/`_INFOTAINMENT_VERIFY_PLANS` in `bluetooth.py`; only clearly-derivable absolute commands are covered - lock/unlock, `set_charge_limit`, `set_charging_amps`, `adjust_volume` absolute, `set_temps`, `auto_conditioning_start/stop`) and returns success on a match, raises `BluetoothCommandFailed` on a proven mismatch, or returns `None` (still unresolved) if the read itself couldn't complete - `None` falls through to `raise_unconfirmed`. Commands with no plan (true toggles, relative steps, ack-only actions) always fall through regardless of `confirmation`. The legacy `optimistic`/`verify_commands` boolean surface is deprecated: both warn (`DeprecationWarning`) and map onto `confirmation` (a positional bool in the `confirmation` slot is treated as old `verify_commands`; dominance order preserved - `optimistic=True` wins if both are set), and remain as read-only properties (`confirmation == "optimistic"`/`"verify"`) for existing readers. See `docs/bluetooth_vehicles.md` for the user-facing table and defaults. - **Broadcast-as-confirmation races the ack wait for lock/unlock**: the vehicle keeps emitting unsolicited VCSEC status broadcasts on the same notification subscription even when it emits no addressed ack for a lock/unlock actuation - live-verified at scale (see the timeout-rate study referenced from `docs/bluetooth_vehicles.md`). `_send`'s `confirm_broadcast` param (threaded through `Commands._command`/`_sendVehicleSecurity`, ignored by the Fleet-signed transport) arms a per-domain watcher in `_on_message` (`_broadcast_watchers`, `bluetooth.py`) that decodes broadcast frames via `_decode_vcsec_status` and races them against the addressed-reply wait in `_await_response_or_broadcast`; first to satisfy the plan's predicate wins, and only the addressed-reply path can raise a car-side rejection. A mismatching broadcast doesn't fail fast (it's appended to `mismatches`, not resolved) since a later broadcast in the same window could still confirm success - but if the whole window elapses with a mismatch as the last word and nothing else confirming, `_await_response_or_broadcast` raises `BluetoothCommandFailed` instead of the ambiguous timeout. This reuses the exact same `_vcsec_verify_plan` predicate as the `"verify"` rung above (one source of truth), applied to a broadcast's decoded `VehicleStatus` instead of a follow-up read; it currently covers only lock/unlock, the one VCSEC actuation with an observed status broadcast. Low-level race tests drive the real (unmocked) `_send` state machine directly - see `tests/test_ble_broadcast_confirmation.py`. +- **Persistent broadcast listeners (`tesla_fleet_api/tesla/vehicle/broadcast.py`)**: `VehicleBluetooth` fans the same VCSEC status broadcasts out to long-lived per-field listeners, not just the one-shot confirmation ladder above - one source of broadcast truth, dispatched from the same `_on_message`. Every `VehicleStatus` leaf field is a well-defined protobuf enum/int, so it gets a typed `listen_` method (`listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, `listen_user_presence`, the 8 door/trunk/charge-port/tonneau closure listeners, `listen_tonneau_percent_open`); anything not decoded into `VehicleStatus` (other VCSEC payloads like `CommandStatus`/whitelist/faults, and any future infotainment-domain broadcast - none observed today) falls back to the untyped `listen_broadcast(domain, callback)`. Closure/tonneau-percent listeners gate on `HasField` since those are submessages with real proto3 presence tracking; the three scalar enum fields (`vehicleLockState`/`vehicleSleepStatus`/`userPresence`) have none, so they fire on every status broadcast rather than only on change. Each `listen_*` returns an `unsubscribe()` closure; registries live for the `VehicleBluetooth` instance's lifetime and are unaffected by reconnects, matching `_queues`. Not consumed by Home Assistant yet - HA gets state via Teslemetry streaming already. See `docs/bluetooth_vehicles.md` and `tests/test_ble_broadcast_listeners.py`. - **`BluetoothUnconfirmedCommand` vs `BluetoothCommandFailed` (`exceptions.py`), and how `Router` treats each**: `_sendVehicleSecurity`/`_sendInfotainment` (`bluetooth.py`, the mutating-command seam, unconditionally) wrap a caught `BluetoothTimeout` into `BluetoothUnconfirmedCommand` when the ladder is genuinely unresolved - either the write succeeded but the ack/broadcast was lost, or the write entered backend I/O and failed with delivery unprovable, so the vehicle may have executed the command. With default `raise_unconfirmed=False` that unresolved outcome returns best-effort success; with `raise_unconfirmed=True` the `BluetoothUnconfirmedCommand` reaches the caller. `BluetoothCommandFailed` is the other, distinct outcome: a state check (the `"verify"` rung's read, or a mismatching broadcast still standing at window-end) actively *proved* the command did not apply - it deliberately does **not** subclass `BluetoothTimeout`/`BluetoothUnconfirmedCommand`. `Router._dispatch` (`router/base.py`) special-cases only `BluetoothUnconfirmedCommand` to skip its normal per-command failover and re-raise immediately - replaying an already-possibly-executed command risks double-execution. `BluetoothCommandFailed` carries no such risk (the command is proven not to have applied), so it falls through `Router`'s ordinary `except (Exception, TeslaFleetError)` clause and fails over like any other error - no `Router` code change was needed for this, only the exception type's placement outside the `BluetoothTimeout` hierarchy. A plain read (`_getVehicleSecurity`/`_getInfotainment`) still raises unadorned `BluetoothTimeout` on the same kind of wait timeout, since a read has no side effect to be unconfirmed about, and a *provably pre-submission* write failure still raises the unrelated `BluetoothTransportError` (see the write-delivery-certainty entry below - a submitted-then-ambiguous write is not this case). - **Write-delivery certainty splits `BluetoothTransportError` from `BluetoothTimeout` at the GATT write in `_send`**: `write_gatt_char` failures are not uniformly `BluetoothTransportError`. `BleakCharacteristicNotFoundError` (bleak resolves `WRITE_UUID` synchronously, before any backend I/O) is the only case provably pre-submission, so it alone stays `BluetoothTransportError` and is safe for `Router` to retry. Every other `BleakError`/`TimeoutError` from that call happens inside backend I/O (D-Bus/CoreBluetooth/an ESPHome proxy) where delivery can't be proven either way - field data measured 2 of 3 such write failures had already executed on the vehicle - so `_send` instead races any already-armed broadcast watcher for the rest of the window (a matching broadcast can still confirm success despite the failed write) and, failing that, raises plain `BluetoothTimeout`. Because `BluetoothUnconfirmedCommand` subclasses `BluetoothTimeout`, this lands in the exact same `except BluetoothTimeout` ladder in `_sendVehicleSecurity`/`_sendInfotainment` as a lost post-write ack, with no separate exception type needed; `_send_optimistic` gets the equivalent treatment explicitly since it bypasses that ladder (see above). A read is unaffected - `_getVehicleSecurity`/`_getInfotainment` don't override the ladder, so a write-ambiguous read propagates plain `BluetoothTimeout` and `Router` fails over on it normally, which is safe since a read has no double-execution risk. Tests: `tests/test_ble_send_transport.py` (`SendTransportErrorTests`), `tests/test_ble_broadcast_confirmation.py` (`WriteFailureBroadcastRaceTests`), `tests/test_ble_write_timeout_router.py`. - **`wake_up()` is best-effort; confirm readiness with an INFO read**: `wake_up()` is a VCSEC actuation, so a terminal ack now returns promptly when observed, but an unresolved wake remains only an inconclusive wake signal, not command failure (`BluetoothUnconfirmedCommand` when `raise_unconfirmed=True`, best-effort success by default). Confirm readiness by retrying a cheap INFO read instead (see the boot-delay gotcha above for why the first INFO read still needs its own retry/backoff). Hold one connection across a whole batch of related commands rather than reconnecting between each - reconnecting costs ~123% more per operation with no demonstrated wake-preservation benefit from the connection alone. diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index c18df66..7756857 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -472,6 +472,66 @@ endpoint, because requesting multiple endpoints together can exceed the vehicle's signed-command response-size cap and raise `TeslaFleetMessageFaultResponseSizeExceedsMTU`. +## Broadcast Listeners + +The vehicle's VCSEC computer emits unsolicited `VehicleStatus` broadcasts on +the same BLE notification subscription used for command replies, independent +of any command you send. `VehicleBluetooth` fans these out to persistent +per-field listeners, so you can receive vehicle-state changes passively +instead of polling the state readers above. + +Every `VehicleStatus` leaf field is a well-defined protobuf enum or int, so +each has its own typed listener method, similar in spirit to +[python-teslemetry-stream](https://github.com/Teslemetry/python-teslemetry-stream)'s +`listen_` surface: + +- `listen_vehicle_lock_state(callback)` - `VehicleLockState_E` +- `listen_vehicle_sleep_status(callback)` - `VehicleSleepStatus_E` +- `listen_user_presence(callback)` - `UserPresence_E` +- `listen_front_driver_door(callback)` - `ClosureState_E` +- `listen_front_passenger_door(callback)` - `ClosureState_E` +- `listen_rear_driver_door(callback)` - `ClosureState_E` +- `listen_rear_passenger_door(callback)` - `ClosureState_E` +- `listen_front_trunk(callback)` - `ClosureState_E` +- `listen_rear_trunk(callback)` - `ClosureState_E` +- `listen_charge_port(callback)` - `ClosureState_E` +- `listen_tonneau(callback)` - `ClosureState_E` +- `listen_tonneau_percent_open(callback)` - `int` + +Each `listen_*` method takes a synchronous `callback(value)` and returns an +`unsubscribe()` closure: + +```python +def on_lock_state(state): + print(f"Lock state: {state}") + +unsubscribe = vehicle.listen_vehicle_lock_state(on_lock_state) +... +unsubscribe() +``` + +The door/trunk/charge-port/tonneau listeners and `listen_tonneau_percent_open` +only fire on broadcasts that actually carry the corresponding submessage +(`closureStatuses`/`detailedClosureStatus`) - not every status broadcast +includes them. `listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, and +`listen_user_presence` fire on every `VehicleStatus` broadcast, since proto3 +gives no presence tracking for a scalar enum field. + +Anything not decoded into `VehicleStatus` - other VCSEC broadcast payloads +(`CommandStatus`, whitelist events, faults) and any future +infotainment-domain broadcast - falls back to the untyped `listen_broadcast`, +which delivers the raw `RoutableMessage` for a given `Domain`: + +```python +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain + +unsubscribe = vehicle.listen_broadcast(Domain.DOMAIN_VEHICLE_SECURITY, print) +``` + +Listeners are plain Python callables registered on the `VehicleBluetooth` +instance - they persist across reconnects and are torn down only by calling +the `unsubscribe()` closure returned at registration. + ## Media Commands `VehicleBluetooth` inherits the signed media commands from `Commands`, so media diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 204f224..349b55a 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -26,6 +26,7 @@ TeslaFleetError, WhitelistOperationStatus, ) +from tesla_fleet_api.tesla.vehicle.broadcast import BroadcastListeners from tesla_fleet_api.tesla.vehicle.commands import ( Commands, infotainment_command_name, @@ -298,7 +299,9 @@ def _plan_auto_conditioning(action: VehicleAction) -> VerifyPlan | None: } -class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]): +class VehicleBluetooth( + BroadcastListeners, Commands[BluetoothParentT], Generic[BluetoothParentT] +): """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing. Callers can catch failures from this class with a single ``TeslaFleetError``, @@ -489,6 +492,7 @@ def __init__( Domain.DOMAIN_INFOTAINMENT: asyncio.Queue(), } self._broadcast_watchers = {} + self._init_broadcast_listeners() self.device = device self._connect_lock = asyncio.Lock() self._buffer = ReassemblingBuffer(self._on_message) @@ -655,9 +659,15 @@ def _on_message(self, msg: RoutableMessage) -> None: if msg.to_destination.routing_address != self._from_destination: LOGGER.debug("Ignoring broadcast message (not addressed to us)") - watcher = self._broadcast_watchers.get(msg.from_destination.domain) + domain = msg.from_destination.domain + watcher = self._broadcast_watchers.get(domain) if watcher is not None: watcher(msg) + self._dispatch_domain_listeners(domain, msg) + if domain == Domain.DOMAIN_VEHICLE_SECURITY and self._status_listeners: + status = _decode_vcsec_status(msg) + if status is not None: + self._dispatch_status_listeners(status) return queue = self._queues.get(msg.from_destination.domain) diff --git a/tesla_fleet_api/tesla/vehicle/broadcast.py b/tesla_fleet_api/tesla/vehicle/broadcast.py new file mode 100644 index 0000000..704d830 --- /dev/null +++ b/tesla_fleet_api/tesla/vehicle/broadcast.py @@ -0,0 +1,209 @@ +"""Persistent listeners for unsolicited BLE vehicle-status broadcasts. + +VCSEC emits ``VehicleStatus`` broadcasts on the same notification +subscription used for command replies (``_on_message`` in ``bluetooth.py``). +Every leaf field of ``VehicleStatus`` is a well-defined protobuf enum or int, +so each gets its own typed ``listen_`` method here, mirroring +python-teslemetry-stream's per-field listener surface. Anything not decoded +into ``VehicleStatus`` - other VCSEC broadcast payloads (``CommandStatus``, +whitelist events, faults) and any future infotainment-domain broadcast - +falls back to the untyped ``listen_broadcast``. +""" + +from __future__ import annotations + +from typing import Callable, TypeVar + +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( + Domain, + RoutableMessage, +) +from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import ( + ClosureState_E, + UserPresence_E, + VehicleLockState_E, + VehicleSleepStatus_E, + VehicleStatus, +) + +Unsubscribe = Callable[[], None] + +T = TypeVar("T") + + +class BroadcastListeners: + """Typed and untyped listener registries fed by ``_on_message`` broadcasts. + + Both registries are created once and live for the lifetime of the + ``VehicleBluetooth`` instance - they are plain Python callables, not + connection-scoped resources, so they need no reset on reconnect and leak + nothing beyond what a caller keeps registered without unsubscribing. + """ + + _status_listeners: list[Callable[[VehicleStatus], None]] + _domain_listeners: dict[Domain, list[Callable[[RoutableMessage], None]]] + + def _init_broadcast_listeners(self) -> None: + self._status_listeners = [] + self._domain_listeners = {} + + def _register(self, registry: list[T], item: T) -> Unsubscribe: + registry.append(item) + + def unsubscribe() -> None: + try: + registry.remove(item) + except ValueError: + pass + + return unsubscribe + + def _dispatch_domain_listeners(self, domain: Domain, msg: RoutableMessage) -> None: + for callback in list(self._domain_listeners.get(domain, ())): + callback(msg) + + def _dispatch_status_listeners(self, status: VehicleStatus) -> None: + for callback in list(self._status_listeners): + callback(status) + + # -- Generic, untyped -------------------------------------------------- + + def listen_broadcast( + self, domain: Domain, callback: Callable[[RoutableMessage], None] + ) -> Unsubscribe: + """Listen for any unsolicited broadcast on ``domain``. + + Catch-all for broadcasts the typed listeners below don't cover: + non-``VehicleStatus`` VCSEC payloads (``CommandStatus``, whitelist + events, faults) and any infotainment-domain broadcast. + """ + return self._register(self._domain_listeners.setdefault(domain, []), callback) + + # -- Typed VCSEC VehicleStatus fields ------------------------------------ + + def listen_vehicle_lock_state( + self, callback: Callable[[VehicleLockState_E], None] + ) -> Unsubscribe: + """Listen for the vehicle's overall lock state.""" + return self._register( + self._status_listeners, lambda status: callback(status.vehicleLockState) + ) + + def listen_vehicle_sleep_status( + self, callback: Callable[[VehicleSleepStatus_E], None] + ) -> Unsubscribe: + """Listen for the vehicle's sleep status.""" + return self._register( + self._status_listeners, lambda status: callback(status.vehicleSleepStatus) + ) + + def listen_user_presence( + self, callback: Callable[[UserPresence_E], None] + ) -> Unsubscribe: + """Listen for driver-presence status.""" + return self._register( + self._status_listeners, lambda status: callback(status.userPresence) + ) + + def listen_tonneau_percent_open( + self, callback: Callable[[int], None] + ) -> Unsubscribe: + """Listen for the tonneau's open percentage. + + Only fires on broadcasts that carry ``detailedClosureStatus`` - unlike + the scalar enum fields above, proto3 tracks presence for this + submessage, so a broadcast that omits it is skipped rather than + reported as 0%. + """ + + def on_status(status: VehicleStatus) -> None: + if status.HasField("detailedClosureStatus"): + callback(status.detailedClosureStatus.tonneauPercentOpen) + + return self._register(self._status_listeners, on_status) + + def listen_front_driver_door( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the front driver door's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.frontDriverDoor) + + return self._register(self._status_listeners, on_status) + + def listen_front_passenger_door( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the front passenger door's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.frontPassengerDoor) + + return self._register(self._status_listeners, on_status) + + def listen_rear_driver_door( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the rear driver door's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.rearDriverDoor) + + return self._register(self._status_listeners, on_status) + + def listen_rear_passenger_door( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the rear passenger door's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.rearPassengerDoor) + + return self._register(self._status_listeners, on_status) + + def listen_front_trunk( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the front trunk's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.frontTrunk) + + return self._register(self._status_listeners, on_status) + + def listen_rear_trunk( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the rear trunk's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.rearTrunk) + + return self._register(self._status_listeners, on_status) + + def listen_charge_port( + self, callback: Callable[[ClosureState_E], None] + ) -> Unsubscribe: + """Listen for the charge port's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.chargePort) + + return self._register(self._status_listeners, on_status) + + def listen_tonneau(self, callback: Callable[[ClosureState_E], None]) -> Unsubscribe: + """Listen for the tonneau's closure state.""" + + def on_status(status: VehicleStatus) -> None: + if status.HasField("closureStatuses"): + callback(status.closureStatuses.tonneau) + + return self._register(self._status_listeners, on_status) diff --git a/tests/test_ble_broadcast_listeners.py b/tests/test_ble_broadcast_listeners.py new file mode 100644 index 0000000..6270926 --- /dev/null +++ b/tests/test_ble_broadcast_listeners.py @@ -0,0 +1,446 @@ +"""Tests for the persistent BLE broadcast listener API (``BroadcastListeners``). + +Fans unsolicited VCSEC status broadcasts out to typed per-field listeners +plus a generic per-domain listener, reusing the exact same ``_on_message`` +routing path as the one-shot confirmation-ladder watcher - no second +notification subscription. Broadcasts are injected directly via +``vehicle._on_message``, matching ``test_ble_broadcast_confirmation.py``. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, cast +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) +from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import ( + ClosureState_E, + ClosureStatuses, + CommandStatus, + DetailedClosureStatus, + FromVCSECMessage, + OperationStatus_E, + UserPresence_E, + VehicleLockState_E, + VehicleSleepStatus_E, + VehicleStatus, +) + +VIN = "5YJXCAE43LF123456" +DOMAIN = Domain.DOMAIN_VEHICLE_SECURITY + + +def _make_vehicle(**kwargs: Any) -> VehicleBluetooth[Any]: + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + vehicle = VehicleBluetooth(parent, VIN, **kwargs) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle.client = MagicMock() + vehicle.client.write_gatt_char = AsyncMock() + + # Mark both signed-command sessions ready so _command skips the + # handshake round-trip (which would otherwise also go through _send). + sessions = cast("dict[int, Any]", getattr(vehicle, "_sessions")) + for session in sessions.values(): + session.epoch = b"\x00" * 16 + session.hmac = b"\x00" * 32 + session.delta = 0 + session.sharedKey = b"\x00" * 16 + return vehicle + + +def _status_broadcast(status: VehicleStatus) -> RoutableMessage: + """An unsolicited (unaddressed) VCSEC status broadcast.""" + body = FromVCSECMessage(vehicleStatus=status) + return RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=body.SerializeToString(), + ) + + +def _command_status_broadcast() -> RoutableMessage: + """A non-``VehicleStatus`` VCSEC broadcast. + + ``_decode_vcsec_status`` only decodes the ``vehicleStatus`` oneof member, + so this is invisible to every typed listener - exactly the "rest" the + generic listener exists for. + """ + body = FromVCSECMessage( + commandStatus=CommandStatus( + operationStatus=OperationStatus_E.OPERATIONSTATUS_OK + ) + ) + return RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=body.SerializeToString(), + ) + + +def _addressed_status( + vehicle: VehicleBluetooth[Any], status: VehicleStatus +) -> RoutableMessage: + """An addressed (non-broadcast) VCSEC status reply - must not reach listeners.""" + body = FromVCSECMessage(vehicleStatus=status) + return RoutableMessage( + to_destination=Destination( + domain=DOMAIN, routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=body.SerializeToString(), + ) + + +async def _wait_until_watching( + vehicle: VehicleBluetooth[Any], domain: Domain, timeout: float = 1.0 +) -> None: + """Block until the one-shot confirmation ladder has armed its watcher.""" + + async def poll() -> None: + while domain not in vehicle._broadcast_watchers: + await asyncio.sleep(0) + + await asyncio.wait_for(poll(), timeout=timeout) + + +class TypedFieldListenerTests(IsolatedAsyncioTestCase): + async def test_lock_state_listener_fires_with_typed_value(self) -> None: + vehicle = _make_vehicle() + seen: list[Any] = [] + vehicle.listen_vehicle_lock_state(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + async def test_sleep_status_listener_fires(self) -> None: + vehicle = _make_vehicle() + seen: list[Any] = [] + vehicle.listen_vehicle_sleep_status(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleSleepStatus=VehicleSleepStatus_E.VEHICLE_SLEEP_STATUS_ASLEEP + ) + ) + ) + + self.assertEqual(seen, [VehicleSleepStatus_E.VEHICLE_SLEEP_STATUS_ASLEEP]) + + async def test_user_presence_listener_fires(self) -> None: + vehicle = _make_vehicle() + seen: list[Any] = [] + vehicle.listen_user_presence(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus(userPresence=UserPresence_E.VEHICLE_USER_PRESENCE_PRESENT) + ) + ) + + self.assertEqual(seen, [UserPresence_E.VEHICLE_USER_PRESENCE_PRESENT]) + + async def test_multiple_field_listeners_fire_from_one_broadcast(self) -> None: + vehicle = _make_vehicle() + lock_seen: list[Any] = [] + presence_seen: list[Any] = [] + vehicle.listen_vehicle_lock_state(lock_seen.append) + vehicle.listen_user_presence(presence_seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED, + userPresence=UserPresence_E.VEHICLE_USER_PRESENCE_PRESENT, + ) + ) + ) + + self.assertEqual(lock_seen, [VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED]) + self.assertEqual(presence_seen, [UserPresence_E.VEHICLE_USER_PRESENCE_PRESENT]) + + +class ClosureListenerTests(IsolatedAsyncioTestCase): + async def test_all_eight_closure_listeners_read_their_own_field(self) -> None: + vehicle = _make_vehicle() + results: dict[str, Any] = {} + for name in [ + "front_driver_door", + "front_passenger_door", + "rear_driver_door", + "rear_passenger_door", + "front_trunk", + "rear_trunk", + "charge_port", + "tonneau", + ]: + listener = getattr(vehicle, f"listen_{name}") + listener(lambda value, name=name: results.__setitem__(name, value)) + + status = VehicleStatus( + closureStatuses=ClosureStatuses( + frontDriverDoor=ClosureState_E.CLOSURESTATE_OPEN, + frontPassengerDoor=ClosureState_E.CLOSURESTATE_AJAR, + rearDriverDoor=ClosureState_E.CLOSURESTATE_CLOSED, + rearPassengerDoor=ClosureState_E.CLOSURESTATE_CLOSING, + frontTrunk=ClosureState_E.CLOSURESTATE_OPENING, + rearTrunk=ClosureState_E.CLOSURESTATE_FAILED_UNLATCH, + chargePort=ClosureState_E.CLOSURESTATE_OPEN, + tonneau=ClosureState_E.CLOSURESTATE_UNKNOWN, + ) + ) + vehicle._on_message(_status_broadcast(status)) + + self.assertEqual(results["front_driver_door"], ClosureState_E.CLOSURESTATE_OPEN) + self.assertEqual( + results["front_passenger_door"], ClosureState_E.CLOSURESTATE_AJAR + ) + self.assertEqual( + results["rear_driver_door"], ClosureState_E.CLOSURESTATE_CLOSED + ) + self.assertEqual( + results["rear_passenger_door"], ClosureState_E.CLOSURESTATE_CLOSING + ) + self.assertEqual(results["front_trunk"], ClosureState_E.CLOSURESTATE_OPENING) + self.assertEqual( + results["rear_trunk"], ClosureState_E.CLOSURESTATE_FAILED_UNLATCH + ) + self.assertEqual(results["charge_port"], ClosureState_E.CLOSURESTATE_OPEN) + self.assertEqual(results["tonneau"], ClosureState_E.CLOSURESTATE_UNKNOWN) + + async def test_closure_listener_does_not_fire_without_closure_statuses( + self, + ) -> None: + vehicle = _make_vehicle() + seen: list[Any] = [] + vehicle.listen_front_driver_door(seen.append) + + # A status broadcast that carries only vehicleLockState has no + # closureStatuses submessage - proto3 tracks presence for it. + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, []) + + +class TonneauPercentOpenTests(IsolatedAsyncioTestCase): + async def test_fires_when_detailed_closure_status_present(self) -> None: + vehicle = _make_vehicle() + seen: list[int] = [] + vehicle.listen_tonneau_percent_open(seen.append) + + status = VehicleStatus( + detailedClosureStatus=DetailedClosureStatus(tonneauPercentOpen=42) + ) + vehicle._on_message(_status_broadcast(status)) + + self.assertEqual(seen, [42]) + + async def test_does_not_fire_without_detailed_closure_status(self) -> None: + vehicle = _make_vehicle() + seen: list[int] = [] + vehicle.listen_tonneau_percent_open(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, []) + + +class UnsubscribeTests(IsolatedAsyncioTestCase): + async def test_unsubscribe_stops_delivery(self) -> None: + vehicle = _make_vehicle() + seen: list[Any] = [] + unsubscribe = vehicle.listen_vehicle_lock_state(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + unsubscribe() + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED + ) + ) + ) + + self.assertEqual(seen, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + async def test_double_unsubscribe_is_a_no_op(self) -> None: + vehicle = _make_vehicle() + unsubscribe = vehicle.listen_vehicle_lock_state(lambda _: None) + unsubscribe() + unsubscribe() # must not raise + + async def test_unsubscribe_only_removes_its_own_listener(self) -> None: + vehicle = _make_vehicle() + seen_a: list[Any] = [] + seen_b: list[Any] = [] + unsubscribe_a = vehicle.listen_vehicle_lock_state(seen_a.append) + vehicle.listen_vehicle_lock_state(seen_b.append) + + unsubscribe_a() + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen_a, []) + self.assertEqual(seen_b, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + +class GenericBroadcastListenerTests(IsolatedAsyncioTestCase): + async def test_generic_listener_receives_raw_message(self) -> None: + vehicle = _make_vehicle() + seen: list[RoutableMessage] = [] + vehicle.listen_broadcast(DOMAIN, seen.append) + + msg = _status_broadcast( + VehicleStatus(vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED) + ) + vehicle._on_message(msg) + + self.assertEqual(seen, [msg]) + + async def test_generic_listener_covers_non_vehicle_status_payloads(self) -> None: + vehicle = _make_vehicle() + typed_seen: list[Any] = [] + raw_seen: list[RoutableMessage] = [] + vehicle.listen_vehicle_lock_state(typed_seen.append) + vehicle.listen_broadcast(DOMAIN, raw_seen.append) + + vehicle._on_message(_command_status_broadcast()) + + self.assertEqual(typed_seen, []) + self.assertEqual(len(raw_seen), 1) + + async def test_generic_listener_scoped_to_its_domain(self) -> None: + vehicle = _make_vehicle() + seen: list[RoutableMessage] = [] + vehicle.listen_broadcast(Domain.DOMAIN_INFOTAINMENT, seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, []) + + +class AddressedMessagesNeverReachListenersTests(IsolatedAsyncioTestCase): + async def test_addressed_reply_does_not_fire_listeners(self) -> None: + vehicle = _make_vehicle() + typed_seen: list[Any] = [] + raw_seen: list[RoutableMessage] = [] + vehicle.listen_vehicle_lock_state(typed_seen.append) + vehicle.listen_broadcast(DOMAIN, raw_seen.append) + + vehicle._on_message( + _addressed_status( + vehicle, + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ), + ) + ) + + self.assertEqual(typed_seen, []) + self.assertEqual(raw_seen, []) + + +class CoexistsWithConfirmationLadderTests(IsolatedAsyncioTestCase): + async def test_persistent_listener_fires_alongside_one_shot_watcher(self) -> None: + """A broadcast resolving an in-flight command's confirmation ladder + must still reach a persistent listener registered separately - the + two mechanisms share ``_on_message`` but must not clobber each other. + """ + vehicle = _make_vehicle() + vehicle._actuation_timeout = 5.0 + seen: list[Any] = [] + vehicle.listen_vehicle_lock_state(seen.append) + + task = asyncio.ensure_future(vehicle.door_lock()) + await _wait_until_watching(vehicle, DOMAIN) + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + result = await asyncio.wait_for(task, timeout=1.0) + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + self.assertEqual(seen, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + +class ListenerLifecycleTests(IsolatedAsyncioTestCase): + async def test_listeners_survive_disconnect(self) -> None: + vehicle = _make_vehicle() + vehicle.client.disconnect = AsyncMock() + seen: list[Any] = [] + vehicle.listen_vehicle_lock_state(seen.append) + + await vehicle.disconnect() + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + async def test_registries_are_not_shared_across_vehicle_instances(self) -> None: + vehicle_a = _make_vehicle() + vehicle_b = _make_vehicle() + seen_a: list[Any] = [] + vehicle_a.listen_vehicle_lock_state(seen_a.append) + + vehicle_b._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen_a, []) From dbdb5483923a76c2dd8dcef5d30ef68b8b274e38 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 18:42:29 +1000 Subject: [PATCH 2/3] no-mistakes(review): Isolate BLE broadcast listener failures --- tesla_fleet_api/tesla/vehicle/broadcast.py | 15 ++++- tests/test_ble_broadcast_listeners.py | 69 ++++++++++++++++++++++ 2 files changed, 82 insertions(+), 2 deletions(-) diff --git a/tesla_fleet_api/tesla/vehicle/broadcast.py b/tesla_fleet_api/tesla/vehicle/broadcast.py index 704d830..433442d 100644 --- a/tesla_fleet_api/tesla/vehicle/broadcast.py +++ b/tesla_fleet_api/tesla/vehicle/broadcast.py @@ -14,6 +14,7 @@ from typing import Callable, TypeVar +from tesla_fleet_api.const import LOGGER from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( Domain, RoutableMessage, @@ -30,6 +31,8 @@ T = TypeVar("T") +_FATAL_CALLBACK_ERRORS = (KeyboardInterrupt, SystemExit) + class BroadcastListeners: """Typed and untyped listener registries fed by ``_on_message`` broadcasts. @@ -58,13 +61,21 @@ def unsubscribe() -> None: return unsubscribe + def _dispatch_callback(self, callback: Callable[[T], None], value: T) -> None: + try: + callback(value) + except _FATAL_CALLBACK_ERRORS: + raise + except BaseException: + LOGGER.exception("BLE broadcast listener callback failed") + def _dispatch_domain_listeners(self, domain: Domain, msg: RoutableMessage) -> None: for callback in list(self._domain_listeners.get(domain, ())): - callback(msg) + self._dispatch_callback(callback, msg) def _dispatch_status_listeners(self, status: VehicleStatus) -> None: for callback in list(self._status_listeners): - callback(status) + self._dispatch_callback(callback, status) # -- Generic, untyped -------------------------------------------------- diff --git a/tests/test_ble_broadcast_listeners.py b/tests/test_ble_broadcast_listeners.py index 6270926..c19dcb4 100644 --- a/tests/test_ble_broadcast_listeners.py +++ b/tests/test_ble_broadcast_listeners.py @@ -16,6 +16,7 @@ from cryptography.hazmat.primitives.asymmetric import ec +from tesla_fleet_api.exceptions import BluetoothTimeout from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( Destination, @@ -363,6 +364,74 @@ async def test_generic_listener_scoped_to_its_domain(self) -> None: self.assertEqual(seen, []) +class ListenerExceptionIsolationTests(IsolatedAsyncioTestCase): + async def test_failing_generic_listener_does_not_block_later_listeners( + self, + ) -> None: + vehicle = _make_vehicle() + seen: list[RoutableMessage] = [] + + def fail(_: RoutableMessage) -> None: + raise RuntimeError("listener failed") + + vehicle.listen_broadcast(DOMAIN, fail) + vehicle.listen_broadcast(DOMAIN, seen.append) + + msg = _status_broadcast( + VehicleStatus(vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED) + ) + vehicle._on_message(msg) + + self.assertEqual(seen, [msg]) + + async def test_failing_typed_listener_does_not_block_later_listeners(self) -> None: + vehicle = _make_vehicle() + seen: list[Any] = [] + + def fail(_: VehicleLockState_E) -> None: + raise BluetoothTimeout() + + vehicle.listen_vehicle_lock_state(fail) + vehicle.listen_vehicle_lock_state(seen.append) + + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + self.assertEqual(seen, [VehicleLockState_E.VEHICLELOCKSTATE_LOCKED]) + + async def test_failing_listener_does_not_block_later_addressed_reply( + self, + ) -> None: + vehicle = _make_vehicle() + + def fail(_: VehicleLockState_E) -> None: + raise RuntimeError("listener failed") + + vehicle.listen_vehicle_lock_state(fail) + vehicle._on_message( + _status_broadcast( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ) + ) + + reply = _addressed_status( + vehicle, + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED + ), + ) + vehicle._on_message(reply) + + self.assertIs(vehicle._queues[DOMAIN].get_nowait(), reply) + + class AddressedMessagesNeverReachListenersTests(IsolatedAsyncioTestCase): async def test_addressed_reply_does_not_fire_listeners(self) -> None: vehicle = _make_vehicle() From 269cbd0f53c79fada868eedca88153fc5e0d2e64 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sun, 12 Jul 2026 18:53:20 +1000 Subject: [PATCH 3/3] no-mistakes(document): Document BLE broadcast listeners --- AGENTS.md | 2 +- README.md | 5 +++++ docs/bluetooth_vehicles.md | 7 +++++-- tesla_fleet_api/tesla/vehicle/bluetooth.py | 6 ++++++ tesla_fleet_api/tesla/vehicle/broadcast.py | 14 +++++++------- 5 files changed, 24 insertions(+), 10 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ad9f591..7c37c87 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -129,7 +129,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te - **BLE mutating-command timeout is inconclusive - never assume "the write didn't land"**: this corrects an earlier version of this note, which observed `adjust_volume` write timeouts on one rig leaving the car unmutated and concluded a write timeout means the write never landed. Later live testing disproved that as a general rule: `door_unlock` and `door_lock` each raised a timeout yet both physically executed - a VCSEC state read after each confirmed the lock state had flipped. Reads (state readers, `wake_up()`'s effect) came back reliably all night; only mutating VCSEC/RKE actions showed this false-negative pattern, most likely because the vehicle doesn't reliably return an ack `_send()` observes within the timeout, not because the write failed. Treat `BluetoothUnconfirmedCommand` (a `BluetoothTimeout` subclass) from any mutating BLE command as **inconclusive, not failure**: snapshot state before acting, then verify the outcome with a follow-up state read whenever a mutation times out. Never blind-retry a non-idempotent command (toggles like `media_toggle_playback`, volume steps, schedule add/remove) on timeout alone - see the retry-double-execution entry below for why the library's own retry has the same exposure. The `adjust_volume`/`TimeoutAPIError` root cause from the original observation is still unexplained; it just isn't evidence that timed-out writes never land. VCSEC actuations now use the shorter `_actuation_timeout` when their terminal ack is lost. - **The BLE mutating-command confirmation ladder is one `confirmation` enum + one `raise_unconfirmed` bool**: `VehicleBluetooth.confirmation` (`"optimistic" | "ack" | "verify"`, default `"ack"`; threaded through `Vehicles`/`VehiclesBluetooth.create*`) picks how many of write → ack-or-broadcast wait → state-read confirmation run; `raise_unconfirmed` (default `False`) picks what happens when the ladder still can't tell. `"optimistic"` short-circuits `_sendVehicleSecurity`/`_sendInfotainment` (`bluetooth.py`) to `_send_optimistic()`, which signs and writes but never waits for any reply - a provably pre-submission write failure still raises `BluetoothTransportError` unconditionally, but a submitted-then-ambiguous write (see the write-delivery-certainty entry below) follows `raise_unconfirmed` like every other rung instead of being consulted as "nothing else." `"verify"` adds a post-timeout state-read rung: on an unresolved ack/broadcast wait, `_resolve_timeout()` reads the mapped prover state (`_vcsec_verify_plan`/`_INFOTAINMENT_VERIFY_PLANS` in `bluetooth.py`; only clearly-derivable absolute commands are covered - lock/unlock, `set_charge_limit`, `set_charging_amps`, `adjust_volume` absolute, `set_temps`, `auto_conditioning_start/stop`) and returns success on a match, raises `BluetoothCommandFailed` on a proven mismatch, or returns `None` (still unresolved) if the read itself couldn't complete - `None` falls through to `raise_unconfirmed`. Commands with no plan (true toggles, relative steps, ack-only actions) always fall through regardless of `confirmation`. The legacy `optimistic`/`verify_commands` boolean surface is deprecated: both warn (`DeprecationWarning`) and map onto `confirmation` (a positional bool in the `confirmation` slot is treated as old `verify_commands`; dominance order preserved - `optimistic=True` wins if both are set), and remain as read-only properties (`confirmation == "optimistic"`/`"verify"`) for existing readers. See `docs/bluetooth_vehicles.md` for the user-facing table and defaults. - **Broadcast-as-confirmation races the ack wait for lock/unlock**: the vehicle keeps emitting unsolicited VCSEC status broadcasts on the same notification subscription even when it emits no addressed ack for a lock/unlock actuation - live-verified at scale (see the timeout-rate study referenced from `docs/bluetooth_vehicles.md`). `_send`'s `confirm_broadcast` param (threaded through `Commands._command`/`_sendVehicleSecurity`, ignored by the Fleet-signed transport) arms a per-domain watcher in `_on_message` (`_broadcast_watchers`, `bluetooth.py`) that decodes broadcast frames via `_decode_vcsec_status` and races them against the addressed-reply wait in `_await_response_or_broadcast`; first to satisfy the plan's predicate wins, and only the addressed-reply path can raise a car-side rejection. A mismatching broadcast doesn't fail fast (it's appended to `mismatches`, not resolved) since a later broadcast in the same window could still confirm success - but if the whole window elapses with a mismatch as the last word and nothing else confirming, `_await_response_or_broadcast` raises `BluetoothCommandFailed` instead of the ambiguous timeout. This reuses the exact same `_vcsec_verify_plan` predicate as the `"verify"` rung above (one source of truth), applied to a broadcast's decoded `VehicleStatus` instead of a follow-up read; it currently covers only lock/unlock, the one VCSEC actuation with an observed status broadcast. Low-level race tests drive the real (unmocked) `_send` state machine directly - see `tests/test_ble_broadcast_confirmation.py`. -- **Persistent broadcast listeners (`tesla_fleet_api/tesla/vehicle/broadcast.py`)**: `VehicleBluetooth` fans the same VCSEC status broadcasts out to long-lived per-field listeners, not just the one-shot confirmation ladder above - one source of broadcast truth, dispatched from the same `_on_message`. Every `VehicleStatus` leaf field is a well-defined protobuf enum/int, so it gets a typed `listen_` method (`listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, `listen_user_presence`, the 8 door/trunk/charge-port/tonneau closure listeners, `listen_tonneau_percent_open`); anything not decoded into `VehicleStatus` (other VCSEC payloads like `CommandStatus`/whitelist/faults, and any future infotainment-domain broadcast - none observed today) falls back to the untyped `listen_broadcast(domain, callback)`. Closure/tonneau-percent listeners gate on `HasField` since those are submessages with real proto3 presence tracking; the three scalar enum fields (`vehicleLockState`/`vehicleSleepStatus`/`userPresence`) have none, so they fire on every status broadcast rather than only on change. Each `listen_*` returns an `unsubscribe()` closure; registries live for the `VehicleBluetooth` instance's lifetime and are unaffected by reconnects, matching `_queues`. Not consumed by Home Assistant yet - HA gets state via Teslemetry streaming already. See `docs/bluetooth_vehicles.md` and `tests/test_ble_broadcast_listeners.py`. +- **Persistent broadcast listeners (`tesla_fleet_api/tesla/vehicle/broadcast.py`)**: `VehicleBluetooth` fans the same VCSEC status broadcasts out to long-lived per-field listeners, not just the one-shot confirmation ladder above - one source of broadcast truth, dispatched from the same `_on_message`. Every `VehicleStatus` leaf field is a well-defined protobuf enum/int, so it gets a typed `listen_` method (`listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, `listen_user_presence`, the 8 door/trunk/charge-port/tonneau closure listeners, `listen_tonneau_percent_open`); anything not decoded into `VehicleStatus` (other VCSEC payloads like `CommandStatus`/whitelist/faults, and any future infotainment-domain broadcast - none observed today) has no typed listener and is covered by the untyped `listen_broadcast(domain, callback)`, which receives every raw unsolicited broadcast for that domain including `VehicleStatus`. Closure/tonneau-percent listeners gate on `HasField` since those are submessages with real proto3 presence tracking; the three scalar enum fields (`vehicleLockState`/`vehicleSleepStatus`/`userPresence`) have none, so they fire on every status broadcast rather than only on change. Each `listen_*` returns an `unsubscribe()` closure; registries live for the `VehicleBluetooth` instance's lifetime and are unaffected by reconnects, matching `_queues`. Listener callback exceptions are logged and isolated from later listeners/message routing, except `KeyboardInterrupt`/`SystemExit`. Not consumed by Home Assistant yet - HA gets state via Teslemetry streaming already. See `docs/bluetooth_vehicles.md` and `tests/test_ble_broadcast_listeners.py`. - **`BluetoothUnconfirmedCommand` vs `BluetoothCommandFailed` (`exceptions.py`), and how `Router` treats each**: `_sendVehicleSecurity`/`_sendInfotainment` (`bluetooth.py`, the mutating-command seam, unconditionally) wrap a caught `BluetoothTimeout` into `BluetoothUnconfirmedCommand` when the ladder is genuinely unresolved - either the write succeeded but the ack/broadcast was lost, or the write entered backend I/O and failed with delivery unprovable, so the vehicle may have executed the command. With default `raise_unconfirmed=False` that unresolved outcome returns best-effort success; with `raise_unconfirmed=True` the `BluetoothUnconfirmedCommand` reaches the caller. `BluetoothCommandFailed` is the other, distinct outcome: a state check (the `"verify"` rung's read, or a mismatching broadcast still standing at window-end) actively *proved* the command did not apply - it deliberately does **not** subclass `BluetoothTimeout`/`BluetoothUnconfirmedCommand`. `Router._dispatch` (`router/base.py`) special-cases only `BluetoothUnconfirmedCommand` to skip its normal per-command failover and re-raise immediately - replaying an already-possibly-executed command risks double-execution. `BluetoothCommandFailed` carries no such risk (the command is proven not to have applied), so it falls through `Router`'s ordinary `except (Exception, TeslaFleetError)` clause and fails over like any other error - no `Router` code change was needed for this, only the exception type's placement outside the `BluetoothTimeout` hierarchy. A plain read (`_getVehicleSecurity`/`_getInfotainment`) still raises unadorned `BluetoothTimeout` on the same kind of wait timeout, since a read has no side effect to be unconfirmed about, and a *provably pre-submission* write failure still raises the unrelated `BluetoothTransportError` (see the write-delivery-certainty entry below - a submitted-then-ambiguous write is not this case). - **Write-delivery certainty splits `BluetoothTransportError` from `BluetoothTimeout` at the GATT write in `_send`**: `write_gatt_char` failures are not uniformly `BluetoothTransportError`. `BleakCharacteristicNotFoundError` (bleak resolves `WRITE_UUID` synchronously, before any backend I/O) is the only case provably pre-submission, so it alone stays `BluetoothTransportError` and is safe for `Router` to retry. Every other `BleakError`/`TimeoutError` from that call happens inside backend I/O (D-Bus/CoreBluetooth/an ESPHome proxy) where delivery can't be proven either way - field data measured 2 of 3 such write failures had already executed on the vehicle - so `_send` instead races any already-armed broadcast watcher for the rest of the window (a matching broadcast can still confirm success despite the failed write) and, failing that, raises plain `BluetoothTimeout`. Because `BluetoothUnconfirmedCommand` subclasses `BluetoothTimeout`, this lands in the exact same `except BluetoothTimeout` ladder in `_sendVehicleSecurity`/`_sendInfotainment` as a lost post-write ack, with no separate exception type needed; `_send_optimistic` gets the equivalent treatment explicitly since it bypasses that ladder (see above). A read is unaffected - `_getVehicleSecurity`/`_getInfotainment` don't override the ladder, so a write-ambiguous read propagates plain `BluetoothTimeout` and `Router` fails over on it normally, which is safe since a read has no double-execution risk. Tests: `tests/test_ble_send_transport.py` (`SendTransportErrorTests`), `tests/test_ble_broadcast_confirmation.py` (`WriteFailureBroadcastRaceTests`), `tests/test_ble_write_timeout_router.py`. - **`wake_up()` is best-effort; confirm readiness with an INFO read**: `wake_up()` is a VCSEC actuation, so a terminal ack now returns promptly when observed, but an unresolved wake remains only an inconclusive wake signal, not command failure (`BluetoothUnconfirmedCommand` when `raise_unconfirmed=True`, best-effort success by default). Confirm readiness by retrying a cheap INFO read instead (see the boot-delay gotcha above for why the first INFO read still needs its own retry/backoff). Hold one connection across a whole batch of related commands rather than reconnecting between each - reconnecting costs ~123% more per operation with no demonstrated wake-preservation benefit from the connection alone. diff --git a/README.md b/README.md index defbe67..eb56a7c 100644 --- a/README.md +++ b/README.md @@ -187,6 +187,11 @@ failures (including `bleak.exc.BleakError` and builtin `TimeoutError` from ESPHome proxies) and response-wait timeouts through the same library error hierarchy. +`VehicleBluetooth` can also register persistent BLE broadcast listeners for +unsolicited VCSEC `VehicleStatus` updates. Use typed `listen_*` helpers for the +decoded vehicle-status fields, or `listen_broadcast(domain, callback)` for raw +per-domain broadcast messages. + ### Routing and Failover The `Router` class composes an ordered list of two-or-more backends that share a common method surface and dispatches each method call down the chain, automatically failing over on most errors. `VehicleRouter` and `EnergySiteRouter` are thin entity-specific subclasses. A common setup is a local `VehicleBluetooth` primary with a cloud fallback (e.g. a `TeslemetryVehicle`), so commands go over Bluetooth when the vehicle is reachable and route to the cloud otherwise: diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index 7756857..77af9db 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -519,8 +519,9 @@ gives no presence tracking for a scalar enum field. Anything not decoded into `VehicleStatus` - other VCSEC broadcast payloads (`CommandStatus`, whitelist events, faults) and any future -infotainment-domain broadcast - falls back to the untyped `listen_broadcast`, -which delivers the raw `RoutableMessage` for a given `Domain`: +infotainment-domain broadcast - has no typed listener surface. Use the untyped +`listen_broadcast`, which delivers every raw unsolicited `RoutableMessage` for +a given `Domain`, including decoded `VehicleStatus` broadcasts: ```python from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain @@ -531,6 +532,8 @@ unsubscribe = vehicle.listen_broadcast(Domain.DOMAIN_VEHICLE_SECURITY, print) Listeners are plain Python callables registered on the `VehicleBluetooth` instance - they persist across reconnects and are torn down only by calling the `unsubscribe()` closure returned at registration. +Callback exceptions are logged and do not stop later listeners or normal +message routing; `KeyboardInterrupt` and `SystemExit` still propagate. ## Media Commands diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 349b55a..d3009ef 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -385,6 +385,12 @@ class VehicleBluetooth( per-command table (one source of truth), just applied to a broadcast frame or a follow-up read respectively. + The same unsolicited broadcast stream also feeds persistent listeners + inherited from ``BroadcastListeners``: typed ``listen_*`` methods for + decoded ``VehicleStatus`` fields and ``listen_broadcast`` for raw + per-domain broadcast messages. These listeners are local callables, persist + across reconnects, and are removed by the returned unsubscribe closure. + ``raise_unconfirmed`` (default ``False``) is the orthogonal question of what to do once a ladder genuinely cannot resolve - the ack/broadcast wait (and, under ``"verify"``, the prover read) neither confirmed nor diff --git a/tesla_fleet_api/tesla/vehicle/broadcast.py b/tesla_fleet_api/tesla/vehicle/broadcast.py index 433442d..82a0a7b 100644 --- a/tesla_fleet_api/tesla/vehicle/broadcast.py +++ b/tesla_fleet_api/tesla/vehicle/broadcast.py @@ -4,10 +4,11 @@ subscription used for command replies (``_on_message`` in ``bluetooth.py``). Every leaf field of ``VehicleStatus`` is a well-defined protobuf enum or int, so each gets its own typed ``listen_`` method here, mirroring -python-teslemetry-stream's per-field listener surface. Anything not decoded -into ``VehicleStatus`` - other VCSEC broadcast payloads (``CommandStatus``, -whitelist events, faults) and any future infotainment-domain broadcast - -falls back to the untyped ``listen_broadcast``. +python-teslemetry-stream's per-field listener surface. The untyped +``listen_broadcast`` receives every raw broadcast for a domain, including +``VehicleStatus`` and payloads that have no typed listener surface: other VCSEC +broadcast payloads (``CommandStatus``, whitelist events, faults) and any future +infotainment-domain broadcast. """ from __future__ import annotations @@ -84,9 +85,8 @@ def listen_broadcast( ) -> Unsubscribe: """Listen for any unsolicited broadcast on ``domain``. - Catch-all for broadcasts the typed listeners below don't cover: - non-``VehicleStatus`` VCSEC payloads (``CommandStatus``, whitelist - events, faults) and any infotainment-domain broadcast. + This is a catch-all for broadcasts the typed listeners below don't + cover, but it also receives typed ``VehicleStatus`` broadcasts. """ return self._register(self._domain_listeners.setdefault(domain, []), callback)