/
knx_entity.py
59 lines (43 loc) · 1.86 KB
/
knx_entity.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
"""Base class for KNX devices."""
from xknx.devices import Climate as XknxClimate, Device as XknxDevice
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
class KnxEntity(Entity):
"""Representation of a KNX entity."""
def __init__(self, device: XknxDevice):
"""Set up device."""
self._device = device
@property
def name(self):
"""Return the name of the KNX device."""
return self._device.name
@property
def available(self):
"""Return True if entity is available."""
return self.hass.data[DOMAIN].connected
@property
def unique_id(self):
"""Return the unique id of the device if enabled."""
if self.hass.data[DOMAIN].use_unique_id:
return self._device.unique_id
return None
@property
def should_poll(self):
"""No polling needed within KNX."""
return False
async def async_update(self):
"""Request a state update from KNX bus."""
await self._device.sync()
async def after_update_callback(self, device: XknxDevice):
"""Call after device was updated."""
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Store register state change callback."""
self._device.register_device_updated_cb(self.after_update_callback)
if isinstance(self._device, XknxClimate) and self._device.mode is not None:
self._device.mode.register_device_updated_cb(self.after_update_callback)
async def async_will_remove_from_hass(self) -> None:
"""Disconnect device object when removed."""
self._device.unregister_device_updated_cb(self.after_update_callback)
if isinstance(self._device, XknxClimate) and self._device.mode is not None:
self._device.mode.unregister_device_updated_cb(self.after_update_callback)