Add component for binary sensor groups (#55365)
* Add component for binary sensor groups https://github.com/home-assistant/home-assistant.io/pull/19239 * Accidental push over prior commit * Add test for any case * Add unavailable attribute and tests for unique_id * Added tests for attributes link to documentation: https://github.com/home-assistant/home-assistant.io/pull/19297
This commit is contained in:
Brian Egge
GitHub
parent
ee616ed992
commit
e638e5bb42
3 changed files with 285 additions and 1 deletions
|
|
@ -56,7 +56,7 @@ ATTR_ALL = "all"
|
|||
SERVICE_SET = "set"
|
||||
SERVICE_REMOVE = "remove"
|
||||
|
||||
PLATFORMS = ["light", "cover", "notify"]
|
||||
PLATFORMS = ["light", "cover", "notify", "binary_sensor"]
|
||||
|
||||
REG_KEY = f"{DOMAIN}_registry"
|
||||
|
||||
|
|
|
|||
133
homeassistant/components/group/binary_sensor.py
Normal file
133
homeassistant/components/group/binary_sensor.py
Normal file
|
|
@ -0,0 +1,133 @@
|
|||
"""This platform allows several binary sensor to be grouped into one binary sensor."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DEVICE_CLASSES_SCHEMA,
|
||||
DOMAIN as BINARY_SENSOR_DOMAIN,
|
||||
PLATFORM_SCHEMA,
|
||||
BinarySensorEntity,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_DEVICE_CLASS,
|
||||
CONF_ENTITIES,
|
||||
CONF_NAME,
|
||||
CONF_UNIQUE_ID,
|
||||
STATE_ON,
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
from homeassistant.core import CoreState, Event, HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.event import async_track_state_change_event
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from . import GroupEntity
|
||||
|
||||
DEFAULT_NAME = "Binary Sensor Group"
|
||||
|
||||
CONF_ALL = "all"
|
||||
REG_KEY = f"{BINARY_SENSOR_DOMAIN}_registry"
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_ENTITIES): cv.entities_domain(BINARY_SENSOR_DOMAIN),
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_UNIQUE_ID): cv.string,
|
||||
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASSES_SCHEMA,
|
||||
vol.Optional(CONF_ALL): cv.boolean,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Set up the Group Binary Sensor platform."""
|
||||
async_add_entities(
|
||||
[
|
||||
BinarySensorGroup(
|
||||
config.get(CONF_UNIQUE_ID),
|
||||
config[CONF_NAME],
|
||||
config.get(CONF_DEVICE_CLASS),
|
||||
config[CONF_ENTITIES],
|
||||
config.get(CONF_ALL),
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class BinarySensorGroup(GroupEntity, BinarySensorEntity):
|
||||
"""Representation of a BinarySensorGroup."""
|
||||
|
||||
_attr_assumed_state: bool = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
unique_id: str | None,
|
||||
name: str,
|
||||
device_class: str | None,
|
||||
entity_ids: list[str],
|
||||
mode: str | None,
|
||||
) -> None:
|
||||
"""Initialize a BinarySensorGroup entity."""
|
||||
super().__init__()
|
||||
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._device_class = device_class
|
||||
self._state: str | None = None
|
||||
self.mode = any
|
||||
if mode:
|
||||
self.mode = all
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Register callbacks."""
|
||||
|
||||
async def async_state_changed_listener(event: Event) -> None:
|
||||
"""Handle child updates."""
|
||||
self.async_set_context(event.context)
|
||||
await self.async_defer_or_update_ha_state()
|
||||
|
||||
self.async_on_remove(
|
||||
async_track_state_change_event(
|
||||
self.hass, self._entity_ids, async_state_changed_listener
|
||||
)
|
||||
)
|
||||
|
||||
if self.hass.state == CoreState.running:
|
||||
await self.async_update()
|
||||
return
|
||||
|
||||
await super().async_added_to_hass()
|
||||
|
||||
async def async_update(self) -> None:
|
||||
"""Query all members and determine the binary sensor group state."""
|
||||
all_states = [self.hass.states.get(x) for x in self._entity_ids]
|
||||
filtered_states: list[str] = [x.state for x in all_states if x is not None]
|
||||
self._attr_available = any(
|
||||
state != STATE_UNAVAILABLE for state in filtered_states
|
||||
)
|
||||
if STATE_UNAVAILABLE in filtered_states:
|
||||
self._attr_is_on = None
|
||||
else:
|
||||
states = list(map(lambda x: x == STATE_ON, filtered_states))
|
||||
state = self.mode(states)
|
||||
self._attr_is_on = state
|
||||
self.async_write_ha_state()
|
||||
|
||||
@property
|
||||
def device_class(self) -> str | None:
|
||||
"""Return the sensor class of the binary sensor."""
|
||||
return self._device_class
|
||||
151
tests/components/group/test_binary_sensor.py
Normal file
151
tests/components/group/test_binary_sensor.py
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
"""The tests for the Group Binary Sensor platform."""
|
||||
from os import path
|
||||
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
||||
from homeassistant.components.group import DOMAIN
|
||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
|
||||
async def test_default_state(hass):
|
||||
"""Test binary_sensor group default state."""
|
||||
hass.states.async_set("binary_sensor.kitchen", "on")
|
||||
hass.states.async_set("binary_sensor.bedroom", "on")
|
||||
await async_setup_component(
|
||||
hass,
|
||||
BINARY_SENSOR_DOMAIN,
|
||||
{
|
||||
BINARY_SENSOR_DOMAIN: {
|
||||
"platform": DOMAIN,
|
||||
"entities": ["binary_sensor.kitchen", "binary_sensor.bedroom"],
|
||||
"name": "Bedroom Group",
|
||||
"unique_id": "unique_identifier",
|
||||
"device_class": "presence",
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("binary_sensor.bedroom_group")
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
assert state.attributes.get(ATTR_ENTITY_ID) == [
|
||||
"binary_sensor.kitchen",
|
||||
"binary_sensor.bedroom",
|
||||
]
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
entry = entity_registry.async_get("binary_sensor.bedroom_group")
|
||||
assert entry
|
||||
assert entry.unique_id == "unique_identifier"
|
||||
assert entry.original_name == "Bedroom Group"
|
||||
assert entry.device_class == "presence"
|
||||
|
||||
|
||||
async def test_state_reporting_all(hass):
|
||||
"""Test the state reporting."""
|
||||
await async_setup_component(
|
||||
hass,
|
||||
BINARY_SENSOR_DOMAIN,
|
||||
{
|
||||
BINARY_SENSOR_DOMAIN: {
|
||||
"platform": DOMAIN,
|
||||
"entities": ["binary_sensor.test1", "binary_sensor.test2"],
|
||||
"name": "Binary Sensor Group",
|
||||
"device_class": "presence",
|
||||
"all": "true",
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_ON)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_ON)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_OFF)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_ON)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_UNAVAILABLE)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
|
||||
)
|
||||
|
||||
|
||||
async def test_state_reporting_any(hass):
|
||||
"""Test the state reporting."""
|
||||
await async_setup_component(
|
||||
hass,
|
||||
BINARY_SENSOR_DOMAIN,
|
||||
{
|
||||
BINARY_SENSOR_DOMAIN: {
|
||||
"platform": DOMAIN,
|
||||
"entities": ["binary_sensor.test1", "binary_sensor.test2"],
|
||||
"name": "Binary Sensor Group",
|
||||
"device_class": "presence",
|
||||
"all": "false",
|
||||
"unique_id": "unique_identifier",
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# binary sensors have state off if unavailable
|
||||
hass.states.async_set("binary_sensor.test1", STATE_ON)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_ON)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_OFF)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_OFF
|
||||
|
||||
hass.states.async_set("binary_sensor.test1", STATE_ON)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get("binary_sensor.binary_sensor_group").state == STATE_ON
|
||||
|
||||
# binary sensors have state off if unavailable
|
||||
hass.states.async_set("binary_sensor.test1", STATE_UNAVAILABLE)
|
||||
hass.states.async_set("binary_sensor.test2", STATE_UNAVAILABLE)
|
||||
await hass.async_block_till_done()
|
||||
assert (
|
||||
hass.states.get("binary_sensor.binary_sensor_group").state == STATE_UNAVAILABLE
|
||||
)
|
||||
|
||||
entity_registry = er.async_get(hass)
|
||||
entry = entity_registry.async_get("binary_sensor.binary_sensor_group")
|
||||
assert entry
|
||||
assert entry.unique_id == "unique_identifier"
|
||||
|
||||
|
||||
def _get_fixtures_base_path():
|
||||
return path.dirname(path.dirname(path.dirname(__file__)))
|
||||
Loading…
Add table
Add a link
Reference in a new issue