File: remote_value_climate_mode.py

package info (click to toggle)
python-xknx 3.6.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 4,012 kB
  • sloc: python: 39,710; javascript: 8,556; makefile: 27; sh: 12
file content (398 lines) | stat: -rw-r--r-- 14,343 bytes parent folder | download
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
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
"""
Module for managing an climate mode remote values.

DPT .
"""

from __future__ import annotations

from abc import abstractmethod
from typing import TYPE_CHECKING, Any, TypeVar

from xknx.dpt import (
    DPTArray,
    DPTBinary,
    DPTHVACContrMode,
    DPTHVACMode,
    DPTHVACStatus,
)
from xknx.dpt.dpt_1 import HeatCool
from xknx.dpt.dpt_20 import HVACControllerMode, HVACOperationMode, HVACStatus
from xknx.exceptions import ConversionError, CouldNotParseTelegram

from .remote_value import (
    GroupAddressesType,
    RemoteValue,
    RVCallbackType,
)

if TYPE_CHECKING:
    from xknx.xknx import XKNX


HVACModeT = TypeVar(
    "HVACModeT",
    HVACControllerMode,
    HVACOperationMode,
    HVACStatus,
    HVACOperationMode | None,
)


class RemoteValueClimateModeBase(RemoteValue[HVACModeT]):
    """Base class for binary climate mode remote values."""

    @abstractmethod
    def supported_operation_modes(self) -> list[HVACOperationMode]:
        """Return a list of all supported operation modes."""

    @abstractmethod
    def set_operation_mode(self, mode: HVACOperationMode) -> None:
        """Set operation mode. Return if not supported."""

    @abstractmethod
    def supported_controller_modes(self) -> list[HVACControllerMode]:
        """Return a list of all supported controller modes."""

    @abstractmethod
    def set_controller_mode(self, mode: HVACControllerMode) -> None:
        """Set controller mode. Return if not supported."""


class RemoteValueOperationMode(RemoteValueClimateModeBase[HVACOperationMode]):
    """Abstraction for remote value of KNX climate operation modes."""

    dpt_class = DPTHVACMode

    def __init__(
        self,
        xknx: XKNX,
        group_address: GroupAddressesType = None,
        group_address_state: GroupAddressesType = None,
        sync_state: bool | int | float | str = True,
        device_name: str | None = None,
        feature_name: str = "Climate mode",
        after_update_cb: RVCallbackType[HVACOperationMode] | None = None,
    ) -> None:
        """Initialize remote value of KNX climate mode."""
        super().__init__(
            xknx,
            group_address=group_address,
            group_address_state=group_address_state,
            sync_state=sync_state,
            device_name=device_name,
            feature_name=feature_name,
            after_update_cb=after_update_cb,
        )

    def supported_operation_modes(self) -> list[HVACOperationMode]:
        """Return a list of all supported operation modes."""
        return self.dpt_class.get_valid_values()

    def set_operation_mode(self, mode: HVACOperationMode) -> None:
        """Set operation mode. Return if not supported."""
        if mode not in self.supported_operation_modes():
            return
        super().set(mode)

    def supported_controller_modes(self) -> list[HVACControllerMode]:
        """Return a list of all supported controller modes."""
        return []

    def set_controller_mode(self, mode: HVACControllerMode) -> None:
        """Set controller mode. Return if not supported."""
        return


class RemoteValueControllerMode(RemoteValueClimateModeBase[HVACControllerMode]):
    """Abstraction for remote value of KNX climate controller modes."""

    dpt_class = DPTHVACContrMode

    def __init__(
        self,
        xknx: XKNX,
        group_address: GroupAddressesType = None,
        group_address_state: GroupAddressesType = None,
        sync_state: bool | int | float | str = True,
        device_name: str | None = None,
        feature_name: str = "Controller Mode",
        after_update_cb: RVCallbackType[HVACControllerMode] | None = None,
    ) -> None:
        """Initialize remote value of KNX climate mode."""
        super().__init__(
            xknx,
            group_address=group_address,
            group_address_state=group_address_state,
            sync_state=sync_state,
            device_name=device_name,
            feature_name=feature_name,
            after_update_cb=after_update_cb,
        )

    def supported_operation_modes(self) -> list[HVACOperationMode]:
        """Return a list of all supported operation modes."""
        return []

    def set_operation_mode(self, mode: HVACOperationMode) -> None:
        """Set operation mode. Return if not supported."""
        return

    def supported_controller_modes(self) -> list[HVACControllerMode]:
        """Return a list of all supported controller modes."""
        return DPTHVACContrMode.get_valid_values()

    def set_controller_mode(self, mode: HVACControllerMode) -> None:
        """Set controller mode. Return if not supported."""
        if mode not in self.supported_controller_modes():
            return
        super().set(mode)


