Base event platform

tests/test_platform_event.py

@@ -0,0 +1,164 @@
+"""Test the ZHA event platform."""
+
+from typing import Any
+
+import pytest
+from zigpy.profiles import zha
+from zigpy.zcl.clusters import general
+
+from tests.common import (
+    SIG_EP_INPUT,
+    SIG_EP_OUTPUT,
+    SIG_EP_PROFILE,
+    SIG_EP_TYPE,
+    create_mock_zigpy_device,
+    join_zigpy_device,
+)
+from zha.application import Platform
+from zha.application.gateway import Gateway
+from zha.application.platforms import EntityStateChangedEvent
+from zha.application.platforms.event import (
+    BaseEvent,
+    EntityEventTriggeredEvent,
+    EventState,
+    TriggeredEvent,
+)
+from zha.application.platforms.event.const import (
+    ATTR_MULTI_PRESS_COUNT,
+    ButtonEventType,
+    EventDeviceClass,
+)
+from zha.zigbee.device import Device
+
+
+class FakeEvent(BaseEvent):
+    """Event entity with a fixed set of event types."""
+
+    _unique_id_suffix = "fake"
+    _attr_device_class = EventDeviceClass.BUTTON
+    _attr_event_types = [
+        ButtonEventType.PRESS_END,
+        ButtonEventType.MULTI_PRESS_END,
+    ]
+
+    def trigger(
+        self, event_type: str, event_attributes: dict[str, Any] | None = None
+    ) -> None:
+        """Trigger an event, as a concrete subclass would."""
+        self._trigger_event(event_type, event_attributes)
+
+
+class FakeDoorbellEvent(FakeEvent):
+    """Doorbell event entity that cannot ring."""
+
+    _attr_device_class = EventDeviceClass.DOORBELL
+
+
+@pytest.fixture
+async def zha_device(zha_gateway: Gateway) -> Device:
+    """Return a joined device to attach event entities to."""
+    zigpy_device = create_mock_zigpy_device(
+        zha_gateway,
+        {
+            1: {
+                SIG_EP_INPUT: [general.Basic.cluster_id],
+                SIG_EP_OUTPUT: [general.OnOff.cluster_id],
+                SIG_EP_TYPE: zha.DeviceType.NON_COLOR_CONTROLLER,
+                SIG_EP_PROFILE: zha.PROFILE_ID,
+            }
+        },
+    )
+    return await join_zigpy_device(zha_gateway, zigpy_device)
+
+
+def create_event_entity(zha_device: Device, entity_class: type[FakeEvent]) -> FakeEvent:
+    """Create an event entity on the first endpoint of a device."""
+    endpoint = zha_device.endpoints[1]
+
+    return entity_class(
+        endpoint=endpoint,
+        device=zha_device,
+        cluster=endpoint.zigpy_endpoint.out_clusters[general.OnOff.cluster_id],
+    )
+
+
+@pytest.fixture
+def entity(zha_device: Device) -> FakeEvent:
+    """Return an event entity for a joined device."""
+    return create_event_entity(zha_device, FakeEvent)
+
+
+async def test_event_state(entity: FakeEvent) -> None:
+    """Test that the state of an event entity only describes its capabilities."""
+    assert entity.PLATFORM == Platform.EVENT
+    assert entity.device_class == EventDeviceClass.BUTTON
+    assert entity.event_types == ["press_end", "multi_press_end"]
+
+    state = entity.state
+    assert isinstance(state, EventState)
+    assert state.event_types == ["press_end", "multi_press_end"]
+
+
+async def test_trigger_event(entity: FakeEvent) -> None:
+    """Test triggering events."""
+    events: list[EntityEventTriggeredEvent] = []
+    unsub = entity.on_event(EntityEventTriggeredEvent.event, events.append)
+
+    # Nothing is delivered until an event is actually triggered
+    assert events == []
+
+    entity.trigger(ButtonEventType.MULTI_PRESS_END, {ATTR_MULTI_PRESS_COUNT: 2})
+    assert events == [
+        EntityEventTriggeredEvent(
+            platform=Platform.EVENT,
+            unique_id=entity.unique_id,
+            device_ieee=entity.device.ieee,
+            endpoint_id=1,
+            triggered=TriggeredEvent(
+                event_type="multi_press_end",
+                event_attributes={"multi_press_count": 2},
+            ),
+        )
+    ]
+
+    # An event identical to the previous one is still delivered
+    entity.trigger(ButtonEventType.MULTI_PRESS_END, {ATTR_MULTI_PRESS_COUNT: 2})
+    assert len(events) == 2
+    assert events[0] == events[1]
+
+    # Event attributes are optional
+    entity.trigger(ButtonEventType.PRESS_END)
+    assert events[-1].triggered == TriggeredEvent(
+        event_type="press_end", event_attributes={}
+    )
+
+    unsub()
+    entity.trigger(ButtonEventType.PRESS_END)
+    assert len(events) == 3
+
+
+async def test_trigger_event_does_not_change_state(entity: FakeEvent) -> None:
+    """Test that triggering an event is not a state change."""
+    state_changes: list[EntityStateChangedEvent] = []
+    entity.subscribe_state(state_changes.append)
+    assert len(state_changes) == 1
+
+    entity.trigger(ButtonEventType.PRESS_END)
+    assert len(state_changes) == 1
+
+
+async def test_doorbell_must_ring(zha_device: Device) -> None:
+    """Test that a doorbell event entity has to support the ring event type."""
+    with pytest.raises(ValueError, match="does not support the 'ring' event type"):
+        create_event_entity(zha_device, FakeDoorbellEvent)
+
+
+async def test_trigger_unsupported_event(entity: FakeEvent) -> None:
+    """Test that triggering an unsupported event type fails."""
+    events: list[EntityEventTriggeredEvent] = []
+    entity.on_event(EntityEventTriggeredEvent.event, events.append)
+
+    with pytest.raises(ValueError, match="Invalid event type long_press_end"):
+        entity.trigger(ButtonEventType.LONG_PRESS_END)
+
+    assert events == []

