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: 2 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,8 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
- **Naming**: camelCase for class instance attributes that mirror API structure (`energySites`, `createFleet`). Snake_case for method names that are API endpoints.
- **BLE discovery gotcha**: a Tesla vehicle advertises no 128-bit service UUID pre-connect — only its VIN-derived local name (`^S[a-f0-9]{16}[CDRP]$`), and only in the scan response, not the `ADV_IND`. `SERVICE_UUID` (`tesla_fleet_api/tesla/vehicle/bluetooth.py`) exists only as a GATT service after connecting. Never pass `service_uuids=[SERVICE_UUID]` as a `BleakScanner` discovery-time filter — it hides the vehicle on a direct BlueZ adapter (an ESPHome proxy doesn't enforce that filter the same way, which can mask the bug in testing). Scan unfiltered with active scanning and match by name; keep `SERVICE_UUID` for post-connect GATT use only.
- **BLE domain-routing gotcha**: `Domain` (`proto/universal_message.proto`) has more values (`DOMAIN_BROADCAST`, `DOMAIN_AUTHD`, ...) than `VehicleBluetooth._queues` has keys (only `DOMAIN_VEHICLE_SECURITY`/`DOMAIN_INFOTAINMENT`). `_on_message` (`tesla_fleet_api/tesla/vehicle/bluetooth.py`) must look up `_queues` with `.get()` and drop unrecognized domains rather than indexing directly — indexing raises `KeyError` inside the `ReassemblingBuffer` callback, aborting reassembly of any further already-buffered messages in that notification.
- **BLE infotainment boot-delay gotcha**: `wake_up()` (VCSEC) returns as soon as the vehicle-security computer acks it, well before the infotainment computer is ready to complete a signed-command handshake. An INFO-domain read/command issued immediately after `wake_up()` can raise `BluetoothTimeout` on the handshake through no fault of the command itself. Live-verified: waiting ~10s after `wake_up()` before the first INFO read is sufficient on the test rig; callers doing INFO work right after waking should retry-with-backoff rather than treat one timeout as failure.
- **BLE `vehicle_data()` response-size cap**: the vehicle's signed-command implementation enforces its own response-size limit independent of the BLE transport's packet reassembly. Live-verified: a single-endpoint `vehicle_data()` call (or any of the dedicated per-substate readers like `charge_state()`) succeeds, but requesting as few as two `BluetoothVehicleData` endpoints together reliably raises `TeslaFleetMessageFaultResponseSizeExceedsMTU` (`exceptions.py`). This is why `vehicle_data()`'s `endpoints` arg has no all-endpoints default (unlike the cloud method) - prefer the per-substate readers, or a single-endpoint `vehicle_data()` call, over a multi-endpoint composite. Auto-chunking (split under the cap, merge replies) would fix this properly but is not implemented.

## Maintaining this file

Expand Down
48 changes: 43 additions & 5 deletions docs/bluetooth_vehicles.md
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,11 @@ async def main():
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.

## Open/Close Individual Doors (Bluetooth Only)

The individual door closure commands are Bluetooth-only and are not available via Fleet API signed commands.
Expand Down Expand Up @@ -112,20 +117,53 @@ asyncio.run(main())

## Get Vehicle Data

You can get data from a `VehicleBluetooth` instance using the `vehicle_data` method. Here's a basic example to get data from a `VehicleBluetooth` instance:
You can get data from a `VehicleBluetooth` instance using the individual state
reader methods. Each reader sends one BLE vehicle-data request and returns the
typed protobuf state from the signed-command reply.

```python
import asyncio
from tesla_fleet_api import TeslaBluetooth, BluetoothVehicleData
from tesla_fleet_api import TeslaBluetooth

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("<vin>")
data = await vehicle.vehicle_data([BluetoothVehicleData.CHARGE_STATE, BluetoothVehicleData.CLIMATE_STATE])
print(f"Vehicle data for VIN: {vehicle.vin}")
print(data)
charge_state = await vehicle.charge_state()
print(f"Battery level for VIN {vehicle.vin}: {charge_state.battery_level}")

asyncio.run(main())
```

Available BLE state readers:

- `vehicle_state()`
- `charge_state()`
- `climate_state()`
- `drive_state()`
- `location_state()`
- `closures_state()`
- `charge_schedule_state()`
- `preconditioning_schedule_state()`
- `tire_pressure_state()`
- `media_state()`
- `media_detail_state()`
- `software_update_state()`
- `parental_controls_state()`

For explicit composite requests, use `vehicle_data(endpoints)` with
`BluetoothVehicleData` values:

```python
from tesla_fleet_api import BluetoothVehicleData

data = await vehicle.vehicle_data([BluetoothVehicleData.CHARGE_STATE])
```

Unlike `VehicleFleet.vehicle_data()`, the BLE `vehicle_data()` method requires
an explicit `endpoints` list and returns a `VehicleData` protobuf message, not a
REST JSON `dict`. Prefer the individual state readers, or a single explicit
endpoint, because requesting multiple endpoints together can exceed the
vehicle's signed-command response-size cap and raise
`TeslaFleetMessageFaultResponseSizeExceedsMTU`.
35 changes: 33 additions & 2 deletions tesla_fleet_api/tesla/vehicle/bluetooth.py
Original file line number Diff line number Diff line change
Expand Up @@ -540,13 +540,44 @@ async def pair(
return

async def wake_up(self):
"""Wake up the vehicle."""
"""Wake up the vehicle security computer.

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
backoff.
"""
return await self._sendVehicleSecurity(
UnsignedMessage(RKEAction=RKEAction_E.RKE_ACTION_WAKE_VEHICLE)
)

async def vehicle_data(self, endpoints: list[BluetoothVehicleData]) -> VehicleData:
"""Get vehicle data."""
"""Get vehicle data over the BLE infotainment channel.

This is a BLE-specific method, not a re-implementation of the cloud
one, and it diverges from ``VehicleFleet.vehicle_data()`` in ways a
cloud user porting to BLE should not be surprised by:

- ``endpoints`` takes ``BluetoothVehicleData`` (proto request names
like ``"GetChargeState"``), not the cloud ``VehicleDataEndpoint``
(REST names like ``"charge_state"``) - the two enums are not
interchangeable.
- The return type is the ``VehicleData`` protobuf message built from
the signed-command reply, not a REST JSON ``dict``. Use
``tesla_fleet_api.tesla.bluetooth.toDict``/``toJson`` to convert it
if a dict is needed.
- Unlike the cloud method, ``endpoints`` has no "all endpoints"
default: requesting more than one sub-state in a single call risks
exceeding the vehicle's signed-command response-size cap, raised as
``TeslaFleetMessageFaultResponseSizeExceedsMTU`` - observed live
with as few as two endpoints requested together. The individual
state readers (``charge_state()``, ``climate_state()``, etc.) each
issue their own single-endpoint request and are not subject to
this cap; prefer them, or a narrow explicit ``endpoints`` subset,
over a large composite call. A future enhancement could split a
multi-endpoint request into multiple under-the-cap round trips and
merge the replies, but that auto-chunking is not implemented here.
"""
return await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
Expand Down
10 changes: 10 additions & 0 deletions tests/ble_mocked_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
CommandStatus,
FromVCSECMessage,
VehicleStatus,
)
from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
OperationStatus_E as VcsecOperationStatus_E,
Expand All @@ -57,6 +58,15 @@ def vcsec_ok_reply() -> RoutableMessage:
)


