mirror of
https://github.com/home-assistant/core.git
synced 2026-02-15 07:36:16 +00:00
Add Shelly support for virtual buttons (#151940)
This commit is contained in:
@@ -11,6 +11,7 @@ from aioshelly.const import BLU_TRV_IDENTIFIER, MODEL_BLU_GATEWAY_G3, RPC_GENERA
|
||||
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
|
||||
|
||||
from homeassistant.components.button import (
|
||||
DOMAIN as BUTTON_PLATFORM,
|
||||
ButtonDeviceClass,
|
||||
ButtonEntity,
|
||||
ButtonEntityDescription,
|
||||
@@ -26,7 +27,14 @@ from homeassistant.util import slugify
|
||||
from .const import DOMAIN, LOGGER, SHELLY_GAS_MODELS
|
||||
from .coordinator import ShellyBlockCoordinator, ShellyConfigEntry, ShellyRpcCoordinator
|
||||
from .entity import get_entity_block_device_info, get_entity_rpc_device_info
|
||||
from .utils import get_blu_trv_device_info, get_device_entry_gen, get_rpc_key_ids
|
||||
from .utils import (
|
||||
async_remove_orphaned_entities,
|
||||
get_blu_trv_device_info,
|
||||
get_device_entry_gen,
|
||||
get_rpc_entity_name,
|
||||
get_rpc_key_ids,
|
||||
get_virtual_component_ids,
|
||||
)
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
@@ -87,6 +95,13 @@ BLU_TRV_BUTTONS: Final[list[ShellyButtonDescription]] = [
|
||||
),
|
||||
]
|
||||
|
||||
VIRTUAL_BUTTONS: Final[list[ShellyButtonDescription]] = [
|
||||
ShellyButtonDescription[ShellyRpcCoordinator](
|
||||
key="button",
|
||||
press_action="single_push",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@callback
|
||||
def async_migrate_unique_ids(
|
||||
@@ -138,7 +153,7 @@ async def async_setup_entry(
|
||||
hass, config_entry.entry_id, partial(async_migrate_unique_ids, coordinator)
|
||||
)
|
||||
|
||||
entities: list[ShellyButton | ShellyBluTrvButton] = []
|
||||
entities: list[ShellyButton | ShellyBluTrvButton | ShellyVirtualButton] = []
|
||||
|
||||
entities.extend(
|
||||
ShellyButton(coordinator, button)
|
||||
@@ -146,10 +161,20 @@ async def async_setup_entry(
|
||||
if button.supported(coordinator)
|
||||
)
|
||||
|
||||
if blutrv_key_ids := get_rpc_key_ids(coordinator.device.status, BLU_TRV_IDENTIFIER):
|
||||
if TYPE_CHECKING:
|
||||
assert isinstance(coordinator, ShellyRpcCoordinator)
|
||||
if not isinstance(coordinator, ShellyRpcCoordinator):
|
||||
async_add_entities(entities)
|
||||
return
|
||||
|
||||
# add virtual buttons
|
||||
if virtual_button_ids := get_rpc_key_ids(coordinator.device.status, "button"):
|
||||
entities.extend(
|
||||
ShellyVirtualButton(coordinator, button, id_)
|
||||
for id_ in virtual_button_ids
|
||||
for button in VIRTUAL_BUTTONS
|
||||
)
|
||||
|
||||
# add BLU TRV buttons
|
||||
if blutrv_key_ids := get_rpc_key_ids(coordinator.device.status, BLU_TRV_IDENTIFIER):
|
||||
entities.extend(
|
||||
ShellyBluTrvButton(coordinator, button, id_)
|
||||
for id_ in blutrv_key_ids
|
||||
@@ -159,6 +184,19 @@ async def async_setup_entry(
|
||||
|
||||
async_add_entities(entities)
|
||||
|
||||
# the user can remove virtual components from the device configuration, so
|
||||
# we need to remove orphaned entities
|
||||
virtual_button_component_ids = get_virtual_component_ids(
|
||||
coordinator.device.config, BUTTON_PLATFORM
|
||||
)
|
||||
async_remove_orphaned_entities(
|
||||
hass,
|
||||
config_entry.entry_id,
|
||||
coordinator.mac,
|
||||
BUTTON_PLATFORM,
|
||||
virtual_button_component_ids,
|
||||
)
|
||||
|
||||
|
||||
class ShellyBaseButton(
|
||||
CoordinatorEntity[ShellyRpcCoordinator | ShellyBlockCoordinator], ButtonEntity
|
||||
@@ -273,3 +311,32 @@ class ShellyBluTrvButton(ShellyBaseButton):
|
||||
assert method is not None
|
||||
|
||||
await method(self._id)
|
||||
|
||||
|
||||
class ShellyVirtualButton(ShellyBaseButton):
|
||||
"""Defines a Shelly virtual component button."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: ShellyRpcCoordinator,
|
||||
description: ShellyButtonDescription,
|
||||
_id: int,
|
||||
) -> None:
|
||||
"""Initialize Shelly virtual component button."""
|
||||
super().__init__(coordinator, description)
|
||||
|
||||
self._attr_unique_id = f"{coordinator.mac}-{description.key}:{_id}"
|
||||
self._attr_device_info = get_entity_rpc_device_info(coordinator)
|
||||
self._attr_name = get_rpc_entity_name(
|
||||
coordinator.device, f"{description.key}:{_id}"
|
||||
)
|
||||
self._id = _id
|
||||
|
||||
async def _press_method(self) -> None:
|
||||
"""Press method."""
|
||||
if TYPE_CHECKING:
|
||||
assert isinstance(self.coordinator, ShellyRpcCoordinator)
|
||||
|
||||
await self.coordinator.device.button_trigger(
|
||||
self._id, self.entity_description.press_action
|
||||
)
|
||||
|
||||
@@ -265,9 +265,10 @@ DEVICES_WITHOUT_FIRMWARE_CHANGELOG = (
|
||||
|
||||
CONF_GEN = "gen"
|
||||
|
||||
VIRTUAL_COMPONENTS = ("boolean", "enum", "input", "number", "text")
|
||||
VIRTUAL_COMPONENTS = ("boolean", "button", "enum", "input", "number", "text")
|
||||
VIRTUAL_COMPONENTS_MAP = {
|
||||
"binary_sensor": {"types": ["boolean"], "modes": ["label"]},
|
||||
"button": {"types": ["button"], "modes": ["button"]},
|
||||
"number": {"types": ["number"], "modes": ["field", "slider"]},
|
||||
"select": {"types": ["enum"], "modes": ["dropdown"]},
|
||||
"sensor": {"types": ["enum", "number", "text"], "modes": ["label"]},
|
||||
|
||||
@@ -631,6 +631,11 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
|
||||
"""Handle device events."""
|
||||
events: list[dict[str, Any]] = event_data["events"]
|
||||
for event in events:
|
||||
# filter out button events as they are triggered by button entities
|
||||
component = event.get("component")
|
||||
if component is not None and component.startswith("button"):
|
||||
continue
|
||||
|
||||
event_type = event.get("event")
|
||||
if event_type is None:
|
||||
continue
|
||||
|
||||
@@ -96,3 +96,51 @@
|
||||
'state': 'unknown',
|
||||
})
|
||||
# ---
|
||||
# name: test_rpc_device_virtual_button[button.test_name_button-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': set({
|
||||
}),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'button',
|
||||
'entity_category': None,
|
||||
'entity_id': 'button.test_name_button',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Button',
|
||||
'platform': 'shelly',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': None,
|
||||
'unique_id': '123456789ABC-button:200',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_rpc_device_virtual_button[button.test_name_button-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'friendly_name': 'Test name Button',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'button.test_name_button',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'unknown',
|
||||
})
|
||||
# ---
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Tests for Shelly button platform."""
|
||||
|
||||
from copy import deepcopy
|
||||
from unittest.mock import Mock
|
||||
|
||||
from aioshelly.const import MODEL_BLU_GATEWAY_G3
|
||||
@@ -13,9 +14,10 @@ from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
|
||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.device_registry import DeviceRegistry
|
||||
from homeassistant.helpers.entity_registry import EntityRegistry
|
||||
|
||||
from . import init_integration
|
||||
from . import init_integration, register_device, register_entity
|
||||
|
||||
|
||||
async def test_block_button(
|
||||
@@ -278,3 +280,65 @@ async def test_rpc_blu_trv_button_auth_error(
|
||||
assert "context" in flow
|
||||
assert flow["context"].get("source") == SOURCE_REAUTH
|
||||
assert flow["context"].get("entry_id") == entry.entry_id
|
||||
|
||||
|
||||
async def test_rpc_device_virtual_button(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: EntityRegistry,
|
||||
mock_rpc_device: Mock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
"""Test a virtual button for RPC device."""
|
||||
config = deepcopy(mock_rpc_device.config)
|
||||
config["button:200"] = {
|
||||
"name": "Button",
|
||||
"meta": {"ui": {"view": "button"}},
|
||||
}
|
||||
monkeypatch.setattr(mock_rpc_device, "config", config)
|
||||
|
||||
status = deepcopy(mock_rpc_device.status)
|
||||
status["button:200"] = {"value": None}
|
||||
monkeypatch.setattr(mock_rpc_device, "status", status)
|
||||
|
||||
await init_integration(hass, 3)
|
||||
entity_id = "button.test_name_button"
|
||||
|
||||
assert (state := hass.states.get(entity_id))
|
||||
assert state == snapshot(name=f"{entity_id}-state")
|
||||
|
||||
assert (entry := entity_registry.async_get(entity_id))
|
||||
assert entry == snapshot(name=f"{entity_id}-entry")
|
||||
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
SERVICE_PRESS,
|
||||
{ATTR_ENTITY_ID: entity_id},
|
||||
blocking=True,
|
||||
)
|
||||
mock_rpc_device.button_trigger.assert_called_once_with(200, "single_push")
|
||||
|
||||
|
||||
async def test_rpc_remove_virtual_button_when_orphaned(
|
||||
hass: HomeAssistant,
|
||||
entity_registry: EntityRegistry,
|
||||
device_registry: DeviceRegistry,
|
||||
mock_rpc_device: Mock,
|
||||
) -> None:
|
||||
"""Check whether the virtual button 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,
|
||||
BUTTON_DOMAIN,
|
||||
"test_name_button_200",
|
||||
"button:200",
|
||||
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
|
||||
|
||||
@@ -553,6 +553,57 @@ async def test_rpc_click_event(
|
||||
}
|
||||
|
||||
|
||||
async def test_rpc_ignore_virtual_click_event(
|
||||
hass: HomeAssistant,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
mock_rpc_device: Mock,
|
||||
events: list[Event],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Test RPC virtual click events are ignored as they are triggered by the integration."""
|
||||
await init_integration(hass, 2)
|
||||
|
||||
# Generate a virtual button event
|
||||
inject_rpc_device_event(
|
||||
monkeypatch,
|
||||
mock_rpc_device,
|
||||
{
|
||||
"events": [
|
||||
{
|
||||
"component": "button:200",
|
||||
"id": 200,
|
||||
"event": "single_push",
|
||||
"ts": 1757358109.89,
|
||||
}
|
||||
],
|
||||
"ts": 757358109.89,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(events) == 0
|
||||
|
||||
# Generate valid event
|
||||
inject_rpc_device_event(
|
||||
monkeypatch,
|
||||
mock_rpc_device,
|
||||
{
|
||||
"events": [
|
||||
{
|
||||
"data": [],
|
||||
"event": "single_push",
|
||||
"id": 0,
|
||||
"ts": 1668522399.2,
|
||||
}
|
||||
],
|
||||
"ts": 1668522399.2,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(events) == 1
|
||||
|
||||
|
||||
async def test_rpc_update_entry_sleep_period(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
|
||||
Reference in New Issue
Block a user