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

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
- **`ReassemblingBuffer` resets on a >1s inter-chunk gap, not just on decode failure**: `bluetooth.py`'s `ReassemblingBuffer.receive_data` discards any in-progress partial frame if the next chunk arrives more than `STALE_CHUNK_TIMEOUT` (1s) after the previous one, mirroring Tesla's official Go SDK (`teslamotors/vehicle-command`, `pkg/connector/ble/ble.go`'s `rxTimeout`). Without this, a chunk dropped mid-message left a stale partial in the buffer that got prepended to the next message, corrupting it until a lucky decode failure resynced. This is a frame-integrity hardening, not a fix for the separate ack-loss behavior documented above (that's a stalled/silent link, which no buffer-side reset can recover).
- **`pair()` confirms whitelisting two ways: one-shot reply OR verify-by-state poll**: the whitelist-op success is a single VCSEC frame, and it lands on a dead session (lost forever) if the BLE link cycles while the user walks to the car to approve - live-observed on HA/macOS CoreBluetooth, where `tesla_fleet_api` logged "Reconnecting to S..." every ~100s during a pending pair, and the plain one-shot wait hung despite car-side completion. `pair()` (`bluetooth.py`) keeps the reply as the fast path (waits one `poll_interval` for it) but, on a lost reply, polls `_pair_probe()` every `poll_interval` until an overall `timeout` (default 300s) elapses. The probe is a VCSEC `_handshake` with our own public key: it succeeds only once the key is whitelisted and faults `NotOnWhitelistFault` until then - `_pair_probe` maps any `TeslaFleetError` (incl. transport failures from a mid-wait reconnect) to "not yet", so polling survives reconnects. The whitelist op is written **exactly once** - never re-sent - because a re-send re-prompts the user (and the retry/double-execute hazard documented above applies). Deadline with neither path confirming raises a typed `BluetoothTimeout`. Default behavior, no new knob (`poll_interval` is a defaulted param).
- **Idle BLE keepalive (`keepalive_interval`)**: an idle held BLE link to the vehicle drops at ~42s mean lifetime (link supervision timeout ~720ms underneath, bluetoothd-verified on macOS CoreBluetooth); a single trivial passive GATT read every ~20s extends lifetime ~10x (~400s observed). `VehicleBluetooth.__init__`'s `keepalive_interval` (default `DEFAULT_KEEPALIVE_INTERVAL` = 20.0, `None`/`0` disables; threaded through `Vehicles`/`VehiclesBluetooth.create*`) starts one asyncio task per connection (`_keepalive_loop`, `bluetooth.py`) that reads `VERSION_UUID` only after `keepalive_interval` seconds of genuine GATT idleness. **Idle-triggered, not periodic**: `_last_activity` is bumped on every `_send` write and every `_on_notify` frame, so an active session never gets extra traffic; the loop recomputes the wait each pass and fires only when idle. The read is **bounded** (`_keepalive_timeout`, 2s) and **best-effort** - a prior un-timed RSSI read hung indefinitely against a sleeping car, so every attempt carries a timeout and swallows all failures (`_keepalive_read` catches `Exception`, never `CancelledError`); a failed keepalive never raises into user code, never triggers reconnect (the existing `connect_if_needed` machinery owns recovery), and never wakes the car. Task lifecycle is tied to the connection: started at the end of `connect()` (after `start_notify`), cancelled-and-awaited in `disconnect()` and restarted cleanly on reconnect (`_start_keepalive`/`_stop_keepalive`). **Sleep tradeoff**: these reads keep an *awake* car awake and defer vehicle sleep - consumers wanting the car to sleep should disable keepalive or disconnect when idle. Tests in `tests/test_ble_keepalive.py`.
- **Cross-transport parity (cloud REST `VehicleFleet` vs BLE `Commands`)**: the same-named command on both paths should build a semantically equivalent instruction from identical args - a divergence there is a bug, but response *bodies* legitimately differ (REST JSON dict vs decoded protobuf) and are not. `tests/test_cross_transport_parity.py` locks the equivalence in with mocked-both-transports tests. Known **non-bug FORM differences** (do not "fix"): `set_scheduled_departure`'s `preconditioning_enabled`/`off_peak_charging_enabled` (no proto fields), `window_control` lat/lon and `navigation_sc_request` `id` (no proto fields), `navigation_request`'s `type`/`locale`/`timestamp_ms` (REST share-intent framing), and `media_volume_up` (no Tesla REST endpoint - BLE-only; cloud raises volume via `adjust_volume`). One **open divergence left unfixed** pending live verification: `clear_pin_to_drive_admin` builds `DrivingClearSpeedLimitPinAction` (speed-limit PIN, not PIN-to-Drive - suspected mismapping, security-sensitive). `navigation_gps_request`'s prior `order` signature mismatch is resolved: both transports now default `order` to `0` (`REMOTE_NAV_TRIP_ORDER_UNKNOWN`, a defined proto enum value) when the caller omits it, so cloud always sends an explicit `order` rather than `null`.
- **Cross-transport parity (cloud REST `VehicleFleet` vs BLE `Commands`)**: the same-named command on both paths should build a semantically equivalent instruction from identical args - a divergence there is a bug, but response *bodies* legitimately differ (REST JSON dict vs decoded protobuf) and are not. `tests/test_cross_transport_parity.py` locks the equivalence in with mocked-both-transports tests. Known **non-bug FORM differences** (do not "fix"): `set_scheduled_departure`'s `preconditioning_enabled`/`off_peak_charging_enabled` (no proto fields), `window_control` lat/lon and `navigation_sc_request` `id` (no proto fields), `navigation_request`'s `type`/`locale`/`timestamp_ms` (REST share-intent framing), and `media_volume_up` (no Tesla REST endpoint - BLE-only; cloud raises volume via `adjust_volume`), and (newly accepted) `clear_pin_to_drive_admin`'s `pin` param (no proto field on `VehicleControlResetPinToDriveAdminAction` - cloud still sends it in the REST body, BLE ignores it). `clear_pin_to_drive_admin`'s prior mismapping is fixed and live-verified: it now builds `VehicleControlResetPinToDriveAdminAction` (delegating to `reset_pin_to_drive_admin`), not the wrong `DrivingClearSpeedLimitPinAction` (speed-limit PIN, a different feature) it built before - the vehicle itself rejected a live call to the old build with reason `speed_limit_mode_active`, meaningful only to Speed Limit Mode, confirming the mismapping. `navigation_gps_request`'s prior `order` signature mismatch is resolved: both transports now default `order` to `0` (`REMOTE_NAV_TRIP_ORDER_UNKNOWN`, a defined proto enum value) when the caller omits it, so cloud always sends an explicit `order` rather than `null`.
- **Per-command debug logging chokepoints and the `command=` name it derives**: `LOGGER.debug` lines of the form `command=<name> transport=<t> result=...` are emitted from exactly four places, not per-method - `Commands._sendVehicleSecurity`/`_getVehicleSecurity`/`_sendInfotainment`/`_getInfotainment` (`commands.py`, covers both BLE and Fleet-signed) and `TeslaFleetApi._request` (`fleet.py`, covers Fleet/Teslemetry/Tessie REST). `transport` comes from a `_transport_name` `ClassVar` set per concrete class (`"bluetooth"`/`"fleet"`/`"teslemetry"`/`"tessie"`), mirroring the existing `_auth_method` pattern - add that ClassVar to any new `Commands`/`TeslaFleetApi` subclass. For BLE/Fleet-signed, `command` is **not** the Python method name; it's derived from the populated protobuf oneof field (`vcsec_command_name`/`infotainment_command_name` in `commands.py`), e.g. `door_lock()` logs as `RKE_ACTION_LOCK` and `set_charge_limit()` as `chargingSetLimitAction` - deliberately robust to call-site changes since it reads the message being sent, not the call stack. `VehicleBluetooth`'s `verify_commands` resolution logs a second, separate line (`verify_commands=resolved`/`unresolved`) rather than duplicating the base class's raw-attempt line. `Router._dispatch` (`router/base.py`) logs `command=... backend=<ClassName> result=...` per backend tried, independent of the above. See `docs/bluetooth_vehicles.md`'s "Troubleshooting: Enable Debug Logging" section for the user-facing format; `tests/test_command_logging.py` locks in the exact line shapes.
- **`_log_request_result` (`fleet.py`) must tolerate any JSON-legal REST body, not just dicts**: it runs after the HTTP request already succeeded, so it's a logging convenience only - a non-dict body (`null`, a list, a bare scalar; live case: Teslemetry's `list_authorized_clients` returning `null`) must never raise there. It guards with `isinstance(data, dict)` before calling `.get()`, logging `result=success` and returning for anything else. Regression tests in `tests/test_command_logging.py` (`test_null_json_body_returns_none_without_raising` etc.) cover null/list/scalar bodies.
- **Typed accessor pattern for undocumented raw-dict responses**: `TeslemetryEnergySite.find_authorized_clients()` (`teslemetry/energysite.py`) is the library's first frozen-dataclass typed wrapper over a raw `dict[str, Any]`/`None`/`list` REST response, added so API-parsing logic (envelope unwrap, field lookup, shape validation, enum typing) lives in the library instead of each consumer (e.g. Home Assistant's config flow) reimplementing it. `_authorized_clients_list()` does one precise, non-recursive envelope unwrap (`{"response": {"authorized_clients": [...]}}` or `{"response": {"clients": [...]}}`, or a bare list with no envelope) rather than a multi-key/depth-first search - the `clients` key variant was added after a live capture (Tesla Release 953) showed the endpoint's real key differs from the originally-documented `authorized_clients`, confirmed against a populated 5-entry sample. `AuthorizedClient` models only the two fields a pairing flow actually reads (`public_key`, `state`), each accepting only the specific key-name variant pairs empirically evidenced, not speculative extras. Two rules any future typed accessor over an undocumented response shape must keep, demonstrated here even though `AuthorizedClientState` (`const.py`) has no falsy member: (1) field lookup must check key presence (`key in payload`), never `payload.get(key) or default` - a legal falsy value is not "missing", and a present-but-unrecognized enum value (`_normalize_state()`) is returned raw rather than coerced to `None`; (2) a `None` body and an unrecognized response shape (not a dict/list, or an envelope that unwraps to neither accepted list key) are malformed data, not "zero clients" - Tesla's endpoint intermittently returns HTTP 200 with a null body, so `_authorized_clients_list()` raises `InvalidResponse` (`exceptions.py`) for both rather than collapsing them to `[]`; only a genuinely empty list under either accepted key parses to `AuthorizedClients.clients == []` without raising. Tesla has not published an OpenAPI schema for pairing endpoints, so `const.py`'s enums are the schema of record; widen `AuthorizedClient`'s modeled fields only against a further live sample, not speculatively. The raw `list_authorized_clients()` method is kept alongside as the untyped escape hatch and still returns a null body unchanged rather than raising. Tests: `tests/test_teslemetry_authorized_clients.py`.
Expand Down
2 changes: 1 addition & 1 deletion tesla_fleet_api.egg-info/PKG-INFO
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: tesla_fleet_api
Version: 1.7.4
Version: 1.7.6
Summary: Tesla Fleet API library for Python
Author-email: Brett Adams <[email protected]>
License-Expression: Apache-2.0
Expand Down
1 change: 1 addition & 0 deletions tesla_fleet_api.egg-info/SOURCES.txt
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@ tests/test_ble_write_timeout_router.py
tests/test_command_counter_lock.py
tests/test_command_logging.py
tests/test_cross_transport_parity.py
tests/test_energysite_island_mode.py
tests/test_find_vehicle_scan_filter.py
tests/test_firmware_at_least.py
tests/test_fleet_auth_refresh.py
Expand Down
28 changes: 17 additions & 11 deletions tesla_fleet_api/tesla/vehicle/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -987,17 +987,23 @@ async def charge_stop(self) -> dict[str, Any]:
)
)

