Add support for Shelly virtual boolean component (#119932)

Co-authored-by: Maciej Bieniek <478555+bieniu@users.noreply.github.com>
This commit is contained in:
Maciej Bieniek 2024-07-10 23:51:51 +02:00 committed by GitHub
parent 311b1e236a
commit 70f05e5f13
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 448 additions and 7 deletions

View File

@ -8,6 +8,7 @@ from typing import Final, cast
from aioshelly.const import RPC_GENERATIONS
from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_PLATFORM,
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
@ -33,7 +34,9 @@ from .entity import (
async_setup_entry_rpc,
)
from .utils import (
async_remove_orphaned_virtual_entities,
get_device_entry_gen,
get_virtual_component_ids,
is_block_momentary_input,
is_rpc_momentary_input,
)
@ -215,6 +218,11 @@ RPC_SENSORS: Final = {
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
),
"boolean": RpcBinarySensorDescription(
key="boolean",
sub_key="value",
has_entity_name=True,
),
}
@ -234,9 +242,26 @@ async def async_setup_entry(
RpcSleepingBinarySensor,
)
else:
coordinator = config_entry.runtime_data.rpc
assert coordinator
async_setup_entry_rpc(
hass, config_entry, async_add_entities, RPC_SENSORS, RpcBinarySensor
)
# the user can remove virtual components from the device configuration, so
# we need to remove orphaned entities
virtual_binary_sensor_ids = get_virtual_component_ids(
coordinator.device.config, BINARY_SENSOR_PLATFORM
)
async_remove_orphaned_virtual_entities(
hass,
config_entry.entry_id,
coordinator.mac,
BINARY_SENSOR_PLATFORM,
"boolean",
virtual_binary_sensor_ids,
)
return
if config_entry.data[CONF_SLEEP_PERIOD]:

View File

