diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index 9404400..1d94f50 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -470,6 +470,9 @@ Available BLE state readers: - `light_show_state()` - `suspension_state()` - `child_presence_detection_state()` +- `vehicle_image_state()` (paged image bytes; see + [Vehicle Image State](fleet_api_signed_commands.md#vehicle-image-state) for + its shared BLE/Fleet signed-command contract) For explicit composite requests, use `vehicle_data(endpoints)` with `BluetoothVehicleData` values: diff --git a/docs/fleet_api_signed_commands.md b/docs/fleet_api_signed_commands.md index c36121c..cf357e9 100644 --- a/docs/fleet_api_signed_commands.md +++ b/docs/fleet_api_signed_commands.md @@ -190,6 +190,24 @@ state after the call instead of relying on the number of send attempts. `adjust_volume(volume)` accepts absolute volume values from `0.0` through `11.0`, matching the Fleet API command validation. +## Vehicle Image State + +`vehicle_image_state(image_type, *, chunk_size=16384)` retrieves a rendered +vehicle image over the signed-command channel and returns its assembled +`bytes`. It is available on both `VehicleSigned` and `VehicleBluetooth`. + +```python +from tesla_fleet_api.const import VehicleImageType + +image = await vehicle.vehicle_image_state( + VehicleImageType.AUTOPILOT_VISUALIZATION_WRAP +) +``` + +Use `VehicleImageType.LICENSE_PLATE` for the license-plate placement image. +The method first reads the image's declared size, then requests and reassembles +the data in `chunk_size` pages. `chunk_size` must be a positive integer. + ## Troubleshooting: Debug Logging Enable the `tesla_fleet_api` logger at `DEBUG` to log each signed command's diff --git a/tesla_fleet_api.egg-info/SOURCES.txt b/tesla_fleet_api.egg-info/SOURCES.txt index c1defc6..73f22d8 100644 --- a/tesla_fleet_api.egg-info/SOURCES.txt +++ b/tesla_fleet_api.egg-info/SOURCES.txt @@ -32,6 +32,17 @@ tesla_fleet_api/tesla/vehicle/fleet.py tesla_fleet_api/tesla/vehicle/signed.py tesla_fleet_api/tesla/vehicle/vehicle.py tesla_fleet_api/tesla/vehicle/vehicles.py +tesla_fleet_api/tesla/vehicle/proto/__init__.py +tesla_fleet_api/tesla/vehicle/proto/car_server_pb2.py +tesla_fleet_api/tesla/vehicle/proto/common_pb2.py +tesla_fleet_api/tesla/vehicle/proto/errors_pb2.py +tesla_fleet_api/tesla/vehicle/proto/keys_pb2.py +tesla_fleet_api/tesla/vehicle/proto/managed_charging_pb2.py +tesla_fleet_api/tesla/vehicle/proto/session_pb2.py +tesla_fleet_api/tesla/vehicle/proto/signatures_pb2.py +tesla_fleet_api/tesla/vehicle/proto/universal_message_pb2.py +tesla_fleet_api/tesla/vehicle/proto/vcsec_pb2.py +tesla_fleet_api/tesla/vehicle/proto/vehicle_pb2.py tesla_fleet_api/teslemetry/__init__.py tesla_fleet_api/teslemetry/energysite.py tesla_fleet_api/teslemetry/teslemetry.py @@ -43,9 +54,12 @@ tests/test_auto_seat_climate.py tests/test_ble_broadcast_confirmation.py tests/test_ble_broadcast_listeners.py tests/test_ble_charging_commands.py +tests/test_ble_charging_utility_commands.py tests/test_ble_climate_commands.py tests/test_ble_command_verification.py tests/test_ble_confirmation_mode.py +tests/test_ble_connectivity_diagnostics_commands.py +tests/test_ble_destructive_commands.py tests/test_ble_expects_data.py tests/test_ble_keepalive.py tests/test_ble_message_routing.py @@ -53,11 +67,15 @@ tests/test_ble_mocked_closures_locks.py tests/test_ble_mocked_commands.py tests/test_ble_mocked_media_commands.py tests/test_ble_mocked_state_readers.py +tests/test_ble_mocked_state_readers_new.py +tests/test_ble_nav_messaging_media_commands.py tests/test_ble_nav_misc_commands.py +tests/test_ble_niche_commands.py tests/test_ble_optimistic_and_best_effort.py tests/test_ble_pair.py tests/test_ble_reassembling_buffer.py tests/test_ble_send_transport.py +tests/test_ble_ui_unit_preference_commands.py tests/test_ble_unconfirmed_command.py tests/test_ble_write_timeout_router.py tests/test_command_counter_lock.py @@ -68,8 +86,12 @@ tests/test_find_vehicle_scan_filter.py tests/test_firmware_at_least.py tests/test_fleet_auth_refresh.py tests/test_power_mode_commands.py +tests/test_proto_compatibility.py +tests/test_proto_coverage_lock.py tests/test_router.py +tests/test_session_info_authentication.py tests/test_tesla_private_key.py tests/test_teslemetry_authorized_clients.py tests/test_teslemetry_gateway_address.py -tests/test_tessie_vehicle_params.py \ No newline at end of file +tests/test_tessie_vehicle_params.py +tests/test_vehicle_image_state.py \ No newline at end of file diff --git a/tesla_fleet_api/const.py b/tesla_fleet_api/const.py index cc275d5..c296774 100644 --- a/tesla_fleet_api/const.py +++ b/tesla_fleet_api/const.py @@ -76,6 +76,16 @@ class AutoSeat(IntEnum): FRONT_RIGHT = 2 +class VehicleImageType(IntEnum): + """Rendered vehicle image kinds available from ``vehicle_image_state()``. + + Values match the proto ``VehicleImageStateType`` enum (``APVIZ_*`` names). + """ + + AUTOPILOT_VISUALIZATION_WRAP = 1 + LICENSE_PLATE = 2 + + class Level(IntEnum): """Level options""" diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 7257ff0..c82dd72 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -52,6 +52,7 @@ TirePressureUnit, EnergyDisplayFormat, PhoneFontSize, + VehicleImageType, ) # Protocol @@ -60,7 +61,11 @@ AutoStwHeatAction, BoomboxAction, DrivingClearSpeedLimitPinAdminAction, + GetVehicleData, + GetVehicleImageState, Response, + VehicleImageDataChunkRequest, + VehicleImageRequest, ) from tesla_protocol.command.signatures_pb2 import ( Session_Info_Status, @@ -980,6 +985,99 @@ async def _getInfotainment(self, command: Action) -> VehicleData: _log_command_result(name, self._transport_name, reply) return reply["response"] + async def vehicle_image_state( + self, + image_type: VehicleImageType | int, + *, + chunk_size: int = 16384, + ) -> bytes: + """Fetch a rendered vehicle image (e.g. AP visualization wrap or + license-plate placement) over the signed-command channel and return + its assembled bytes. + + Unlike every other ``GetVehicleData`` sub-state, image data is + transferred in paged chunks rather than a single reply: this first + requests the image's metadata (an ``ID`` request, which reports its + total size), then issues ``DATA`` requests advancing by + ``chunk_size`` until the full byte range has been retrieved, + reassembling the chunks in order. Works over both the BLE and + Fleet API signed transports since it only relies on + ``_getInfotainment``. + """ + if ( + not isinstance(chunk_size, int) # pyright: ignore[reportUnnecessaryIsInstance] + or isinstance(chunk_size, bool) + or chunk_size <= 0 + ): + raise ValueError("chunk_size must be a positive integer") + + id_reply = await self._getInfotainment( + Action( + vehicleAction=VehicleAction( + getVehicleData=GetVehicleData( + getVehicleImageState=GetVehicleImageState( + imageRequests=[ + VehicleImageRequest( + dataType=VehicleImageRequest.Type.ID, + imageType=image_type, # pyright: ignore[reportArgumentType] + ) + ] + ) + ) + ) + ) + ) + total_size = id_reply.vehicle_image_state.vehicle_images[0].total_image_size + + data = bytearray() + offset = 0 + while offset < total_size: + chunk_reply = await self._getInfotainment( + Action( + vehicleAction=VehicleAction( + getVehicleData=GetVehicleData( + getVehicleImageState=GetVehicleImageState( + imageRequests=[ + VehicleImageRequest( + dataType=VehicleImageRequest.Type.DATA, + imageType=image_type, # pyright: ignore[reportArgumentType] + chunkRequest=VehicleImageDataChunkRequest( + chunk_offset=offset, + chunk_size=min( + chunk_size, total_size - offset + ), + ), + ) + ] + ) + ) + ) + ) + ) + asset_data = chunk_reply.vehicle_image_state.vehicle_images[0].asset_data + if asset_data.start_offset != offset: + raise ValueError( + f"Unexpected vehicle image chunk offset " + f"{asset_data.start_offset}; expected {offset}" + ) + chunk = asset_data.data + if not chunk: + raise ValueError( + f"Vehicle image ended after {offset} of {total_size} bytes" + ) + data += chunk + offset += len(chunk) + if offset > total_size: + raise ValueError( + f"Vehicle image exceeded declared size of {total_size} bytes" + ) + + if len(data) != total_size: + raise ValueError( + f"Vehicle image contained {len(data)} of {total_size} bytes" + ) + return bytes(data) + async def handshakeVehicleSecurity(self) -> None: """Perform a handshake with the vehicle security domain.""" await self._handshake(Domain.DOMAIN_VEHICLE_SECURITY) diff --git a/tests/test_proto_coverage_lock.py b/tests/test_proto_coverage_lock.py index 6712b90..21633ad 100644 --- a/tests/test_proto_coverage_lock.py +++ b/tests/test_proto_coverage_lock.py @@ -54,10 +54,7 @@ ALTERNATE_PATH_VEHICLE_ACTION_FIELDS | STREAMING_SUBSCRIPTION_VEHICLE_ACTION_FIELDS ) -# Explicitly chunked binary image-data transfer (paged offset/size requests), -# unlike every other GetVehicleData sub-state's single no-arg read. Needs its -# own chunking/reassembly design, not a one-shot reader. -KNOWN_UNWRAPPED_GET_VEHICLE_DATA_FIELDS = frozenset({"getVehicleImageState"}) +KNOWN_UNWRAPPED_GET_VEHICLE_DATA_FIELDS: frozenset[str] = frozenset() def _vehicle_action_fields(): @@ -105,9 +102,11 @@ def test_every_field_is_wrapped_or_allowlisted(self) -> None: if field.name in KNOWN_UNWRAPPED_GET_VEHICLE_DATA_FIELDS: continue type_name = field.message_type.name if field.message_type else None - if field.name in BLUETOOTH_SOURCE: + if field.name in BLUETOOTH_SOURCE or field.name in COMMANDS_SOURCE: continue - if type_name and type_name in BLUETOOTH_SOURCE: + if type_name and ( + type_name in BLUETOOTH_SOURCE or type_name in COMMANDS_SOURCE + ): continue unwrapped.append(field.name) self.assertEqual( diff --git a/tests/test_vehicle_image_state.py b/tests/test_vehicle_image_state.py new file mode 100644 index 0000000..f964d91 --- /dev/null +++ b/tests/test_vehicle_image_state.py @@ -0,0 +1,165 @@ +"""``vehicle_image_state`` (INFO read, defined on the shared ``Commands`` class) +over the mocked BLE transport. + +Unlike every other ``GetVehicleData`` sub-state, image bytes are paged: an +``ID`` request reports the total size, then ``DATA`` requests advance by +``chunk_size`` until the full range is retrieved. Defined on ``Commands`` +(not ``VehicleBluetooth``) so it is available on both signed transports - +BLE and the Fleet API signed-command relay (``VehicleSigned``). +""" + +from tesla_protocol.command.car_server_pb2 import Action, VehicleImageRequest +from tesla_protocol.command.vehicle_pb2 import ( + VehicleData, + VehicleImage, + VehicleImageData, + VehicleImageState, +) + +from tesla_fleet_api.const import VehicleImageType +from tesla_fleet_api.tesla.vehicle.commands import Commands +from tesla_fleet_api.tesla.vehicle.signed import VehicleSigned + +from ble_mocked_transport import ( + MockedBleTransportTestCase, + decrypt_sent_command, + infotainment_vehicle_data_reply, +) + + +def _image_state_reply( + total_size: int, data: bytes = b"", start_offset: int = 0 +) -> VehicleData: + return VehicleData( + vehicle_image_state=VehicleImageState( + vehicle_images=[ + VehicleImage( + total_image_size=total_size, + asset_data=VehicleImageData(data=data, start_offset=start_offset), + ) + ] + ) + ) + + +class VehicleImageStateTests(MockedBleTransportTestCase): + async def test_pages_chunks_and_reassembles_full_image(self) -> None: + vehicle, send = self.make_vehicle() + send.side_effect = [ + infotainment_vehicle_data_reply(_image_state_reply(total_size=10)), + infotainment_vehicle_data_reply( + _image_state_reply(total_size=10, data=b"abcdef") + ), + infotainment_vehicle_data_reply( + _image_state_reply(total_size=10, data=b"ghij", start_offset=6) + ), + ] + + result = await vehicle.vehicle_image_state( + VehicleImageType.AUTOPILOT_VISUALIZATION_WRAP, chunk_size=6 + ) + + self.assertEqual(result, b"abcdefghij") + self.assertEqual(send.await_count, 3) + + id_msg = send.await_args_list[0].args[0] + id_action = Action.FromString(decrypt_sent_command(vehicle, id_msg)) + id_request = ( + id_action.vehicleAction.getVehicleData.getVehicleImageState.imageRequests[0] + ) + self.assertEqual(id_request.dataType, VehicleImageRequest.Type.ID) + self.assertEqual( + id_request.imageType, VehicleImageType.AUTOPILOT_VISUALIZATION_WRAP + ) + + chunk1_msg = send.await_args_list[1].args[0] + chunk1_action = Action.FromString(decrypt_sent_command(vehicle, chunk1_msg)) + chunk1_request = chunk1_action.vehicleAction.getVehicleData.getVehicleImageState.imageRequests[ + 0 + ] + self.assertEqual(chunk1_request.dataType, VehicleImageRequest.Type.DATA) + self.assertEqual(chunk1_request.chunkRequest.chunk_offset, 0) + self.assertEqual(chunk1_request.chunkRequest.chunk_size, 6) + + chunk2_msg = send.await_args_list[2].args[0] + chunk2_action = Action.FromString(decrypt_sent_command(vehicle, chunk2_msg)) + chunk2_request = chunk2_action.vehicleAction.getVehicleData.getVehicleImageState.imageRequests[ + 0 + ] + self.assertEqual(chunk2_request.chunkRequest.chunk_offset, 6) + self.assertEqual(chunk2_request.chunkRequest.chunk_size, 4) + + async def test_zero_size_image_makes_no_data_requests(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_vehicle_data_reply( + _image_state_reply(total_size=0) + ) + + result = await vehicle.vehicle_image_state(VehicleImageType.LICENSE_PLATE) + + self.assertEqual(result, b"") + send.assert_awaited_once() + + async def test_rejects_invalid_chunk_size_before_requesting_metadata( + self, + ) -> None: + vehicle, send = self.make_vehicle() + + for chunk_size in (0, -1, 1.5, True): + with self.subTest(chunk_size=chunk_size): + with self.assertRaisesRegex( + ValueError, "chunk_size must be a positive integer" + ): + await vehicle.vehicle_image_state( + VehicleImageType.LICENSE_PLATE, + chunk_size=chunk_size, # pyright: ignore[reportArgumentType] + ) + + send.assert_not_awaited() + + async def test_rejects_empty_chunk_before_total_size(self) -> None: + vehicle, send = self.make_vehicle() + send.side_effect = [ + infotainment_vehicle_data_reply(_image_state_reply(total_size=4)), + infotainment_vehicle_data_reply(_image_state_reply(total_size=4)), + ] + + with self.assertRaisesRegex( + ValueError, "Vehicle image ended after 0 of 4 bytes" + ): + await vehicle.vehicle_image_state(VehicleImageType.LICENSE_PLATE) + + async def test_rejects_chunk_with_unexpected_offset(self) -> None: + vehicle, send = self.make_vehicle() + send.side_effect = [ + infotainment_vehicle_data_reply(_image_state_reply(total_size=4)), + infotainment_vehicle_data_reply( + _image_state_reply(total_size=4, data=b"abcd", start_offset=1) + ), + ] + + with self.assertRaisesRegex( + ValueError, "Unexpected vehicle image chunk offset 1; expected 0" + ): + await vehicle.vehicle_image_state(VehicleImageType.LICENSE_PLATE) + + async def test_rejects_data_beyond_declared_size(self) -> None: + vehicle, send = self.make_vehicle() + send.side_effect = [ + infotainment_vehicle_data_reply(_image_state_reply(total_size=3)), + infotainment_vehicle_data_reply( + _image_state_reply(total_size=3, data=b"abcd") + ), + ] + + with self.assertRaisesRegex( + ValueError, "Vehicle image exceeded declared size of 3 bytes" + ): + await vehicle.vehicle_image_state(VehicleImageType.LICENSE_PLATE) + + def test_available_on_the_fleet_api_signed_transport_too(self) -> None: + """Defined on ``Commands`` (not ``VehicleBluetooth``), so it is + inherited by ``VehicleSigned`` (Fleet API signed-command relay) as + well as by BLE - the same signed-command wrapper, two transports.""" + self.assertTrue(issubclass(VehicleSigned, Commands)) + self.assertTrue(hasattr(VehicleSigned, "vehicle_image_state"))