Add door locks to Subaru integration (#52852)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/subaru/__init__.py

@@ -10,6 +10,7 @@ from homeassistant.const import CONF_DEVICE_ID, CONF_PASSWORD, CONF_PIN, CONF_US
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import aiohttp_client
+from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import (
@@ -21,6 +22,7 @@ from .const import (
     ENTRY_COORDINATOR,
     ENTRY_VEHICLES,
     FETCH_INTERVAL,
+    MANUFACTURER,
     PLATFORMS,
     UPDATE_INTERVAL,
     VEHICLE_API_GEN,
@@ -154,3 +156,12 @@ def get_vehicle_info(controller, vin):
         VEHICLE_LAST_UPDATE: 0,
     }
     return info
+
+
+def get_device_info(vehicle_info):
+    """Return DeviceInfo object based on vehicle info."""
+    return DeviceInfo(
+        identifiers={(DOMAIN, vehicle_info[VEHICLE_VIN])},
+        manufacturer=MANUFACTURER,
+        name=vehicle_info[VEHICLE_NAME],
+    )

homeassistant/components/subaru/const.py

@@ -1,4 +1,6 @@
 """Constants for the Subaru integration."""
+from subarulink.const import ALL_DOORS, DRIVERS_DOOR, TAILGATE_DOOR
+
 from homeassistant.const import Platform
 
 DOMAIN = "subaru"
@@ -32,9 +34,25 @@ API_GEN_2 = "g2"
 MANUFACTURER = "Subaru Corp."
 
 PLATFORMS = [
+    Platform.LOCK,
     Platform.SENSOR,
 ]
 
+SERVICE_LOCK = "lock"
+SERVICE_UNLOCK = "unlock"
+SERVICE_UNLOCK_SPECIFIC_DOOR = "unlock_specific_door"
+
+ATTR_DOOR = "door"
+
+UNLOCK_DOOR_ALL = "all"
+UNLOCK_DOOR_DRIVERS = "driver"
+UNLOCK_DOOR_TAILGATE = "tailgate"
+UNLOCK_VALID_DOORS = {
+    UNLOCK_DOOR_ALL: ALL_DOORS,
+    UNLOCK_DOOR_DRIVERS: DRIVERS_DOOR,
+    UNLOCK_DOOR_TAILGATE: TAILGATE_DOOR,
+}
+
 ICONS = {
     "Avg Fuel Consumption": "mdi:leaf",
     "EV Range": "mdi:ev-station",

homeassistant/components/subaru/lock.py

@@ -0,0 +1,91 @@
+"""Support for Subaru door locks."""
+import logging
+
+import voluptuous as vol
+
+from homeassistant.components.lock import LockEntity
+from homeassistant.const import SERVICE_LOCK, SERVICE_UNLOCK
+from homeassistant.helpers import entity_platform
+
+from . import DOMAIN, get_device_info
+from .const import (
+    ATTR_DOOR,
+    ENTRY_CONTROLLER,
+    ENTRY_VEHICLES,
+    SERVICE_UNLOCK_SPECIFIC_DOOR,
+    UNLOCK_DOOR_ALL,
+    UNLOCK_VALID_DOORS,
+    VEHICLE_HAS_REMOTE_SERVICE,
+    VEHICLE_NAME,
+    VEHICLE_VIN,
+)
+from .remote_service import async_call_remote_service
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up the Subaru locks by config_entry."""
+    entry = hass.data[DOMAIN][config_entry.entry_id]
+    controller = entry[ENTRY_CONTROLLER]
+    vehicle_info = entry[ENTRY_VEHICLES]
+    async_add_entities(
+        SubaruLock(vehicle, controller)
+        for vehicle in vehicle_info.values()
+        if vehicle[VEHICLE_HAS_REMOTE_SERVICE]
+    )
+
+    platform = entity_platform.async_get_current_platform()
+
+    platform.async_register_entity_service(
+        SERVICE_UNLOCK_SPECIFIC_DOOR,
+        {vol.Required(ATTR_DOOR): vol.In(UNLOCK_VALID_DOORS)},
+        "async_unlock_specific_door",
+    )
+
+
+class SubaruLock(LockEntity):
+    """
+    Representation of a Subaru door lock.
+
+    Note that the Subaru API currently does not support returning the status of the locks. Lock status is always unknown.
+    """
+
+    def __init__(self, vehicle_info, controller):
+        """Initialize the locks for the vehicle."""
+        self.controller = controller
+        self.vehicle_info = vehicle_info
+        vin = vehicle_info[VEHICLE_VIN]
+        self.car_name = vehicle_info[VEHICLE_NAME]
+        self._attr_name = f"{self.car_name} Door Locks"
+        self._attr_unique_id = f"{vin}_door_locks"
+        self._attr_device_info = get_device_info(vehicle_info)
+
+    async def async_lock(self, **kwargs):
+        """Send the lock command."""
+        _LOGGER.debug("Locking doors for: %s", self.car_name)
+        await async_call_remote_service(
+            self.controller,
+            SERVICE_LOCK,
+            self.vehicle_info,
+        )
+
+    async def async_unlock(self, **kwargs):
+        """Send the unlock command."""
+        _LOGGER.debug("Unlocking doors for: %s", self.car_name)
+        await async_call_remote_service(
+            self.controller,
+            SERVICE_UNLOCK,
+            self.vehicle_info,
+            UNLOCK_VALID_DOORS[UNLOCK_DOOR_ALL],
+        )
+
+    async def async_unlock_specific_door(self, door):
+        """Send the unlock command for a specified door."""
+        _LOGGER.debug("Unlocking %s door for: %s", door, self.car_name)
+        await async_call_remote_service(
+            self.controller,
+            SERVICE_UNLOCK,
+            self.vehicle_info,
+            UNLOCK_VALID_DOORS[door],
+        )

homeassistant/components/subaru/remote_service.py

@@ -0,0 +1,33 @@
+"""Remote vehicle services for Subaru integration."""
+import logging
+
+from subarulink.exceptions import SubaruException
+
+from homeassistant.exceptions import HomeAssistantError
+
+from .const import SERVICE_UNLOCK, VEHICLE_NAME, VEHICLE_VIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_call_remote_service(controller, cmd, vehicle_info, arg=None):
+    """Execute subarulink remote command."""
+    car_name = vehicle_info[VEHICLE_NAME]
+    vin = vehicle_info[VEHICLE_VIN]
+
+    _LOGGER.debug("Sending %s command command to %s", cmd, car_name)
+    success = False
+    err_msg = ""
+    try:
+        if cmd == SERVICE_UNLOCK:
+            success = await getattr(controller, cmd)(vin, arg)
+        else:
+            success = await getattr(controller, cmd)(vin)
+    except SubaruException as err:
+        err_msg = err.message
+
+    if success:
+        _LOGGER.debug("%s command successfully completed for %s", cmd, car_name)
+        return
+
+    raise HomeAssistantError(f"Service {cmd} failed for {car_name}: {err_msg}")

homeassistant/components/subaru/services.yaml

@@ -0,0 +1,19 @@
+unlock_specific_door:
+  name: Unlock Specific Door
+  description: Unlocks specific door(s)
+  target:
+    entity:
+      domain: lock
+      integration: subaru
+  fields:
+    door:
+      name: Door
+      description: "One of the following: 'all', 'driver', 'tailgate'"
+      example: driver
+      required: true
+      selector:
+        select:
+          options:
+            - "all"
+            - "driver"
+            - "tailgate"

tests/components/subaru/test_lock.py

@@ -0,0 +1,86 @@
+"""Test Subaru locks."""
+from unittest.mock import patch
+
+from pytest import raises
+from voluptuous.error import MultipleInvalid
+
+from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN
+from homeassistant.components.subaru.const import (
+    ATTR_DOOR,
+    DOMAIN as SUBARU_DOMAIN,
+    SERVICE_UNLOCK_SPECIFIC_DOOR,
+    UNLOCK_DOOR_DRIVERS,
+)
+from homeassistant.const import ATTR_ENTITY_ID, SERVICE_LOCK, SERVICE_UNLOCK
+from homeassistant.exceptions import HomeAssistantError
+
+from .conftest import MOCK_API
+
+MOCK_API_LOCK = f"{MOCK_API}lock"
+MOCK_API_UNLOCK = f"{MOCK_API}unlock"
+DEVICE_ID = "lock.test_vehicle_2_door_locks"
+
+
+async def test_device_exists(hass, ev_entry):
+    """Test subaru lock entity exists."""
+    entity_registry = await hass.helpers.entity_registry.async_get_registry()
+    entry = entity_registry.async_get(DEVICE_ID)
+    assert entry
+
+
+async def test_lock_cmd(hass, ev_entry):
+    """Test subaru lock function."""
+    with patch(MOCK_API_LOCK) as mock_lock:
+        await hass.services.async_call(
+            LOCK_DOMAIN, SERVICE_LOCK, {ATTR_ENTITY_ID: DEVICE_ID}, blocking=True
+        )
+        await hass.async_block_till_done()
+        mock_lock.assert_called_once()
+
+
+async def test_unlock_cmd(hass, ev_entry):
+    """Test subaru unlock function."""
+    with patch(MOCK_API_UNLOCK) as mock_unlock:
+        await hass.services.async_call(
+            LOCK_DOMAIN, SERVICE_UNLOCK, {ATTR_ENTITY_ID: DEVICE_ID}, blocking=True
+        )
+        await hass.async_block_till_done()
+        mock_unlock.assert_called_once()
+
+
+async def test_lock_cmd_fails(hass, ev_entry):
+    """Test subaru lock request that initiates but fails."""
+    with patch(MOCK_API_LOCK, return_value=False) as mock_lock, raises(
+        HomeAssistantError
+    ):
+        await hass.services.async_call(
+            LOCK_DOMAIN, SERVICE_UNLOCK, {ATTR_ENTITY_ID: DEVICE_ID}, blocking=True
+        )
+        await hass.async_block_till_done()
+        mock_lock.assert_called_once()
+
+
+async def test_unlock_specific_door(hass, ev_entry):
+    """Test subaru unlock specific door function."""
+    with patch(MOCK_API_UNLOCK) as mock_unlock:
+        await hass.services.async_call(
+            SUBARU_DOMAIN,
+            SERVICE_UNLOCK_SPECIFIC_DOOR,
+            {ATTR_ENTITY_ID: DEVICE_ID, ATTR_DOOR: UNLOCK_DOOR_DRIVERS},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        mock_unlock.assert_called_once()
+
+
+async def test_unlock_specific_door_invalid(hass, ev_entry):
+    """Test subaru unlock specific door function."""
+    with patch(MOCK_API_UNLOCK) as mock_unlock, raises(MultipleInvalid):
+        await hass.services.async_call(
+            SUBARU_DOMAIN,
+            SERVICE_UNLOCK_SPECIFIC_DOOR,
+            {ATTR_ENTITY_ID: DEVICE_ID, ATTR_DOOR: "bad_value"},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+        mock_unlock.assert_not_called()