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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
|
"""
Module for managing operation and controller modes.
Operation modes can be 'auto', 'comfort', 'standby', 'economy', 'protection' and use either a binary DPT or DPT 20.102.
Controller modes use DPT 20.105.
"""
from __future__ import annotations
from collections.abc import Iterator
from typing import TYPE_CHECKING, Any
from xknx.dpt.dpt_1 import HeatCool
from xknx.dpt.dpt_20 import HVACControllerMode, HVACOperationMode, HVACStatus
from xknx.exceptions import DeviceIllegalValue
from xknx.remote_value import GroupAddressesType
from xknx.remote_value.remote_value_climate_mode import (
RemoteValueBinaryHeatCool,
RemoteValueBinaryOperationMode,
RemoteValueClimateModeBase,
RemoteValueControllerMode,
RemoteValueHVACStatus,
RemoteValueOperationMode,
)
from .device import Device, DeviceCallbackType
if TYPE_CHECKING:
from xknx.telegram import Telegram
from xknx.xknx import XKNX
class ClimateMode(Device):
"""Class for managing the climate mode."""
def __init__(
self,
xknx: XKNX,
name: str,
group_address_operation_mode: GroupAddressesType = None,
group_address_operation_mode_state: GroupAddressesType = None,
group_address_operation_mode_protection: GroupAddressesType = None,
group_address_operation_mode_economy: GroupAddressesType = None,
group_address_operation_mode_comfort: GroupAddressesType = None,
group_address_operation_mode_standby: GroupAddressesType = None,
group_address_controller_status: GroupAddressesType = None,
group_address_controller_status_state: GroupAddressesType = None,
group_address_controller_mode: GroupAddressesType = None,
group_address_controller_mode_state: GroupAddressesType = None,
group_address_heat_cool: GroupAddressesType = None,
group_address_heat_cool_state: GroupAddressesType = None,
sync_state: bool | int | float | str = True,
operation_modes: list[str | HVACOperationMode] | None = None,
controller_modes: list[str | HVACControllerMode] | None = None,
device_updated_cb: DeviceCallbackType[ClimateMode] | None = None,
) -> None:
"""Initialize ClimateMode class."""
super().__init__(xknx, name, device_updated_cb)
self.remote_value_operation_mode = RemoteValueOperationMode(
xknx,
group_address=group_address_operation_mode,
group_address_state=group_address_operation_mode_state,
sync_state=sync_state,
device_name=name,
feature_name="Operation mode",
after_update_cb=self._set_internal_operation_mode,
)
self.remote_value_controller_mode = RemoteValueControllerMode(
xknx,
group_address=group_address_controller_mode,
group_address_state=group_address_controller_mode_state,
sync_state=sync_state,
device_name=name,
feature_name="Controller mode",
after_update_cb=self._set_internal_controller_mode,
)
self.remote_value_controller_status = RemoteValueHVACStatus(
xknx,
group_address=group_address_controller_status,
group_address_state=group_address_controller_status_state,
sync_state=sync_state,
device_name=name,
feature_name="Controller status",
after_update_cb=self._set_internal_modes_from_status,
)
self.remote_value_operation_mode_comfort = RemoteValueBinaryOperationMode(
xknx,
group_address=group_address_operation_mode_comfort,
group_address_state=group_address_operation_mode_comfort,
sync_state=sync_state,
device_name=name,
feature_name="Operation mode Comfort",
operation_mode=HVACOperationMode.COMFORT,
after_update_cb=self._set_internal_operation_mode,
)
self.remote_value_operation_mode_standby = RemoteValueBinaryOperationMode(
xknx,
group_address=group_address_operation_mode_standby,
group_address_state=group_address_operation_mode_standby,
sync_state=sync_state,
device_name=name,
feature_name="Operation mode Standby",
operation_mode=HVACOperationMode.STANDBY,
after_update_cb=self._set_internal_operation_mode,
)
self.remote_value_operation_mode_economy = RemoteValueBinaryOperationMode(
xknx,
group_address=group_address_operation_mode_economy,
group_address_state=group_address_operation_mode_economy,
sync_state=sync_state,
device_name=name,
feature_name="Operation mode Economy",
operation_mode=HVACOperationMode.ECONOMY,
after_update_cb=self._set_internal_operation_mode,
)
self.remote_value_operation_mode_protection = RemoteValueBinaryOperationMode(
xknx,
group_address=group_address_operation_mode_protection,
group_address_state=group_address_operation_mode_protection,
sync_state=sync_state,
device_name=name,
feature_name="Operation mode Protection",
operation_mode=HVACOperationMode.BUILDING_PROTECTION,
after_update_cb=self._set_internal_operation_mode,
)
self.remote_value_heat_cool = RemoteValueBinaryHeatCool(
xknx,
group_address=group_address_heat_cool,
group_address_state=group_address_heat_cool_state,
sync_state=sync_state,
device_name=name,
feature_name="Heat/Cool",
controller_mode=HVACControllerMode.HEAT,
after_update_cb=self._set_internal_controller_mode,
)
self.operation_mode = HVACOperationMode.STANDBY
self.controller_mode = HVACControllerMode.HEAT
self._operation_modes = self.gather_operation_modes(only_writable=True)
if operation_modes is not None:
custom_operation_modes = []
for op_mode in operation_modes:
if isinstance(op_mode, str):
custom_operation_modes.append(
HVACOperationMode[op_mode.replace(" ", "_").upper()]
)
elif isinstance(op_mode, HVACOperationMode):
custom_operation_modes.append(op_mode)
self._operation_modes = [
mode for mode in custom_operation_modes if mode in self._operation_modes
]
self._controller_modes = self.gather_controller_modes(only_writable=True)
if controller_modes is not None:
custom_controller_modes = []
for ct_mode in controller_modes:
if isinstance(ct_mode, str):
custom_controller_modes.append(
HVACControllerMode[ct_mode.replace(" ", "_").upper()]
)
elif isinstance(ct_mode, HVACControllerMode):
custom_controller_modes.append(ct_mode)
self._controller_modes = [
mode
for mode in custom_controller_modes
if mode in self._controller_modes
]
self.supports_operation_mode = bool(
self.gather_operation_modes(only_writable=False)
)
self.supports_controller_mode = bool(
self.gather_controller_modes(only_writable=False)
)
def _iter_remote_values(
self,
) -> Iterator[RemoteValueClimateModeBase[Any]]:
"""Iterate climate mode RemoteValue classes."""
yield self.remote_value_operation_mode
yield self.remote_value_controller_mode
yield self.remote_value_controller_status
yield self.remote_value_heat_cool
yield self.remote_value_operation_mode_comfort
yield self.remote_value_operation_mode_economy
yield self.remote_value_operation_mode_protection
yield self.remote_value_operation_mode_standby
def _set_internal_operation_mode(
self, operation_mode: HVACOperationMode | None
) -> None:
"""Set internal value of operation mode. Call hooks if operation mode was changed."""
if operation_mode is not None and operation_mode != self.operation_mode:
self.operation_mode = operation_mode
self.after_update()
def _set_internal_controller_mode(
self, controller_mode: HVACControllerMode
) -> None:
"""Set internal value of controller mode. Call hooks if controller mode was changed."""
if controller_mode != self.controller_mode:
self.controller_mode = controller_mode
self.after_update()
def _set_internal_modes_from_status(self, status: HVACStatus) -> None:
"""Set internal values from HVACStatus."""
updated = False
if status.mode != self.operation_mode:
self.operation_mode = status.mode
updated = True
contr_mode_heat_cool = (
HVACControllerMode.HEAT
if status.heat_cool is HeatCool.HEAT
else HVACControllerMode.COOL
)
if contr_mode_heat_cool != self.controller_mode:
self.controller_mode = contr_mode_heat_cool
updated = True
if updated:
self.after_update()
async def set_operation_mode(self, operation_mode: HVACOperationMode) -> None:
"""Set the operation mode of a thermostat. Send new operation_mode to BUS and update internal state."""
if (
not self.supports_operation_mode
or operation_mode not in self._operation_modes
):
raise DeviceIllegalValue(
"operation (preset) mode not supported", str(operation_mode)
)
for rv in self._iter_remote_values():
if rv.writable:
rv.set_operation_mode(operation_mode)
self._set_internal_operation_mode(operation_mode)
async def set_controller_mode(self, controller_mode: HVACControllerMode) -> None:
"""Set the controller mode of a thermostat. Send new controller mode to the bus and update internal state."""
if (
not self.supports_controller_mode
or controller_mode not in self._controller_modes
):
raise DeviceIllegalValue(
"controller (HVAC) mode not supported", str(controller_mode)
)
for rv in self._iter_remote_values():
if rv.writable:
rv.set_controller_mode(controller_mode)
self._set_internal_controller_mode(controller_mode)
@property
def operation_modes(self) -> list[HVACOperationMode]:
"""Return all configured operation modes."""
return self._operation_modes
@property
def controller_modes(self) -> list[HVACControllerMode]:
"""Return all configured controller modes."""
return self._controller_modes
def gather_operation_modes(
self, only_writable: bool = True
) -> list[HVACOperationMode]:
"""Gather operation modes from RemoteValues."""
operation_modes: list[HVACOperationMode] = []
for rv in self._iter_remote_values():
if rv.initialized:
if only_writable and not rv.writable:
continue
operation_modes.extend(rv.supported_operation_modes())
# remove duplicates
return list(set(operation_modes))
def gather_controller_modes(
self, only_writable: bool = True
) -> list[HVACControllerMode]:
"""Gather controller modes from RemoteValues."""
controller_modes: list[HVACControllerMode] = []
for rv in self._iter_remote_values():
if rv.initialized:
if only_writable and not rv.writable:
continue
controller_modes.extend(rv.supported_controller_modes())
# remove duplicates
return list(set(controller_modes))
def process_group_write(self, telegram: Telegram) -> None:
"""Process incoming and outgoing GROUP WRITE telegram."""
for rv in self._iter_remote_values():
rv.process(telegram)
def __str__(self) -> str:
"""Return object as readable string."""
return (
f'<ClimateMode name="{self.name}" '
f"operation_mode={self.remote_value_operation_mode.group_addr_str()} "
f"controller_mode={self.remote_value_controller_mode.group_addr_str()} "
f"controller_status={self.remote_value_controller_status.group_addr_str()} "
"/>"
)
|
