evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add preview support to binary sensor group (#98872)

Browse source
This commit is contained in:
Erik Montnemery 2023-08-23 14:24:48 +02:00 • committed by GitHub
parent e3b945a8d0
commit 6be20b5408
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 270 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

25
homeassistant/components/group/binary_sensor.py
View file

@ -1,6 +1,9 @@
"""Platform allowing several binary sensor to be grouped into one binary sensor."""
from __future__ import annotations
from collections.abc import Callable, Mapping
from typing import Any
import voluptuous as vol
from homeassistant.components.binary_sensor import (
@ -21,7 +24,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import CALLBACK_TYPE, 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 (
@ -113,6 +116,26 @@ class BinarySensorGroup(GroupEntity, BinarySensorEntity):
if mode:
self.mode = all
@callback
def async_start_preview(
self,
preview_callback: Callable[[str, Mapping[str, Any]], None],
) -> CALLBACK_TYPE:
"""Render a preview."""
@callback
def async_state_changed_listener(
event: EventType[EventStateChangedData] | None,
) -> None:
"""Handle child updates."""
self.async_update_group_state()
preview_callback(*self._async_generate_attributes())
async_state_changed_listener(None)
return async_track_state_change_event(
self.hass, self._entity_ids, async_state_changed_listener
)
async def async_added_to_hass(self) -> None:
"""Register callbacks."""

142
homeassistant/components/group/config_flow.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine, Mapping
from functools import partial
from typing import Any, cast
from typing import Any, Literal, cast
import voluptuous as vol
@ -22,7 +22,7 @@ from homeassistant.helpers.schema_config_entry_flow import (
)
from . import DOMAIN
from .binary_sensor import CONF_ALL
from .binary_sensor import CONF_ALL, BinarySensorGroup
from .const import CONF_HIDE_MEMBERS, CONF_IGNORE_NON_NUMERIC
from .sensor import SensorGroup
@ -73,7 +73,9 @@ def basic_group_config_schema(domain: str | list[str]) -> vol.Schema:
)
async def binary_sensor_options_schema(handler: SchemaCommonFlowHandler) -> vol.Schema:
async def binary_sensor_options_schema(
handler: SchemaCommonFlowHandler | None,
) -> vol.Schema:
"""Generate options schema."""
return (await basic_group_options_schema("binary_sensor", handler)).extend(
{
@ -170,6 +172,7 @@ CONFIG_FLOW = {
"binary_sensor": SchemaFlowFormStep(
BINARY_SENSOR_CONFIG_SCHEMA,
validate_user_input=set_group_type("binary_sensor"),
preview="group_binary_sensor",
),
"cover": SchemaFlowFormStep(
basic_group_config_schema("cover"),
@ -205,7 +208,10 @@ CONFIG_FLOW = {
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(next_step=choose_options_step),
"binary_sensor": SchemaFlowFormStep(binary_sensor_options_schema),
"binary_sensor": SchemaFlowFormStep(
binary_sensor_options_schema,
preview="group_binary_sensor",
),
"cover": SchemaFlowFormStep(partial(basic_group_options_schema, "cover")),
"fan": SchemaFlowFormStep(partial(basic_group_options_schema, "fan")),
"light": SchemaFlowFormStep(partial(light_switch_options_schema, "light")),
@ -260,6 +266,7 @@ class GroupConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
def async_setup_preview(hass: HomeAssistant) -> None:
"""Set up preview WS API."""
websocket_api.async_register_command(hass, ws_preview_sensor)
websocket_api.async_register_command(hass, ws_preview_binary_sensor)
def _async_hide_members(
@ -275,6 +282,86 @@ def _async_hide_members(
registry.async_update_entity(entity_id, hidden_by=hidden_by)
@callback
def _async_handle_ws_preview(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
config_schema: vol.Schema,
options_schema: vol.Schema,
create_preview_entity: Callable[
[Literal["config_flow", "options_flow"], str, dict[str, Any]],
BinarySensorGroup | SensorGroup,
],
) -> None:
"""Generate a preview."""
if msg["flow_type"] == "config_flow":
validated = config_schema(msg["user_input"])
name = validated["name"]
else:
validated = options_schema(msg["user_input"])
flow_status = hass.config_entries.options.async_get(msg["flow_id"])
config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
if not config_entry:
raise HomeAssistantError
name = config_entry.options["name"]
@callback
def async_preview_updated(state: str, attributes: Mapping[str, Any]) -> None:
"""Forward config entry state events to websocket."""
connection.send_message(
websocket_api.event_message(
msg["id"], {"state": state, "attributes": attributes}
)
)
preview_entity = create_preview_entity(msg["flow_type"], name, validated)
preview_entity.hass = hass
connection.send_result(msg["id"])
connection.subscriptions[msg["id"]] = preview_entity.async_start_preview(
async_preview_updated
)
@websocket_api.websocket_command(
{
vol.Required("type"): "group/binary_sensor/start_preview",
vol.Required("flow_id"): str,
vol.Required("flow_type"): vol.Any("config_flow", "options_flow"),
vol.Required("user_input"): dict,
}
)
@websocket_api.async_response
async def ws_preview_binary_sensor(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Generate a preview."""
def create_preview_binary_sensor(
flow_type: Literal["config_flow", "options_flow"],
name: str,
validated_config: dict[str, Any],
) -> BinarySensorGroup:
"""Create a preview sensor."""
return BinarySensorGroup(
None,
name,
None,
validated_config[CONF_ENTITIES],
validated_config[CONF_ALL],
)
_async_handle_ws_preview(
hass,
connection,
msg,
BINARY_SENSOR_CONFIG_SCHEMA,
await binary_sensor_options_schema(None),
create_preview_binary_sensor,
)
@websocket_api.websocket_command(
{
vol.Required("type"): "group/sensor/start_preview",
@ -288,41 +375,34 @@ async def ws_preview_sensor(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Generate a preview."""
if msg["flow_type"] == "config_flow":
validated = SENSOR_CONFIG_SCHEMA(msg["user_input"])
ignore_non_numeric = False
name = validated["name"]
else:
validated = (await sensor_options_schema("sensor", None))(msg["user_input"])
flow_status = hass.config_entries.options.async_get(msg["flow_id"])
config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
if not config_entry:
raise HomeAssistantError
ignore_non_numeric = validated[CONF_IGNORE_NON_NUMERIC]
name = config_entry.options["name"]
@callback
def async_preview_updated(state: str, attributes: Mapping[str, Any]) -> None:
"""Forward config entry state events to websocket."""
connection.send_message(
websocket_api.event_message(
msg["id"], {"state": state, "attributes": attributes}
def create_preview_sensor(
flow_type: Literal["config_flow", "options_flow"],
name: str,
validated_config: dict[str, Any],
) -> SensorGroup:
"""Create a preview sensor."""
ignore_non_numeric = (
False
if flow_type == "config_flow"
else validated_config[CONF_IGNORE_NON_NUMERIC]
)
)
sensor = SensorGroup(
return SensorGroup(
None,
name,
validated[CONF_ENTITIES],
validated_config[CONF_ENTITIES],
ignore_non_numeric,
validated[CONF_TYPE],
validated_config[CONF_TYPE],
None,
None,
None,
)
sensor.hass = hass
connection.send_result(msg["id"])
connection.subscriptions[msg["id"]] = sensor.async_start_preview(
async_preview_updated
_async_handle_ws_preview(
hass,
connection,
msg,
SENSOR_CONFIG_SCHEMA,
await sensor_options_schema("sensor", None),
create_preview_sensor,
)

140
tests/components/group/test_config_flow.py
View file

@ -449,13 +449,129 @@ async def test_options_flow_hides_members(
assert registry.async_get(f"{group_type}.three").hidden_by == hidden_by
async def test_config_flow_binary_sensor_preview(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the config flow preview."""
client = await hass_ws_client(hass)
input_entities = ["binary_sensor.input_one", "binary_sensor.input_two"]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.MENU
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"next_step_id": "binary_sensor"},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "binary_sensor"
assert result["errors"] is None
assert result["preview"] == "group_binary_sensor"
await client.send_json_auto_id(
{
"type": "group/binary_sensor/start_preview",
"flow_id": result["flow_id"],
"flow_type": "config_flow",
"user_input": {
"name": "My binary sensor group",
"entities": input_entities,
"all": True,
},
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await client.receive_json()
assert msg["event"] == {
"attributes": {"friendly_name": "My binary sensor group"},
"state": "unavailable",
}
hass.states.async_set("binary_sensor.input_one", "on")
hass.states.async_set("binary_sensor.input_two", "off")
msg = await client.receive_json()
assert msg["event"] == {
"attributes": {
"entity_id": ["binary_sensor.input_one", "binary_sensor.input_two"],
"friendly_name": "My binary sensor group",
},
"state": "off",
}
async def test_option_flow_binary_sensor_preview(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the option flow preview."""
client = await hass_ws_client(hass)
input_entities = ["binary_sensor.input_one", "binary_sensor.input_two"]
# Setup the config entry
config_entry = MockConfigEntry(
data={},
domain=DOMAIN,
options={
"all": True,
"entities": input_entities,
"group_type": "binary_sensor",
"hide_members": False,
"name": "My group",
},
title="My min_max",
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
assert result["preview"] == "group_binary_sensor"
hass.states.async_set("binary_sensor.input_one", "on")
hass.states.async_set("binary_sensor.input_two", "off")
await client.send_json_auto_id(
{
"type": "group/binary_sensor/start_preview",
"flow_id": result["flow_id"],
"flow_type": "options_flow",
"user_input": {
"entities": input_entities,
"all": False,
},
}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await client.receive_json()
assert msg["event"] == {
"attributes": {
"entity_id": input_entities,
"friendly_name": "My group",
},
"state": "on",
}
async def test_config_flow_sensor_preview(
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the config flow preview."""
client = await hass_ws_client(hass)
input_sensors = ["sensor.input_one", "sensor.input_two"]
input_entities = ["sensor.input_one", "sensor.input_two"]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -479,7 +595,7 @@ async def test_config_flow_sensor_preview(
"flow_type": "config_flow",
"user_input": {
"name": "My sensor group",
"entities": input_sensors,
"entities": input_entities,
"type": "max",
},
}
@ -503,7 +619,7 @@ async def test_config_flow_sensor_preview(
msg = await client.receive_json()
assert msg["event"] == {
"attributes": {
"entity_id": ["sensor.input_one", "sensor.input_two"],
"entity_id": input_entities,
"friendly_name": "My sensor group",
"icon": "mdi:calculator",
"max_entity_id": "sensor.input_two",
@ -518,12 +634,14 @@ async def test_option_flow_sensor_preview(
"""Test the option flow preview."""
client = await hass_ws_client(hass)
input_entities = ["sensor.input_one", "sensor.input_two"]
# Setup the config entry
config_entry = MockConfigEntry(
data={},
domain=DOMAIN,
options={
"entities": ["sensor.input_one", "sensor.input_two"],
"entities": input_entities,
"group_type": "sensor",
"hide_members": False,
"name": "My sensor group",
@ -535,8 +653,6 @@ async def test_option_flow_sensor_preview(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
input_sensors = ["sensor.input_one", "sensor.input_two"]
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
@ -551,7 +667,7 @@ async def test_option_flow_sensor_preview(
"flow_id": result["flow_id"],
"flow_type": "options_flow",
"user_input": {
"entities": input_sensors,
"entities": input_entities,
"type": "min",
},
}
@ -563,7 +679,7 @@ async def test_option_flow_sensor_preview(
msg = await client.receive_json()
assert msg["event"] == {
"attributes": {
"entity_id": ["sensor.input_one", "sensor.input_two"],
"entity_id": input_entities,
"friendly_name": "My sensor group",
"icon": "mdi:calculator",
"min_entity_id": "sensor.input_one",
@ -578,12 +694,14 @@ async def test_option_flow_sensor_preview_config_entry_removed(
"""Test the option flow preview where the config entry is removed."""
client = await hass_ws_client(hass)
input_entities = ["sensor.input_one", "sensor.input_two"]
# Setup the config entry
config_entry = MockConfigEntry(
data={},
domain=DOMAIN,
options={
"entities": ["sensor.input_one", "sensor.input_two"],
"entities": input_entities,
"group_type": "sensor",
"hide_members": False,
"name": "My sensor group",
@ -595,8 +713,6 @@ async def test_option_flow_sensor_preview_config_entry_removed(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
input_sensors = ["sensor.input_one", "sensor.input_two"]
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
@ -610,7 +726,7 @@ async def test_option_flow_sensor_preview_config_entry_removed(
"flow_id": result["flow_id"],
"flow_type": "options_flow",
"user_input": {
"entities": input_sensors,
"entities": input_entities,
"type": "min",
},
}