jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 19:27:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor ZHA binary_sensor (#30138)

Browse Source 
* Refactor ZHA binary_sensor.

Use ZHA entity class registry for channel specific implementations.

* Remove registries.BINARY_SENSOR_TYPES dict.

* Address PR comments.
This commit is contained in:
Alexei Chetroi 2019-12-22 13:24:57 -05:00 committed by David F. Mulcahey
parent ed0ee3100d
commit 83768be814
4 changed files with 69 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
homeassistant/components/zha/binary_sensor.py
View File

@ -1,4 +1,5 @@
"""Binary sensors on Zigbee Home Automation networks."""
import functools
import logging
from homeassistant.components.binary_sensor import (
@ -18,20 +19,16 @@ from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .core.const import (
CHANNEL_ACCELEROMETER,
CHANNEL_OCCUPANCY,
CHANNEL_ON_OFF,
CHANNEL_ZONE,
DATA_ZHA,
DATA_ZHA_DISPATCHERS,
SENSOR_ACCELERATION,
SENSOR_OCCUPANCY,
SENSOR_OPENING,
SENSOR_TYPE,
SIGNAL_ATTR_UPDATED,
UNKNOWN,
ZHA_DISCOVERY_NEW,
ZONE,
)
from .core.registries import ZHA_ENTITIES, MatchRule
from .entity import ZhaEntity
_LOGGER = logging.getLogger(__name__)
@ -46,19 +43,13 @@ CLASS_MAPPING = {
0x002D: DEVICE_CLASS_VIBRATION,
}
async def get_ias_device_class(channel):
"""Get the HA device class from the channel."""
zone_type = await channel.get_attribute_value("zone_type")
return CLASS_MAPPING.get(zone_type)
STRICT_MATCH = functools.partial(ZHA_ENTITIES.strict_match, DOMAIN)
DEVICE_CLASS_REGISTRY = {
UNKNOWN: None,
SENSOR_OPENING: DEVICE_CLASS_OPENING,
ZONE: get_ias_device_class,
SENSOR_OCCUPANCY: DEVICE_CLASS_OCCUPANCY,
SENSOR_ACCELERATION: DEVICE_CLASS_MOVING,
CHANNEL_ACCELEROMETER: DEVICE_CLASS_MOVING,
CHANNEL_OCCUPANCY: DEVICE_CLASS_OCCUPANCY,
CHANNEL_ON_OFF: DEVICE_CLASS_OPENING,
}
@ -94,7 +85,12 @@ async def _async_setup_entities(
"""Set up the ZHA binary sensors."""
entities = []
for discovery_info in discovery_infos:
entities.append(BinarySensor(**discovery_info))
zha_dev = discovery_info["zha_device"]
channels = discovery_info["channels"]
entity = ZHA_ENTITIES.get_entity(DOMAIN, zha_dev, channels, BinarySensor)
if entity:
entities.append(entity(**discovery_info))
async_add_entities(entities, update_before_add=True)
@ -102,43 +98,25 @@ async def _async_setup_entities(
class BinarySensor(ZhaEntity, BinarySensorDevice):
"""ZHA BinarySensor."""
_device_class = None
DEVICE_CLASS = None
def __init__(self, **kwargs):
def __init__(self, unique_id, zha_device, channels, **kwargs):
"""Initialize the ZHA binary sensor."""
super().__init__(**kwargs)
self._device_state_attributes = {}
self._zone_channel = self.cluster_channels.get(CHANNEL_ZONE)
self._on_off_channel = self.cluster_channels.get(CHANNEL_ON_OFF)
self._attr_channel = self.cluster_channels.get(CHANNEL_OCCUPANCY)
self._zha_sensor_type = kwargs[SENSOR_TYPE]
super().__init__(unique_id, zha_device, channels, **kwargs)
self._channel = channels[0]
self._device_class = self.DEVICE_CLASS
async def _determine_device_class(self):
"""Determine the device class for this binary sensor."""
device_class_supplier = DEVICE_CLASS_REGISTRY.get(self._zha_sensor_type)
if callable(device_class_supplier):
channel = self.cluster_channels.get(self._zha_sensor_type)
if channel is None:
return None
return await device_class_supplier(channel)
return device_class_supplier
async def get_device_class(self):
"""Get the HA device class from the channel."""
pass
async def async_added_to_hass(self):
"""Run when about to be added to hass."""
self._device_class = await self._determine_device_class()
await super().async_added_to_hass()
if self._on_off_channel:
await self.async_accept_signal(
self._on_off_channel, SIGNAL_ATTR_UPDATED, self.async_set_state
)
if self._zone_channel:
await self.async_accept_signal(
self._zone_channel, SIGNAL_ATTR_UPDATED, self.async_set_state
)
if self._attr_channel:
await self.async_accept_signal(
self._attr_channel, SIGNAL_ATTR_UPDATED, self.async_set_state
)
await self.get_device_class()
await self.async_accept_signal(
self._channel, SIGNAL_ATTR_UPDATED, self.async_set_state
)
@callback
def async_restore_last_state(self, last_state):
@ -148,7 +126,7 @@ class BinarySensor(ZhaEntity, BinarySensorDevice):
@property
def is_on(self) -> bool:
"""Return if the switch is on based on the statemachine."""
"""Return True if the switch is on based on the state machine."""
if self._state is None:
return False
return self._state
@ -166,13 +144,43 @@ class BinarySensor(ZhaEntity, BinarySensorDevice):
async def async_update(self):
"""Attempt to retrieve on off state from the binary sensor."""
await super().async_update()
if self._on_off_channel:
self._state = await self._on_off_channel.get_attribute_value("on_off")
if self._zone_channel:
value = await self._zone_channel.get_attribute_value("zone_status")
if value is not None:
self._state = value & 3
if self._attr_channel:
self._state = await self._attr_channel.get_attribute_value(
self._attr_channel.value_attribute
)
attribute = getattr(self._channel, "value_attribute", "on_off")
self._state = await self._channel.get_attribute_value(attribute)
@STRICT_MATCH(MatchRule(channel_names={CHANNEL_ACCELEROMETER}))
class Accelerometer(BinarySensor):
"""ZHA BinarySensor."""
DEVICE_CLASS = DEVICE_CLASS_MOTION
@STRICT_MATCH(MatchRule(channel_names={CHANNEL_OCCUPANCY}))
class Occupancy(BinarySensor):
"""ZHA BinarySensor."""
DEVICE_CLASS = DEVICE_CLASS_OCCUPANCY
@STRICT_MATCH(MatchRule(channel_names={CHANNEL_ON_OFF}))
class Opening(BinarySensor):
"""ZHA BinarySensor."""
DEVICE_CLASS = DEVICE_CLASS_OPENING
@STRICT_MATCH(MatchRule(channel_names={CHANNEL_ZONE}))
class IASZone(BinarySensor):
"""ZHA IAS BinarySensor."""
async def get_device_class(self) -> None:
"""Get the HA device class from the channel."""
zone_type = await self._channel.get_attribute_value("zone_type")
self._device_class = CLASS_MAPPING.get(zone_type)
async def async_update(self):
"""Attempt to retrieve on off state from the binary sensor."""
await super().async_update()
value = await self._channel.get_attribute_value("zone_status")
if value is not None:
self._state = value & 3

1
homeassistant/components/zha/core/const.py
View File

@ -43,6 +43,7 @@ ATTR_WARNING_DEVICE_STROBE_INTENSITY = "intensity"
BAUD_RATES = [2400, 4800, 9600, 14400, 19200, 38400, 57600, 115200, 128000, 256000]
CHANNEL_ACCELEROMETER = "accelerometer"
CHANNEL_ATTRIBUTE = "attribute"
CHANNEL_BASIC = "basic"
CHANNEL_COLOR = "light_color"

25
homeassistant/components/zha/core/discovery.py
View File

@ -11,21 +11,12 @@ import zigpy.profiles
from zigpy.zcl.clusters.general import OnOff, PowerConfiguration
from homeassistant import const as ha_const
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .channels import AttributeListeningChannel, EventRelayChannel, ZDOChannel
from .const import (
COMPONENTS,
CONF_DEVICE_CONFIG,
DATA_ZHA,
SENSOR_TYPE,
UNKNOWN,
ZHA_DISCOVERY_NEW,
)
from .const import COMPONENTS, CONF_DEVICE_CONFIG, DATA_ZHA, ZHA_DISCOVERY_NEW
from .registries import (
BINARY_SENSOR_TYPES,
CHANNEL_ONLY_CLUSTERS,
COMPONENT_CLUSTERS,
DEVICE_CLASS,
@ -160,15 +151,6 @@ def _async_handle_profile_match(
"component": component,
}
if component == BINARY_SENSOR:
discovery_info.update({SENSOR_TYPE: UNKNOWN})
for cluster_id in profile_clusters:
if cluster_id in BINARY_SENSOR_TYPES:
discovery_info.update(
{SENSOR_TYPE: BINARY_SENSOR_TYPES.get(cluster_id, UNKNOWN)}
)
break
return discovery_info
@ -288,9 +270,4 @@ def _async_handle_single_cluster_match(
"component": component,
}
if component == BINARY_SENSOR:
discovery_info.update(
{SENSOR_TYPE: BINARY_SENSOR_TYPES.get(cluster.cluster_id, UNKNOWN)}
)
return discovery_info

21
homeassistant/components/zha/core/registries.py
View File

@ -30,20 +30,10 @@ from homeassistant.components.switch import DOMAIN as SWITCH
# importing channels updates registries
from . import channels # noqa: F401 pylint: disable=unused-import
from .const import (
CONTROLLER,
SENSOR_ACCELERATION,
SENSOR_OCCUPANCY,
SENSOR_OPENING,
ZHA_GW_RADIO,
ZHA_GW_RADIO_DESCRIPTION,
ZONE,
RadioType,
)
from .const import CONTROLLER, ZHA_GW_RADIO, ZHA_GW_RADIO_DESCRIPTION, RadioType
from .decorators import CALLABLE_T, DictRegistry, SetRegistry
BINARY_SENSOR_CLUSTERS = SetRegistry()
BINARY_SENSOR_TYPES = {}
BINDABLE_CLUSTERS = SetRegistry()
CHANNEL_ONLY_CLUSTERS = SetRegistry()
CLUSTER_REPORT_CONFIGS = {}
@ -104,15 +94,6 @@ def establish_device_mappings():
BINARY_SENSOR_CLUSTERS.add(SMARTTHINGS_ACCELERATION_CLUSTER)
BINARY_SENSOR_TYPES.update(
{
SMARTTHINGS_ACCELERATION_CLUSTER: SENSOR_ACCELERATION,
zcl.clusters.general.OnOff.cluster_id: SENSOR_OPENING,
zcl.clusters.measurement.OccupancySensing.cluster_id: SENSOR_OCCUPANCY,
zcl.clusters.security.IasZone.cluster_id: ZONE,
}
)
DEVICE_CLASS[zigpy.profiles.zha.PROFILE_ID].update(
{
SMARTTHINGS_ARRIVAL_SENSOR_DEVICE_TYPE: DEVICE_TRACKER,