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
1 change: 1 addition & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `.
- **`_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`.
- **`networking_status`'s `ipv4_config` fields are raw big-endian uint32 ints, not strings**: the wire format applies to all of `ipv4_config.address`/`subnet_mask`/`gateway`, but `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes only `address` - confirmed against a live Powerwall 3 capture where `3232235914` decodes network-byte-order (`struct.pack(">I", ...)`) to `192.168.1.138`, not little-endian. It considers only `eth`/`wifi` (never `gsm` - cellular isn't a LAN path), preferring whichever has `active_route` set and a decodable address, else falling back to the first of the two (in that order) with any decodable address; `0`/`0xFFFFFFFF` are treated as undecodable so an unconfigured interface never shadows a real one with `0.0.0.0`. A `{"response": null}` envelope raises `InvalidResponse` (the endpoint's known intermittent malformed mode), while a well-formed response with no usable interface returns `None` (not a raise). Tests: `tests/test_teslemetry_gateway_address.py`.
- **`VehicleAction`/`GetVehicleData` proto coverage is locked by test, not just by convention**: `tests/test_proto_coverage_lock.py` walks both descriptors and fails if any field has no wrapper (`commands.py`) or reader (`bluetooth.py`) and isn't on one of its two small, reasoned allowlists - keep that test in sync with any future `tesla-protocol` bump rather than special-casing new fields elsewhere. The only fields deliberately left unwrapped today are the 7-field push-style subscription/streaming family (`createStreamSession`/`streamMessage`/`vehicleDataSubscription`/`vehicleDataAck`/`vitalsSubscription`/`vitalsAck`/`cancelVehicleDataSubscription`, needs a persistent INFOTAINMENT-domain broadcast dispatcher like `broadcast.py`'s VCSEC listeners) and `getVehicleImageState` (needs chunked binary-transfer paging) - both are separate, unscoped design work, not oversights. CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, specifically to avoid confusion with the pre-existing `vehicle_state()` (VCSEC `VehicleStatus`, a different message/domain). `set_rate_tariff`/`add_managed_charging_site` (`commands.py`) take `tesla_protocol` message types directly for their deeply-nested arguments rather than a parallel flattened dataclass API. `pii_key_request`/`pseudonym_sync_request`/`tesla_auth_response`/`setup_cloud_profile_with_local_profile_uuid`/`get_local_profiles_for_vault_uuid` are wrapped with no known third-party consumer use case, purely for full-proto-coverage completeness.

## Maintaining this file

Expand Down
9 changes: 8 additions & 1 deletion docs/bluetooth_vehicles.md
Original file line number Diff line number Diff line change
Expand Up @@ -450,7 +450,7 @@ asyncio.run(main())

Available BLE state readers:

- `vehicle_state()`
- `vehicle_state()` (VCSEC `VehicleStatus` - not `legacy_vehicle_state()` below)
- `charge_state()`
- `climate_state()`
- `drive_state()`
Expand All @@ -463,6 +463,13 @@ Available BLE state readers:
- `media_detail_state()`
- `software_update_state()`
- `parental_controls_state()`
- `gui_settings()`
- `parked_accessory_state()`
- `legacy_vehicle_state()` (CarServer's `GetVehicleState` - a different message from `vehicle_state()` above)
- `alert_state()`
- `light_show_state()`
- `suspension_state()`
- `child_presence_detection_state()`

For explicit composite requests, use `vehicle_data(endpoints)` with
`BluetoothVehicleData` values:
Expand Down
7 changes: 7 additions & 0 deletions tesla_fleet_api/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -323,3 +323,10 @@ class BluetoothVehicleData(StrEnum):
MEDIA_DETAIL_STATE = "GetMediaDetailState"
SOFTWARE_UPDATE_STATE = "GetSoftwareUpdateState"
PARENTAL_CONTROLS_STATE = "GetParentalControlsState"
GUI_SETTINGS = "GetGuiSettings"
PARKED_ACCESSORY_STATE = "GetParkedAccessoryState"
LEGACY_VEHICLE_STATE = "GetVehicleState"
ALERT_STATE = "GetAlertState"
LIGHT_SHOW_STATE = "GetLightShowState"
SUSPENSION_STATE = "GetSuspensionState"
CHILD_PRESENCE_DETECTION_STATE = "GetChildPresenceDetectionState"
134 changes: 134 additions & 0 deletions tesla_fleet_api/tesla/vehicle/bluetooth.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,19 +36,26 @@
# Protocol
from tesla_protocol.command.car_server_pb2 import (
Action,
GetAlertState,
GetChargeScheduleState,
GetChargeState,
GetChildPresenceDetectionState,
GetClimateState,
GetClosuresState,
GetDriveState,
GetGuiSettings,
GetLightShowState,
GetLocationState,
GetMediaDetailState,
GetMediaState,
GetParentalControlsState,
GetParkedAccessoryState,
GetPreconditioningScheduleState,
GetSoftwareUpdateState,
GetSuspensionState,
GetTirePressureState,
GetVehicleData,
GetVehicleState,
VehicleAction,
)
from tesla_protocol.command.keys_pb2 import Role
Expand All @@ -74,19 +81,26 @@
WhitelistOperation,
)
from tesla_protocol.command.vehicle_pb2 import (
AlertState,
ChargeScheduleState,
ChargeState,
ChildPresenceDetectionState,
ClimateState,
ClosuresState,
DriveState,
GuiSettings,
LightShowState,
LocationState,
MediaDetailState,
MediaState,
ParentalControlsState,
ParkedAccessoryState,
PreconditioningScheduleState,
SoftwareUpdateState,
SuspensionState,
TirePressureState,
VehicleData,
VehicleState,
)

SERVICE_UUID = "00000211-b2d1-43f0-9b88-960cebf8b91e"
Expand Down Expand Up @@ -1504,6 +1518,28 @@ async def vehicle_data(self, endpoints: list[BluetoothVehicleData]) -> VehicleDa
getParentalControlsState=GetParentalControlsState()
if BluetoothVehicleData.PARENTAL_CONTROLS_STATE in endpoints
else None,
getGuiSettings=GetGuiSettings()
if BluetoothVehicleData.GUI_SETTINGS in endpoints
else None,
getParkedAccessoryState=GetParkedAccessoryState()
if BluetoothVehicleData.PARKED_ACCESSORY_STATE in endpoints
else None,
getVehicleState=GetVehicleState()
if BluetoothVehicleData.LEGACY_VEHICLE_STATE in endpoints
else None,
getAlertState=GetAlertState()
if BluetoothVehicleData.ALERT_STATE in endpoints
else None,
getLightShowState=GetLightShowState()
if BluetoothVehicleData.LIGHT_SHOW_STATE in endpoints
else None,
getSuspensionState=GetSuspensionState()
if BluetoothVehicleData.SUSPENSION_STATE in endpoints
else None,
getChildPresenceDetectionState=GetChildPresenceDetectionState()
if BluetoothVehicleData.CHILD_PRESENCE_DETECTION_STATE
in endpoints
else None,
)
)
)
Expand Down Expand Up @@ -1669,6 +1705,104 @@ async def parental_controls_state(self) -> ParentalControlsState:
)
).parental_controls_state

async def gui_settings(self) -> GuiSettings:
"""Return the current GUI display settings over BLE."""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(getGuiSettings=GetGuiSettings())
)
)
)
).gui_settings

async def parked_accessory_state(self) -> ParkedAccessoryState:
"""Return the current parked-accessory (e.g. awning) state over BLE."""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(
getParkedAccessoryState=GetParkedAccessoryState()
)
)
)
)
).parked_accessory_state

async def legacy_vehicle_state(self) -> VehicleState:
"""Return CarServer's legacy vehicle-state surface over BLE.

Named to match the reply field (``VehicleData.legacy_vehicle_state``),
not ``vehicle_state()``: that name is already taken by the VCSEC
``VehicleStatus`` reader below, a different message from a different
domain.
"""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(getVehicleState=GetVehicleState())
)
)
)
).legacy_vehicle_state

async def alert_state(self) -> AlertState:
"""Return the current active vehicle alerts over BLE."""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(getAlertState=GetAlertState())
)
)
)
).alert_state

async def light_show_state(self) -> LightShowState:
"""Return the current light show state over BLE."""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(
getLightShowState=GetLightShowState()
)
)
)
)
).light_show_state

async def suspension_state(self) -> SuspensionState:
"""Return the current adaptive suspension state over BLE."""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(
getSuspensionState=GetSuspensionState()
)
)
)
)
).suspension_state

async def child_presence_detection_state(self) -> ChildPresenceDetectionState:
"""Return the current child presence detection state over BLE."""
return (
await self._getInfotainment(
Action(
vehicleAction=VehicleAction(
getVehicleData=GetVehicleData(
getChildPresenceDetectionState=GetChildPresenceDetectionState()
)
)
)
)
).child_presence_detection_state

