aiokonstsmide

13async def find_devices(timeout: float = 5.0) -> AsyncGenerator[str, None]:
14    """
15    Scans for available Konstsmide Bluetooth devices.
16
17    This function is an [asynchronous generator](https://peps.python.org/pep-0525/) and can be used with `async for`.
18
19    :param timeout: Time in seconds to scan for devices
20
21    :return: An asynchronous generator with addresses of found Konstsmide devices
22    """
23    for device in await BleakScanner.discover(timeout=timeout, return_adv=False):
24        if device.name and device.name.strip().lower() == DEVICE_NAME:
25            yield device.address

28async def check_address(address: str, timeout: float = 5.0) -> bool:
29    """
30    Checks if the given address is a valid reachable Konstsmide device.
31
32    :param address: The address of the device to check
33    :param timeout: Timeout in seconds
34
35    :return: True if the address is a valid device, False otherwise
36    """
37    device = await BleakScanner.find_device_by_address(address, timeout=timeout)
38    if device and device.name and device.name.strip().lower() == DEVICE_NAME:
39        return True
40    return False

19async def connect(
20    address: str,
21    password: Optional[str] = None,
22    on: bool = True,
23    function: message.Function = message.Function.Steady,
24    brightness: int = 100,
25    flash_speed: int = 50,
26    timeout: float = 5.0,
27) -> "Device":
28    """
29    Connects to the device with the given address.
30
31    Make sure to call `Device.disconnect` once you're finished,
32    otherwise it will not be possible to connect to the device anymore
33    unless the power is cut.
34
35    :param address: The address of the device to connect to
36    :param password: The password of the device
37    :param on: If the device should be turned on or off after connecting
38    :param function: The function to set after connecting
39    :param brightness: The brightness to set after connecting, in the range 0 (dim) - 100 (bright)
40    :param flash_speed: The flash speed to set after connecting, in the range 0 (slow) - 100 (fast)
41    :param timeout: Timeout in seconds
42
43    :return: A Device instance connected to the device with the given address
44    """
45    device = Device(address, password, on, function, brightness, flash_speed)
46    await device.connect(timeout)
47    return device

 60class Device:
 61    """
 62    Represents a Konstsmide Bluetooth device.
 63    """
 64
 65    def __init__(
 66        self,
 67        address: str,
 68        password: Optional[str] = None,
 69        on: bool = True,
 70        function: message.Function = message.Function.Steady,
 71        brightness: int = 100,
 72        flash_speed: int = 50,
 73    ):
 74        """
 75        Initializes a Device instance.
 76
 77        :param address: The address of the device to connect to
 78        :param password: The password of the device
 79        :param on: If the device should be turned on or off after connecting
 80        :param function: The function to set after connecting
 81        :param brightness: The brightness to set after connecting, in the range 0 (dim) - 100 (bright)
 82        :param flash_speed: The flash speed to set after connecting, in the range 0 (slow) - 100 (fast)
 83        """
 84        self.__logger = logging.getLogger(f"{__package__}({address})")
 85        self.__address = address
 86        self.__password = password or "123456"
 87        self.__status = Status(on, function, brightness, flash_speed)
 88        self.__client: BleakClient = None
 89        self.__reconnect = True
 90
 91    async def connect(self, timeout: float = 5.0):
 92        """
 93        Establishes a connection to the device.
 94
 95        :param timeout: The timeout in seconds
 96        """
 97        if not self.__client:
 98            if not await check_address(self.__address):
 99                raise DeviceNotFoundError