async def clear_pin_to_drive_admin(self, pin: str | None = None):
"""Deactivates PIN to Drive and resets the associated PIN for vehicles running firmware versions 2023.44+. This command is only accessible to fleet managers or owners."""
return await self._sendInfotainment(
Action(
vehicleAction=VehicleAction(
drivingClearSpeedLimitPinAction=DrivingClearSpeedLimitPinAction(
pin=pin
)
)
)
)
async def clear_pin_to_drive_admin(self, pin: str | None = None) -> dict[str, Any]:
"""Deactivates PIN to Drive and resets the associated PIN for vehicles running firmware versions 2023.44+. This command is only accessible to fleet managers or owners.

``pin`` is accepted for cross-transport signature parity with the
cloud REST endpoint, but the signed ``VehicleControlResetPinToDriveAdminAction``
has no pin field, so it is not sent (same pattern as other documented
cross-transport form gaps - see CLAUDE.md). Live-verified: the
previous implementation built ``DrivingClearSpeedLimitPinAction``,
the Speed-Limit-Mode pin clear, not a PIN-to-Drive action at all -
confirmed by the vehicle itself rejecting a live call with reason
``speed_limit_mode_active``, a condition meaningful only to Speed
Limit Mode. This action requires proof of Tesla account credentials
(fleet-manager/owner tier) - calling it over raw BLE signing (no
OAuth session) always raises ``TeslaFleetMessageFaultCommandRequiresAccountCredentials``;
use the Fleet-API-relayed ``VehicleSigned`` transport instead.
"""
return await self.reset_pin_to_drive_admin()

