Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for event groups (#98463) 

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/98962/head
Franck Nijhof 2023-08-24 12:49:38 +02:00 committed by GitHub
parent 87dd18cc2e
commit 0d013767ee
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 349 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
homeassistant/components/group/config_flow.py
View File

@ -137,6 +137,7 @@ async def light_switch_options_schema(
GROUP_TYPES = [
"binary_sensor",
"cover",
"event",
"fan",
"light",
"lock",
@ -178,6 +179,10 @@ CONFIG_FLOW = {
basic_group_config_schema("cover"),
validate_user_input=set_group_type("cover"),
),
"event": SchemaFlowFormStep(
basic_group_config_schema("event"),
validate_user_input=set_group_type("event"),
),
"fan": SchemaFlowFormStep(
basic_group_config_schema("fan"),
validate_user_input=set_group_type("fan"),
@ -213,6 +218,7 @@ OPTIONS_FLOW = {
preview="group",
),
"cover": SchemaFlowFormStep(partial(basic_group_options_schema, "cover")),
"event": SchemaFlowFormStep(partial(basic_group_options_schema, "event")),
"fan": SchemaFlowFormStep(partial(basic_group_options_schema, "fan")),
"light": SchemaFlowFormStep(partial(light_switch_options_schema, "light")),
"lock": SchemaFlowFormStep(partial(basic_group_options_schema, "lock")),

180
homeassistant/components/group/event.py Normal file
View File

@ -0,0 +1,180 @@
"""Platform allowing several event entities to be grouped into one event."""
from __future__ import annotations
import itertools
import voluptuous as vol
from homeassistant.components.event import (
ATTR_EVENT_TYPE,
ATTR_EVENT_TYPES,
DOMAIN,
PLATFORM_SCHEMA,
EventEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
CONF_ENTITIES,
CONF_NAME,
CONF_UNIQUE_ID,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import (
EventStateChangedData,
async_track_state_change_event,
)
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, EventType
from . import GroupEntity
DEFAULT_NAME = "Event group"
# No limit on parallel updates to enable a group calling another group
PARALLEL_UPDATES = 0
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_ENTITIES): cv.entities_domain(DOMAIN),
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_UNIQUE_ID): cv.string,
}
)
async def async_setup_platform(
_: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
__: DiscoveryInfoType | None = None,
) -> None:
"""Set up the event group platform."""
async_add_entities(
[
EventGroup(
config.get(CONF_UNIQUE_ID),
config[CONF_NAME],
config[CONF_ENTITIES],
)
]
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Initialize event group config entry."""
registry = er.async_get(hass)
entities = er.async_validate_entity_ids(
registry, config_entry.options[CONF_ENTITIES]
)
async_add_entities(
[
EventGroup(
config_entry.entry_id,
config_entry.title,
entities,
)
]
)
class EventGroup(GroupEntity, EventEntity):
"""Representation of an event group."""
_attr_available = False
_attr_should_poll = False
def __init__(
self,
unique_id: str | None,
name: str,
entity_ids: list[str],
) -> None:
"""Initialize an event group."""
self._entity_ids = entity_ids
self._attr_name = name
self._attr_extra_state_attributes = {ATTR_ENTITY_ID: entity_ids}
self._attr_unique_id = unique_id
self._attr_event_types = []
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
@callback
def async_state_changed_listener(
event: EventType[EventStateChangedData],
) -> None:
"""Handle child updates."""
if not self.hass.is_running:
return
self.async_set_context(event.context)
# Update all properties of the group
self.async_update_group_state()
# Re-fire if one of the members fires an event, but only
# if the original state was not unavailable or unknown.
if (
(old_state := event.data["old_state"])
and old_state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
and (new_state := event.data["new_state"])
and new_state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
and (event_type := new_state.attributes.get(ATTR_EVENT_TYPE))
):
event_attributes = new_state.attributes.copy()
# We should not propagate the event properties as
# fired event attributes.
del event_attributes[ATTR_EVENT_TYPE]
del event_attributes[ATTR_EVENT_TYPES]
event_attributes.pop(ATTR_DEVICE_CLASS, None)
event_attributes.pop(ATTR_FRIENDLY_NAME, None)
# Fire the group event
self._trigger_event(event_type, event_attributes)
self.async_write_ha_state()
self.async_on_remove(
async_track_state_change_event(
self.hass, self._entity_ids, async_state_changed_listener
)
)
await super().async_added_to_hass()
@callback
def async_update_group_state(self) -> None:
"""Query all members and determine the event group properties."""
states = [
state
for entity_id in self._entity_ids
if (state := self.hass.states.get(entity_id)) is not None
]
# None of the members are available
if not states:
self._attr_available = False
return
# Gather and combine all possible event types from all entities
self._attr_event_types = list(
set(
itertools.chain.from_iterable(
state.attributes.get(ATTR_EVENT_TYPES, []) for state in states
)
)
)
# Set group as unavailable if all members are unavailable or missing
self._attr_available = any(state.state != STATE_UNAVAILABLE for state in states)

9
homeassistant/components/group/strings.json
View File

@ -8,6 +8,7 @@
"menu_options": {
"binary_sensor": "Binary sensor group",
"cover": "Cover group",
"event": "Event group",
"fan": "Fan group",
"light": "Light group",
"lock": "Lock group",
@ -34,6 +35,14 @@
"name": "[%key:component::group::config::step::binary_sensor::data::name%]"
}
},
"event": {
"title": "[%key:component::group::config::step::user::title%]",
"data": {
"entities": "[%key:component::group::config::step::binary_sensor::data::entities%]",
"hide_members": "[%key:component::group::config::step::binary_sensor::data::hide_members%]",
"name": "[%key:component::group::config::step::binary_sensor::data::name%]"
}
},
"fan": {
"title": "[%key:component::group::config::step::user::title%]",
"data": {

16
tests/components/group/test_config_flow.py
View File

@ -6,6 +6,7 @@ import pytest
from homeassistant import config_entries
from homeassistant.components.group import DOMAIN, async_setup_entry
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import entity_registry as er
@ -28,6 +29,18 @@ from tests.typing import WebSocketGenerator
("binary_sensor", "on", "on", {}, {}, {"all": False}, {}),
("binary_sensor", "on", "on", {}, {"all": True}, {"all": True}, {}),
("cover", "open", "open", {}, {}, {}, {}),
(
"event",
STATE_UNKNOWN,
"2021-01-01T23:59:59.123+00:00",
{
"event_type": "single_press",
"event_types": ["single_press", "double_press"],
},
{},
{},
{},
),
("fan", "on", "on", {}, {}, {}, {}),
("light", "on", "on", {}, {}, {}, {}),
("lock", "locked", "locked", {}, {}, {}, {}),
@ -122,6 +135,7 @@ async def test_config_flow(
(
("binary_sensor", {"all": False}),
("cover", {}),
("event", {}),
("fan", {}),
("light", {}),
("lock", {}),
@ -194,6 +208,7 @@ def get_suggested(schema, key):
(
("binary_sensor", "on", {"all": False}, {}),
("cover", "open", {}, {}),
("event", "2021-01-01T23:59:59.123+00:00", {}, {}),
("fan", "on", {}, {}),
("light", "on", {"all": False}, {}),
("lock", "locked", {}, {}),
@ -377,6 +392,7 @@ async def test_all_options(
(
("binary_sensor", {"all": False}),
("cover", {}),
("event", {}),
("fan", {}),
("light", {}),
("lock", {}),

138
tests/components/group/test_event.py Normal file
View File

@ -0,0 +1,138 @@
"""The tests for the group event platform."""
from pytest_unordered import unordered
from homeassistant.components.event import DOMAIN as EVENT_DOMAIN
from homeassistant.components.event.const import ATTR_EVENT_TYPE, ATTR_EVENT_TYPES
from homeassistant.components.group import DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
async def test_default_state(hass: HomeAssistant) -> None:
"""Test event group default state."""
await async_setup_component(
hass,
EVENT_DOMAIN,
{
EVENT_DOMAIN: {
"platform": DOMAIN,
"entities": ["event.button_1", "event.button_2"],
"name": "Remote control",
"unique_id": "unique_identifier",
}
},
)
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state == STATE_UNAVAILABLE
hass.states.async_set(
"event.button_1",
"2021-01-01T23:59:59.123+00:00",
{"event_type": "double_press", "event_types": ["single_press", "double_press"]},
)
await hass.async_block_till_done()
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state == STATE_UNKNOWN
assert state.attributes.get(ATTR_ENTITY_ID) == ["event.button_1", "event.button_2"]
assert not state.attributes.get(ATTR_EVENT_TYPE)
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
["single_press", "double_press"]
)
# State changed
hass.states.async_set(
"event.button_1",
"2021-01-01T23:59:59.123+00:00",
{"event_type": "single_press", "event_types": ["single_press", "double_press"]},
)
await hass.async_block_till_done()
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state
assert state.attributes.get(ATTR_ENTITY_ID) == ["event.button_1", "event.button_2"]
assert state.attributes.get(ATTR_EVENT_TYPE) == "single_press"
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
["single_press", "double_press"]
)
# State changed, second remote came online
hass.states.async_set(
"event.button_2",
"2021-01-01T23:59:59.123+00:00",
{"event_type": "double_press", "event_types": ["double_press", "triple_press"]},
)
await hass.async_block_till_done()
# State should be single_press, because button coming online is not an event
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state
assert state.attributes.get(ATTR_ENTITY_ID) == ["event.button_1", "event.button_2"]
assert state.attributes.get(ATTR_EVENT_TYPE) == "single_press"
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
["single_press", "double_press", "triple_press"]
)
# State changed, now it fires an event
hass.states.async_set(
"event.button_2",
"2021-01-01T23:59:59.123+00:00",
{
"event_type": "triple_press",
"event_types": ["double_press", "triple_press"],
"device_class": "doorbell",
},
)
await hass.async_block_till_done()
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state
assert state.attributes.get(ATTR_ENTITY_ID) == ["event.button_1", "event.button_2"]
assert state.attributes.get(ATTR_EVENT_TYPE) == "triple_press"
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
["single_press", "double_press", "triple_press"]
)
assert ATTR_DEVICE_CLASS not in state.attributes
# Mark button 1 unavailable
hass.states.async_set("event.button_1", STATE_UNAVAILABLE)
await hass.async_block_till_done()
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state
assert state.attributes.get(ATTR_ENTITY_ID) == ["event.button_1", "event.button_2"]
assert state.attributes.get(ATTR_EVENT_TYPE) == "triple_press"
assert state.attributes.get(ATTR_EVENT_TYPES) == unordered(
["double_press", "triple_press"]
)
# Mark button 2 unavailable
hass.states.async_set("event.button_2", STATE_UNAVAILABLE)
await hass.async_block_till_done()
state = hass.states.get("event.remote_control")
assert state is not None
assert state.state == STATE_UNAVAILABLE
entity_registry = er.async_get(hass)
entry = entity_registry.async_get("event.remote_control")
assert entry
assert entry.unique_id == "unique_identifier"