Add support for SwitchBot Lock (#84673)

* Added support for SwitchBot Lock

* Updated PySwitchbot to 0.32.1

* Updated .coveragerc

* Removed unnecessary condition

* Using library method to verify encryption key

* Added config flow tests

* Remove link from config flow description

* Added one more test for config flow

* Updated CODEOWNERS

.coveragerc

@@ -1256,6 +1256,7 @@ omit =
     homeassistant/components/switchbot/light.py
     homeassistant/components/switchbot/sensor.py
     homeassistant/components/switchbot/switch.py
+    homeassistant/components/switchbot/lock.py
     homeassistant/components/switchmate/switch.py
     homeassistant/components/syncthing/__init__.py
     homeassistant/components/syncthing/sensor.py

CODEOWNERS

@@ -1136,8 +1136,8 @@ build.json @home-assistant/supervisor
 /tests/components/switch_as_x/ @home-assistant/core
 /homeassistant/components/switchbee/ @jafar-atili
 /tests/components/switchbee/ @jafar-atili
-/homeassistant/components/switchbot/ @bdraco @danielhiversen @RenierM26 @murtas @Eloston
+/homeassistant/components/switchbot/ @bdraco @danielhiversen @RenierM26 @murtas @Eloston @dsypniewski
-/tests/components/switchbot/ @bdraco @danielhiversen @RenierM26 @murtas @Eloston
+/tests/components/switchbot/ @bdraco @danielhiversen @RenierM26 @murtas @Eloston @dsypniewski
 /homeassistant/components/switcher_kis/ @tomerfi @thecode
 /tests/components/switcher_kis/ @tomerfi @thecode
 /homeassistant/components/switchmate/ @danielhiversen @qiz-li

homeassistant/components/switchbot/__init__.py

@@ -19,6 +19,8 @@ from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 
 from .const import (
+    CONF_ENCRYPTION_KEY,
+    CONF_KEY_ID,
     CONF_RETRY_COUNT,
     CONNECTABLE_SUPPORTED_MODEL_TYPES,
     DEFAULT_RETRY_COUNT,
@@ -43,6 +45,7 @@ PLATFORMS_BY_TYPE = {
     SupportedModels.CONTACT.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
     SupportedModels.MOTION.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
     SupportedModels.HUMIDIFIER.value: [Platform.HUMIDIFIER, Platform.SENSOR],
+    SupportedModels.LOCK.value: [Platform.BINARY_SENSOR, Platform.LOCK],
 }
 CLASS_BY_DEVICE = {
     SupportedModels.CEILING_LIGHT.value: switchbot.SwitchbotCeilingLight,
@@ -52,6 +55,7 @@ CLASS_BY_DEVICE = {
     SupportedModels.BULB.value: switchbot.SwitchbotBulb,
     SupportedModels.LIGHT_STRIP.value: switchbot.SwitchbotLightStrip,
     SupportedModels.HUMIDIFIER.value: switchbot.SwitchbotHumidifier,
+    SupportedModels.LOCK.value: switchbot.SwitchbotLock,
 }
 
 
@@ -94,11 +98,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     await switchbot.close_stale_connections(ble_device)
     cls = CLASS_BY_DEVICE.get(sensor_type, switchbot.SwitchbotDevice)
-    device = cls(
+    if cls is switchbot.SwitchbotLock:
-        device=ble_device,
+        try:
-        password=entry.data.get(CONF_PASSWORD),
+            device = switchbot.SwitchbotLock(
-        retry_count=entry.options[CONF_RETRY_COUNT],
+                device=ble_device,
-    )
+                key_id=entry.data.get(CONF_KEY_ID),
+                encryption_key=entry.data.get(CONF_ENCRYPTION_KEY),
+                retry_count=entry.options[CONF_RETRY_COUNT],
+            )
+        except ValueError as error:
+            raise ConfigEntryNotReady(
+                "Invalid encryption configuration provided"
+            ) from error
+    else:
+        device = cls(
+            device=ble_device,
+            password=entry.data.get(CONF_PASSWORD),
+            retry_count=entry.options[CONF_RETRY_COUNT],
+        )
 
     coordinator = hass.data[DOMAIN][entry.entry_id] = SwitchbotDataUpdateCoordinator(
         hass,

homeassistant/components/switchbot/binary_sensor.py

@@ -44,6 +44,28 @@ BINARY_SENSOR_TYPES: dict[str, BinarySensorEntityDescription] = {
         name="Light",
         device_class=BinarySensorDeviceClass.LIGHT,
     ),
+    "door_open": BinarySensorEntityDescription(
+        key="door_status",
+        name="Door status",
+        device_class=BinarySensorDeviceClass.DOOR,
+    ),
+    "unclosed_alarm": BinarySensorEntityDescription(
+        key="unclosed_alarm",
+        name="Door unclosed alarm",
+        entity_category=EntityCategory.DIAGNOSTIC,
+        device_class=BinarySensorDeviceClass.PROBLEM,
+    ),
+    "unlocked_alarm": BinarySensorEntityDescription(
+        key="unlocked_alarm",
+        name="Door unlocked alarm",
+        entity_category=EntityCategory.DIAGNOSTIC,
+        device_class=BinarySensorDeviceClass.PROBLEM,
+    ),
+    "auto_lock_paused": BinarySensorEntityDescription(
+        key="auto_lock_paused",
+        name="Door auto-lock paused",
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
 }
 
 

homeassistant/components/switchbot/config_flow.py

@@ -4,7 +4,12 @@ from __future__ import annotations
 import logging
 from typing import Any
 
-from switchbot import SwitchBotAdvertisement, parse_advertisement_data
+from switchbot import (
+    SwitchBotAdvertisement,
+    SwitchbotLock,
+    SwitchbotModel,
+    parse_advertisement_data,
+)
 import voluptuous as vol
 
 from homeassistant.components.bluetooth import (
@@ -17,6 +22,8 @@ from homeassistant.core import callback
 from homeassistant.data_entry_flow import AbortFlow, FlowResult
 
 from .const import (
+    CONF_ENCRYPTION_KEY,
+    CONF_KEY_ID,
     CONF_RETRY_COUNT,
     CONNECTABLE_SUPPORTED_MODEL_TYPES,
     DEFAULT_RETRY_COUNT,
@@ -144,6 +151,39 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
             },
         )
 
+    async def async_step_lock_key(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the encryption key step."""
+        errors = {}
+        assert self._discovered_adv is not None
+        if user_input is not None:
+            if not await SwitchbotLock.verify_encryption_key(
+                self._discovered_adv.device,
+                user_input.get(CONF_KEY_ID),
+                user_input.get(CONF_ENCRYPTION_KEY),
+            ):
+                errors = {
+                    CONF_KEY_ID: "key_id_invalid",
+                    CONF_ENCRYPTION_KEY: "encryption_key_invalid",
+                }
+            else:
+                return await self._async_create_entry_from_discovery(user_input)
+
+        return self.async_show_form(
+            step_id="lock_key",
+            errors=errors,
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_KEY_ID): str,
+                    vol.Required(CONF_ENCRYPTION_KEY): str,
+                }
+            ),
+            description_placeholders={
+                "name": name_from_discovery(self._discovered_adv),
+            },
+        )
+
     @callback
     def _async_discover_devices(self) -> None:
         current_addresses = self._async_current_ids()
@@ -188,6 +228,8 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
         if user_input is not None:
             device_adv = self._discovered_advs[user_input[CONF_ADDRESS]]
             await self._async_set_device(device_adv)
+            if device_adv.data.get("modelName") == SwitchbotModel.LOCK:
+                return await self.async_step_lock_key()
             if device_adv.data["isEncrypted"]:
                 return await self.async_step_password()
             return await self._async_create_entry_from_discovery(user_input)
@@ -198,6 +240,8 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
             # or simply confirm it
             device_adv = list(self._discovered_advs.values())[0]
             await self._async_set_device(device_adv)
+            if device_adv.data.get("modelName") == SwitchbotModel.LOCK:
+                return await self.async_step_lock_key()
             if device_adv.data["isEncrypted"]:
                 return await self.async_step_password()
             return await self.async_step_confirm()

homeassistant/components/switchbot/const.py

@@ -24,6 +24,7 @@ class SupportedModels(StrEnum):
     PLUG = "plug"
     MOTION = "motion"
     HUMIDIFIER = "humidifier"
+    LOCK = "lock"
 
 
 CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@@ -34,6 +35,7 @@ CONNECTABLE_SUPPORTED_MODEL_TYPES = {
     SwitchbotModel.LIGHT_STRIP: SupportedModels.LIGHT_STRIP,
     SwitchbotModel.CEILING_LIGHT: SupportedModels.CEILING_LIGHT,
     SwitchbotModel.HUMIDIFIER: SupportedModels.HUMIDIFIER,
+    SwitchbotModel.LOCK: SupportedModels.LOCK,
 }
 
 NON_CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@@ -56,6 +58,8 @@ DEFAULT_RETRY_COUNT = 3
 
 # Config Options
 CONF_RETRY_COUNT = "retry_count"
+CONF_KEY_ID = "key_id"
+CONF_ENCRYPTION_KEY = "encryption_key"
 
 # Deprecated config Entry Options to be removed in 2023.4
 CONF_TIME_BETWEEN_UPDATE_COMMAND = "update_time"

homeassistant/components/switchbot/lock.py

@@ -0,0 +1,55 @@
+"""Support for SwitchBot lock platform."""
+from typing import Any
+
+import switchbot
+from switchbot.const import LockStatus
+
+from homeassistant.components.lock import LockEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN
+from .coordinator import SwitchbotDataUpdateCoordinator
+from .entity import SwitchbotEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up Switchbot lock based on a config entry."""
+    coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([(SwitchBotLock(coordinator))])
+
+
+# noinspection PyAbstractClass
+class SwitchBotLock(SwitchbotEntity, LockEntity):
+    """Representation of a Switchbot lock."""
+
+    _device: switchbot.SwitchbotLock
+
+    def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+        self._async_update_attrs()
+
+    def _async_update_attrs(self) -> None:
+        """Update the entity attributes."""
+        status = self._device.get_lock_status()
+        self._attr_is_locked = status is LockStatus.LOCKED
+        self._attr_is_locking = status is LockStatus.LOCKING
+        self._attr_is_unlocking = status is LockStatus.UNLOCKING
+        self._attr_is_jammed = status in {
+            LockStatus.LOCKING_STOP,
+            LockStatus.UNLOCKING_STOP,
+        }
+
+    async def async_lock(self, **kwargs: Any) -> None:
+        """Lock the lock."""
+        self._last_run_success = await self._device.lock()
+        self.async_write_ha_state()
+
+    async def async_unlock(self, **kwargs: Any) -> None:
+        """Unlock the lock."""
+        self._last_run_success = await self._device.unlock()
+        self.async_write_ha_state()

homeassistant/components/switchbot/manifest.json

@@ -2,7 +2,7 @@
   "domain": "switchbot",
   "name": "SwitchBot",
   "documentation": "https://www.home-assistant.io/integrations/switchbot",
-  "requirements": ["PySwitchbot==0.31.0"],
+  "requirements": ["PySwitchbot==0.33.0"],
   "config_flow": true,
   "dependencies": ["bluetooth"],
   "codeowners": [
@@ -10,7 +10,8 @@
     "@danielhiversen",
     "@RenierM26",
     "@murtas",
-    "@Eloston"
+    "@Eloston",
+    "@dsypniewski"
   ],
   "bluetooth": [
     {

homeassistant/components/switchbot/strings.json

@@ -15,9 +15,19 @@
         "data": {
           "password": "[%key:common::config_flow::data::password%]"
         }
+      },
+      "lock_key": {
+        "description": "The {name} device requires encryption key, details on how to obtain it can be found in the documentation.",
+        "data": {
+          "key_id": "Key ID",
+          "encryption_key": "Encryption key"
+        }
       }
     },
-    "error": {},
+    "error": {
+      "key_id_invalid": "Key ID or Encryption key is invalid",
+      "encryption_key_invalid": "Key ID or Encryption key is invalid"
+    },
     "abort": {
       "already_configured_device": "[%key:common::config_flow::abort::already_configured_device%]",
       "no_unconfigured_devices": "No unconfigured devices found.",

homeassistant/components/switchbot/translations/en.json

@@ -22,7 +22,18 @@
                 "data": {
                     "address": "Device address"
                 }
+            },
+            "lock_key": {
+            "description": "The {name} device requires encryption key, details on how to obtain it can be found in the documentation.",
+            "data": {
+                "key_id": "Key ID",
+                "encryption_key": "Encryption key"
             }
+      }
+        },
+        "error": {
+            "key_id_invalid": "Key ID or Encryption key is invalid",
+            "encryption_key_invalid": "Key ID or Encryption key is invalid"
         }
     },
     "options": {

requirements_all.txt

@@ -40,7 +40,7 @@ PyRMVtransport==0.3.3
 PySocks==1.7.1
 
 # homeassistant.components.switchbot
-PySwitchbot==0.31.0
+PySwitchbot==0.33.0
 
 # homeassistant.components.transport_nsw
 PyTransportNSW==0.1.1

requirements_test_all.txt

@@ -36,7 +36,7 @@ PyRMVtransport==0.3.3
 PySocks==1.7.1
 
 # homeassistant.components.switchbot
-PySwitchbot==0.31.0
+PySwitchbot==0.33.0
 
 # homeassistant.components.transport_nsw
 PyTransportNSW==0.1.1

tests/components/switchbot/__init__.py

@@ -168,6 +168,26 @@ WOSENSORTH_SERVICE_INFO = BluetoothServiceInfoBleak(
     connectable=False,
 )
 
+
+WOLOCK_SERVICE_INFO = BluetoothServiceInfoBleak(
+    name="WoLock",
+    manufacturer_data={2409: b"\xf1\t\x9fE\x1a]\xda\x83\x00 "},
+    service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"o\x80d"},
+    service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
+    address="aa:bb:cc:dd:ee:ff",
+    rssi=-60,
+    source="local",
+    advertisement=generate_advertisement_data(
+        local_name="WoLock",
+        manufacturer_data={2409: b"\xf1\t\x9fE\x1a]\xda\x83\x00 "},
+        service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"o\x80d"},
+        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
+    ),
+    device=BLEDevice("aa:bb:cc:dd:ee:ff", "WoLock"),
+    time=0,
+    connectable=True,
+)
+
 NOT_SWITCHBOT_INFO = BluetoothServiceInfoBleak(
     name="unknown",
     service_uuids=[],

tests/components/switchbot/test_config_flow.py

@@ -2,7 +2,11 @@
 
 from unittest.mock import patch
 
-from homeassistant.components.switchbot.const import CONF_RETRY_COUNT
+from homeassistant.components.switchbot.const import (
+    CONF_ENCRYPTION_KEY,
+    CONF_KEY_ID,
+    CONF_RETRY_COUNT,
+)
 from homeassistant.config_entries import SOURCE_BLUETOOTH, SOURCE_USER
 from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
 from homeassistant.data_entry_flow import FlowResultType
@@ -15,6 +19,7 @@ from . import (
     WOHAND_SERVICE_ALT_ADDRESS_INFO,
     WOHAND_SERVICE_INFO,
     WOHAND_SERVICE_INFO_NOT_CONNECTABLE,
+    WOLOCK_SERVICE_INFO,
     WOSENSORTH_SERVICE_INFO,
     init_integration,
     patch_async_setup_entry,
@@ -322,6 +327,107 @@ async def test_user_setup_single_bot_with_password(hass):
     assert len(mock_setup_entry.mock_calls) == 1
 
 
+async def test_user_setup_wolock(hass):
+    """Test the user initiated form for a lock."""
+
+    with patch(
+        "homeassistant.components.switchbot.config_flow.async_discovered_service_info",
+        return_value=[WOLOCK_SERVICE_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "lock_key"
+    assert result["errors"] == {}
+
+    with patch_async_setup_entry() as mock_setup_entry, patch(
+        "switchbot.SwitchbotLock.verify_encryption_key", return_value=True
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_KEY_ID: "ff",
+                CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert result2["title"] == "Lock EEFF"
+    assert result2["data"] == {
+        CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
+        CONF_KEY_ID: "ff",
+        CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
+        CONF_SENSOR_TYPE: "lock",
+    }
+
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_user_setup_wolock_or_bot(hass):
+    """Test the user initiated form for a lock."""
+
+    with patch(
+        "homeassistant.components.switchbot.config_flow.async_discovered_service_info",
+        return_value=[
+            WOLOCK_SERVICE_INFO,
+            WOHAND_SERVICE_ALT_ADDRESS_INFO,
+        ],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        USER_INPUT,
+    )
+    await hass.async_block_till_done()
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "lock_key"
+    assert result["errors"] == {}
+
+
+async def test_user_setup_wolock_invalid_encryption_key(hass):
+    """Test the user initiated form for a lock with invalid encryption key."""
+
+    with patch(
+        "homeassistant.components.switchbot.config_flow.async_discovered_service_info",
+        return_value=[WOLOCK_SERVICE_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "lock_key"
+    assert result["errors"] == {}
+
+    with patch_async_setup_entry() as mock_setup_entry, patch(
+        "switchbot.SwitchbotLock.verify_encryption_key", return_value=False
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_KEY_ID: "",
+                CONF_ENCRYPTION_KEY: "",
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["step_id"] == "lock_key"
+    assert result2["errors"] == {
+        CONF_KEY_ID: "key_id_invalid",
+        CONF_ENCRYPTION_KEY: "encryption_key_invalid",
+    }
+
+    assert len(mock_setup_entry.mock_calls) == 0
+
+
 async def test_user_setup_wosensor(hass):
     """Test the user initiated form with password and valid mac."""
     with patch(