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
|
"""
Module for managing a fan via KNX.
It provides functionality for
* setting fan to specific speed / step
* reading the current speed from KNX bus.
"""
from __future__ import annotations
from collections.abc import Iterator
from enum import Enum
import logging
from typing import TYPE_CHECKING, Any
from xknx.remote_value import (
GroupAddressesType,
RemoteValue,
RemoteValueDptValue1Ucount,
RemoteValueScaling,
RemoteValueSwitch,
)
from .device import Device, DeviceCallbackType
if TYPE_CHECKING:
from xknx.telegram import Telegram
from xknx.xknx import XKNX
DEFAULT_TURN_ON_SPEED = 50
logger = logging.getLogger("xknx.log")
class FanSpeedMode(Enum):
"""Enum for setting the fan speed mode."""
PERCENT = 1
STEP = 2
class Fan(Device):
"""Class for managing a fan."""
def __init__(
self,
xknx: XKNX,
name: str,
group_address_speed: GroupAddressesType = None,
group_address_speed_state: GroupAddressesType = None,
group_address_oscillation: GroupAddressesType = None,
group_address_oscillation_state: GroupAddressesType = None,
group_address_switch: GroupAddressesType = None,
group_address_switch_state: GroupAddressesType = None,
sync_state: bool | int | float | str = True,
device_updated_cb: DeviceCallbackType[Fan] | None = None,
max_step: int | None = None,
) -> None:
"""Initialize fan class."""
super().__init__(xknx, name, device_updated_cb)
self.speed: RemoteValueDptValue1Ucount | RemoteValueScaling
self.mode = FanSpeedMode.STEP if max_step else FanSpeedMode.PERCENT
self.max_step = max_step
# If there is a dedicated switch GA, it controls the on/off behavior of the fan.
# Otherwise the speed GA of the fan implicitly controls the on/off behavior instead.
# `self.switch.initialized`` can be used to check which setup is used.
self.switch = RemoteValueSwitch(
xknx,
group_address_switch,
group_address_switch_state,
sync_state=sync_state,
device_name=self.name,
feature_name="Switch",
after_update_cb=self.after_update,
)
if self.mode == FanSpeedMode.STEP:
self.speed = RemoteValueDptValue1Ucount(
xknx,
group_address_speed,
group_address_speed_state,
sync_state=sync_state,
device_name=self.name,
feature_name="Speed",
after_update_cb=self.after_update,
)
else:
self.speed = RemoteValueScaling(
xknx,
group_address_speed,
group_address_speed_state,
sync_state=sync_state,
device_name=self.name,
feature_name="Speed",
after_update_cb=self.after_update,
range_from=0,
range_to=100,
)
self.oscillation = RemoteValueSwitch(
xknx,
group_address_oscillation,
group_address_oscillation_state,
sync_state=sync_state,
device_name=self.name,
feature_name="Oscillation",
after_update_cb=self.after_update,
)
def _iter_remote_values(self) -> Iterator[RemoteValue[Any]]:
"""Iterate the devices RemoteValue classes."""
yield from (self.switch, self.speed, self.oscillation)
@property
def supports_oscillation(self) -> bool:
"""Return if fan supports oscillation."""
return self.oscillation.initialized
@property
def is_on(self) -> bool:
"""Return the current fan state of the device."""
if self.switch.initialized:
return bool(self.switch.value)
return bool(self.current_speed)
async def turn_on(self, speed: int | None = None) -> None:
"""Turn on fan."""
if self.switch.initialized:
self.switch.on()
# For a switch GA fan, we only use an explicitly provided speed, but not
# arbitrarily set a default speed here, compared to the speed GA based fans below.
if speed is not None:
await self.set_speed(speed)
else:
await self.set_speed(speed or DEFAULT_TURN_ON_SPEED)
async def turn_off(self) -> None:
"""Turn off fan."""
if self.switch.initialized:
self.switch.off()
else:
await self.set_speed(0)
async def set_speed(self, speed: int) -> None:
"""Set the fan to a designated speed."""
self.speed.set(speed)
async def set_oscillation(self, oscillation: bool) -> None:
"""Set the fan oscillation mode on or off."""
self.oscillation.set(oscillation)
def process_group_write(self, telegram: Telegram) -> None:
"""Process incoming and outgoing GROUP WRITE telegram."""
self.switch.process(telegram)
self.speed.process(telegram)
self.oscillation.process(telegram)
@property
def current_speed(self) -> int | None:
"""Return current speed of fan."""
return self.speed.value
@property
def current_oscillation(self) -> bool | None:
"""Return true if the fan is oscillating."""
return self.oscillation.value
def __str__(self) -> str:
"""Return object as readable string."""
str_oscillation = (
f" oscillation={self.oscillation.group_addr_str()}"
if self.supports_oscillation
else ""
)
return f'<Fan name="{self.name}" speed={self.speed.group_addr_str()}{str_oscillation} />'
|
