Use VCSEC broadcasts for Teslemetry vehicle covers

homeassistant/components/teslemetry/cover.py

@@ -6,6 +6,9 @@
 from tesla_fleet_api import firmware_at_least
 from tesla_fleet_api.const import Scope, SunRoofCommand, Trunk, WindowCommand
 from tesla_fleet_api.router import VehicleRouter
+
+# pylint: disable-next=no-name-in-module
+from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import ClosureState_E
 from tesla_fleet_api.teslemetry import Vehicle
 from teslemetry_stream import Signal
 from teslemetry_stream.const import WindowState
@@ -15,11 +18,12 @@
     CoverEntity,
     CoverEntityFeature,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.helpers.restore_state import RestoreEntity
 
 from . import TeslemetryConfigEntry
+from .ble import TeslemetryVehicleBluetoothEntity
 from .entity import (
     TeslemetryRootEntity,
     TeslemetryVehiclePollingEntity,
@@ -34,6 +38,19 @@
 PARALLEL_UPDATES = 0
 
 
+def _closure_is_open(value: int) -> bool | None:
+    """Map a VCSEC closure enum onto an open/closed state.
+
+    UNKNOWN and FAILED_UNLATCH carry no usable state, so they are unavailable.
+    """
+    if value in (
+        ClosureState_E.CLOSURESTATE_UNKNOWN,
+        ClosureState_E.CLOSURESTATE_FAILED_UNLATCH,
+    ):
+        return None
+    return value != ClosureState_E.CLOSURESTATE_CLOSED
+
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: TeslemetryConfigEntry,
@@ -50,7 +67,9 @@ async def async_setup_entry(
                 for vehicle in entry.runtime_data.vehicles
             ),
             (
-                TeslemetryVehiclePollingChargePortEntity(
+                TeslemetryBluetoothChargePortEntity(vehicle, entry.runtime_data.scopes)
+                if vehicle.ble is not None
+                else TeslemetryVehiclePollingChargePortEntity(
                     vehicle, entry.runtime_data.scopes
                 )
                 if vehicle.poll or not firmware_at_least(vehicle.firmware, "2024.44.25")
@@ -60,7 +79,9 @@ async def async_setup_entry(
                 for vehicle in entry.runtime_data.vehicles
             ),
             (
-                TeslemetryVehiclePollingFrontTrunkEntity(
+                TeslemetryBluetoothFrontTrunkEntity(vehicle, entry.runtime_data.scopes)
+                if vehicle.ble is not None
+                else TeslemetryVehiclePollingFrontTrunkEntity(
                     vehicle, entry.runtime_data.scopes
                 )
                 if vehicle.poll or not firmware_at_least(vehicle.firmware, "2024.26")
@@ -70,7 +91,9 @@ async def async_setup_entry(
                 for vehicle in entry.runtime_data.vehicles
             ),
             (
-                TeslemetryVehiclePollingRearTrunkEntity(
+                TeslemetryBluetoothRearTrunkEntity(vehicle, entry.runtime_data.scopes)
+                if vehicle.ble is not None
+                else TeslemetryVehiclePollingRearTrunkEntity(
                     vehicle, entry.runtime_data.scopes
                 )
                 if vehicle.poll or not firmware_at_least(vehicle.firmware, "2024.26")
@@ -519,3 +542,96 @@ async def async_stop_cover(self, **kwargs: Any) -> None:
         await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.STOP))
         self._attr_is_closed = False
         self.async_write_ha_state()
+
+
+class TeslemetryBluetoothClosureCover(TeslemetryVehicleBluetoothEntity):
+    """Mixin rendering a VCSEC closure broadcast as a cover's closed state."""
+
+    _attr_is_closed: bool | None = None
+
+    @callback
+    @override
+    def _handle_broadcast(self, value: Any, generation: int) -> None:
+        """Store the broadcast open state and render it as closed/not-closed."""
+        self._value = value
+        self._generation = generation
+        self._attr_is_closed = None if value is None else not value
+        self.async_write_ha_state()
+
+
+class TeslemetryBluetoothChargePortEntity(
+    TeslemetryBluetoothClosureCover, TeslemetryChargePortEntity
+):
+    """Bluetooth cover entity for the charge port door."""
+
+    def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the cover."""
+        super().__init__(vehicle, "charge_state_charge_port_door_open")
+        self.scoped = any(
+            scope in scopes
+            for scope in (Scope.VEHICLE_CMDS, Scope.VEHICLE_CHARGING_CMDS)
+        )
+        if not self.scoped:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    @override
+    async def async_added_to_hass(self) -> None:
+        """Register the charge port closure broadcast listener."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            self.manager.async_on_broadcast(
+                lambda ble, callback: ble.listen_charge_port(callback),
+                _closure_is_open,
+                self._handle_broadcast,
+            )
+        )
+
+
+class TeslemetryBluetoothFrontTrunkEntity(
+    TeslemetryBluetoothClosureCover, TeslemetryFrontTrunkEntity
+):
+    """Bluetooth cover entity for the front trunk."""
+
+    def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the cover."""
+        super().__init__(vehicle, "vehicle_state_ft")
+        self.scoped = Scope.VEHICLE_CMDS in scopes
+        if not self.scoped:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    @override
+    async def async_added_to_hass(self) -> None:
+        """Register the front trunk closure broadcast listener."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            self.manager.async_on_broadcast(
+                lambda ble, callback: ble.listen_front_trunk(callback),
+                _closure_is_open,
+                self._handle_broadcast,
+            )
+        )
+
+
+class TeslemetryBluetoothRearTrunkEntity(
+    TeslemetryBluetoothClosureCover, TeslemetryRearTrunkEntity
+):
+    """Bluetooth cover entity for the rear trunk."""
+
+    def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the cover."""
+        super().__init__(vehicle, "vehicle_state_rt")
+        self.scoped = Scope.VEHICLE_CMDS in scopes
+        if not self.scoped:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    @override
+    async def async_added_to_hass(self) -> None:
+        """Register the rear trunk closure broadcast listener."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            self.manager.async_on_broadcast(
+                lambda ble, callback: ble.listen_rear_trunk(callback),
+                _closure_is_open,
+                self._handle_broadcast,
+            )
+        )