def vcsec_vehicle_status_reply(vehicle_status: VehicleStatus) -> RoutableMessage:
"""A canned VCSEC reply carrying a ``vehicleStatus`` (as returned by an InformationRequest)."""
body = FromVCSECMessage(vehicleStatus=vehicle_status)
return RoutableMessage(
from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY),
protobuf_message_as_bytes=body.SerializeToString(),
)


def infotainment_action_ok_reply() -> RoutableMessage:
"""A canned infotainment reply as returned by a signed action that succeeded."""
body = Response(
Expand Down
221 changes: 221 additions & 0 deletions tests/test_ble_mocked_state_readers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,221 @@
"""Regression tests for all 14 BLE state readers over the mocked transport.

Each reader is defined on ``VehicleBluetooth`` (not inherited from
``Commands``) and composes a ``GetVehicleData`` request; these tests feed a
canned reply and assert the reader decodes it to the expected typed proto.
"""

from tesla_fleet_api.const import BluetoothVehicleData
from tesla_fleet_api.tesla.vehicle.proto.car_server_pb2 import Action
from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
VehicleLockState_E,
VehicleStatus,
)
from tesla_fleet_api.tesla.vehicle.proto.vehicle_pb2 import (
ChargeScheduleState,
ChargeState,
ClimateState,
ClosuresState,
DriveState,
LocationState,
MediaDetailState,
MediaState,
ParentalControlsState,
PreconditioningScheduleState,
SoftwareUpdateState,
TirePressureState,
VehicleData,
)

from ble_mocked_transport import (
MockedBleTransportTestCase,
decrypt_sent_command,
infotainment_vehicle_data_reply,
vcsec_vehicle_status_reply,
)


class ChargeStateTests(MockedBleTransportTestCase):
async def test_charge_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(charge_state=ChargeState(battery_level=42))
)
result = await vehicle.charge_state()
self.assertIsInstance(result, ChargeState)
self.assertEqual(result.battery_level, 42)


class ClimateStateTests(MockedBleTransportTestCase):
async def test_climate_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(climate_state=ClimateState(inside_temp_celsius=21.5))
)
result = await vehicle.climate_state()
self.assertIsInstance(result, ClimateState)
self.assertAlmostEqual(result.inside_temp_celsius, 21.5)


