Compare commits

...

1 Commits

Author SHA1 Message Date
abmantis
eaa851c5ef Implement doorbell.rang trigger 2026-04-16 20:24:56 +01:00
5 changed files with 288 additions and 16 deletions

View File

@@ -14,6 +14,9 @@
}
},
"triggers": {
"rang": {
"trigger": "mdi:doorbell"
},
"received": {
"trigger": "mdi:eye-check"
}

View File

@@ -30,6 +30,10 @@
},
"title": "Event",
"triggers": {
"rang": {
"description": "Triggers after one or more doorbells rang.",
"name": "Doorbell rang"
},
"received": {
"description": "Triggers after one or more event entities receive a matching event.",
"fields": {

View File

@@ -1,5 +1,7 @@
"""Provides triggers for events."""
from typing import TYPE_CHECKING
import voluptuous as vol
from homeassistant.const import CONF_OPTIONS, STATE_UNAVAILABLE, STATE_UNKNOWN
@@ -13,7 +15,8 @@ from homeassistant.helpers.trigger import (
TriggerConfig,
)
from .const import ATTR_EVENT_TYPE, DOMAIN
from . import EventDeviceClass
from .const import ATTR_EVENT_TYPE, DOMAIN, DoorbellEventType
CONF_EVENT_TYPE = "event_type"
@@ -28,16 +31,22 @@ EVENT_RECEIVED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.extend(
)
class EventReceivedTrigger(EntityTriggerBase):
"""Trigger for event entity when it receives a matching event."""
class EventReceivedTriggerBase(EntityTriggerBase):
"""Base trigger for event entity when it receives a matching event."""
_domain_specs = {DOMAIN: DomainSpec()}
_schema = EVENT_RECEIVED_TRIGGER_SCHEMA
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
def __init__(
self, hass: HomeAssistant, config: TriggerConfig, event_types: set[str]
) -> None:
"""Initialize the event received trigger."""
super().__init__(hass, config)
self._event_types = set(self._options[CONF_EVENT_TYPE])
self._event_types = event_types
def is_valid_state(self, state: State) -> bool:
"""Check if the event type is valid and matches one of the configured types."""
return (
state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
and state.attributes.get(ATTR_EVENT_TYPE) in self._event_types
)
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is valid and different from the current state."""
@@ -49,16 +58,34 @@ class EventReceivedTrigger(EntityTriggerBase):
return from_state.state != to_state.state
def is_valid_state(self, state: State) -> bool:
"""Check if the event type is valid and matches one of the configured types."""
return (
state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
and state.attributes.get(ATTR_EVENT_TYPE) in self._event_types
)
class EventReceivedTrigger(EventReceivedTriggerBase):
"""Trigger for event entity when it receives a matching event."""
_domain_specs = {DOMAIN: DomainSpec()}
_schema = EVENT_RECEIVED_TRIGGER_SCHEMA
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the event received trigger."""
if TYPE_CHECKING:
assert config.options is not None
super().__init__(hass, config, set(config.options[CONF_EVENT_TYPE]))
class DoorbellRangTrigger(EventReceivedTriggerBase):
"""Trigger for doorbell event entity when a ring event is received."""
_domain_specs = {DOMAIN: DomainSpec(device_class=EventDeviceClass.DOORBELL)}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the ring event trigger."""
super().__init__(hass, config, {DoorbellEventType.RING})
TRIGGERS: dict[str, type[Trigger]] = {
"received": EventReceivedTrigger,
"rang": DoorbellRangTrigger,
}

View File

@@ -14,3 +14,8 @@ received:
- unavailable
- unknown
multiple: true
rang:
target:
entity:
domain: event
device_class: doorbell

View File

@@ -3,10 +3,11 @@
import pytest
from homeassistant.components.event.const import ATTR_EVENT_TYPE
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.const import ATTR_DEVICE_CLASS, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from tests.components.common import (
BasicTriggerStateDescription,
TriggerStateDescription,
arm_trigger,
assert_trigger_gated_by_labs_flag,
@@ -22,7 +23,7 @@ async def target_events(hass: HomeAssistant) -> dict[str, list[str]]:
return await target_entities(hass, "event")
@pytest.mark.parametrize("trigger_key", ["event.received"])
@pytest.mark.parametrize("trigger_key", ["event.received", "event.rang"])
async def test_event_triggers_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
) -> None:
@@ -285,3 +286,235 @@ async def test_event_state_trigger(
await hass.async_block_till_done()
assert len(calls) == (entities_in_target - 1) * state["count"]
calls.clear()
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("event"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
# Doorbell rang with ring event_type
(
"event.rang",
[
{
"included_state": {
"state": STATE_UNAVAILABLE,
"attributes": {ATTR_DEVICE_CLASS: "doorbell"},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:00.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:01.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
],
),
# From unknown - first valid state after unknown is triggered
(
"event.rang",
[
{
"included_state": {
"state": STATE_UNKNOWN,
"attributes": {ATTR_DEVICE_CLASS: "doorbell"},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:00.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
{
"included_state": {
"state": "2026-01-01T00:00:01.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
{
"included_state": {
"state": STATE_UNKNOWN,
"attributes": {ATTR_DEVICE_CLASS: "doorbell"},
},
"count": 0,
},
],
),
# Same ring event fires again (different timestamps)
(
"event.rang",
[
{
"included_state": {
"state": "2026-01-01T00:00:00.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:01.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
{
"included_state": {
"state": "2026-01-01T00:00:02.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
],
),
# To unavailable - should not trigger, and first state after unavailable is skipped
(
"event.rang",
[
{
"included_state": {
"state": "2026-01-01T00:00:00.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 0,
},
{
"included_state": {
"state": STATE_UNAVAILABLE,
"attributes": {ATTR_DEVICE_CLASS: "doorbell"},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:01.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:02.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
],
),
# Non-ring event_type should not trigger
(
"event.rang",
[
{
"included_state": {
"state": STATE_UNAVAILABLE,
"attributes": {ATTR_DEVICE_CLASS: "doorbell"},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:00.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "other_event",
},
},
"count": 0,
},
{
"included_state": {
"state": "2026-01-01T00:00:01.000+00:00",
"attributes": {
ATTR_DEVICE_CLASS: "doorbell",
ATTR_EVENT_TYPE: "ring",
},
},
"count": 1,
},
],
),
],
)
async def test_doorbell_rang_trigger(
hass: HomeAssistant,
target_events: dict[str, list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[BasicTriggerStateDescription],
) -> None:
"""Test that the doorbell rang trigger fires when a doorbell ring event is received."""
calls: list[str] = []
other_entity_ids = set(target_events["included_entities"]) - {entity_id}
# Set all events to the initial state
for eid in target_events["included_entities"]:
set_or_remove_state(hass, eid, states[0]["included_state"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, None, trigger_target_config, calls)
for state in states[1:]:
included_state = state["included_state"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(calls) == state["count"]
for call in calls:
assert call == entity_id
calls.clear()
# Check if changing other events also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(calls) == (entities_in_target - 1) * state["count"]
calls.clear()