Add cover platform to Tesla Fleet (#126411)

Add cover platform

homeassistant/components/tesla_fleet/__init__.py

@@ -42,6 +42,7 @@ from .oauth import TeslaSystemImplementation
 PLATFORMS: Final = [
     Platform.BINARY_SENSOR,
     Platform.CLIMATE,
+    Platform.COVER,
     Platform.DEVICE_TRACKER,
     Platform.MEDIA_PLAYER,
     Platform.SELECT,

homeassistant/components/tesla_fleet/cover.py

@@ -0,0 +1,253 @@
+"""Cover platform for Tesla Fleet integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from tesla_fleet_api.const import Scope, SunRoofCommand, Trunk, WindowCommand
+
+from homeassistant.components.cover import (
+    CoverDeviceClass,
+    CoverEntity,
+    CoverEntityFeature,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import TeslaFleetConfigEntry
+from .entity import TeslaFleetVehicleEntity
+from .helpers import handle_vehicle_command
+from .models import TeslaFleetVehicleData
+
+OPEN = 1
+CLOSED = 0
+
+PARALLEL_UPDATES = 0
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: TeslaFleetConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the TeslaFleet cover platform from a config entry."""
+
+    async_add_entities(
+        klass(vehicle, entry.runtime_data.scopes)
+        for (klass) in (
+            TeslaFleetWindowEntity,
+            TeslaFleetChargePortEntity,
+            TeslaFleetFrontTrunkEntity,
+            TeslaFleetRearTrunkEntity,
+            TeslaFleetSunroofEntity,
+        )
+        for vehicle in entry.runtime_data.vehicles
+    )
+
+
+class TeslaFleetWindowEntity(TeslaFleetVehicleEntity, CoverEntity):
+    """Cover entity for the windows."""
+
+    _attr_device_class = CoverDeviceClass.WINDOW
+
+    def __init__(self, data: TeslaFleetVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the cover."""
+        super().__init__(data, "windows")
+        self.scoped = Scope.VEHICLE_CMDS in scopes
+        self._attr_supported_features = (
+            CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
+        )
+        if not self.scoped or self.vehicle.signing:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    def _async_update_attrs(self) -> None:
+        """Update the entity attributes."""
+        fd = self.get("vehicle_state_fd_window")
+        fp = self.get("vehicle_state_fp_window")
+        rd = self.get("vehicle_state_rd_window")
+        rp = self.get("vehicle_state_rp_window")
+
+        # Any open set to open
+        if OPEN in (fd, fp, rd, rp):
+            self._attr_is_closed = False
+        # All closed set to closed
+        elif CLOSED == fd == fp == rd == rp:
+            self._attr_is_closed = True
+        # Otherwise, set to unknown
+        else:
+            self._attr_is_closed = None
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Vent windows."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(
+            self.api.window_control(command=WindowCommand.VENT)
+        )
+        self._attr_is_closed = False
+        self.async_write_ha_state()
+
+    async def async_close_cover(self, **kwargs: Any) -> None:
+        """Close windows."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(
+            self.api.window_control(command=WindowCommand.CLOSE)
+        )
+        self._attr_is_closed = True
+        self.async_write_ha_state()
+
+
+class TeslaFleetChargePortEntity(TeslaFleetVehicleEntity, CoverEntity):
+    """Cover entity for the charge port."""
+
+    _attr_device_class = CoverDeviceClass.DOOR
+
+    def __init__(self, vehicle: TeslaFleetVehicleData, 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)
+        )
+        self._attr_supported_features = (
+            CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
+        )
+        if not self.scoped or self.vehicle.signing:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    def _async_update_attrs(self) -> None:
+        """Update the entity attributes."""
+        self._attr_is_closed = not self._value
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Open charge port."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.charge_port_door_open())
+        self._attr_is_closed = False
+        self.async_write_ha_state()
+
+    async def async_close_cover(self, **kwargs: Any) -> None:
+        """Close charge port."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.charge_port_door_close())
+        self._attr_is_closed = True
+        self.async_write_ha_state()
+
+
+class TeslaFleetFrontTrunkEntity(TeslaFleetVehicleEntity, CoverEntity):
+    """Cover entity for the front trunk."""
+
+    _attr_device_class = CoverDeviceClass.DOOR
+
+    def __init__(self, vehicle: TeslaFleetVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the cover."""
+        super().__init__(vehicle, "vehicle_state_ft")
+
+        self.scoped = Scope.VEHICLE_CMDS in scopes
+        self._attr_supported_features = CoverEntityFeature.OPEN
+        if not self.scoped or self.vehicle.signing:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    def _async_update_attrs(self) -> None:
+        """Update the entity attributes."""
+        self._attr_is_closed = self._value == CLOSED
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Open front trunk."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.actuate_trunk(Trunk.FRONT))
+        self._attr_is_closed = False
+        self.async_write_ha_state()
+
+
+class TeslaFleetRearTrunkEntity(TeslaFleetVehicleEntity, CoverEntity):
+    """Cover entity for the rear trunk."""
+
+    _attr_device_class = CoverDeviceClass.DOOR
+
+    def __init__(self, vehicle: TeslaFleetVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the cover."""
+        super().__init__(vehicle, "vehicle_state_rt")
+
+        self.scoped = Scope.VEHICLE_CMDS in scopes
+        self._attr_supported_features = (
+            CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
+        )
+        if not self.scoped or self.vehicle.signing:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    def _async_update_attrs(self) -> None:
+        """Update the entity attributes."""
+        value = self._value
+        if value == CLOSED:
+            self._attr_is_closed = True
+        elif value == OPEN:
+            self._attr_is_closed = False
+        else:
+            self._attr_is_closed = None
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Open rear trunk."""
+        if self.is_closed is not False:
+            await self.wake_up_if_asleep()
+            await handle_vehicle_command(self.api.actuate_trunk(Trunk.REAR))
+            self._attr_is_closed = False
+            self.async_write_ha_state()
+
+    async def async_close_cover(self, **kwargs: Any) -> None:
+        """Close rear trunk."""
+        if self.is_closed is not True:
+            await self.wake_up_if_asleep()
+            await handle_vehicle_command(self.api.actuate_trunk(Trunk.REAR))
+            self._attr_is_closed = True
+            self.async_write_ha_state()
+
+
+class TeslaFleetSunroofEntity(TeslaFleetVehicleEntity, CoverEntity):
+    """Cover entity for the sunroof."""
+
+    _attr_device_class = CoverDeviceClass.WINDOW
+    _attr_supported_features = (
+        CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP
+    )
+    _attr_entity_registry_enabled_default = False
+
+    def __init__(self, vehicle: TeslaFleetVehicleData, scopes: list[Scope]) -> None:
+        """Initialize the sensor."""
+        super().__init__(vehicle, "vehicle_state_sun_roof_state")
+
+        self.scoped = Scope.VEHICLE_CMDS in scopes
+        if not self.scoped or self.vehicle.signing:
+            self._attr_supported_features = CoverEntityFeature(0)
+
+    def _async_update_attrs(self) -> None:
+        """Update the entity attributes."""
+        value = self._value
+        if value in (None, "unknown"):
+            self._attr_is_closed = None
+        else:
+            self._attr_is_closed = value == "closed"
+
+        self._attr_current_cover_position = self.get(
+            "vehicle_state_sun_roof_percent_open"
+        )
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Open sunroof."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.VENT))
+        self._attr_is_closed = False
+        self.async_write_ha_state()
+
+    async def async_close_cover(self, **kwargs: Any) -> None:
+        """Close sunroof."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.CLOSE))
+        self._attr_is_closed = True
+        self.async_write_ha_state()
+
+    async def async_stop_cover(self, **kwargs: Any) -> None:
+        """Close sunroof."""
+        await self.wake_up_if_asleep()
+        await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.STOP))
+        self._attr_is_closed = False
+        self.async_write_ha_state()

homeassistant/components/tesla_fleet/icons.json

@@ -52,6 +52,11 @@
         }
       }
     },
+    "cover": {
+      "charge_state_charge_port_door_open": {
+        "default": "mdi:ev-plug-ccs2"
+      }
+    },
     "device_tracker": {
       "location": {
         "default": "mdi:map-marker"

homeassistant/components/tesla_fleet/strings.json

@@ -125,6 +125,23 @@
         }
       }
     },
+    "cover": {
+      "charge_state_charge_port_door_open": {
+        "name": "Charge port door"
+      },
+      "vehicle_state_ft": {
+        "name": "Frunk"
+      },
+      "vehicle_state_rt": {
+        "name": "Trunk"
+      },
+      "vehicle_state_sun_roof_state": {
+        "name": "Sunroof"
+      },
+      "windows": {
+        "name": "Windows"
+      }
+    },
     "device_tracker": {
       "location": {
         "name": "Location"

tests/components/tesla_fleet/test_cover.py

@@ -0,0 +1,240 @@
+"""Test the Teslemetry cover platform."""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from syrupy import SnapshotAssertion
+from tesla_fleet_api.exceptions import VehicleOffline
+
+from homeassistant.components.cover import (
+    DOMAIN as COVER_DOMAIN,
+    SERVICE_CLOSE_COVER,
+    SERVICE_OPEN_COVER,
+    SERVICE_STOP_COVER,
+)
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    STATE_CLOSED,
+    STATE_OPEN,
+    STATE_UNKNOWN,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from . import assert_entities, setup_platform
+from .const import COMMAND_OK, VEHICLE_DATA_ALT
+
+from tests.common import MockConfigEntry
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_cover(
+    hass: HomeAssistant,
+    snapshot: SnapshotAssertion,
+    entity_registry: er.EntityRegistry,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the cover entities are correct."""
+
+    await setup_platform(hass, normal_config_entry, [Platform.COVER])
+    assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_cover_alt(
+    hass: HomeAssistant,
+    snapshot: SnapshotAssertion,
+    entity_registry: er.EntityRegistry,
+    mock_vehicle_data: AsyncMock,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the cover entities are correct with alternate values."""
+
+    mock_vehicle_data.return_value = VEHICLE_DATA_ALT
+    await setup_platform(hass, normal_config_entry, [Platform.COVER])
+    assert_entities(hass, normal_config_entry.entry_id, entity_registry, snapshot)
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_cover_readonly(
+    hass: HomeAssistant,
+    snapshot: SnapshotAssertion,
+    entity_registry: er.EntityRegistry,
+    readonly_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the cover entities are correct without scopes."""
+
+    await setup_platform(hass, readonly_config_entry, [Platform.COVER])
+    assert_entities(hass, readonly_config_entry.entry_id, entity_registry, snapshot)
+
+
+async def test_cover_offline(
+    hass: HomeAssistant,
+    mock_vehicle_data: AsyncMock,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the cover entities are correct when offline."""
+
+    mock_vehicle_data.side_effect = VehicleOffline
+    await setup_platform(hass, normal_config_entry, [Platform.COVER])
+    state = hass.states.get("cover.test_windows")
+    assert state.state == STATE_UNKNOWN
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_cover_services(
+    hass: HomeAssistant,
+    normal_config_entry: MockConfigEntry,
+) -> None:
+    """Tests that the cover entities are correct."""
+
+    await setup_platform(hass, normal_config_entry, [Platform.COVER])
+
+    # Vent Windows
+    entity_id = "cover.test_windows"
+    with patch(
+        "homeassistant.components.teslemetry.VehicleSpecific.window_control",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_OPEN_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_OPEN
+
+        call.reset_mock()
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_CLOSE_COVER,
+            {ATTR_ENTITY_ID: ["cover.test_windows"]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_CLOSED
+
+    # Charge Port Door
+    entity_id = "cover.test_charge_port_door"
+    with patch(
+        "homeassistant.components.teslemetry.VehicleSpecific.charge_port_door_open",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_OPEN_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_OPEN
+
+    with patch(
+        "homeassistant.components.teslemetry.VehicleSpecific.charge_port_door_close",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_CLOSE_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_CLOSED
+
+    # Frunk
+    entity_id = "cover.test_frunk"
+    with patch(
+        "homeassistant.components.teslemetry.VehicleSpecific.actuate_trunk",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_OPEN_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_OPEN
+
+    # Trunk
+    entity_id = "cover.test_trunk"
+    with patch(
+        "homeassistant.components.teslemetry.VehicleSpecific.actuate_trunk",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_OPEN_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_OPEN
+
+        call.reset_mock()
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_CLOSE_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_CLOSED
+
+    # Sunroof
+    entity_id = "cover.test_sunroof"
+    with patch(
+        "homeassistant.components.teslemetry.VehicleSpecific.sun_roof_control",
+        return_value=COMMAND_OK,
+    ) as call:
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_OPEN_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_OPEN
+
+        call.reset_mock()
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_STOP_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_OPEN
+
+        call.reset_mock()
+        await hass.services.async_call(
+            COVER_DOMAIN,
+            SERVICE_CLOSE_COVER,
+            {ATTR_ENTITY_ID: [entity_id]},
+            blocking=True,
+        )
+        call.assert_called_once()
+        state = hass.states.get(entity_id)
+        assert state
+        assert state.state is STATE_CLOSED