-
-
Notifications
You must be signed in to change notification settings - Fork 96
/
expose.py
170 lines (145 loc) · 5.1 KB
/
expose.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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
"""Exposures to KNX bus."""
from __future__ import annotations
from typing import Callable
from xknx import XKNX
from xknx.devices import DateTime, ExposeSensor
from homeassistant.const import (
CONF_ENTITY_ID,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.typing import ConfigType, StateType
from .const import KNX_ADDRESS
from .schema import ExposeSchema
@callback
def create_knx_exposure(
hass: HomeAssistant, xknx: XKNX, config: ConfigType
) -> KNXExposeSensor | KNXExposeTime:
"""Create exposures from config."""
address = config[KNX_ADDRESS]
expose_type = config[ExposeSchema.CONF_XKNX_EXPOSE_TYPE]
attribute = config.get(ExposeSchema.CONF_XKNX_EXPOSE_ATTRIBUTE)
default = config.get(ExposeSchema.CONF_XKNX_EXPOSE_DEFAULT)
exposure: KNXExposeSensor | KNXExposeTime
if (
isinstance(expose_type, str)
and expose_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES
):
exposure = KNXExposeTime(xknx, expose_type, address)
else:
entity_id = config[CONF_ENTITY_ID]
exposure = KNXExposeSensor(
hass,
xknx,
expose_type,
entity_id,
attribute,
default,
address,
)
return exposure
class KNXExposeSensor:
"""Object to Expose Home Assistant entity to KNX bus."""
def __init__(
self,
hass: HomeAssistant,
xknx: XKNX,
expose_type: int | str,
entity_id: str,
attribute: str | None,
default: StateType,
address: str,
) -> None:
"""Initialize of Expose class."""
self.hass = hass
self.xknx = xknx
self.type = expose_type
self.entity_id = entity_id
self.expose_attribute = attribute
self.expose_default = default
self.address = address
self._remove_listener: Callable[[], None] | None = None
self.device: ExposeSensor = self.async_register()
@callback
def async_register(self) -> ExposeSensor:
"""Register listener."""
if self.expose_attribute is not None:
_name = self.entity_id + "__" + self.expose_attribute
else:
_name = self.entity_id
device = ExposeSensor(
self.xknx,
name=_name,
group_address=self.address,
value_type=self.type,
)
self._remove_listener = async_track_state_change_event(
self.hass, [self.entity_id], self._async_entity_changed
)
return device
@callback
def shutdown(self) -> None:
"""Prepare for deletion."""
if self._remove_listener is not None:
self._remove_listener()
self._remove_listener = None
self.device.shutdown()
async def _async_entity_changed(self, event: Event) -> None:
"""Handle entity change."""
new_state = event.data.get("new_state")
if new_state is None:
return
if new_state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
return
old_state = event.data.get("old_state")
if self.expose_attribute is None:
if old_state is None or old_state.state != new_state.state:
# don't send same value sequentially
await self._async_set_knx_value(new_state.state)
return
new_attribute = new_state.attributes.get(self.expose_attribute)
if old_state is not None:
old_attribute = old_state.attributes.get(self.expose_attribute)
if old_attribute == new_attribute:
# don't send same value sequentially
return
await self._async_set_knx_value(new_attribute)
async def _async_set_knx_value(self, value: StateType) -> None:
"""Set new value on xknx ExposeSensor."""
assert self.device is not None
if value is None:
if self.expose_default is None:
return
value = self.expose_default
if self.type == "binary":
if value in (1, STATE_ON, "True"):
value = True
elif value in (0, STATE_OFF, "False"):
value = False
await self.device.set(value)
class KNXExposeTime:
"""Object to Expose Time/Date object to KNX bus."""
def __init__(self, xknx: XKNX, expose_type: str, address: str) -> None:
"""Initialize of Expose class."""
self.xknx = xknx
self.expose_type = expose_type
self.address = address
self.device: DateTime = self.async_register()
@callback
def async_register(self) -> DateTime:
"""Register listener."""
return DateTime(
self.xknx,
name=self.expose_type.capitalize(),
broadcast_type=self.expose_type.upper(),
localtime=True,
group_address=self.address,
)
@callback
def shutdown(self) -> None:
"""Prepare for deletion."""
self.device.shutdown()