@ -238,3 +238,8 @@ DEVICES_WITHOUT_FIRMWARE_CHANGELOG = (
CONF_GEN = "gen"
SHELLY_PLUS_RGBW_CHANNELS = 4
VIRTUAL_COMPONENTS_MAP = {
"binary_sensor": {"type": "boolean", "mode": "label"},
"switch": {"type": "boolean", "mode": "toggle"},
}

View File

@ -551,7 +551,7 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
for event_callback in self._event_listeners:
event_callback(event)
if event_type == "config_changed":
if event_type in ("component_added", "component_removed", "config_changed"):
self.update_sleep_period()
LOGGER.info(
"Config for %s changed, reloading entry in %s seconds",
@ -739,6 +739,7 @@ class ShellyRpcPollingCoordinator(ShellyCoordinatorBase[RpcDevice]):
LOGGER.debug("Polling Shelly RPC Device - %s", self.name)
try:
await self.device.update_status()
await self.device.get_dynamic_components()
except (DeviceConnectionError, RpcCallError) as err:
raise UpdateFailed(f"Device disconnected: {err!r}") from err
except InvalidAuthError:

View File

@ -505,6 +505,8 @@ class ShellyRpcAttributeEntity(ShellyRpcEntity, Entity):
self._attr_unique_id = f"{super().unique_id}-{attribute}"
self._attr_name = get_rpc_entity_name(coordinator.device, key, description.name)
self._last_value = None
id_key = key.split(":")[-1]
self._id = int(id_key) if id_key.isnumeric() else None
@property
def sub_status(self) -> Any:

View File

@ -9,7 +9,7 @@
"iot_class": "local_push",
"loggers": ["aioshelly"],
"quality_scale": "platinum",
"requirements": ["aioshelly==11.0.0"],
"requirements": ["aioshelly==11.1.0"],
"zeroconf": [
{
"type": "_http._tcp.local.",

View File

@ -8,7 +8,11 @@ from typing import Any, cast
from aioshelly.block_device import Block
from aioshelly.const import MODEL_2, MODEL_25, MODEL_WALL_DISPLAY, RPC_GENERATIONS
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.components.switch import (
DOMAIN as SWITCH_PLATFORM,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.const import STATE_ON, EntityCategory
from homeassistant.core import HomeAssistant, State, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -19,15 +23,20 @@ from .const import CONF_SLEEP_PERIOD, MOTION_MODELS
from .coordinator import ShellyBlockCoordinator, ShellyConfigEntry, ShellyRpcCoordinator
from .entity import (
BlockEntityDescription,
RpcEntityDescription,
ShellyBlockEntity,
ShellyRpcAttributeEntity,
ShellyRpcEntity,
ShellySleepingBlockAttributeEntity,
async_setup_entry_attribute_entities,
async_setup_rpc_attribute_entities,
)
from .utils import (
async_remove_orphaned_virtual_entities,
async_remove_shelly_entity,
get_device_entry_gen,
get_rpc_key_ids,
get_virtual_component_ids,
is_block_channel_type_light,
is_rpc_channel_type_light,
is_rpc_thermostat_internal_actuator,
@ -47,6 +56,17 @@ MOTION_SWITCH = BlockSwitchDescription(
)
@dataclass(frozen=True, kw_only=True)
class RpcSwitchDescription(RpcEntityDescription, SwitchEntityDescription):
"""Class to describe a RPC virtual switch."""
RPC_VIRTUAL_SWITCH = RpcSwitchDescription(
key="boolean",
sub_key="value",
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ShellyConfigEntry,
@ -148,6 +168,28 @@ def async_setup_rpc_entry(
unique_id = f"{coordinator.mac}-switch:{id_}"
async_remove_shelly_entity(hass, "light", unique_id)
async_setup_rpc_attribute_entities(
hass,
config_entry,
async_add_entities,
{"boolean": RPC_VIRTUAL_SWITCH},
RpcVirtualSwitch,
)
# the user can remove virtual components from the device configuration, so we need
# to remove orphaned entities
virtual_switch_ids = get_virtual_component_ids(
coordinator.device.config, SWITCH_PLATFORM
)
async_remove_orphaned_virtual_entities(
hass,
config_entry.entry_id,
coordinator.mac,
SWITCH_PLATFORM,
"boolean",
virtual_switch_ids,
)
if not switch_ids:
return
@ -255,3 +297,23 @@ class RpcRelaySwitch(ShellyRpcEntity, SwitchEntity):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off relay."""
await self.call_rpc("Switch.Set", {"id": self._id, "on": False})
class RpcVirtualSwitch(ShellyRpcAttributeEntity, SwitchEntity):
"""Entity that controls a virtual boolean component on RPC based Shelly devices."""
entity_description: RpcSwitchDescription
_attr_has_entity_name = True
@property
def is_on(self) -> bool:
"""If switch is on."""
return bool(self.attribute_value)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on relay."""
await self.call_rpc("Boolean.Set", {"id": self._id, "value": True})
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off relay."""
await self.call_rpc("Boolean.Set", {"id": self._id, "value": False})

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from datetime import datetime, timedelta
from ipaddress import IPv4Address
import re
from types import MappingProxyType
from typing import Any, cast
@ -52,6 +53,7 @@ from .const import (
SHBTN_MODELS,
SHIX3_1_INPUTS_EVENTS_TYPES,
UPTIME_DEVIATION,
VIRTUAL_COMPONENTS_MAP,
)
@ -321,6 +323,8 @@ def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
return f"{device_name} {key.replace(':', '_')}"
if key.startswith("em1"):
return f"{device_name} EM{key.split(':')[-1]}"
if key.startswith("boolean:"):
return key.replace(":", " ").title()
return device_name
return entity_name
@ -497,3 +501,55 @@ def async_remove_shelly_rpc_entities(
def is_rpc_thermostat_mode(ident: int, status: dict[str, Any]) -> bool:
"""Return True if 'thermostat:<IDent>' is present in the status."""
return f"thermostat:{ident}" in status
def get_virtual_component_ids(config: dict[str, Any], platform: str) -> list[str]:
"""Return a list of virtual component IDs for a platform."""
component = VIRTUAL_COMPONENTS_MAP.get(platform)
if not component:
return []
return [
k
for k, v in config.items()
if k.startswith(component["type"])
and v["meta"]["ui"]["view"] == component["mode"]
]
@callback
def async_remove_orphaned_virtual_entities(
hass: HomeAssistant,
config_entry_id: str,
mac: str,
platform: str,
virt_comp_type: str,
virt_comp_ids: list[str],
) -> None:
"""Remove orphaned virtual entities."""
orphaned_entities = []
entity_reg = er.async_get(hass)
device_reg = dr.async_get(hass)
if not (
devices := device_reg.devices.get_devices_for_config_entry_id(config_entry_id)
):
return
device_id = devices[0].id
entities = er.async_entries_for_device(entity_reg, device_id, True)
for entity in entities:
if not entity.entity_id.startswith(platform):
continue
if virt_comp_type not in entity.unique_id:
continue
# we are looking for the component ID, e.g. boolean:201
if not (match := re.search(r"[a-z]+:\d+", entity.unique_id)):
continue
virt_comp_id = match.group()
if virt_comp_id not in virt_comp_ids:
orphaned_entities.append(f"{virt_comp_id}-{virt_comp_type}")
if orphaned_entities:
async_remove_shelly_rpc_entities(hass, platform, mac, orphaned_entities)

View File

@ -365,7 +365,7 @@ aioruuvigateway==0.1.0
aiosenz==1.0.0
# homeassistant.components.shelly
aioshelly==11.0.0
aioshelly==11.1.0
# homeassistant.components.skybell
aioskybell==22.7.0

View File

@ -338,7 +338,7 @@ aioruuvigateway==0.1.0
aiosenz==1.0.0
# homeassistant.components.shelly
aioshelly==11.0.0
aioshelly==11.1.0
# homeassistant.components.skybell
aioskybell==22.7.0

View File

@ -23,6 +23,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import (
CONNECTION_NETWORK_MAC,
DeviceEntry,
DeviceRegistry,
format_mac,
)
@ -111,6 +112,7 @@ def register_entity(
unique_id: str,
config_entry: ConfigEntry | None = None,
capabilities: Mapping[str, Any] | None = None,
device_id: str | None = None,
) -> str:
"""Register enabled entity, return entity_id."""
entity_registry = er.async_get(hass)
@ -122,6 +124,7 @@ def register_entity(
disabled_by=None,
config_entry=config_entry,
capabilities=capabilities,
device_id=device_id,
)
return f"{domain}.{object_id}"
@ -145,9 +148,11 @@ def get_entity_state(hass: HomeAssistant, entity_id: str) -> str:
return entity.state
def register_device(device_registry: DeviceRegistry, config_entry: ConfigEntry) -> None:
def register_device(
device_registry: DeviceRegistry, config_entry: ConfigEntry
) -> DeviceEntry:
"""Register Shelly device."""
device_registry.async_get_or_create(
return device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(CONNECTION_NETWORK_MAC, format_mac(MOCK_MAC))},
)

View File

@ -1,5 +1,6 @@
"""Tests for Shelly binary sensor platform."""
from copy import deepcopy
from unittest.mock import Mock
from aioshelly.const import MODEL_MOTION
@ -10,6 +11,7 @@ from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAI
from homeassistant.components.shelly.const import UPDATE_PERIOD_MULTIPLIER
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
@ -353,3 +355,125 @@ async def test_rpc_restored_sleeping_binary_sensor_no_last_state(
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
async def test_rpc_device_virtual_binary_sensor_with_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test a virtual binary sensor for RPC device."""
config = deepcopy(mock_rpc_device.config)
config["boolean:203"] = {
"name": "Virtual binary sensor",
"meta": {"ui": {"view": "label"}},
}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:203"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
entity_id = "binary_sensor.test_name_virtual_binary_sensor"
await init_integration(hass, 3)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-boolean:203-boolean"
monkeypatch.setitem(mock_rpc_device.status["boolean:203"], "value", False)
mock_rpc_device.mock_update()
assert hass.states.get(entity_id).state == STATE_OFF
async def test_rpc_device_virtual_binary_sensor_without_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test a virtual binary sensor for RPC device."""
config = deepcopy(mock_rpc_device.config)
config["boolean:203"] = {"name": None, "meta": {"ui": {"view": "label"}}}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:203"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
entity_id = "binary_sensor.test_name_boolean_203"
await init_integration(hass, 3)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-boolean:203-boolean"
async def test_rpc_remove_virtual_binary_sensor_when_mode_toggle(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: DeviceRegistry,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test if the virtual binary sensor will be removed if the mode has been changed to a toggle."""
config = deepcopy(mock_rpc_device.config)
config["boolean:200"] = {"name": None, "meta": {"ui": {"view": "toggle"}}}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:200"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
config_entry = await init_integration(hass, 3, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
entity_id = register_entity(
hass,
BINARY_SENSOR_DOMAIN,
"test_name_boolean_200",
"boolean:200-boolean",
config_entry,
device_id=device_entry.id,
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
entry = entity_registry.async_get(entity_id)
assert not entry
async def test_rpc_remove_virtual_binary_sensor_when_orphaned(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: DeviceRegistry,
mock_rpc_device: Mock,
) -> None:
"""Check whether the virtual binary sensor will be removed if it has been removed from the device configuration."""
config_entry = await init_integration(hass, 3, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
entity_id = register_entity(
hass,
BINARY_SENSOR_DOMAIN,
"test_name_boolean_200",
"boolean:200-boolean",
config_entry,
device_id=device_entry.id,
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
entry = entity_registry.async_get(entity_id)
assert not entry

View File

@ -25,6 +25,7 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
@ -430,3 +431,163 @@ async def test_wall_display_relay_mode(
entry = entity_registry.async_get(switch_entity_id)
assert entry
assert entry.unique_id == "123456789ABC-switch:0"
async def test_rpc_device_virtual_switch_with_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test a virtual switch for RPC device."""
config = deepcopy(mock_rpc_device.config)
config["boolean:200"] = {
"name": "Virtual switch",
"meta": {"ui": {"view": "toggle"}},
}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:200"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
entity_id = "switch.test_name_virtual_switch"
await init_integration(hass, 3)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-boolean:200-boolean"
monkeypatch.setitem(mock_rpc_device.status["boolean:200"], "value", False)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_rpc_device.mock_update()
assert hass.states.get(entity_id).state == STATE_OFF
monkeypatch.setitem(mock_rpc_device.status["boolean:200"], "value", True)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_rpc_device.mock_update()
assert hass.states.get(entity_id).state == STATE_ON
async def test_rpc_device_virtual_switch_without_name(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test a virtual switch for RPC device."""
config = deepcopy(mock_rpc_device.config)
config["boolean:200"] = {"name": None, "meta": {"ui": {"view": "toggle"}}}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:200"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
entity_id = "switch.test_name_boolean_200"
await init_integration(hass, 3)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.unique_id == "123456789ABC-boolean:200-boolean"
async def test_rpc_device_virtual_binary_sensor(
hass: HomeAssistant,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test that a switch entity has not been created for a virtual binary sensor."""
config = deepcopy(mock_rpc_device.config)
config["boolean:200"] = {"name": None, "meta": {"ui": {"view": "label"}}}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:200"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
entity_id = "switch.test_name_boolean_200"
await init_integration(hass, 3)
state = hass.states.get(entity_id)
assert not state
async def test_rpc_remove_virtual_switch_when_mode_label(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: DeviceRegistry,
mock_rpc_device: Mock,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Test if the virtual switch will be removed if the mode has been changed to a label."""
config = deepcopy(mock_rpc_device.config)
config["boolean:200"] = {"name": None, "meta": {"ui": {"view": "label"}}}
monkeypatch.setattr(mock_rpc_device, "config", config)
status = deepcopy(mock_rpc_device.status)
status["boolean:200"] = {"value": True}
monkeypatch.setattr(mock_rpc_device, "status", status)
config_entry = await init_integration(hass, 3, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
entity_id = register_entity(
hass,
SWITCH_DOMAIN,
"test_name_boolean_200",
"boolean:200-boolean",
config_entry,
device_id=device_entry.id,
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
entry = entity_registry.async_get(entity_id)
assert not entry
async def test_rpc_remove_virtual_switch_when_orphaned(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: DeviceRegistry,
mock_rpc_device: Mock,
) -> None:
"""Check whether the virtual switch will be removed if it has been removed from the device configuration."""
config_entry = await init_integration(hass, 3, skip_setup=True)
device_entry = register_device(device_registry, config_entry)
entity_id = register_entity(
hass,
SWITCH_DOMAIN,
"test_name_boolean_200",
"boolean:200-boolean",
config_entry,
device_id=device_entry.id,
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
entry = entity_registry.async_get(entity_id)
assert not entry