async def door_lock(self) -> dict[str, Any]:
"""Locks the vehicle."""
Expand Down
69 changes: 69 additions & 0 deletions tests/test_ble_mocked_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,75 @@ async def test_sends_hvac_temperature_action_and_decodes_ok_reply(self) -> None:
self.assertAlmostEqual(hvac.passenger_temp_celsius, 19.0)


class ClearPinToDriveAdminTests(MockedBleTransportTestCase):
"""``clear_pin_to_drive_admin`` must build the PIN-to-Drive admin reset,
not the Speed-Limit-Mode pin clear (live-verified mismapping, see
report history for capstone F1 / task tfa-ble-pin-clear-verify)."""

async def test_builds_reset_pin_to_drive_admin_action_not_speed_limit_action(
self,
) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_action_ok_reply()

result = await vehicle.clear_pin_to_drive_admin(pin="1234")

self.assertEqual(result, {"response": {"result": True, "reason": ""}})

send.assert_awaited_once()
await_args = send.await_args
assert await_args is not None
sent_msg = await_args.args[0]
self.assertEqual(sent_msg.to_destination.domain, Domain.DOMAIN_INFOTAINMENT)

plaintext = decrypt_sent_command(vehicle, sent_msg)
action = Action.FromString(plaintext)
self.assertEqual(
action.vehicleAction.WhichOneof("vehicle_action_msg"),
"vehicleControlResetPinToDriveAdminAction",
)