100
101            def on_disconnect(client: BleakClient):
102                if self.__reconnect:
103                    self.__logger.debug("Device disconnected, trying to reconnect")
104                    asyncio.create_task(self.connect(timeout))
105
106            self.__client = BleakClient(
107                self.__address,
108                disconnected_callback=on_disconnect,
109                timeout=timeout,
110            )
111
112        self.__reconnect = True
113        if not self.__client.is_connected:
114            await self.__client.connect()
115            if self.__client.is_connected:
116                self.__logger.debug("Device connected, sending password")
117                await self.__write(message.password_input(self.__password))
118                self.__logger.debug("Synchronizing status")
119                await self.__sync_status()
120                self.__logger.debug("Synchronizing time")
121                await self.sync_time()
122            else:
123                self.__logger.error("Failed to connect to device")
124
125    async def __sync_status(self):
126        """
127        Synchronizes the status of the device.
128        Since the status of the device can't be read,
129        the device status is set according to the internal status.
130        """
131        init_state = self.__status.on
132
133        await self.control(
134            function=self.__status.function,
135            brightness=self.__status.brightness,
136            flash_speed=self.__status.flash_speed,
137        )
138
139        if init_state:
140            await self.on()
141        else:
142            await self.off()
143
144    async def disconnect(self):
145        """Disconnects from the device."""
146        self.__reconnect = False
147        if self.__client and self.__client.is_connected:
148            await self.__client.disconnect()
149
150    async def __aenter__(self):
151        await self.connect()
152        return self
153
154    async def __aexit__(self, _exc_type, _exc_val, _exc_tb):
155        await self.disconnect()
156
157    @property
158    def is_on(self) -> bool:
159        """`True` if the device is currently on, else `False`."""
160        return self.__status.on
161
162    @property
163    def brightness(self) -> int:
164        """The current brightness of the device."""
165        return self.__status.brightness
166
167    @property
168    def flash_speed(self) -> int:
169        """The current flash speed of the device."""
170        return self.__status.flash_speed
171
172    @property
173    def function(self) -> message.Function:
174        """The current function of the device."""
175        return self.__status.function
176
177    async def on(self):
178        """Turn on the device."""
179        self.__logger.debug("Turning on")
180        self.__status.on = True
181        await self.__write(message.on_off(self.__status.on))
182
183    async def off(self):
184        """Turn off the device."""
185        self.__logger.debug("Turning off")
186        self.__status.on = False
187        await self.__write(message.on_off(self.__status.on))
188
189    async def toggle(self):
190        """Toggle between on and off."""
191        if self.__status.on:
192            await self.off()
193        else:
194            await self.on()
195
196    async def control(
197        self,
198        function: Optional[message.Function] = None,
199        brightness: Optional[int] = None,
200        flash_speed: Optional[int] = None,
201    ):
202        """
203        Control the devices function, brightness and flash speed.
204        If a parameter is None, the current value will be kept.
205
206        :param function: The function to set
207        :param brightness: The brightness to set, in the range 0 (dim) - 100 (bright)
208        :param flash_speed: The flash speed to set, in the range 0 (slow) - 100 (fast)
209        """
210        if function:
211            self.__status.function = function
212        if brightness:
213            self.__status.brightness = brightness
214        if flash_speed:
215            self.__status.flash_speed = flash_speed
216
217        # Control turns on the device automatically
218        self.__status.on = True
219
220        self.__logger.debug(
221            f"Setting function {self.__status.function.name} with brightness {self.__status.brightness} and flash speed {self.__status.flash_speed}"
222        )
223        await self.__write(
224            message.control(
225                self.__status.function,
226                self.__status.brightness,
227                self.__status.flash_speed,
228            ),
229        )
230
231    async def deactivate_timer(self, num: Optional[int] = None):
232        """
233        Deactivates one specific or all timers on the device.
234
235        :param num: The timer to deactivate, in the range 0 - 7 or `None` for all timers
236        """
237        if num is not None:
238            await self.timer(
239                num,
240                False,
241                False,
242                0,
243                0,
244                message.Function.Steady,
245                [],
246            )
247        else:
248            for i in range(8):
249                await self.__write(
250                    message.timer(
251                        i,
252                        False,
253                        False,
254                        0,
255                        0,
256                        message.Function.Steady,
257                        [],
258                        self.__status.brightness,
259                    )
260                )
261
262    async def timer(
263        self,
264        num: int,
265        active: bool,
266        turn_on: bool,
267        hour: int,
268        minute: int,
269        function: message.Function,
270        repeat: Union[message.Repeat, List[message.Repeat]],
271    ):
272        """
273        Configures a timer on the device.
274        The device has 8 built-in timers which can be set individually as desired.
275
276        :param num: The timer to set, in the range 0 - 7
277        :param active: `True` if the timer should be activated or `False` if it should be deactivated
278        :param turn_on: `True` if the device should be turned on the timer is triggered, `False` otherwise
279        :param hour: The hour (0-23) at which the timer triggers
280        :param minute: The minute (0-59) at which the timer triggers
281        :param function: The function to set when the timer is triggered
282        :param repeat: On which weekdays the timer triggers
283        """
284
285        # Make sure time is synchronized
286        await self.sync_time()
287
288        # Create timer
289        if isinstance(repeat, message.Repeat):
290            repeat = [repeat]
291
292        await self.__write(
293            message.timer(
294                num,
295                active,
296                turn_on,
297                hour,
298                minute,
299                function,
300                repeat,
301                self.__status.brightness,
302            )
303        )
304
305    async def sync_time(self):
306        """
307        Sends an RTC message to the device to synchronize the time.
308        This is needed for timers to work correctly.
309
310        Time is synchronized implicitly when connecting to the device or changing a timer.
311        """
312        await self.__write(message.rtc(datetime.now()))
313
314    async def __write(self, message: bytes):
315        """Writes the given message to the device."""
316        if self.__client and self.__client.is_connected:
317            self.__logger.debug(f"Sending message to device: {message.hex()}")
318            enc_msg = codec.encode(message)
319            await self.__client.write_gatt_char(CHARACTERISTIC, enc_msg)
320        else:
321            self.__logger.error(
322                "Tried to send message to device, but it's disconnected!"
323            )

23class Function(Enum):
24    """Functions which the device supports."""
25
26    Keep = 0
27    """Keep the currently set function, only available for timers."""
28    Combination = 1
29    InWaves = 2
30    Sequential = 3
31    SloGlo = 4
32    Chasing = 5
33    SlowFade = 6
34    Twinkle = 7
35    Steady = 8
36    FlashAlternating = 9
37    """This function is not supported for timers."""
38    FlashSynchronous = 10
39    """This function is not supported for timers."""

42class Repeat(Enum):
43    """Weekdays which can be used for repeating timers."""
44
45    Sunday = 1
46    Monday = 2
47    Tuesday = 4
48    Wednesday = 8
49    Thursday = 16
50    Friday = 32
51    Saturday = 64
52    Weekend = 65
53    Weekdays = 62
54    Everyday = 127

7class AioKonstmideError(Exception):
8    """Base error."""

19class DeviceNotFoundError(AioKonstmideError):
20    """The device couldn't be found or is not a valid Konstsmide Bluetooth device."""

11class EncodeError(AioKonstmideError):
12    """Tried to encode an invalid message."""

15class DecodeError(AioKonstmideError):
16    """Tried to decode an invalid message."""