diff --git a/AGENTS.md b/AGENTS.md index 62bae66..c570d36 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -127,6 +127,8 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te - **`charge_standard()` rejects `already_standard`**: live-verified - calling it while `charge_state().charge_limit_soc` already equals `charge_limit_soc_std` gets `{"result": False, "reason": "already_standard"}` rather than a no-op success. Not a library bug; callers/tests exercising this command need the limit to actually differ from the std preset first (e.g. via `charge_max_range()` or `set_charge_limit()`). - **BLE media state-observability gotcha**: `MediaState.now_playing_artist/title` and all of `MediaDetailState` (`now_playing_album/station/source_string/elapsed/duration`) were observed empty/zero on the test car while Spotify was actively `Playing` at nonzero volume - these legacy fields are apparently only populated for certain sources (e.g. USB/Bluetooth), not Spotify. Don't assume `media_next_track`/`media_prev_track`/`media_next_fav`/`media_prev_fav` are state-observable via these readers; verify by ACK (`{"result": True}`) and pair with the inverse command when the fingerprint doesn't change. `audio_volume`/`media_playback_status` (for `adjust_volume`/`media_volume_up`/`media_volume_down`/`media_toggle_playback`) were populated correctly and are reliable provers. - **BLE write-vs-read reliability asymmetry**: on the test rig, plain GATT reads (state readers, `wake_up()`) succeeded consistently, but every signed-command *write* (e.g. `adjust_volume`) reliably hit `aioesphomeapi.core.TimeoutAPIError` waiting for `BluetoothGATTWriteResponse` (or, once, GATT error 133) across 5 separate fresh connections in one session. Reads never showed the same failure. A client-side write timeout did not leave the car mutated (re-read after each failure showed the pre-command value unchanged), so treat it as the write never landing rather than a lost response to an applied change - but always re-read to confirm before assuming so. Root cause not identified; worth a dedicated transport probe before the next live-actuation attempt on this rig. +- **`wake_up()`'s own ACK is an unreliable signal - almost always a false-negative timeout**: live-verified 9/9 across two independent sessions - `wake_up()` raises `BluetoothTimeout` regardless of whether the vehicle was asleep, already awake, freshly connected, or held open for minutes, yet the wake (or no-op) reliably took effect - an immediate follow-up state read on the same connection succeeds. Never treat a `wake_up()` `BluetoothTimeout` as command failure; call it best-effort (catch and ignore) and 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. +- **`navigation_gps_request`'s `order` param is a raw int, not a callable enum**: `commands.py` used to build it as `NavigationGpsRequest.RemoteNavTripOrder(order)`, treating the protobuf nested-enum wrapper (`EnumTypeWrapper`) as if it were a callable Python `enum.IntEnum` class - it isn't, so every call raised `TypeError` before any message was sent (found live during PR-8; this method had never been exercised over BLE before). Fixed to pass `order=order` directly (matching the working sibling `navigation_gps_destination_request`), which protobuf accepts as a bare int for an enum field at runtime. ## Maintaining this file diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index f9e6242..9fc0c4a 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -65,22 +65,29 @@ You can wake up a `VehicleBluetooth` instance using the `wake_up` method. Here's ```python import asyncio from tesla_fleet_api import TeslaBluetooth +from tesla_fleet_api.exceptions import BluetoothTimeout async def main(): tesla_bluetooth = TeslaBluetooth() device = await tesla_bluetooth.find_vehicle() private_key = await tesla_bluetooth.get_private_key("path/to/private_key.pem") vehicle = tesla_bluetooth.vehicles.create("") - await vehicle.wake_up() - print(f"Woke up VehicleBluetooth instance for VIN: {vehicle.vin}") + try: + await vehicle.wake_up() + except BluetoothTimeout: + pass + print(f"Sent wake request to VehicleBluetooth instance for VIN: {vehicle.vin}") asyncio.run(main()) ``` -`wake_up()` returns when the vehicle-security computer acknowledges the wake -request. The infotainment computer can take longer to become ready, so -INFO-domain reads immediately after waking, such as `charge_state()` or -`vehicle_data()`, should retry `BluetoothTimeout` with backoff. +`wake_up()` is best-effort over BLE: a `BluetoothTimeout` from the wake request +can be a false negative even when the vehicle wakes successfully. Confirm +readiness by retrying a cheap INFO-domain read, such as `charge_state()`, with +backoff. The infotainment computer can also take longer to become ready than +the vehicle-security computer, so INFO-domain reads immediately after waking +should retry `BluetoothTimeout` with backoff. Keep one BLE connection open +across related commands when possible instead of reconnecting for each command. ## Climate Commands @@ -138,6 +145,27 @@ command with `already_standard`. Avoid actuating `charge_port_door_open()` or `charge_port_door_close()` against a plugged-in vehicle unless someone can reseat the cable if needed. +## Navigation, Dashcam, and Power Commands + +`VehicleBluetooth` can send the signed navigation, dashcam, and power-mode +commands over BLE: + +- `navigation_request(value)` +- `navigation_gps_request(lat, lon, order)` +- `navigation_sc_request(order)` +- `navigation_waypoints_request(waypoints)` +- `navigation_gps_destination_request(lat, lon, destination, order)` +- `dashcam_save_clip()` +- `flash_lights()` +- `set_keep_accessory_power_mode(on)` +- `set_low_power_mode(on)` + +For the GPS navigation methods, `order` is the Tesla/protobuf remote-nav order +integer: `1` replaces the trip, `2` prepends a stop, and `3` appends a stop. +These commands are ACK-only over BLE; the library returns the vehicle's command +acknowledgement, but there is no separate BLE state prover for the navigation +destination, dashcam clip save, flash-lights action, or power-mode toggle. + ## Open/Close Individual Doors (Bluetooth Only) The individual door closure commands are Bluetooth-only and are not available via Fleet API signed commands. diff --git a/docs/fleet_api_signed_commands.md b/docs/fleet_api_signed_commands.md index c91e125..3daf444 100644 --- a/docs/fleet_api_signed_commands.md +++ b/docs/fleet_api_signed_commands.md @@ -208,6 +208,20 @@ async def main(): asyncio.run(main()) ``` +## Navigation and Dashcam Commands + +Signed commands include the navigation and dashcam command methods: + +- `navigation_request(value)` +- `navigation_gps_request(lat, lon, order)` +- `navigation_sc_request(order)` +- `navigation_waypoints_request(waypoints)` +- `navigation_gps_destination_request(lat, lon, destination, order)` +- `dashcam_save_clip()` + +For the GPS navigation methods, `order` is the Tesla/protobuf remote-nav order +integer: `1` replaces the trip, `2` prepends a stop, and `3` appends a stop. + ## Honk Horn You can honk the horn of a specific vehicle using its VIN: diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 78d7272..0712ea8 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -542,6 +542,9 @@ async def pair( async def wake_up(self): """Wake up the vehicle security computer. + A ``BluetoothTimeout`` from this command can be a false negative even + when the vehicle wakes successfully, so callers should treat wake as + best-effort and confirm readiness with a retried INFO-domain read. The infotainment computer may still need a short delay before it can complete signed-command handshakes, so callers that issue INFO-domain reads immediately after waking should retry ``BluetoothTimeout`` with diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 42ffaec..d193dd7 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -949,14 +949,18 @@ async def media_volume_up(self) -> dict[str, Any]: async def navigation_gps_request( self, lat: float, lon: float, order: int ) -> dict[str, Any]: - """Start navigation to given coordinates. Order can be used to specify order of multiple stops.""" + """Start navigation to coordinates. + + ``order`` is the Tesla/protobuf remote-nav order integer: 1 replaces + the trip, 2 prepends a stop, and 3 appends a stop. + """ return await self._sendInfotainment( Action( vehicleAction=VehicleAction( navigationGpsRequest=NavigationGpsRequest( lat=lat, lon=lon, - order=NavigationGpsRequest.RemoteNavTripOrder(order), + order=order, # pyright: ignore[reportArgumentType] ) ) ) @@ -2166,7 +2170,11 @@ async def get_charge_on_solar(self) -> dict[str, Any]: async def navigation_gps_destination_request( self, lat: float, lon: float, destination: str, order: int ) -> dict[str, Any]: - """Navigates to a GPS destination with a named destination string.""" + """Navigate to coordinates with a named destination string. + + ``order`` is the Tesla/protobuf remote-nav order integer: 1 replaces + the trip, 2 prepends a stop, and 3 appends a stop. + """ return await self._sendInfotainment( Action( vehicleAction=VehicleAction( diff --git a/tesla_fleet_api/tesla/vehicle/fleet.py b/tesla_fleet_api/tesla/vehicle/fleet.py index 877ea8e..c958697 100644 --- a/tesla_fleet_api/tesla/vehicle/fleet.py +++ b/tesla_fleet_api/tesla/vehicle/fleet.py @@ -196,7 +196,11 @@ async def media_volume_down(self) -> dict[str, Any]: async def navigation_gps_request( self, lat: float, lon: float, order: int | None = None ) -> dict[str, Any]: - """Start navigation to given coordinates. Order can be used to specify order of multiple stops.""" + """Start navigation to coordinates. + + ``order`` is the Tesla remote-nav order integer: 1 replaces the trip, + 2 prepends a stop, and 3 appends a stop. + """ return await self._request( Method.POST, f"api/1/vehicles/{self.vin}/command/navigation_gps_request", diff --git a/tests/test_ble_nav_misc_commands.py b/tests/test_ble_nav_misc_commands.py new file mode 100644 index 0000000..f413d94 --- /dev/null +++ b/tests/test_ble_nav_misc_commands.py @@ -0,0 +1,192 @@ +"""Nav/dashcam/power-mode command group (PR-8) over the mocked BLE transport. + +Live-verified against a test car over a local BLE proxy in a single held BLE +connection using the best-effort-wake + poll-confirm strategy (a ``wake_up()`` +``BluetoothTimeout`` is a known false negative - see AGENTS.md): +``navigation_request``, ``navigation_gps_request``, +``navigation_sc_request``, ``navigation_waypoints_request``, +``navigation_gps_destination_request``, ``dashcam_save_clip``, +``flash_lights``, ``set_keep_accessory_power_mode`` (True then False), +``set_low_power_mode`` (True then False) - all 11 calls ACKed +``{"result": True}``. None of these have a BLE state prover (momentary or +unobservable), so live "verify" is ACK-only, per the master plan; the nav +commands set a harmless destination and need no restore, and the two +power-mode commands were toggled back off to leave no lasting state change. + +Live-discovered bug (fixed in this PR, not a test-only find): +``navigation_gps_request`` called ``NavigationGpsRequest.RemoteNavTripOrder(order)``, +treating the protobuf nested-enum wrapper as if it were a callable Python +enum class - ``EnumTypeWrapper`` isn't callable, so every live call raised +``TypeError`` before a message was ever sent. Fixed to pass the raw int +``order`` directly, matching the working sibling +``navigation_gps_destination_request``. +""" + +from tesla_fleet_api.tesla.vehicle.proto.car_server_pb2 import ( + Action, + NavigationGpsDestinationRequest, + NavigationGpsRequest, +) +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain + +from ble_mocked_transport import ( + MockedBleTransportTestCase, + decrypt_sent_command, + infotainment_action_ok_reply, +) + +REPLACE_GPS = NavigationGpsRequest.RemoteNavTripOrder.REMOTE_NAV_TRIP_ORDER_REPLACE +REPLACE_GPS_DEST = ( + NavigationGpsDestinationRequest.RemoteNavTripOrder.REMOTE_NAV_TRIP_ORDER_REPLACE +) + + +def _decode_vehicle_action(vehicle, sent_msg): + plaintext = decrypt_sent_command(vehicle, sent_msg) + action = Action.FromString(plaintext) + assert sent_msg.to_destination.domain == Domain.DOMAIN_INFOTAINMENT + assert sent_msg.signature_data.HasField("AES_GCM_Personalized_data") + return action.vehicleAction + + +class NavigationRequestTests(MockedBleTransportTestCase): + async def test_sends_destination_string(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.navigation_request("1 Infinite Loop, Cupertino, CA") + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.navigationRequest.destination, + "1 Infinite Loop, Cupertino, CA", + ) + + +class NavigationGpsRequestTests(MockedBleTransportTestCase): + """Regression test for the ``EnumTypeWrapper`` order bug found live.""" + + async def test_sends_lat_lon_and_order(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.navigation_gps_request(37.3230, -122.0322, REPLACE_GPS) + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + gps = vehicle_action.navigationGpsRequest + self.assertAlmostEqual(gps.lat, 37.3230) + self.assertAlmostEqual(gps.lon, -122.0322) + self.assertEqual(gps.order, REPLACE_GPS) + + +class NavigationScRequestTests(MockedBleTransportTestCase): + async def test_sends_supercharger_order(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.navigation_sc_request(1) + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual(vehicle_action.navigationSuperchargerRequest.order, 1) + + +class NavigationWaypointsRequestTests(MockedBleTransportTestCase): + async def test_sends_waypoints_string(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + waypoints = "37.3230,-122.0322;37.4419,-122.1430" + result = await vehicle.navigation_waypoints_request(waypoints) + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual(vehicle_action.navigationWaypointsRequest.waypoints, waypoints) + + +class NavigationGpsDestinationRequestTests(MockedBleTransportTestCase): + async def test_sends_lat_lon_destination_and_order(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.navigation_gps_destination_request( + 37.4419, -122.1430, "Palo Alto, CA", REPLACE_GPS_DEST + ) + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + dest = vehicle_action.navigationGpsDestinationRequest + self.assertAlmostEqual(dest.lat, 37.4419) + self.assertAlmostEqual(dest.lon, -122.1430) + self.assertEqual(dest.destination, "Palo Alto, CA") + self.assertEqual(dest.order, REPLACE_GPS_DEST) + + +class DashcamSaveClipTests(MockedBleTransportTestCase): + async def test_sends_dashcam_save_clip_action(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.dashcam_save_clip() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("dashcamSaveClipAction")) + + +class FlashLightsTests(MockedBleTransportTestCase): + async def test_sends_flash_lights_action(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.flash_lights() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("vehicleControlFlashLightsAction")) + + +class SetKeepAccessoryPowerModeTests(MockedBleTransportTestCase): + async def test_on_sends_keep_accessory_power_mode_true(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_keep_accessory_power_mode(True) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue( + vehicle_action.setKeepAccessoryPowerModeAction.keep_accessory_power_mode + ) + + async def test_off_sends_keep_accessory_power_mode_false(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_keep_accessory_power_mode(False) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertFalse( + vehicle_action.setKeepAccessoryPowerModeAction.keep_accessory_power_mode + ) + + +class SetLowPowerModeTests(MockedBleTransportTestCase): + async def test_on_sends_low_power_mode_true(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_low_power_mode(True) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.setLowPowerModeAction.low_power_mode) + + async def test_off_sends_low_power_mode_false(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_low_power_mode(False) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertFalse(vehicle_action.setLowPowerModeAction.low_power_mode)