Use VCSEC broadcasts for Teslemetry vehicle lock

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

tests/components/teslemetry/test_ble.py

@@ -10,9 +10,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 (
@@ -387,3 +389,80 @@ async def test_cover_command_routes_through_api(
         "cover", "open_cover", {"entity_id": cover_id}, blocking=True
     )
     router.charge_port_door_open.assert_awaited_once()
+
+
+@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
+
+    _emit_connection(bluetooth, False)
+    await hass.async_block_till_done()
+    assert hass.states.get(lock_id).state == STATE_UNAVAILABLE
+
+
+async def test_lock_command_routes_through_api(
+    hass: HomeAssistant, entity_registry: er.EntityRegistry
+) -> None:
+    """The broadcast vehicle lock still sends lock through the command router."""
+    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_UNLOCKED,
+    )
+    await hass.async_block_till_done()
+
+    router = entry.runtime_data.vehicles[0].api
+    router.door_lock = AsyncMock(
+        return_value={"response": {"result": True, "reason": ""}}
+    )
+    await hass.services.async_call(
+        "lock", "lock", {"entity_id": lock_id}, blocking=True
+    )
+    router.door_lock.assert_awaited_once()