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

Add event platform to august (#121392)

pull/118335/head
J. Nick Koston 2024-07-07 06:39:58 -07:00 committed by GitHub
parent 4c1cb6cce8
commit dc5ae9e0b2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 400 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
homeassistant/components/august/binary_sensor.py
View File

@ -8,7 +8,7 @@ from datetime import datetime, timedelta
from functools import partial
import logging
from yalexs.activity import ACTION_DOORBELL_CALL_MISSED, Activity, ActivityType
from yalexs.activity import Activity, ActivityType
from yalexs.doorbell import DoorbellDetail
from yalexs.lock import LockDetail, LockDoorStatus
from yalexs.manager.const import ACTIVITY_UPDATE_INTERVAL
@ -26,67 +26,25 @@ from homeassistant.helpers.event import async_call_later
from . import AugustConfigEntry, AugustData
from .entity import AugustDescriptionEntity
from .util import (
retrieve_ding_activity,
retrieve_doorbell_motion_activity,
retrieve_online_state,
retrieve_time_based_activity,
)
_LOGGER = logging.getLogger(__name__)
TIME_TO_DECLARE_DETECTION = timedelta(seconds=ACTIVITY_UPDATE_INTERVAL.total_seconds())
TIME_TO_RECHECK_DETECTION = timedelta(
seconds=ACTIVITY_UPDATE_INTERVAL.total_seconds() * 3
)
def _retrieve_online_state(
data: AugustData, detail: DoorbellDetail | LockDetail
) -> 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
# to consider is available or we will not report motion or dings
if isinstance(detail, DoorbellDetail):
return detail.is_online or detail.is_standby
return detail.bridge_is_online
def _retrieve_time_based_state(
activities: set[ActivityType], data: AugustData, detail: DoorbellDetail
) -> bool:
"""Get the latest state of the sensor."""
stream = data.activity_stream
if latest := stream.get_latest_device_activity(detail.device_id, activities):
return _activity_time_based_state(latest)
return False
_RING_ACTIVITIES = {ActivityType.DOORBELL_DING}
def _retrieve_ding_state(data: AugustData, detail: DoorbellDetail | LockDetail) -> bool:
stream = data.activity_stream
latest = stream.get_latest_device_activity(detail.device_id, _RING_ACTIVITIES)
if latest is None or (
data.push_updates_connected and latest.action == ACTION_DOORBELL_CALL_MISSED
):
return False
return _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() -> datetime:
"""Return time in the format august uses without timezone."""
return datetime.now()
@dataclass(frozen=True, kw_only=True)
class AugustDoorbellBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes August binary_sensor entity."""
value_fn: Callable[[AugustData, DoorbellDetail], bool]
value_fn: Callable[[AugustData, DoorbellDetail | LockDetail], Activity | None]
is_time_based: bool
@ -99,14 +57,14 @@ SENSOR_TYPES_VIDEO_DOORBELL = (
AugustDoorbellBinarySensorEntityDescription(
key="motion",
device_class=BinarySensorDeviceClass.MOTION,
value_fn=partial(_retrieve_time_based_state, {ActivityType.DOORBELL_MOTION}),
value_fn=retrieve_doorbell_motion_activity,
is_time_based=True,
),
AugustDoorbellBinarySensorEntityDescription(
key="image capture",
translation_key="image_capture",
value_fn=partial(
_retrieve_time_based_state, {ActivityType.DOORBELL_IMAGE_CAPTURE}
retrieve_time_based_activity, {ActivityType.DOORBELL_IMAGE_CAPTURE}
),
is_time_based=True,
),
@ -114,7 +72,7 @@ SENSOR_TYPES_VIDEO_DOORBELL = (
key="online",
device_class=BinarySensorDeviceClass.CONNECTIVITY,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=_retrieve_online_state,
value_fn=retrieve_online_state,
is_time_based=False,
),
)
@ -124,7 +82,7 @@ SENSOR_TYPES_DOORBELL: tuple[AugustDoorbellBinarySensorEntityDescription, ...] =
AugustDoorbellBinarySensorEntityDescription(
key="ding",
device_class=BinarySensorDeviceClass.OCCUPANCY,
value_fn=_retrieve_ding_state,
value_fn=retrieve_ding_activity,
is_time_based=True,
),
)
@ -189,10 +147,12 @@ class AugustDoorbellBinarySensor(AugustDescriptionEntity, BinarySensorEntity):
def _update_from_data(self) -> None:
"""Get the latest state of the sensor."""
self._cancel_any_pending_updates()
self._attr_is_on = self.entity_description.value_fn(self._data, self._detail)
self._attr_is_on = bool(
self.entity_description.value_fn(self._data, self._detail)
)
if self.entity_description.is_time_based:
self._attr_available = _retrieve_online_state(self._data, self._detail)
self._attr_available = retrieve_online_state(self._data, self._detail)
self._schedule_update_to_recheck_turn_off_sensor()
else:
self._attr_available = True

1
homeassistant/components/august/const.py
View File

@ -42,6 +42,7 @@ PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.CAMERA,
Platform.EVENT,
Platform.LOCK,
Platform.SENSOR,
]

104
homeassistant/components/august/event.py Normal file
View File

@ -0,0 +1,104 @@
"""Support for august events."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING
from yalexs.activity import Activity
from yalexs.doorbell import DoorbellDetail
from yalexs.lock import LockDetail
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import AugustConfigEntry, AugustData
from .entity import AugustDescriptionEntity
from .util import (
retrieve_ding_activity,
retrieve_doorbell_motion_activity,
retrieve_online_state,
)
@dataclass(kw_only=True, frozen=True)
class AugustEventEntityDescription(EventEntityDescription):
"""Describe august event entities."""
value_fn: Callable[[AugustData, DoorbellDetail | LockDetail], Activity | None]
TYPES_VIDEO_DOORBELL: tuple[AugustEventEntityDescription, ...] = (
AugustEventEntityDescription(
key="motion",
translation_key="motion",
device_class=EventDeviceClass.MOTION,
event_types=["motion"],
value_fn=retrieve_doorbell_motion_activity,
),
)
TYPES_DOORBELL: tuple[AugustEventEntityDescription, ...] = (
AugustEventEntityDescription(
key="doorbell",
translation_key="doorbell",
device_class=EventDeviceClass.DOORBELL,
event_types=["ring"],
value_fn=retrieve_ding_activity,
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: AugustConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the august event platform."""
data = config_entry.runtime_data
entities: list[AugustEventEntity] = []
for lock in data.locks:
detail = data.get_device_detail(lock.device_id)
if detail.doorbell:
entities.extend(
AugustEventEntity(data, lock, description)
for description in TYPES_DOORBELL
)
for doorbell in data.doorbells:
entities.extend(
AugustEventEntity(data, doorbell, description)
for description in TYPES_DOORBELL + TYPES_VIDEO_DOORBELL
)
async_add_entities(entities)
class AugustEventEntity(AugustDescriptionEntity, EventEntity):
"""An august event entity."""
entity_description: AugustEventEntityDescription
_attr_has_entity_name = True
_last_activity: Activity | None = None
@callback
def _update_from_data(self) -> None:
"""Update from data."""
self._attr_available = retrieve_online_state(self._data, self._detail)
current_activity = self.entity_description.value_fn(self._data, self._detail)
if not current_activity or current_activity == self._last_activity:
return
self._last_activity = current_activity
event_types = self.entity_description.event_types
if TYPE_CHECKING:
assert event_types is not None
self._trigger_event(event_type=event_types[0])
self.async_write_ha_state()

20
homeassistant/components/august/strings.json
View File

@ -58,6 +58,26 @@
"operator": {
"name": "Operator"
}
},
"event": {
"doorbell": {
"state_attributes": {
"event_type": {
"state": {
"ring": "Ring"
}
}
}
},
"motion": {
"state_attributes": {
"event_type": {
"state": {
"motion": "Motion"
}
}
}
}
}
}
}

69
homeassistant/components/august/util.py
View File

@ -1,12 +1,24 @@
"""August util functions."""
from __future__ import annotations
from datetime import datetime, timedelta
from functools import partial
import socket
import aiohttp
from yalexs.activity import ACTION_DOORBELL_CALL_MISSED, Activity, ActivityType
from yalexs.doorbell import DoorbellDetail
from yalexs.lock import LockDetail
from yalexs.manager.const import ACTIVITY_UPDATE_INTERVAL
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import aiohttp_client
from . import AugustData
TIME_TO_DECLARE_DETECTION = timedelta(seconds=ACTIVITY_UPDATE_INTERVAL.total_seconds())
@callback
def async_create_august_clientsession(hass: HomeAssistant) -> aiohttp.ClientSession:
@ -22,3 +34,60 @@ def async_create_august_clientsession(hass: HomeAssistant) -> aiohttp.ClientSess
# we can allow IPv6 again
#
return aiohttp_client.async_create_clientsession(hass, family=socket.AF_INET)
def retrieve_time_based_activity(
activities: set[ActivityType], data: AugustData, detail: DoorbellDetail | LockDetail
) -> Activity | None:
"""Get the latest state of the sensor."""
stream = data.activity_stream
if latest := stream.get_latest_device_activity(detail.device_id, activities):
return _activity_time_based(latest)
return False
_RING_ACTIVITIES = {ActivityType.DOORBELL_DING}
def retrieve_ding_activity(
data: AugustData, detail: DoorbellDetail | LockDetail
) -> Activity | None:
"""Get the ring/ding state."""
stream = data.activity_stream
latest = stream.get_latest_device_activity(detail.device_id, _RING_ACTIVITIES)
if latest is None or (
data.push_updates_connected and latest.action == ACTION_DOORBELL_CALL_MISSED
):
return None
return _activity_time_based(latest)
retrieve_doorbell_motion_activity = partial(
retrieve_time_based_activity, {ActivityType.DOORBELL_MOTION}
)
def _activity_time_based(latest: Activity) -> Activity | None:
"""Get the latest state of the sensor."""
start = latest.activity_start_time
end = latest.activity_end_time + TIME_TO_DECLARE_DETECTION
if start <= _native_datetime() <= end:
return latest
return None
def _native_datetime() -> datetime:
"""Return time in the format august uses without timezone."""
return datetime.now()
def retrieve_online_state(
data: AugustData, detail: DoorbellDetail | LockDetail
) -> 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
# to consider is available or we will not report motion or dings
if isinstance(detail, DoorbellDetail):
return detail.is_online or detail.is_standby
return detail.bridge_is_online

4
tests/components/august/mocks.py
View File

@ -58,6 +58,10 @@ def _mock_authenticator(auth_state):
return authenticator
def _timetoken():
return str(time.time_ns())[:-2]
@patch("yalexs.manager.gateway.ApiAsync")
@patch("yalexs.manager.gateway.AuthenticatorAsync.async_authenticate")
async def _mock_setup_august(

12
tests/components/august/test_binary_sensor.py
View File

@ -1,7 +1,6 @@
"""The binary_sensor tests for the august platform."""
import datetime
import time
from unittest.mock import Mock, patch
from yalexs.pubnub_async import AugustPubNub
@ -25,15 +24,12 @@ from .mocks import (
_mock_doorbell_from_fixture,
_mock_doorsense_enabled_august_lock_detail,
_mock_lock_from_fixture,
_timetoken,
)
from tests.common import async_fire_time_changed
def _timetoken():
return str(time.time_ns())[:-2]
async def test_doorsense(hass: HomeAssistant) -> None:
"""Test creation of a lock with doorsense and bridge."""
lock_one = await _mock_lock_from_fixture(
@ -153,7 +149,7 @@ async def test_create_doorbell_with_motion(hass: HomeAssistant) -> None:
new_time = dt_util.utcnow() + datetime.timedelta(seconds=40)
native_time = datetime.datetime.now() + datetime.timedelta(seconds=40)
with patch(
"homeassistant.components.august.binary_sensor._native_datetime",
"homeassistant.components.august.util._native_datetime",
return_value=native_time,
):
async_fire_time_changed(hass, new_time)
@ -252,7 +248,7 @@ async def test_doorbell_update_via_pubnub(hass: HomeAssistant) -> None:
new_time = dt_util.utcnow() + datetime.timedelta(seconds=40)
native_time = datetime.datetime.now() + datetime.timedelta(seconds=40)
with patch(
"homeassistant.components.august.binary_sensor._native_datetime",
"homeassistant.components.august.util._native_datetime",
return_value=native_time,
):
async_fire_time_changed(hass, new_time)
@ -282,7 +278,7 @@ async def test_doorbell_update_via_pubnub(hass: HomeAssistant) -> None:
new_time = dt_util.utcnow() + datetime.timedelta(seconds=40)
native_time = datetime.datetime.now() + datetime.timedelta(seconds=40)
with patch(
"homeassistant.components.august.binary_sensor._native_datetime",
"homeassistant.components.august.util._native_datetime",
return_value=native_time,
):
async_fire_time_changed(hass, new_time)

182
tests/components/august/test_event.py Normal file
View File

@ -0,0 +1,182 @@
"""The event tests for the august."""
import datetime
from unittest.mock import Mock, patch
from yalexs.pubnub_async import AugustPubNub
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util
from .mocks import (
_create_august_with_devices,
_mock_activities_from_fixture,
_mock_doorbell_from_fixture,
_mock_lock_from_fixture,
_timetoken,
)
from tests.common import async_fire_time_changed
async def test_create_doorbell(hass: HomeAssistant) -> None:
"""Test creation of a doorbell."""
doorbell_one = await _mock_doorbell_from_fixture(hass, "get_doorbell.json")
await _create_august_with_devices(hass, [doorbell_one])
motion_state = hass.states.get("event.k98gidt45gul_name_motion")
assert motion_state is not None
assert motion_state.state == STATE_UNKNOWN
doorbell_state = hass.states.get("event.k98gidt45gul_name_doorbell")
assert doorbell_state is not None
assert doorbell_state.state == STATE_UNKNOWN
async def test_create_doorbell_offline(hass: HomeAssistant) -> None:
"""Test creation of a doorbell that is offline."""
doorbell_one = await _mock_doorbell_from_fixture(hass, "get_doorbell.offline.json")
await _create_august_with_devices(hass, [doorbell_one])
motion_state = hass.states.get("event.tmt100_name_motion")
assert motion_state is not None
assert motion_state.state == STATE_UNAVAILABLE
doorbell_state = hass.states.get("event.tmt100_name_doorbell")
assert doorbell_state is not None
assert doorbell_state.state == STATE_UNAVAILABLE
async def test_create_doorbell_with_motion(hass: HomeAssistant) -> None:
"""Test creation of a doorbell."""
doorbell_one = await _mock_doorbell_from_fixture(hass, "get_doorbell.json")
activities = await _mock_activities_from_fixture(
hass, "get_activity.doorbell_motion.json"
)
await _create_august_with_devices(hass, [doorbell_one], activities=activities)
motion_state = hass.states.get("event.k98gidt45gul_name_motion")
assert motion_state is not None
assert motion_state.state != STATE_UNKNOWN
isotime = motion_state.state
doorbell_state = hass.states.get("event.k98gidt45gul_name_doorbell")
assert doorbell_state is not None
assert doorbell_state.state == STATE_UNKNOWN
new_time = dt_util.utcnow() + datetime.timedelta(seconds=40)
native_time = datetime.datetime.now() + datetime.timedelta(seconds=40)
with patch(
"homeassistant.components.august.util._native_datetime",
return_value=native_time,
):
async_fire_time_changed(hass, new_time)
await hass.async_block_till_done()
motion_state = hass.states.get("event.k98gidt45gul_name_motion")
assert motion_state.state == isotime
async def test_doorbell_update_via_pubnub(hass: HomeAssistant) -> None:
"""Test creation of a doorbell that can be updated via pubnub."""
doorbell_one = await _mock_doorbell_from_fixture(hass, "get_doorbell.json")
pubnub = AugustPubNub()
await _create_august_with_devices(hass, [doorbell_one], pubnub=pubnub)
assert doorbell_one.pubsub_channel == "7c7a6672-59c8-3333-ffff-dcd98705cccc"
motion_state = hass.states.get("event.k98gidt45gul_name_motion")
assert motion_state is not None
assert motion_state.state == STATE_UNKNOWN
doorbell_state = hass.states.get("event.k98gidt45gul_name_doorbell")
assert doorbell_state is not None
assert doorbell_state.state == STATE_UNKNOWN
pubnub.message(
pubnub,
Mock(
channel=doorbell_one.pubsub_channel,
timetoken=_timetoken(),
message={
"status": "doorbell_motion_detected",
"data": {
"event": "doorbell_motion_detected",
"image": {
"height": 640,
"width": 480,
"format": "jpg",
"created_at": "2021-03-16T02:36:26.886Z",
"bytes": 14061,
"secure_url": (
"https://dyu7azbnaoi74.cloudfront.net/images/1f8.jpeg"
),
"url": "https://dyu7azbnaoi74.cloudfront.net/images/1f8.jpeg",
"etag": "09e839331c4ea59eef28081f2caa0e90",
},
"doorbellName": "Front Door",
"callID": None,
"origin": "mars-api",
"mutableContent": True,
},
},
),
)
await hass.async_block_till_done()
motion_state = hass.states.get("event.k98gidt45gul_name_motion")
assert motion_state is not None
assert motion_state.state != STATE_UNKNOWN
isotime = motion_state.state
new_time = dt_util.utcnow() + datetime.timedelta(seconds=40)
native_time = datetime.datetime.now() + datetime.timedelta(seconds=40)
with patch(
"homeassistant.components.august.util._native_datetime",
return_value=native_time,
):
async_fire_time_changed(hass, new_time)
await hass.async_block_till_done()
motion_state = hass.states.get("event.k98gidt45gul_name_motion")
assert motion_state is not None
assert motion_state.state != STATE_UNKNOWN
pubnub.message(
pubnub,
Mock(
channel=doorbell_one.pubsub_channel,
timetoken=_timetoken(),
message={
"status": "buttonpush",
},
),
)
await hass.async_block_till_done()
doorbell_state = hass.states.get("event.k98gidt45gul_name_doorbell")
assert doorbell_state is not None
assert doorbell_state.state != STATE_UNKNOWN
isotime = motion_state.state
new_time = dt_util.utcnow() + datetime.timedelta(seconds=40)
native_time = datetime.datetime.now() + datetime.timedelta(seconds=40)
with patch(
"homeassistant.components.august.util._native_datetime",
return_value=native_time,
):
async_fire_time_changed(hass, new_time)
await hass.async_block_till_done()
doorbell_state = hass.states.get("event.k98gidt45gul_name_doorbell")
assert doorbell_state is not None
assert doorbell_state.state != STATE_UNKNOWN
assert motion_state.state == isotime
async def test_create_lock_with_doorbell(hass: HomeAssistant) -> None:
"""Test creation of a lock with a doorbell."""
lock_one = await _mock_lock_from_fixture(hass, "lock_with_doorbell.online.json")
await _create_august_with_devices(hass, [lock_one])
doorbell_state = hass.states.get(
"event.a6697750d607098bae8d6baa11ef8063_name_doorbell"
)
assert doorbell_state is not None
assert doorbell_state.state == STATE_UNKNOWN