class RemoteValueHVACStatus(RemoteValueClimateModeBase[HVACStatus]):
    """Abstraction for remote value of KNX climate HVAC status (Eberle status)."""

    dpt_class = DPTHVACStatus

    def __init__(
        self,
        xknx: XKNX,
        group_address: GroupAddressesType = None,
        group_address_state: GroupAddressesType = None,
        sync_state: bool | int | float | str = True,
        device_name: str | None = None,
        feature_name: str = "Controller status",
        after_update_cb: RVCallbackType[HVACStatus] | None = None,
    ) -> None:
        """Initialize remote value of KNX climate controller status."""
        super().__init__(
            xknx,
            group_address=group_address,
            group_address_state=group_address_state,
            sync_state=sync_state,
            device_name=device_name,
            feature_name=feature_name,
            after_update_cb=after_update_cb,
        )

    def supported_operation_modes(self) -> list[HVACOperationMode]:
        """Return a list of all supported operation modes."""
        return [
            HVACOperationMode.COMFORT,
            HVACOperationMode.STANDBY,
            HVACOperationMode.ECONOMY,
            HVACOperationMode.BUILDING_PROTECTION,
        ]

    def set_operation_mode(self, mode: HVACOperationMode) -> None:
        """Set operation mode. Return if not supported."""
        if mode not in self.supported_operation_modes():
            return
        if self._value is None:
            raise ConversionError(
                "HVACStatus value not initialized. Can not write new mode.",
                device_name=self.device_name,
            )
        new_status = HVACStatus(
            mode=mode,
            dew_point=self._value.dew_point,
            heat_cool=self._value.heat_cool,
            inactive=self._value.inactive,
            frost_alarm=self._value.frost_alarm,
        )
        return super().set(new_status)

    def supported_controller_modes(self) -> list[HVACControllerMode]:
        """Return a list of all supported controller modes."""
        return [HVACControllerMode.HEAT, HVACControllerMode.COOL]

    def set_controller_mode(self, mode: HVACControllerMode) -> None:
        """Set controller mode. Return if not supported."""
        if mode not in self.supported_controller_modes():
            return
        if self._value is None:
            raise ConversionError(
                "HVACStatus value not initialized. Can not write new mode.",
                device_name=self.device_name,
            )
        new_status = HVACStatus(
            mode=self._value.mode,
            dew_point=self._value.dew_point,
            heat_cool=HeatCool.HEAT
            if mode == HVACControllerMode.HEAT
            else HeatCool.COOL,
            inactive=self._value.inactive,
            frost_alarm=self._value.frost_alarm,
        )
        super().set(new_status)


