forked from home-assistant/core
/
lock.py
144 lines (117 loc) · 5.42 KB
/
lock.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
"""Support for August lock."""
import logging
from aiohttp import ClientResponseError
from yalexs.activity import SOURCE_PUBNUB, ActivityType
from yalexs.lock import LockStatus
from yalexs.util import update_lock_detail_from_activity
from homeassistant.components.lock import ATTR_CHANGED_BY, LockEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_BATTERY_LEVEL
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
import homeassistant.util.dt as dt_util
from . import AugustData
from .const import DATA_AUGUST, DOMAIN
from .entity import AugustEntityMixin
_LOGGER = logging.getLogger(__name__)
LOCK_JAMMED_ERR = 531
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up August locks."""
data: AugustData = hass.data[DOMAIN][config_entry.entry_id][DATA_AUGUST]
async_add_entities([AugustLock(data, lock) for lock in data.locks])
class AugustLock(AugustEntityMixin, RestoreEntity, LockEntity):
"""Representation of an August lock."""
def __init__(self, data, device):
"""Initialize the lock."""
super().__init__(data, device)
self._data = data
self._device = device
self._lock_status = None
self._attr_name = device.device_name
self._attr_unique_id = f"{self._device_id:s}_lock"
self._update_from_data()
@property
def _hyper_bridge(self):
"""Check if the lock has a paired hyper bridge."""
return bool(self._detail.bridge and self._detail.bridge.hyper_bridge)
async def async_lock(self, **kwargs):
"""Lock the device."""
if self._data.activity_stream.pubnub.connected:
await self._data.async_lock_async(self._device_id, self._hyper_bridge)
return
await self._call_lock_operation(self._data.async_lock)
async def async_unlock(self, **kwargs):
"""Unlock the device."""
if self._data.activity_stream.pubnub.connected:
await self._data.async_unlock_async(self._device_id, self._hyper_bridge)
return
await self._call_lock_operation(self._data.async_unlock)
async def _call_lock_operation(self, lock_operation):
try:
activities = await lock_operation(self._device_id)
except ClientResponseError as err:
if err.status == LOCK_JAMMED_ERR:
self._detail.lock_status = LockStatus.JAMMED
self._detail.lock_status_datetime = dt_util.utcnow()
else:
raise
else:
for lock_activity in activities:
update_lock_detail_from_activity(self._detail, lock_activity)
if self._update_lock_status_from_detail():
_LOGGER.debug(
"async_signal_device_id_update (from lock operation): %s",
self._device_id,
)
self._data.async_signal_device_id_update(self._device_id)
def _update_lock_status_from_detail(self):
self._attr_available = self._detail.bridge_is_online
if self._lock_status != self._detail.lock_status:
self._lock_status = self._detail.lock_status
return True
return False
@callback
def _update_from_data(self):
"""Get the latest state of the sensor and update activity."""
lock_activity = self._data.activity_stream.get_latest_device_activity(
self._device_id,
{ActivityType.LOCK_OPERATION, ActivityType.LOCK_OPERATION_WITHOUT_OPERATOR},
)
if lock_activity is not None:
self._attr_changed_by = lock_activity.operated_by
update_lock_detail_from_activity(self._detail, lock_activity)
# If the source is pubnub the lock must be online since its a live update
if lock_activity.source == SOURCE_PUBNUB:
self._detail.set_online(True)
bridge_activity = self._data.activity_stream.get_latest_device_activity(
self._device_id, {ActivityType.BRIDGE_OPERATION}
)
if bridge_activity is not None:
update_lock_detail_from_activity(self._detail, bridge_activity)
self._update_lock_status_from_detail()
if self._lock_status is None or self._lock_status is LockStatus.UNKNOWN:
self._attr_is_locked = None
else:
self._attr_is_locked = self._lock_status is LockStatus.LOCKED
self._attr_is_jammed = self._lock_status is LockStatus.JAMMED
self._attr_is_locking = self._lock_status is LockStatus.LOCKING
self._attr_is_unlocking = self._lock_status is LockStatus.UNLOCKING
self._attr_extra_state_attributes = {
ATTR_BATTERY_LEVEL: self._detail.battery_level
}
if self._detail.keypad is not None:
self._attr_extra_state_attributes[
"keypad_battery_level"
] = self._detail.keypad.battery_level
async def async_added_to_hass(self):
"""Restore ATTR_CHANGED_BY on startup since it is likely no longer in the activity log."""
await super().async_added_to_hass()
if not (last_state := await self.async_get_last_state()):
return
if ATTR_CHANGED_BY in last_state.attributes:
self._attr_changed_by = last_state.attributes[ATTR_CHANGED_BY]