async def test_pin_argument_is_not_sent_on_the_wire(self) -> None:
"""``pin`` exists only for cross-transport signature parity with the
cloud REST endpoint - the signed action has no pin field."""
vehicle, send = self.make_vehicle()
send.return_value = infotainment_action_ok_reply()

await vehicle.clear_pin_to_drive_admin(pin="9999")

sent_msg = send.await_args.args[0]
plaintext = decrypt_sent_command(vehicle, sent_msg)
action = Action.FromString(plaintext)
# VehicleControlResetPinToDriveAdminAction is an empty message - no
# pin field exists to assert against, which is itself the point.
self.assertFalse(
action.vehicleAction.HasField("drivingClearSpeedLimitPinAction")
)


class SpeedLimitClearPinTests(MockedBleTransportTestCase):
"""Regression: ``speed_limit_clear_pin`` keeps building
``drivingClearSpeedLimitPinAction`` with the given pin - the action
``clear_pin_to_drive_admin`` was previously (incorrectly) reusing."""

async def test_sends_driving_clear_speed_limit_pin_action(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_action_ok_reply()

await vehicle.speed_limit_clear_pin(pin="1234")

sent_msg = send.await_args.args[0]
plaintext = decrypt_sent_command(vehicle, sent_msg)
action = Action.FromString(plaintext)
self.assertEqual(
action.vehicleAction.WhichOneof("vehicle_action_msg"),
"drivingClearSpeedLimitPinAction",
)
self.assertEqual(
action.vehicleAction.drivingClearSpeedLimitPinAction.pin, "1234"
)


class ChargeStateTypedReplyTests(MockedBleTransportTestCase):
"""``charge_state`` (INFO read, defined on VehicleBluetooth) decodes a typed reply."""

Expand Down
34 changes: 34 additions & 0 deletions tests/test_cross_transport_parity.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,40 @@ async def test_in_range_sends_absolute_volume_on_both_transports(self) -> None:
self.assertAlmostEqual(action.mediaUpdateVolume.volume_absolute_float, 5.0)


class ClearPinToDriveAdminParityTests(MockedBleTransportTestCase):
"""``clear_pin_to_drive_admin`` targets the same feature (PIN-to-Drive
admin reset) on both transports, though the wire forms now legitimately
differ - a documented, accepted gap, not a bug:

Fix (live-verified, see report history / task tfa-ble-pin-clear-verify):
BLE previously built ``DrivingClearSpeedLimitPinAction`` (the
Speed-Limit-Mode pin clear) instead of a PIN-to-Drive action at all -
confirmed by the vehicle rejecting a live call with reason
``speed_limit_mode_active``. BLE now builds
``VehicleControlResetPinToDriveAdminAction``, which has no pin field, so
``pin`` is accepted (cross-transport signature parity) but not sent.
Cloud's REST endpoint still takes ``pin`` in the body - that divergence
is real but expected (same pattern as ``set_scheduled_departure``'s dead
args - see CLAUDE.md).
"""

async def test_cloud_still_sends_pin_ble_ignores_it(self) -> None:
cloud, request = _make_fleet_vehicle(self.VIN)
await cloud.clear_pin_to_drive_admin(pin="1234")
assert request.await_args is not None
self.assertEqual(request.await_args.kwargs["json"], {"pin": "1234"})

ble, send = self.make_vehicle()
send.return_value = infotainment_action_ok_reply()
await ble.clear_pin_to_drive_admin(pin="1234")
action = _sent_vehicle_action(ble, send)
self.assertEqual(
action.WhichOneof("vehicle_action_msg"),
"vehicleControlResetPinToDriveAdminAction",
)
self.assertFalse(action.HasField("drivingClearSpeedLimitPinAction"))


class NavigationGpsRequestParityTests(MockedBleTransportTestCase):
"""``navigation_gps_request``'s ``order`` must default identically on
both transports.
Expand Down
2 changes: 1 addition & 1 deletion uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading