From 6a646128e28bd4d8a90f7a5f2a836895943cab63 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Fri, 31 Jul 2026 14:37:07 +1000 Subject: [PATCH] Use VCSEC broadcasts for Teslemetry vehicle lock Reroute the vehicle lock to its VCSEC lock-state broadcast when the vehicle is BLE paired, reusing the BLE data manager's connection- generation availability. Lock/unlock commands still route through the existing command router, and the charge cable latch stays on the stream. An unrecognized lock enum resolves to unavailable. --- homeassistant/components/teslemetry/lock.py | 64 ++++++++++++++++++++- tests/components/teslemetry/test_ble.py | 53 +++++++++++++++++ 2 files changed, 115 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/teslemetry/lock.py b/homeassistant/components/teslemetry/lock.py index 7b331e4be6f367..caea243b07d435 100644 --- a/homeassistant/components/teslemetry/lock.py +++ b/homeassistant/components/teslemetry/lock.py @@ -6,15 +6,19 @@ from tesla_fleet_api import firmware_at_least from tesla_fleet_api.const import Scope from tesla_fleet_api.router import VehicleRouter + +# pylint: disable-next=no-name-in-module +from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import VehicleLockState_E from tesla_fleet_api.teslemetry import Vehicle from homeassistant.components.lock import LockEntity -from homeassistant.core import HomeAssistant +from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import ServiceValidationError from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.restore_state import RestoreEntity from . import TeslemetryConfigEntry +from .ble import TeslemetryVehicleBluetoothEntity from .const import DOMAIN from .entity import ( TeslemetryRootEntity, @@ -28,6 +32,24 @@ PARALLEL_UPDATES = 0 +_LOCKED_STATES = ( + VehicleLockState_E.VEHICLELOCKSTATE_LOCKED, + VehicleLockState_E.VEHICLELOCKSTATE_INTERNAL_LOCKED, +) +_UNLOCKED_STATES = ( + VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED, + VehicleLockState_E.VEHICLELOCKSTATE_SELECTIVE_UNLOCKED, +) + + +def _lock_is_locked(value: int) -> bool | None: + """Map the VCSEC lock enum; an unrecognized value is unavailable.""" + if value in _LOCKED_STATES: + return True + if value in _UNLOCKED_STATES: + return False + return None + async def async_setup_entry( hass: HomeAssistant, @@ -39,7 +61,11 @@ async def async_setup_entry( async_add_entities( chain( ( - TeslemetryVehiclePollingVehicleLockEntity( + TeslemetryBluetoothVehicleLockEntity( + vehicle, Scope.VEHICLE_CMDS in entry.runtime_data.scopes + ) + if vehicle.ble is not None + else TeslemetryVehiclePollingVehicleLockEntity( vehicle, Scope.VEHICLE_CMDS in entry.runtime_data.scopes ) if vehicle.poll or not firmware_at_least(vehicle.firmware, "2024.26") @@ -139,6 +165,40 @@ def _callback(self, value: bool | None) -> None: self.async_write_ha_state() +class TeslemetryBluetoothVehicleLockEntity( + TeslemetryVehicleBluetoothEntity, TeslemetryVehicleLockEntity +): + """Bluetooth vehicle lock entity for Teslemetry.""" + + _attr_is_locked: bool | None = None + + def __init__(self, data: TeslemetryVehicleData, scoped: bool) -> None: + """Initialize the lock.""" + super().__init__(data, "vehicle_state_locked") + self.scoped = scoped + + @override + async def async_added_to_hass(self) -> None: + """Register the lock-state broadcast listener.""" + await super().async_added_to_hass() + self.async_on_remove( + self.manager.async_on_broadcast( + lambda ble, callback: ble.listen_vehicle_lock_state(callback), + _lock_is_locked, + self._handle_broadcast, + ) + ) + + @callback + @override + def _handle_broadcast(self, value: Any, generation: int) -> None: + """Render the broadcast lock state.""" + self._value = value + self._generation = generation + self._attr_is_locked = value + self.async_write_ha_state() + + class TeslemetryCableLockEntity(TeslemetryRootEntity, LockEntity): """Base cable Lock entity for Teslemetry.""" diff --git a/tests/components/teslemetry/test_ble.py b/tests/components/teslemetry/test_ble.py index 0642bb9dda9c27..e9fd323ca9d881 100644 --- a/tests/components/teslemetry/test_ble.py +++ b/tests/components/teslemetry/test_ble.py @@ -11,9 +11,11 @@ from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import ( ClosureState_E, UserPresence_E, + VehicleLockState_E, VehicleSleepStatus_E, ) +from homeassistant.components.lock import LockState from homeassistant.components.teslemetry.const import CONF_VIN, SUBENTRY_TYPE_VEHICLE from homeassistant.config_entries import ConfigSubentryData from homeassistant.const import ( @@ -329,3 +331,54 @@ async def test_cover_link_loss_marks_unavailable( async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=6)) await hass.async_block_till_done() assert hass.states.get(cover_id).state == STATE_UNAVAILABLE + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (VehicleLockState_E.VEHICLELOCKSTATE_LOCKED, LockState.LOCKED), + (VehicleLockState_E.VEHICLELOCKSTATE_INTERNAL_LOCKED, LockState.LOCKED), + (VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED, LockState.UNLOCKED), + (VehicleLockState_E.VEHICLELOCKSTATE_SELECTIVE_UNLOCKED, LockState.UNLOCKED), + (99, STATE_UNAVAILABLE), + ], +) +async def test_lock_state_conversion( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + raw: int, + expected: str, +) -> None: + """The vehicle lock maps its VCSEC enum; an unrecognized value is unavailable.""" + _entry, bluetooth = await _setup_ble(hass, platforms=(Platform.LOCK,)) + lock_id = entity_registry.async_get_entity_id( + "lock", "teslemetry", f"{VIN}-vehicle_state_locked" + ) + assert lock_id is not None + + _emit(bluetooth.listen_vehicle_lock_state, raw) + await hass.async_block_till_done() + assert hass.states.get(lock_id).state == expected + + +async def test_lock_link_loss_marks_unavailable( + hass: HomeAssistant, entity_registry: er.EntityRegistry +) -> None: + """The broadcast vehicle lock goes unavailable on link loss, no cloud fallback.""" + _entry, bluetooth = await _setup_ble( + hass, connected=True, platforms=(Platform.LOCK,) + ) + lock_id = entity_registry.async_get_entity_id( + "lock", "teslemetry", f"{VIN}-vehicle_state_locked" + ) + + _emit( + bluetooth.listen_vehicle_lock_state, VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + await hass.async_block_till_done() + assert hass.states.get(lock_id).state == LockState.LOCKED + + bluetooth.client.is_connected = False + async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=6)) + await hass.async_block_till_done() + assert hass.states.get(lock_id).state == STATE_UNAVAILABLE