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

Use EntityDescription - august (#56395)

Browse source
This commit is contained in:
Marc Mueller 2021-09-19 01:10:15 +02:00 • committed by GitHub
parent 9b710cad5d
commit a4f6c3336f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 145 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

132
homeassistant/components/august/binary_sensor.py
View file

@ -1,17 +1,29 @@
"""Support for August binary sensors."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
from typing import Callable, cast
from yalexs.activity import ACTION_DOORBELL_CALL_MISSED, SOURCE_PUBNUB, ActivityType
from yalexs.activity import (
ACTION_DOORBELL_CALL_MISSED,
SOURCE_PUBNUB,
Activity,
ActivityType,
)
from yalexs.doorbell import DoorbellDetail
from yalexs.lock import LockDoorStatus
from yalexs.util import update_lock_detail_from_activity
from homeassistant.components.august import AugustData
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_CONNECTIVITY,
DEVICE_CLASS_DOOR,
DEVICE_CLASS_MOTION,
DEVICE_CLASS_OCCUPANCY,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.core import callback
from homeassistant.helpers.event import async_call_later
@ -27,7 +39,7 @@ TIME_TO_RECHECK_DETECTION = timedelta(
)
def _retrieve_online_state(data, detail):
def _retrieve_online_state(data: AugustData, detail: DoorbellDetail) -> bool:
"""Get the latest state of the sensor."""
# The doorbell will go into standby mode when there is no motion
# for a short while. It will wake by itself when needed so we need
@ -36,7 +48,7 @@ def _retrieve_online_state(data, detail):
return detail.is_online or detail.is_standby
def _retrieve_motion_state(data, detail):
def _retrieve_motion_state(data: AugustData, detail: DoorbellDetail) -> bool:
latest = data.activity_stream.get_latest_device_activity(
detail.device_id, {ActivityType.DOORBELL_MOTION}
)
@ -47,7 +59,7 @@ def _retrieve_motion_state(data, detail):
return _activity_time_based_state(latest)
def _retrieve_ding_state(data, detail):
def _retrieve_ding_state(data: AugustData, detail: DoorbellDetail) -> bool:
latest = data.activity_stream.get_latest_device_activity(
detail.device_id, {ActivityType.DOORBELL_DING}
)
@ -64,34 +76,62 @@ def _retrieve_ding_state(data, detail):
return _activity_time_based_state(latest)
def _activity_time_based_state(latest):
def _activity_time_based_state(latest: Activity) -> bool:
"""Get the latest state of the sensor."""
start = latest.activity_start_time
end = latest.activity_end_time + TIME_TO_DECLARE_DETECTION
return start <= _native_datetime() <= end
def _native_datetime():
def _native_datetime() -> datetime:
"""Return time in the format august uses without timezone."""
return datetime.now()
SENSOR_NAME = 0
SENSOR_DEVICE_CLASS = 1
SENSOR_STATE_PROVIDER = 2
SENSOR_STATE_IS_TIME_BASED = 3
@dataclass
class AugustRequiredKeysMixin:
"""Mixin for required keys."""
# sensor_type: [name, device_class, state_provider, is_time_based]
SENSOR_TYPES_DOORBELL = {
"doorbell_ding": ["Ding", DEVICE_CLASS_OCCUPANCY, _retrieve_ding_state, True],
"doorbell_motion": ["Motion", DEVICE_CLASS_MOTION, _retrieve_motion_state, True],
"doorbell_online": [
"Online",
DEVICE_CLASS_CONNECTIVITY,
_retrieve_online_state,
False,
],
}
state_provider: Callable[[AugustData, DoorbellDetail], bool]
is_time_based: bool
@dataclass
class AugustBinarySensorEntityDescription(
BinarySensorEntityDescription, AugustRequiredKeysMixin
):
"""Describes August binary_sensor entity."""
SENSOR_TYPE_DOOR = BinarySensorEntityDescription(
key="door_open",
name="Open",
)
SENSOR_TYPES_DOORBELL: tuple[AugustBinarySensorEntityDescription, ...] = (
AugustBinarySensorEntityDescription(
key="doorbell_ding",
name="Ding",
device_class=DEVICE_CLASS_OCCUPANCY,
state_provider=_retrieve_ding_state,
is_time_based=True,
),
AugustBinarySensorEntityDescription(
key="doorbell_motion",
name="Motion",
device_class=DEVICE_CLASS_MOTION,
state_provider=_retrieve_motion_state,
is_time_based=True,
),
AugustBinarySensorEntityDescription(
key="doorbell_online",
name="Online",
device_class=DEVICE_CLASS_CONNECTIVITY,
state_provider=_retrieve_online_state,
is_time_based=False,
),
)
async def async_setup_entry(hass, config_entry, async_add_entities):
@ -109,16 +149,16 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
continue
_LOGGER.debug("Adding sensor class door for %s", door.device_name)
entities.append(AugustDoorBinarySensor(data, "door_open", door))
entities.append(AugustDoorBinarySensor(data, door, SENSOR_TYPE_DOOR))
for doorbell in data.doorbells:
for sensor_type, sensor in SENSOR_TYPES_DOORBELL.items():
for description in SENSOR_TYPES_DOORBELL:
_LOGGER.debug(
"Adding doorbell sensor class %s for %s",
sensor[SENSOR_DEVICE_CLASS],
description.device_class,
doorbell.device_name,
)
entities.append(AugustDoorbellBinarySensor(data, sensor_type, doorbell))
entities.append(AugustDoorbellBinarySensor(data, doorbell, description))
async_add_entities(entities)
@ -128,14 +168,16 @@ class AugustDoorBinarySensor(AugustEntityMixin, BinarySensorEntity):
_attr_device_class = DEVICE_CLASS_DOOR
def __init__(self, data, sensor_type, device):
def __init__(self, data, device, description: BinarySensorEntityDescription):
"""Initialize the sensor."""
super().__init__(data, device)
self.entity_description = description
self._data = data
self._sensor_type = sensor_type
self._device = device
self._attr_name = f"{device.device_name} Open"
self._attr_unique_id = f"{self._device_id}_open"
self._attr_name = f"{device.device_name} {description.name}"
self._attr_unique_id = (
f"{self._device_id}_{cast(str, description.name).lower()}"
)
self._update_from_data()
@callback
@ -164,41 +206,29 @@ class AugustDoorBinarySensor(AugustEntityMixin, BinarySensorEntity):
class AugustDoorbellBinarySensor(AugustEntityMixin, BinarySensorEntity):
"""Representation of an August binary sensor."""
def __init__(self, data, sensor_type, device):
entity_description: AugustBinarySensorEntityDescription
def __init__(self, data, device, description: AugustBinarySensorEntityDescription):
"""Initialize the sensor."""
super().__init__(data, device)
self.entity_description = description
self._check_for_off_update_listener = None
self._data = data
self._sensor_type = sensor_type
self._attr_device_class = self._sensor_config[SENSOR_DEVICE_CLASS]
self._attr_name = f"{device.device_name} {self._sensor_config[SENSOR_NAME]}"
self._attr_name = f"{device.device_name} {description.name}"
self._attr_unique_id = (
f"{self._device_id}_{self._sensor_config[SENSOR_NAME].lower()}"
f"{self._device_id}_{cast(str, description.name).lower()}"
)
self._update_from_data()
@property
def _sensor_config(self):
"""Return the config for the sensor."""
return SENSOR_TYPES_DOORBELL[self._sensor_type]
@property
def _state_provider(self):
"""Return the state provider for the binary sensor."""
return self._sensor_config[SENSOR_STATE_PROVIDER]
@property
def _is_time_based(self):
"""Return true of false if the sensor is time based."""
return self._sensor_config[SENSOR_STATE_IS_TIME_BASED]
@callback
def _update_from_data(self):
"""Get the latest state of the sensor."""
self._cancel_any_pending_updates()
self._attr_is_on = self._state_provider(self._data, self._detail)
self._attr_is_on = self.entity_description.state_provider(
self._data, self._detail
)
if self._is_time_based:
if self.entity_description.is_time_based:
self._attr_available = _retrieve_online_state(self._data, self._detail)
self._schedule_update_to_recheck_turn_off_sensor()
else: