diff --git a/homeassistant/components/switchbot/__init__.py b/homeassistant/components/switchbot/__init__.py index 7ba3cb84e97..6b8909d3e16 100644 --- a/homeassistant/components/switchbot/__init__.py +++ b/homeassistant/components/switchbot/__init__.py @@ -5,6 +5,7 @@ import logging import switchbot from homeassistant.components import bluetooth +from homeassistant.components.sensor import ConfigType from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( CONF_ADDRESS, @@ -16,7 +17,7 @@ from homeassistant.const import ( ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady -from homeassistant.helpers import device_registry as dr +from homeassistant.helpers import config_validation as cv, device_registry as dr from .const import ( CONF_ENCRYPTION_KEY, @@ -30,6 +31,10 @@ from .const import ( SupportedModels, ) from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator +from .services import async_setup_services + +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) + PLATFORMS_BY_TYPE = { SupportedModels.BULB.value: [Platform.SENSOR, Platform.LIGHT], @@ -113,6 +118,8 @@ PLATFORMS_BY_TYPE = { Platform.BINARY_SENSOR, Platform.BUTTON, ], + SupportedModels.KEYPAD_VISION.value: [Platform.SENSOR, Platform.BINARY_SENSOR], + SupportedModels.KEYPAD_VISION_PRO.value: [Platform.SENSOR, Platform.BINARY_SENSOR], } CLASS_BY_DEVICE = { SupportedModels.CEILING_LIGHT.value: switchbot.SwitchbotCeilingLight, @@ -150,12 +157,20 @@ CLASS_BY_DEVICE = { SupportedModels.GARAGE_DOOR_OPENER.value: switchbot.SwitchbotGarageDoorOpener, SupportedModels.SMART_THERMOSTAT_RADIATOR.value: switchbot.SwitchbotSmartThermostatRadiator, SupportedModels.ART_FRAME.value: switchbot.SwitchbotArtFrame, + SupportedModels.KEYPAD_VISION.value: switchbot.SwitchbotKeypadVision, + SupportedModels.KEYPAD_VISION_PRO.value: switchbot.SwitchbotKeypadVision, } _LOGGER = logging.getLogger(__name__) +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + """Set up the Switchbot Devices component.""" + async_setup_services(hass) + return True + + async def async_setup_entry(hass: HomeAssistant, entry: SwitchbotConfigEntry) -> bool: """Set up Switchbot from a config entry.""" assert entry.unique_id is not None diff --git a/homeassistant/components/switchbot/const.py b/homeassistant/components/switchbot/const.py index 0b31de28db1..c2285b8d814 100644 --- a/homeassistant/components/switchbot/const.py +++ b/homeassistant/components/switchbot/const.py @@ -62,6 +62,8 @@ class SupportedModels(StrEnum): SMART_THERMOSTAT_RADIATOR = "smart_thermostat_radiator" S20_VACUUM = "s20_vacuum" ART_FRAME = "art_frame" + KEYPAD_VISION = "keypad_vision" + KEYPAD_VISION_PRO = "keypad_vision_pro" CONNECTABLE_SUPPORTED_MODEL_TYPES = { @@ -102,6 +104,8 @@ CONNECTABLE_SUPPORTED_MODEL_TYPES = { SwitchbotModel.CLIMATE_PANEL: SupportedModels.CLIMATE_PANEL, SwitchbotModel.SMART_THERMOSTAT_RADIATOR: SupportedModels.SMART_THERMOSTAT_RADIATOR, SwitchbotModel.ART_FRAME: SupportedModels.ART_FRAME, + SwitchbotModel.KEYPAD_VISION: SupportedModels.KEYPAD_VISION, + SwitchbotModel.KEYPAD_VISION_PRO: SupportedModels.KEYPAD_VISION_PRO, } NON_CONNECTABLE_SUPPORTED_MODEL_TYPES = { @@ -142,6 +146,8 @@ ENCRYPTED_MODELS = { SwitchbotModel.GARAGE_DOOR_OPENER, SwitchbotModel.SMART_THERMOSTAT_RADIATOR, SwitchbotModel.ART_FRAME, + SwitchbotModel.KEYPAD_VISION, + SwitchbotModel.KEYPAD_VISION_PRO, } ENCRYPTED_SWITCHBOT_MODEL_TO_CLASS: dict[ @@ -165,6 +171,8 @@ ENCRYPTED_SWITCHBOT_MODEL_TO_CLASS: dict[ SwitchbotModel.GARAGE_DOOR_OPENER: switchbot.SwitchbotRelaySwitch, SwitchbotModel.SMART_THERMOSTAT_RADIATOR: switchbot.SwitchbotSmartThermostatRadiator, SwitchbotModel.ART_FRAME: switchbot.SwitchbotArtFrame, + SwitchbotModel.KEYPAD_VISION: switchbot.SwitchbotKeypadVision, + SwitchbotModel.KEYPAD_VISION_PRO: switchbot.SwitchbotKeypadVision, } HASS_SENSOR_TYPE_TO_SWITCHBOT_MODEL = { diff --git a/homeassistant/components/switchbot/icons.json b/homeassistant/components/switchbot/icons.json index 856b4bc1f98..29aedc20aa3 100644 --- a/homeassistant/components/switchbot/icons.json +++ b/homeassistant/components/switchbot/icons.json @@ -141,5 +141,10 @@ } } } + }, + "services": { + "add_password": { + "service": "mdi:key-plus" + } } } diff --git a/homeassistant/components/switchbot/services.py b/homeassistant/components/switchbot/services.py new file mode 100644 index 00000000000..d959c3ecb33 --- /dev/null +++ b/homeassistant/components/switchbot/services.py @@ -0,0 +1,119 @@ +"""Services for the SwitchBot integration.""" + +from __future__ import annotations + +import voluptuous as vol + +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import ATTR_DEVICE_ID, CONF_SENSOR_TYPE +from homeassistant.core import HomeAssistant, ServiceCall, callback +from homeassistant.exceptions import ServiceValidationError +from homeassistant.helpers import config_validation as cv, device_registry as dr + +from .const import DOMAIN, SupportedModels +from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator + +SERVICE_ADD_PASSWORD = "add_password" + +ATTR_PASSWORD = "password" + +_PASSWORD_VALIDATOR = vol.All(cv.string, cv.matches_regex(r"^\d{6,12}$")) + +SCHEMA_ADD_PASSWORD_SERVICE = vol.Schema( + { + vol.Required(ATTR_DEVICE_ID): cv.string, + vol.Required(ATTR_PASSWORD): _PASSWORD_VALIDATOR, + }, + extra=vol.ALLOW_EXTRA, +) + + +@callback +def _async_get_switchbot_entry_for_device_id( + hass: HomeAssistant, device_id: str +) -> SwitchbotConfigEntry: + """Return the loaded SwitchBot config entry for a device id.""" + device_registry = dr.async_get(hass) + if not (device_entry := device_registry.async_get(device_id)): + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="invalid_device_id", + translation_placeholders={"device_id": device_id}, + ) + + entries = [ + hass.config_entries.async_get_entry(entry_id) + for entry_id in device_entry.config_entries + ] + switchbot_entries = [ + entry for entry in entries if entry is not None and entry.domain == DOMAIN + ] + if not switchbot_entries: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="device_not_belonging", + translation_placeholders={"device_id": device_id}, + ) + + if not ( + loaded_entry := next( + ( + entry + for entry in switchbot_entries + if entry.state is ConfigEntryState.LOADED + ), + None, + ) + ): + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="device_entry_not_loaded", + translation_placeholders={"device_id": device_id}, + ) + + return loaded_entry + + +def _is_supported_keypad(entry: SwitchbotConfigEntry) -> bool: + """Return if the entry is a supported keypad model.""" + allowed_sensor_types = { + SupportedModels.KEYPAD_VISION.value, + SupportedModels.KEYPAD_VISION_PRO.value, + } + return entry.data.get(CONF_SENSOR_TYPE) in allowed_sensor_types + + +@callback +def _async_target( + hass: HomeAssistant, device_id: str +) -> SwitchbotDataUpdateCoordinator: + """Return coordinator for a single target device.""" + entry = _async_get_switchbot_entry_for_device_id(hass, device_id) + if not _is_supported_keypad(entry): + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="not_keypad_vision_device", + ) + + return entry.runtime_data + + +async def async_add_password(call: ServiceCall) -> None: + """Add a password to a SwitchBot keypad device.""" + password: str = call.data[ATTR_PASSWORD] + device_id = call.data[ATTR_DEVICE_ID] + + coordinator = _async_target(call.hass, device_id) + + await coordinator.device.add_password(password) + + +@callback +def async_setup_services(hass: HomeAssistant) -> None: + """Set up the services for the SwitchBot integration.""" + hass.services.async_register( + DOMAIN, + SERVICE_ADD_PASSWORD, + async_add_password, + schema=SCHEMA_ADD_PASSWORD_SERVICE, + ) diff --git a/homeassistant/components/switchbot/services.yaml b/homeassistant/components/switchbot/services.yaml new file mode 100644 index 00000000000..a7fe53c06ab --- /dev/null +++ b/homeassistant/components/switchbot/services.yaml @@ -0,0 +1,14 @@ +add_password: + fields: + device_id: + required: true + example: "c2d01328efd261f586e56d914e3af07e" + selector: + device: + integration: switchbot + password: + required: true + example: "123456" + selector: + text: + type: password diff --git a/homeassistant/components/switchbot/strings.json b/homeassistant/components/switchbot/strings.json index 022dc9a04bc..30da69ea3c2 100644 --- a/homeassistant/components/switchbot/strings.json +++ b/homeassistant/components/switchbot/strings.json @@ -329,9 +329,24 @@ "advertising_state_error": { "message": "{address} is not advertising state" }, + "device_entry_not_loaded": { + "message": "The device ID {device_id} is not loaded." + }, + "device_not_belonging": { + "message": "The device ID {device_id} does not belong to SwitchBot integration." + }, "device_not_found_error": { "message": "Could not find Switchbot {sensor_type} with address {address}" }, + "device_without_config_entry": { + "message": "The device ID {device_id} is not associated with a config entry." + }, + "invalid_device_id": { + "message": "The device ID {device_id} is not a valid device ID." + }, + "not_keypad_vision_device": { + "message": "This service is only supported for SwitchBot Keypad Vision devices." + }, "operation_error": { "message": "An error occurred while performing the action: {error}" }, @@ -352,5 +367,21 @@ } } } + }, + "services": { + "add_password": { + "description": "Add a password to your keypad vision device.", + "fields": { + "device_id": { + "description": "The device ID of the keypad vision device", + "name": "Device ID" + }, + "password": { + "description": "A 6 to 12 digit password", + "name": "Password" + } + }, + "name": "Add password" + } } } diff --git a/tests/components/switchbot/__init__.py b/tests/components/switchbot/__init__.py index fb3e04cc09c..6c796b32075 100644 --- a/tests/components/switchbot/__init__.py +++ b/tests/components/switchbot/__init__.py @@ -1297,3 +1297,53 @@ ART_FRAME_INFO = BluetoothServiceInfoBleak( connectable=True, tx_power=-127, ) + +KEYPAD_VISION_INFO = BluetoothServiceInfoBleak( + name="Keypad Vision", + manufacturer_data={ + 2409: b"\xb0\xe9\xfe\xe5\x04\x1e\xac\xdf\x00\x00\x00\x00\x00\x02" + }, + service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"\x00\x00_\x01\x11\x03x"}, + service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"], + address="AA:BB:CC:DD:EE:FF", + rssi=-60, + source="local", + advertisement=generate_advertisement_data( + local_name="Keypad Vision", + manufacturer_data={ + 2409: b"\xb0\xe9\xfe\xe5\x04\x1e\xac\xdf\x00\x00\x00\x00\x00\x02" + }, + service_data={ + "0000fd3d-0000-1000-8000-00805f9b34fb": b"\x00\x00_\x01\x11\x03x" + }, + service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"], + ), + device=generate_ble_device("AA:BB:CC:DD:EE:FF", "Keypad Vision"), + time=0, + connectable=True, + tx_power=-127, +) + +KEYPAD_VISION_PRO_INFO = BluetoothServiceInfoBleak( + name="Keypad Vision Pro", + manufacturer_data={2409: b"\xb0\xe9\xfe\xde\xb6\x8c+`\x00\x00\x00\x00\x00\x002"}, + service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"\x00\x00`\x01\x11Q\x98"}, + service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"], + address="AA:BB:CC:DD:EE:FF", + rssi=-60, + source="local", + advertisement=generate_advertisement_data( + local_name="Keypad Vision Pro", + manufacturer_data={ + 2409: b"\xb0\xe9\xfe\xde\xb6\x8c+`\x00\x00\x00\x00\x00\x002" + }, + service_data={ + "0000fd3d-0000-1000-8000-00805f9b34fb": b"\x00\x00`\x01\x11Q\x98" + }, + service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"], + ), + device=generate_ble_device("AA:BB:CC:DD:EE:FF", "Keypad Vision Pro"), + time=0, + connectable=True, + tx_power=-127, +) diff --git a/tests/components/switchbot/test_services.py b/tests/components/switchbot/test_services.py new file mode 100644 index 00000000000..85c4b313181 --- /dev/null +++ b/tests/components/switchbot/test_services.py @@ -0,0 +1,250 @@ +"""Test the switchbot services.""" + +from collections.abc import Callable +from unittest.mock import AsyncMock, patch + +import pytest + +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak +from homeassistant.components.switchbot.const import DOMAIN +from homeassistant.components.switchbot.services import ( + SERVICE_ADD_PASSWORD, + async_setup_services, +) +from homeassistant.const import ATTR_DEVICE_ID +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ServiceValidationError +from homeassistant.helpers import device_registry as dr + +from . import ( + KEYPAD_VISION_INFO, + KEYPAD_VISION_PRO_INFO, + SMART_THERMOSTAT_RADIATOR_SERVICE_INFO, +) + +from tests.common import MockConfigEntry +from tests.components.bluetooth import inject_bluetooth_service_info + + +@pytest.mark.parametrize( + ("ble_service_info", "sensor_type"), + [ + (KEYPAD_VISION_INFO, "keypad_vision"), + (KEYPAD_VISION_PRO_INFO, "keypad_vision_pro"), + ], +) +async def test_add_password_service( + hass: HomeAssistant, + mock_entry_encrypted_factory: Callable[[str], MockConfigEntry], + ble_service_info: BluetoothServiceInfoBleak, + sensor_type: str, + device_registry: dr.DeviceRegistry, +) -> None: + """Test the add_password service.""" + inject_bluetooth_service_info(hass, ble_service_info) + + entry = mock_entry_encrypted_factory(sensor_type=sensor_type) + entry.add_to_hass(hass) + + mocked_instance = AsyncMock(return_value=True) + with patch.multiple( + "homeassistant.components.switchbot.switchbot.SwitchbotKeypadVision", + update=AsyncMock(return_value=None), + add_password=mocked_instance, + ): + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + device_entry = dr.async_entries_for_config_entry( + device_registry, entry.entry_id + )[0] + + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_PASSWORD, + { + ATTR_DEVICE_ID: device_entry.id, + "password": "123456", + }, + blocking=True, + ) + + mocked_instance.assert_called_once_with("123456") + + +async def test_device_not_found( + hass: HomeAssistant, + mock_entry_encrypted_factory: Callable[[str], MockConfigEntry], +) -> None: + """Test the add_password service with non-existent device.""" + inject_bluetooth_service_info(hass, KEYPAD_VISION_INFO) + + entry = mock_entry_encrypted_factory(sensor_type="keypad_vision") + entry.add_to_hass(hass) + + with patch.multiple( + "homeassistant.components.switchbot.switchbot.SwitchbotKeypadVision", + update=AsyncMock(return_value=None), + ): + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + with pytest.raises(ServiceValidationError) as err: + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_PASSWORD, + { + ATTR_DEVICE_ID: "nonexistent_device", + "password": "123456", + }, + blocking=True, + ) + + assert err.value.translation_domain == DOMAIN + assert err.value.translation_key == "invalid_device_id" + assert err.value.translation_placeholders == {"device_id": "nonexistent_device"} + + +async def test_device_not_belonging( + hass: HomeAssistant, + mock_entry_encrypted_factory: Callable[[str], MockConfigEntry], + device_registry: dr.DeviceRegistry, +) -> None: + """Test service errors when device belongs to a different integration.""" + inject_bluetooth_service_info(hass, KEYPAD_VISION_INFO) + + entry = mock_entry_encrypted_factory(sensor_type="keypad_vision") + entry.add_to_hass(hass) + + with patch.multiple( + "homeassistant.components.switchbot.switchbot.SwitchbotKeypadVision", + update=AsyncMock(return_value=None), + ): + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + other_entry = MockConfigEntry(domain="not_switchbot", data={}, title="Other") + other_entry.add_to_hass(hass) + + device_entry = device_registry.async_get_or_create( + config_entry_id=other_entry.entry_id, + identifiers={("not_switchbot", "other_unique_id")}, + name="Other device", + ) + + with pytest.raises(ServiceValidationError) as err: + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_PASSWORD, + { + ATTR_DEVICE_ID: device_entry.id, + "password": "123456", + }, + blocking=True, + ) + + assert err.value.translation_domain == DOMAIN + assert err.value.translation_key == "device_not_belonging" + assert err.value.translation_placeholders == {"device_id": device_entry.id} + + +async def test_device_entry_not_loaded( + hass: HomeAssistant, + mock_entry_encrypted_factory: Callable[[str], MockConfigEntry], + device_registry: dr.DeviceRegistry, +) -> None: + """Test service errors when the config entry is not loaded.""" + inject_bluetooth_service_info(hass, KEYPAD_VISION_INFO) + + entry = mock_entry_encrypted_factory(sensor_type="keypad_vision") + entry.add_to_hass(hass) + + with patch.multiple( + "homeassistant.components.switchbot.switchbot.SwitchbotKeypadVision", + update=AsyncMock(return_value=None), + ): + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + second_entry = mock_entry_encrypted_factory(sensor_type="keypad_vision") + second_entry.add_to_hass(hass) + + device_entry = device_registry.async_get_or_create( + config_entry_id=second_entry.entry_id, + identifiers={(DOMAIN, "not_loaded_unique_id")}, + name="Not loaded device", + ) + + with pytest.raises(ServiceValidationError) as err: + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_PASSWORD, + { + ATTR_DEVICE_ID: device_entry.id, + "password": "123456", + }, + blocking=True, + ) + + assert err.value.translation_domain == DOMAIN + assert err.value.translation_key == "device_entry_not_loaded" + assert err.value.translation_placeholders == {"device_id": device_entry.id} + + +async def test_service_unsupported_device( + hass: HomeAssistant, + mock_entry_encrypted_factory: Callable[[str], MockConfigEntry], + device_registry: dr.DeviceRegistry, +) -> None: + """Test service errors when the device does not support the service.""" + inject_bluetooth_service_info(hass, SMART_THERMOSTAT_RADIATOR_SERVICE_INFO) + + entry = mock_entry_encrypted_factory(sensor_type="smart_thermostat_radiator") + entry.add_to_hass(hass) + + with patch.multiple( + "homeassistant.components.switchbot.switchbot.SwitchbotSmartThermostatRadiator", + update=AsyncMock(return_value=None), + ): + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + device_entry = dr.async_entries_for_config_entry(device_registry, entry.entry_id)[0] + + with pytest.raises(ServiceValidationError) as err: + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_PASSWORD, + { + ATTR_DEVICE_ID: device_entry.id, + "password": "123456", + }, + blocking=True, + ) + + assert err.value.translation_domain == DOMAIN + assert err.value.translation_key == "not_keypad_vision_device" + + +async def test_device_without_config_entry_id( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, +) -> None: + """Test service errors when device has no config entry id.""" + async_setup_services(hass) + + entry = MockConfigEntry(domain=DOMAIN, data={}, title="No entry device") + entry.add_to_hass(hass) + + with pytest.raises(ServiceValidationError) as err: + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_PASSWORD, + { + ATTR_DEVICE_ID: "abc", + "password": "123456", + }, + blocking=True, + ) + + assert err.value.translation_domain == DOMAIN + assert err.value.translation_key == "invalid_device_id" + assert err.value.translation_placeholders == {"device_id": "abc"}