Add event platform for Shelly gen2 devices (#99659)

* Add event platform for gen2 devices

* Add tests

* Add removal condition

* Simplify RpcEventDescription; fix availability

* Improve names and docstrings

* Improve the event entity name

* Use async_on_remove()

* Improve tests coverage

* Improve tests coverage

* Prefix the entity name with the device name in the old way

* Black

* Use DeviceInfo object

homeassistant/components/shelly/__init__.py

@@ -73,6 +73,7 @@ RPC_PLATFORMS: Final = [
     Platform.BINARY_SENSOR,
     Platform.BUTTON,
     Platform.COVER,
+    Platform.EVENT,
     Platform.LIGHT,
     Platform.SENSOR,
     Platform.SWITCH,

homeassistant/components/shelly/coordinator.py

@@ -389,6 +389,7 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
         self._connection_lock = asyncio.Lock()
         self._event_listeners: list[Callable[[dict[str, Any]], None]] = []
         self._ota_event_listeners: list[Callable[[dict[str, Any]], None]] = []
+        self._input_event_listeners: list[Callable[[dict[str, Any]], None]] = []
 
         entry.async_on_unload(
             hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
@@ -426,6 +427,19 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
 
         return _unsubscribe
 
+    @callback
+    def async_subscribe_input_events(
+        self, input_event_callback: Callable[[dict[str, Any]], None]
+    ) -> CALLBACK_TYPE:
+        """Subscribe to input events."""
+
+        def _unsubscribe() -> None:
+            self._input_event_listeners.remove(input_event_callback)
+
+        self._input_event_listeners.append(input_event_callback)
+
+        return _unsubscribe
+
     @callback
     def async_subscribe_events(
         self, event_callback: Callable[[dict[str, Any]], None]
@@ -469,6 +483,8 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
                 )
                 self.hass.async_create_task(self._debounced_reload.async_call())
             elif event_type in RPC_INPUTS_EVENTS_TYPES:
+                for event_callback in self._input_event_listeners:
+                    event_callback(event)
                 self.hass.bus.async_fire(
                     EVENT_SHELLY_CLICK,
                     {

homeassistant/components/shelly/event.py

@@ -0,0 +1,107 @@
+"""Event for Shelly."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any, Final
+
+from homeassistant.components.event import (
+    DOMAIN as EVENT_DOMAIN,
+    EventDeviceClass,
+    EventEntity,
+    EventEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import RPC_INPUTS_EVENTS_TYPES
+from .coordinator import ShellyRpcCoordinator, get_entry_data
+from .utils import (
+    async_remove_shelly_entity,
+    get_device_entry_gen,
+    get_rpc_input_name,
+    get_rpc_key_instances,
+    is_rpc_momentary_input,
+)
+
+
+@dataclass
+class ShellyEventDescription(EventEntityDescription):
+    """Class to describe Shelly event."""
+
+    removal_condition: Callable[[dict, dict, str], bool] | None = None
+
+
+RPC_EVENT: Final = ShellyEventDescription(
+    key="input",
+    device_class=EventDeviceClass.BUTTON,
+    event_types=list(RPC_INPUTS_EVENTS_TYPES),
+    removal_condition=lambda config, status, key: not is_rpc_momentary_input(
+        config, status, key
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up sensors for device."""
+    if get_device_entry_gen(config_entry) == 2:
+        coordinator = get_entry_data(hass)[config_entry.entry_id].rpc
+        assert coordinator
+
+        entities = []
+        key_instances = get_rpc_key_instances(coordinator.device.status, RPC_EVENT.key)
+
+        for key in key_instances:
+            if RPC_EVENT.removal_condition and RPC_EVENT.removal_condition(
+                coordinator.device.config, coordinator.device.status, key
+            ):
+                unique_id = f"{coordinator.mac}-{key}"
+                async_remove_shelly_entity(hass, EVENT_DOMAIN, unique_id)
+            else:
+                entities.append(ShellyRpcEvent(coordinator, key, RPC_EVENT))
+
+        async_add_entities(entities)
+
+
+class ShellyRpcEvent(CoordinatorEntity[ShellyRpcCoordinator], EventEntity):
+    """Represent RPC event entity."""
+
+    _attr_should_poll = False
+    entity_description: ShellyEventDescription
+
+    def __init__(
+        self,
+        coordinator: ShellyRpcCoordinator,
+        key: str,
+        description: ShellyEventDescription,
+    ) -> None:
+        """Initialize Shelly entity."""
+        super().__init__(coordinator)
+        self.input_index = int(key.split(":")[-1])
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_NETWORK_MAC, coordinator.mac)}
+        )
+        self._attr_unique_id = f"{coordinator.mac}-{key}"
+        self._attr_name = get_rpc_input_name(coordinator.device, key)
+        self.entity_description = description
+
+    async def async_added_to_hass(self) -> None:
+        """When entity is added to hass."""
+        await super().async_added_to_hass()
+        self.async_on_remove(
+            self.coordinator.async_subscribe_input_events(self._async_handle_event)
+        )
+
+    @callback
+    def _async_handle_event(self, event: dict[str, Any]) -> None:
+        """Handle the demo button event."""
+        if event["id"] == self.input_index:
+            self._trigger_event(event["event"])
+            self.async_write_ha_state()

homeassistant/components/shelly/utils.py

@@ -285,6 +285,16 @@ def get_model_name(info: dict[str, Any]) -> str:
     return cast(str, MODEL_NAMES.get(info["type"], info["type"]))
 
 
+def get_rpc_input_name(device: RpcDevice, key: str) -> str:
+    """Get input name based from the device configuration."""
+    input_config = device.config[key]
+
+    if input_name := input_config.get("name"):
+        return f"{device.name} {input_name}"
+
+    return f"{device.name} {key.replace(':', ' ').capitalize()}"
+
+
 def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
     """Get name based on device and channel name."""
     key = key.replace("emdata", "em")

tests/components/shelly/conftest.py

@@ -191,6 +191,7 @@ MOCK_STATUS_COAP = {
 
 MOCK_STATUS_RPC = {
     "switch:0": {"output": True},
+    "input:0": {"id": 0, "state": None},
     "light:0": {"output": True, "brightness": 53.0},
     "cloud": {"connected": False},
     "cover:0": {

tests/components/shelly/test_event.py

@@ -0,0 +1,70 @@
+"""Tests for Shelly button platform."""
+from __future__ import annotations
+
+from pytest_unordered import unordered
+
+from homeassistant.components.event import (
+    ATTR_EVENT_TYPE,
+    ATTR_EVENT_TYPES,
+    DOMAIN as EVENT_DOMAIN,
+    EventDeviceClass,
+)
+from homeassistant.const import ATTR_DEVICE_CLASS, STATE_UNKNOWN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_registry import async_get
+
+from . import init_integration, inject_rpc_device_event, register_entity
+
+
+async def test_rpc_button(hass: HomeAssistant, mock_rpc_device, monkeypatch) -> None:
+    """Test RPC device event."""
+    await init_integration(hass, 2)
+    entity_id = "event.test_name_input_0"
+    registry = async_get(hass)
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
+        ["btn_down", "btn_up", "double_push", "long_push", "single_push", "triple_push"]
+    )
+    assert state.attributes.get(ATTR_EVENT_TYPE) is None
+    assert state.attributes.get(ATTR_DEVICE_CLASS) == EventDeviceClass.BUTTON
+
+    entry = registry.async_get(entity_id)
+    assert entry
+    assert entry.unique_id == "123456789ABC-input:0"
+
+    inject_rpc_device_event(
+        monkeypatch,
+        mock_rpc_device,
+        {
+            "events": [
+                {
+                    "event": "single_push",
+                    "id": 0,
+                    "ts": 1668522399.2,
+                }
+            ],
+            "ts": 1668522399.2,
+        },
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(entity_id)
+    assert state.attributes.get(ATTR_EVENT_TYPE) == "single_push"
+
+
+async def test_rpc_event_removal(
+    hass: HomeAssistant, mock_rpc_device, monkeypatch
+) -> None:
+    """Test RPC event entity is removed due to removal_condition."""
+    registry = async_get(hass)
+    entity_id = register_entity(hass, EVENT_DOMAIN, "test_name_input_0", "input:0")
+
+    assert registry.async_get(entity_id) is not None
+
+    monkeypatch.setitem(mock_rpc_device.config, "input:0", {"id": 0, "type": "switch"})
+    await init_integration(hass, 2)
+
+    assert registry.async_get(entity_id) is None

tests/components/shelly/test_utils.py

@@ -8,6 +8,7 @@ from homeassistant.components.shelly.utils import (
     get_device_uptime,
     get_number_of_channels,
     get_rpc_channel_name,
+    get_rpc_input_name,
     get_rpc_input_triggers,
     is_block_momentary_input,
 )
@@ -210,6 +211,18 @@ async def test_get_rpc_channel_name(mock_rpc_device) -> None:
     assert get_rpc_channel_name(mock_rpc_device, "input:3") == "Test name switch_3"
 
 
+async def test_get_rpc_input_name(mock_rpc_device, monkeypatch) -> None:
+    """Test get RPC input name."""
+    assert get_rpc_input_name(mock_rpc_device, "input:0") == "Test name Input 0"
+
+    monkeypatch.setitem(
+        mock_rpc_device.config,
+        "input:0",
+        {"id": 0, "type": "button", "name": "Input name"},
+    )
+    assert get_rpc_input_name(mock_rpc_device, "input:0") == "Test name Input name"
+
+
 async def test_get_rpc_input_triggers(mock_rpc_device, monkeypatch) -> None:
     """Test get RPC input triggers."""
     monkeypatch.setattr(mock_rpc_device, "config", {"input:0": {"type": "button"}})