async def vehicle_state(self) -> VehicleStatus:
"""Return the vehicle security-domain status over BLE."""
return await self._getVehicleSecurity(
Expand Down
131 changes: 131 additions & 0 deletions tests/test_ble_mocked_state_readers_new.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
"""Regression tests for the 7 new BLE vehicle-data sub-state readers.

Completes GetVehicleData coverage alongside the existing 12 readers
(``test_ble_mocked_state_readers.py``): gui_settings, parked_accessory_state,
legacy_vehicle_state, alert_state, light_show_state, suspension_state,
child_presence_detection_state.
"""

from tesla_fleet_api.const import BluetoothVehicleData
from tesla_protocol.command.car_server_pb2 import Action
from tesla_protocol.command.vehicle_pb2 import (
AlertState,
ChildPresenceDetectionState,
GuiSettings,
LightShowState,
ParkedAccessoryState,
SuspensionState,
VehicleData,
VehicleState,
)

from ble_mocked_transport import (
MockedBleTransportTestCase,
decrypt_sent_command,
infotainment_vehicle_data_reply,
)


class GuiSettingsTests(MockedBleTransportTestCase):
async def test_gui_settings(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(gui_settings=GuiSettings(gui_24_hour_time=True))
)
result = await vehicle.gui_settings()
self.assertIsInstance(result, GuiSettings)
self.assertTrue(result.gui_24_hour_time)


class ParkedAccessoryStateTests(MockedBleTransportTestCase):
async def test_parked_accessory_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(
parked_accessory_state=ParkedAccessoryState(tent_mode_request=True)
)
)
result = await vehicle.parked_accessory_state()
self.assertIsInstance(result, ParkedAccessoryState)
self.assertTrue(result.tent_mode_request)


class LegacyVehicleStateTests(MockedBleTransportTestCase):
async def test_legacy_vehicle_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(legacy_vehicle_state=VehicleState(car_version="2025.14.3"))
)
result = await vehicle.legacy_vehicle_state()
self.assertIsInstance(result, VehicleState)
self.assertEqual(result.car_version, "2025.14.3")


class AlertStateTests(MockedBleTransportTestCase):
async def test_alert_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(alert_state=AlertState())
)
result = await vehicle.alert_state()
self.assertIsInstance(result, AlertState)


class LightShowStateTests(MockedBleTransportTestCase):
async def test_light_show_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(light_show_state=LightShowState(light_show_active=True))
)
result = await vehicle.light_show_state()
self.assertIsInstance(result, LightShowState)
self.assertTrue(result.light_show_active)


class SuspensionStateTests(MockedBleTransportTestCase):
async def test_suspension_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(suspension_state=SuspensionState(offroad_on=True))
)
result = await vehicle.suspension_state()
self.assertIsInstance(result, SuspensionState)
self.assertTrue(result.offroad_on)


class ChildPresenceDetectionStateTests(MockedBleTransportTestCase):
async def test_child_presence_detection_state(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(
child_presence_detection_state=ChildPresenceDetectionState(
cpd_hvac_active=True
)
)
)
result = await vehicle.child_presence_detection_state()
self.assertIsInstance(result, ChildPresenceDetectionState)
self.assertTrue(result.cpd_hvac_active)


class VehicleDataDispatchNewEndpointsTests(MockedBleTransportTestCase):
"""The 7 new sub-states must wire into ``vehicle_data()``'s endpoints dispatch."""

async def test_new_endpoints_are_requested_when_given(self) -> None:
vehicle, send = self.make_vehicle()
send.return_value = infotainment_vehicle_data_reply(
VehicleData(gui_settings=GuiSettings())
)

await vehicle.vehicle_data([BluetoothVehicleData.GUI_SETTINGS])

sent_msg = send.await_args.args[0]
plaintext = decrypt_sent_command(vehicle, sent_msg)
get_vehicle_data = Action.FromString(plaintext).vehicleAction.getVehicleData
self.assertTrue(get_vehicle_data.HasField("getGuiSettings"))
self.assertFalse(get_vehicle_data.HasField("getParkedAccessoryState"))
self.assertFalse(get_vehicle_data.HasField("getVehicleState"))
self.assertFalse(get_vehicle_data.HasField("getAlertState"))
self.assertFalse(get_vehicle_data.HasField("getLightShowState"))
self.assertFalse(get_vehicle_data.HasField("getSuspensionState"))
self.assertFalse(get_vehicle_data.HasField("getChildPresenceDetectionState"))
Loading
Loading