class RemoteValueBinaryOperationMode(
    RemoteValueClimateModeBase[HVACOperationMode | None]
):
    """Abstraction for remote value of split up KNX climate modes."""

    def __init__(
        self,
        xknx: XKNX,
        group_address: GroupAddressesType = None,
        group_address_state: GroupAddressesType = None,
        sync_state: bool | int | float | str = True,
        device_name: str | None = None,
        feature_name: str = "Climate mode binary",
        after_update_cb: RVCallbackType[HVACOperationMode | None] | None = None,
        operation_mode: HVACOperationMode | None = None,
    ) -> None:
        """Initialize remote value of KNX DPT 1 representing a climate operation mode."""
        if not isinstance(operation_mode, HVACOperationMode):
            raise ConversionError(
                "Invalid operation mode type",
                operation_mode=str(operation_mode),
                device_name=str(device_name),
                feature_name=feature_name,
            )
        self.operation_mode = operation_mode
        super().__init__(
            xknx,
            group_address=group_address,
            group_address_state=group_address_state,
            sync_state=sync_state,
            device_name=device_name,
            feature_name=feature_name,
            after_update_cb=after_update_cb,
        )

    def supported_operation_modes(self) -> list[HVACOperationMode]:
        """Return a list of all supported operation modes."""
        # all binary operation modes off -> Standby according to MDT
        return (
            [self.operation_mode, HVACOperationMode.STANDBY]
            if self.operation_mode is not HVACOperationMode.STANDBY
            else [HVACOperationMode.STANDBY]
        )

    def set_operation_mode(self, mode: HVACOperationMode) -> None:
        """Set operation mode. Return if not supported."""
        super().set(mode)

    def supported_controller_modes(self) -> list[HVACControllerMode]:
        """Return a list of all supported controller modes."""
        return []

    def set_controller_mode(self, mode: HVACControllerMode) -> None:
        """Set controller mode. Return if not supported."""
        return

    def to_knx(self, value: HVACOperationMode | None) -> DPTBinary:
        """Convert value to payload."""
        if isinstance(value, HVACOperationMode):
            # foreign operation modes will set the RemoteValue to False
            return DPTBinary(value == self.operation_mode)
        raise ConversionError(
            "value invalid",
            value=value,
            device_name=self.device_name,
            feature_name=self.feature_name,
        )

    def from_knx(self, payload: DPTArray | DPTBinary) -> HVACOperationMode | None:
        """Convert current payload to value."""
        if payload.value == 1:
            return self.operation_mode
        if payload.value == 0:
            return None
        raise CouldNotParseTelegram(
            "Payload invalid",
            payload=str(payload),
            device_name=self.device_name,
            feature_name=self.feature_name,
        )


class RemoteValueBinaryHeatCool(RemoteValueClimateModeBase[HVACControllerMode]):
    """Abstraction for remote value of heat/cool controller mode."""

    def __init__(
        self,
        xknx: XKNX,
        group_address: GroupAddressesType = None,
        group_address_state: GroupAddressesType = None,
        sync_state: bool | int | float | str = True,
        device_name: str | None = None,
        feature_name: str = "Controller mode Heat/Cool",
        after_update_cb: RVCallbackType[HVACControllerMode] | None = None,
        controller_mode: HVACControllerMode | None = None,
    ) -> None:
        """Initialize remote value of KNX DPT 1 representing a climate controller mode."""
        if not isinstance(controller_mode, HVACControllerMode):
            raise ConversionError(
                "Invalid controller mode type",
                controller_mode=str(controller_mode),
                device_name=str(device_name),
                feature_name=feature_name,
            )
        if controller_mode not in self.supported_controller_modes():
            raise ConversionError(
                "Controller mode not supported for binary mode object",
                controller_mode=str(controller_mode),
                device_name=str(device_name),
                feature_name=feature_name,
            )
        self.controller_mode = controller_mode
        super().__init__(
            xknx,
            group_address=group_address,
            group_address_state=group_address_state,
            sync_state=sync_state,
            device_name=device_name,
            feature_name=feature_name,
            after_update_cb=after_update_cb,
        )

    def supported_operation_modes(self) -> list[HVACOperationMode]:
        """Return a list of all supported operation modes."""
        return []

    def set_operation_mode(self, mode: HVACOperationMode) -> None:
        """Set operation mode. Return if not supported."""
        return

    def supported_controller_modes(self) -> list[HVACControllerMode]:
        """Return a list of all supported controller modes."""
        return [HVACControllerMode.HEAT, HVACControllerMode.COOL]

    def set_controller_mode(self, mode: HVACControllerMode) -> None:
        """Set controller mode. Return if not supported."""
        if mode not in self.supported_controller_modes():
            return
        super().set(mode)

    def to_knx(self, value: Any) -> DPTBinary:
        """Convert value to payload."""
        if isinstance(value, HVACControllerMode):
            # foreign operation modes will set the RemoteValue to False
            return DPTBinary(value == self.controller_mode)
        raise ConversionError(
            "value invalid",
            value=value,
            device_name=self.device_name,
            feature_name=self.feature_name,
        )

    def from_knx(self, payload: DPTArray | DPTBinary) -> HVACControllerMode:
        """Convert current payload to value."""
        if payload.value == 1:
            return self.controller_mode
        if payload.value == 0:
            # return the other operation mode
            return next(
                _op
                for _op in self.supported_controller_modes()
                if _op is not self.controller_mode
            )
        raise CouldNotParseTelegram(
            "Payload invalid",
            payload=str(payload),
            device_name=self.device_name,
            feature_name=self.feature_name,
        )