/
lock.py
141 lines (110 loc) · 4.43 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
"""Locks on Zigbee Home Automation networks."""
import logging
from zigpy.zcl.foundation import Status
from homeassistant.components.lock import (
DOMAIN,
STATE_LOCKED,
STATE_UNLOCKED,
LockDevice,
)
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .core.const import (
CHANNEL_DOORLOCK,
DATA_ZHA,
DATA_ZHA_DISPATCHERS,
SIGNAL_ATTR_UPDATED,
ZHA_DISCOVERY_NEW,
)
from .entity import ZhaEntity
_LOGGER = logging.getLogger(__name__)
""" The first state is Zigbee 'Not fully locked' """
STATE_LIST = [STATE_UNLOCKED, STATE_LOCKED, STATE_UNLOCKED]
VALUE_TO_STATE = dict(enumerate(STATE_LIST))
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Old way of setting up Zigbee Home Automation locks."""
pass
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Zigbee Home Automation Door Lock from config entry."""
async def async_discover(discovery_info):
await _async_setup_entities(
hass, config_entry, async_add_entities, [discovery_info]
)
unsub = async_dispatcher_connect(
hass, ZHA_DISCOVERY_NEW.format(DOMAIN), async_discover
)
hass.data[DATA_ZHA][DATA_ZHA_DISPATCHERS].append(unsub)
locks = hass.data.get(DATA_ZHA, {}).get(DOMAIN)
if locks is not None:
await _async_setup_entities(
hass, config_entry, async_add_entities, locks.values()
)
del hass.data[DATA_ZHA][DOMAIN]
async def _async_setup_entities(
hass, config_entry, async_add_entities, discovery_infos
):
"""Set up the ZHA locks."""
entities = []
for discovery_info in discovery_infos:
entities.append(ZhaDoorLock(**discovery_info))
async_add_entities(entities, update_before_add=True)
class ZhaDoorLock(ZhaEntity, LockDevice):
"""Representation of a ZHA lock."""
_domain = DOMAIN
def __init__(self, unique_id, zha_device, channels, **kwargs):
"""Init this sensor."""
super().__init__(unique_id, zha_device, channels, **kwargs)
self._doorlock_channel = self.cluster_channels.get(CHANNEL_DOORLOCK)
async def async_added_to_hass(self):
"""Run when about to be added to hass."""
await super().async_added_to_hass()
await self.async_accept_signal(
self._doorlock_channel, SIGNAL_ATTR_UPDATED, self.async_set_state
)
@callback
def async_restore_last_state(self, last_state):
"""Restore previous state."""
self._state = VALUE_TO_STATE.get(last_state.state, last_state.state)
@property
def is_locked(self) -> bool:
"""Return true if entity is locked."""
if self._state is None:
return False
return self._state == STATE_LOCKED
@property
def device_state_attributes(self):
"""Return state attributes."""
return self.state_attributes
async def async_lock(self, **kwargs):
"""Lock the lock."""
result = await self._doorlock_channel.lock_door()
if not isinstance(result, list) or result[0] is not Status.SUCCESS:
self.error("Error with lock_door: %s", result)
return
self.async_schedule_update_ha_state()
async def async_unlock(self, **kwargs):
"""Unlock the lock."""
result = await self._doorlock_channel.unlock_door()
if not isinstance(result, list) or result[0] is not Status.SUCCESS:
self.error("Error with unlock_door: %s", result)
return
self.async_schedule_update_ha_state()
async def async_update(self):
"""Attempt to retrieve state from the lock."""
await super().async_update()
await self.async_get_state()
def async_set_state(self, state):
"""Handle state update from channel."""
self._state = VALUE_TO_STATE.get(state, self._state)
self.async_schedule_update_ha_state()
async def async_get_state(self, from_cache=True):
"""Attempt to retrieve state from the lock."""
if self._doorlock_channel:
state = await self._doorlock_channel.get_attribute_value(
"lock_state", from_cache=from_cache
)
if state is not None:
self._state = VALUE_TO_STATE.get(state, self._state)
async def refresh(self, time):
"""Call async_get_state at an interval."""
await self.async_get_state(from_cache=False)