tests/components/teslemetry/test_ble.py

@@ -17,8 +17,10 @@
 from homeassistant.config_entries import ConfigSubentryData
 from homeassistant.const import (
     CONF_ADDRESS,
+    STATE_CLOSED,
     STATE_OFF,
     STATE_ON,
+    STATE_OPEN,
     STATE_UNAVAILABLE,
     Platform,
 )
@@ -54,9 +56,11 @@ def _entry_with_ble() -> MockConfigEntry:
 
 
 async def _setup_ble(
-    hass: HomeAssistant, connected: bool = True
+    hass: HomeAssistant,
+    connected: bool = True,
+    platforms: tuple[Platform, ...] = (Platform.BINARY_SENSOR,),
 ) -> tuple[MockConfigEntry, MagicMock]:
-    """Set up the binary sensor platform for a BLE-paired vehicle.
+    """Set up the given platforms for a BLE-paired vehicle.
 
     Returns the entry and the BLE client mock. The client's ``listen_*`` methods
     record the manager's broadcast callbacks so tests can feed them raw values;
@@ -77,9 +81,7 @@ async def _setup_ble(
         patch(
             "homeassistant.components.teslemetry.helpers.TeslaBluetooth"
         ) as mock_parent,
-        patch(
-            "homeassistant.components.teslemetry.PLATFORMS", [Platform.BINARY_SENSOR]
-        ),
+        patch("homeassistant.components.teslemetry.PLATFORMS", list(platforms)),
     ):
         mock_parent.return_value.get_private_key = AsyncMock()
         mock_parent.return_value.vehicles.createBluetooth.return_value = (
@@ -301,3 +303,87 @@ async def test_no_info_requests(hass: HomeAssistant) -> None:
     bluetooth.vehicle_data.assert_not_called()
     bluetooth.closures_state.assert_not_called()
     bluetooth.climate_state.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("listener", "entity_key"),
+    [
+        ("listen_charge_port", "charge_state_charge_port_door_open"),
+        ("listen_front_trunk", "vehicle_state_ft"),
+        ("listen_rear_trunk", "vehicle_state_rt"),
+    ],
+)
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        (ClosureState_E.CLOSURESTATE_CLOSED, STATE_CLOSED),
+        (ClosureState_E.CLOSURESTATE_OPEN, STATE_OPEN),
+        (ClosureState_E.CLOSURESTATE_AJAR, STATE_OPEN),
+        (ClosureState_E.CLOSURESTATE_OPENING, STATE_OPEN),
+        (ClosureState_E.CLOSURESTATE_CLOSING, STATE_OPEN),
+        (ClosureState_E.CLOSURESTATE_UNKNOWN, STATE_UNAVAILABLE),
+        (ClosureState_E.CLOSURESTATE_FAILED_UNLATCH, STATE_UNAVAILABLE),
+    ],
+)
+async def test_cover_closure_conversion(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    listener: str,
+    entity_key: str,
+    raw: int,
+    expected: str,
+) -> None:
+    """Each broadcast cover maps its closure enum; unknown is unavailable."""
+    _entry, bluetooth = await _setup_ble(hass, platforms=(Platform.COVER,))
+    cover_id = entity_registry.async_get_entity_id(
+        "cover", "teslemetry", f"{VIN}-{entity_key}"
+    )
+    assert cover_id is not None
+
+    _emit(getattr(bluetooth, listener), raw)
+    await hass.async_block_till_done()
+    assert hass.states.get(cover_id).state == expected
+
+
+async def test_cover_link_loss_marks_unavailable(
+    hass: HomeAssistant, entity_registry: er.EntityRegistry
+) -> None:
+    """A broadcast cover goes unavailable on link loss with no cloud fallback."""
+    _entry, bluetooth = await _setup_ble(
+        hass, connected=True, platforms=(Platform.COVER,)
+    )
+    cover_id = entity_registry.async_get_entity_id(
+        "cover", "teslemetry", f"{VIN}-vehicle_state_ft"
+    )
+
+    _emit(bluetooth.listen_front_trunk, ClosureState_E.CLOSURESTATE_CLOSED)
+    await hass.async_block_till_done()
+    assert hass.states.get(cover_id).state == STATE_CLOSED
+
+    _emit_connection(bluetooth, False)
+    await hass.async_block_till_done()
+    assert hass.states.get(cover_id).state == STATE_UNAVAILABLE
+
+
+async def test_cover_command_routes_through_api(
+    hass: HomeAssistant, entity_registry: er.EntityRegistry
+) -> None:
+    """A broadcast cover still sends its command through the command router."""
+    entry, bluetooth = await _setup_ble(
+        hass, connected=True, platforms=(Platform.COVER,)
+    )
+    cover_id = entity_registry.async_get_entity_id(
+        "cover", "teslemetry", f"{VIN}-charge_state_charge_port_door_open"
+    )
+
+    _emit(bluetooth.listen_charge_port, ClosureState_E.CLOSURESTATE_CLOSED)
+    await hass.async_block_till_done()
+
+    router = entry.runtime_data.vehicles[0].api
+    router.charge_port_door_open = AsyncMock(
+        return_value={"response": {"result": True, "reason": ""}}
+    )
+    await hass.services.async_call(
+        "cover", "open_cover", {"entity_id": cover_id}, blocking=True
+    )
+    router.charge_port_door_open.assert_awaited_once()