zha/application/discovery.py

@@ -27,6 +27,7 @@
     climate,
     cover,
     device_tracker,
+    event,
     fan,
     light,
     lock,
@@ -53,6 +54,7 @@
     Platform.CLIMATE,
     Platform.COVER,
     Platform.DEVICE_TRACKER,
+    Platform.EVENT,
     Platform.FAN,
     Platform.LIGHT,
     Platform.LOCK,

zha/application/platforms/event/__init__.py

@@ -0,0 +1,96 @@
+"""Events on Zigbee Home Automation networks."""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import TYPE_CHECKING, Any, Final
+
+from zigpy.types.named import EUI64
+
+from zha.application import Platform
+from zha.application.platforms import BaseEntityState, PlatformEntity
+from zha.application.platforms.event.const import DoorbellEventType, EventDeviceClass
+
+if TYPE_CHECKING:
+    from zha.zigbee.device import Device
+    from zha.zigbee.endpoint import Endpoint
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class EventState(BaseEntityState):
+    """State for event entities."""
+
+    event_types: list[str]
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class TriggeredEvent:
+    """The event an event entity fired."""
+
+    event_type: str
+    event_attributes: dict[str, Any]
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class EntityEventTriggeredEvent:
+    """Event for when an event entity fires."""
+
+    event_type: Final[str] = "entity"
+    event: Final[str] = "event_triggered"
+    platform: str
+    unique_id: str
+    device_ieee: EUI64 | None = None
+    endpoint_id: int | None = None
+    group_id: int | None = None
+    triggered: TriggeredEvent
+
+
+class BaseEvent(PlatformEntity):
+    """Base representation of a ZHA event entity."""
+
+    PLATFORM = Platform.EVENT
+
+    _attr_device_class: EventDeviceClass | None = None
+    _attr_event_types: list[str]
+
+    def __init__(self, endpoint: Endpoint, device: Device, **kwargs: Any) -> None:
+        """Initialize the event entity."""
+        super().__init__(endpoint=endpoint, device=device, **kwargs)
+
+        # Doorbells are expected to ring: the `doorbell.rang` trigger matches on it
+        if (
+            self.device_class == EventDeviceClass.DOORBELL
+            and DoorbellEventType.RING not in self.event_types
+        ):
+            raise ValueError(
+                f"Doorbell event entity {self.unique_id} does not support the"
+                f" '{DoorbellEventType.RING}' event type"
+            )
+
+    @property
+    def event_types(self) -> list[str]:
+        """Return the event types this entity can trigger."""
+        return self._attr_event_types
+
+    @property
+    def state(self) -> EventState:
+        """Return the state of the event entity."""
+        return EventState(**super().state.__dict__, event_types=self.event_types)
+
+    def _trigger_event(
+        self, event_type: str, event_attributes: dict[str, Any] | None = None
+    ) -> None:
+        """Trigger an event, to be called by subclasses."""
+        if event_type not in self.event_types:
+            raise ValueError(f"Invalid event type {event_type} for {self.unique_id}")
+
+        self.emit(
+            EntityEventTriggeredEvent.event,
+            EntityEventTriggeredEvent(
+                **self.identifiers.__dict__,
+                triggered=TriggeredEvent(
+                    event_type=event_type,
+                    event_attributes=event_attributes or {},
+                ),
+            ),
+        )

zha/application/platforms/event/const.py

@@ -0,0 +1,30 @@
+"""Constants for the event platform."""
+
+from enum import StrEnum
+
+ATTR_MULTI_PRESS_COUNT = "multi_press_count"
+
+
+class EventDeviceClass(StrEnum):
+    """Device class for events."""
+
+    DOORBELL = "doorbell"
+    BUTTON = "button"
+    MOTION = "motion"
+
+
+class DoorbellEventType(StrEnum):
+    """Standard event types for the doorbell device class."""
+
+    RING = "ring"
+
+
+class ButtonEventType(StrEnum):
+    """Standard event types for the button device class."""
+
+    PRESS_START = "press_start"
+    PRESS_END = "press_end"
+    LONG_PRESS_START = "long_press_start"
+    LONG_PRESS_END = "long_press_end"
+    MULTI_PRESS_ONGOING = "multi_press_ongoing"
+    MULTI_PRESS_END = "multi_press_end"