diff --git a/homeassistant/components/teslemetry/cover.py b/homeassistant/components/teslemetry/cover.py index d878c12f26c013..5a39656493f9e5 100644 --- a/homeassistant/components/teslemetry/cover.py +++ b/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, + ) + ) diff --git a/tests/components/teslemetry/test_ble.py b/tests/components/teslemetry/test_ble.py index 9daa1fed7f6511..982de5c9ac6b2c 100644 --- a/tests/components/teslemetry/test_ble.py +++ b/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()