class DriveStateTests(MockedBleTransportTestCase):
async def test_drive_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(drive_state=DriveState(speed=55))
)
result = await vehicle.drive_state()
self.assertIsInstance(result, DriveState)
self.assertEqual(result.speed, 55)


class LocationStateTests(MockedBleTransportTestCase):
async def test_location_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(location_state=LocationState(latitude=37.4, longitude=-122.1))
)
result = await vehicle.location_state()
self.assertIsInstance(result, LocationState)
self.assertAlmostEqual(result.latitude, 37.4, places=5)
self.assertAlmostEqual(result.longitude, -122.1, places=5)


class ClosuresStateTests(MockedBleTransportTestCase):
async def test_closures_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(closures_state=ClosuresState(door_open_driver_front=True))
)
result = await vehicle.closures_state()
self.assertIsInstance(result, ClosuresState)
self.assertTrue(result.door_open_driver_front)


class ChargeScheduleStateTests(MockedBleTransportTestCase):
async def test_charge_schedule_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(charge_schedule_state=ChargeScheduleState(charge_buffer=10))
)
result = await vehicle.charge_schedule_state()
self.assertIsInstance(result, ChargeScheduleState)
self.assertEqual(result.charge_buffer, 10)


class PreconditioningScheduleStateTests(MockedBleTransportTestCase):
async def test_preconditioning_schedule_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(
preconditioning_schedule_state=PreconditioningScheduleState(
max_num_precondition_schedules=5
)
)
)
result = await vehicle.preconditioning_schedule_state()
self.assertIsInstance(result, PreconditioningScheduleState)
self.assertEqual(result.max_num_precondition_schedules, 5)


class TirePressureStateTests(MockedBleTransportTestCase):
async def test_tire_pressure_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(tire_pressure_state=TirePressureState(tpms_pressure_fl=2.9))
)
result = await vehicle.tire_pressure_state()
self.assertIsInstance(result, TirePressureState)
self.assertAlmostEqual(result.tpms_pressure_fl, 2.9, places=5)


class MediaStateTests(MockedBleTransportTestCase):
async def test_media_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(
media_state=MediaState(audio_volume=7.0, now_playing_title="Song")
)
)
result = await vehicle.media_state()
self.assertIsInstance(result, MediaState)
self.assertAlmostEqual(result.audio_volume, 7.0)
self.assertEqual(result.now_playing_title, "Song")


class MediaDetailStateTests(MockedBleTransportTestCase):
async def test_media_detail_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(
media_detail_state=MediaDetailState(now_playing_duration=180000)
)
)
result = await vehicle.media_detail_state()
self.assertIsInstance(result, MediaDetailState)
self.assertEqual(result.now_playing_duration, 180000)


class SoftwareUpdateStateTests(MockedBleTransportTestCase):
async def test_software_update_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(software_update_state=SoftwareUpdateState(version="2025.14.3"))
)
result = await vehicle.software_update_state()
self.assertIsInstance(result, SoftwareUpdateState)
self.assertEqual(result.version, "2025.14.3")


class ParentalControlsStateTests(MockedBleTransportTestCase):
async def test_parental_controls_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(
parental_controls_state=ParentalControlsState(
parental_controls_active=True
)
)
)
result = await vehicle.parental_controls_state()
self.assertIsInstance(result, ParentalControlsState)
self.assertTrue(result.parental_controls_active)


class VehicleStateTests(MockedBleTransportTestCase):
"""``vehicle_state`` is the VCSEC ``VehicleStatus`` reader (not the infotainment ``VehicleState``, see §2.3)."""

async def test_vehicle_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = vcsec_vehicle_status_reply(
VehicleStatus(vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED)
)
result = await vehicle.vehicle_state()
self.assertIsInstance(result, VehicleStatus)
self.assertEqual(
result.vehicleLockState, VehicleLockState_E.VEHICLELOCKSTATE_LOCKED
)


class VehicleDataTests(MockedBleTransportTestCase):
"""``vehicle_data`` composite reader; ``endpoints`` is required (no all-endpoints default,
see the response-size-cap note in its docstring)."""

async def test_explicit_endpoints_requests_only_those_substates(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(charge_state=ChargeState(battery_level=77))
)

result = await vehicle.vehicle_data([BluetoothVehicleData.CHARGE_STATE])

self.assertIsInstance(result, VehicleData)
self.assertEqual(result.charge_state.battery_level, 77)

send.assert_awaited_once()
sent_msg = send.await_args.args[0]

plaintext = decrypt_sent_command(vehicle, sent_msg)
action = Action.FromString(plaintext)
get_vehicle_data = action.vehicleAction.getVehicleData
self.assertTrue(get_vehicle_data.HasField("getChargeState"))
self.assertFalse(get_vehicle_data.HasField("getClimateState"))
Loading