feat(vehicle): add signed vehicle image retrieval

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:

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

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,21 +54,28 @@ 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
 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
+tests/test_tessie_vehicle_params.py
+tests/test_vehicle_image_state.py

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"""
 

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)

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(