Add service for switchbot keypad vision (#160659)

Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Retha Runolfsson
2026-01-29 19:23:38 +08:00
committed by GitHub
parent 15ff5d0f74
commit dfe1990484
8 changed files with 493 additions and 1 deletions

View File

@@ -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

View File

@@ -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 = {

View File

@@ -141,5 +141,10 @@
}
}
}
},
"services": {
"add_password": {
"service": "mdi:key-plus"
}
}
}

View File

@@ -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,
)

View File

@@ -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

View File

@@ -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"
}
}
}

View File

@@ -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,
)

View File

@@ -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"}