"""Library to handle connection with Switchbot.""" from __future__ import annotations import asyncio import binascii import logging import time from dataclasses import replace from enum import Enum from typing import Any, Callable, TypeVar, cast from uuid import UUID import async_timeout from bleak import BleakError from bleak.backends.device import BLEDevice from bleak.backends.service import BleakGATTCharacteristic, BleakGATTServiceCollection from bleak.exc import BleakDBusError from bleak_retry_connector import ( BLEAK_RETRY_EXCEPTIONS, BleakClientWithServiceCache, BleakNotFoundError, ble_device_has_changed, establish_connection, ) from ..const import DEFAULT_RETRY_COUNT, DEFAULT_SCAN_TIMEOUT from ..discovery import GetSwitchbotDevices from ..models import SwitchBotAdvertisement _LOGGER = logging.getLogger(__name__) REQ_HEADER = "570f" # Keys common to all device types DEVICE_GET_BASIC_SETTINGS_KEY = "5702" DEVICE_SET_MODE_KEY = "5703" DEVICE_SET_EXTENDED_KEY = REQ_HEADER # Base key when encryption is set KEY_PASSWORD_PREFIX = "571" # How long to hold the connection # to wait for additional commands for # disconnecting the device. DISCONNECT_DELAY = 8.5 class ColorMode(Enum): OFF = 0 COLOR_TEMP = 1 RGB = 2 EFFECT = 3 # If the scanner is in passive mode, we # need to poll the device to get the # battery and a few rarely updating # values. PASSIVE_POLL_INTERVAL = 60 * 60 * 24 class CharacteristicMissingError(Exception): """Raised when a characteristic is missing.""" class SwitchbotOperationError(Exception): """Raised when an operation fails.""" def _sb_uuid(comms_type: str = "service") -> UUID | str: """Return Switchbot UUID.""" _uuid = {"tx": "002", "rx": "003", "service": "d00"} if comms_type in _uuid: return UUID(f"cba20{_uuid[comms_type]}-224d-11e6-9fb8-0002a5d5c51b") return "Incorrect type, choose between: tx, rx or service" READ_CHAR_UUID = _sb_uuid(comms_type="rx") WRITE_CHAR_UUID = _sb_uuid(comms_type="tx") WrapFuncType = TypeVar("WrapFuncType", bound=Callable[..., Any]) def update_after_operation(func: WrapFuncType) -> WrapFuncType: """Define a wrapper to update after an operation.""" async def _async_update_after_operation_wrap( self: SwitchbotBaseDevice, *args: Any, **kwargs: Any ) -> None: ret = await func(self, *args, **kwargs) await self.update() return ret return cast(WrapFuncType, _async_update_after_operation_wrap) def _merge_data(old_data: dict[str, Any], new_data: dict[str, Any]) -> dict[str, Any]: """Merge data but only add None keys if they are missing.""" merged = old_data.copy() for key, value in new_data.items(): if value is not None or key not in old_data: merged[key] = value return merged class SwitchbotBaseDevice: """Base Representation of a Switchbot Device.""" def __init__( self, device: BLEDevice, password: str | None = None, interface: int = 0, **kwargs: Any, ) -> None: """Switchbot base class constructor.""" self._interface = f"hci{interface}" self._device = device self._sb_adv_data: SwitchBotAdvertisement | None = None self._override_adv_data: dict[str, Any] | None = None self._scan_timeout: int = kwargs.pop("scan_timeout", DEFAULT_SCAN_TIMEOUT) self._retry_count: int = kwargs.pop("retry_count", DEFAULT_RETRY_COUNT) self._connect_lock = asyncio.Lock() self._operation_lock = asyncio.Lock() if password is None or password == "": self._password_encoded = None else: self._password_encoded = "%08x" % ( binascii.crc32(password.encode("ascii")) & 0xFFFFFFFF ) self._client: BleakClientWithServiceCache | None = None self._read_char: BleakGATTCharacteristic | None = None self._write_char: BleakGATTCharacteristic | None = None self._disconnect_timer: asyncio.TimerHandle | None = None self._expected_disconnect = False self.loop = asyncio.get_event_loop() self._callbacks: list[Callable[[], None]] = [] self._notify_future: asyncio.Future[bytearray] | None = None self._last_full_update: float = -PASSIVE_POLL_INTERVAL def advertisement_changed(self, advertisement: SwitchBotAdvertisement) -> bool: """Check if the advertisement has changed.""" return bool( not self._sb_adv_data or ble_device_has_changed(self._sb_adv_data.device, advertisement.device) or advertisement.data != self._sb_adv_data.data ) def _commandkey(self, key: str) -> str: """Add password to key if set.""" if self._password_encoded is None: return key key_action = key[3] key_suffix = key[4:] return KEY_PASSWORD_PREFIX + key_action + self._password_encoded + key_suffix async def _send_command(self, key: str, retry: int | None = None) -> bytes | None: """Send command to device and read response.""" if retry is None: retry = self._retry_count command = bytearray.fromhex(self._commandkey(key)) _LOGGER.debug("%s: Scheduling command %s", self.name, command.hex()) max_attempts = retry + 1 if self._operation_lock.locked(): _LOGGER.debug( "%s: Operation already in progress, waiting for it to complete; RSSI: %s", self.name, self.rssi, ) async with self._operation_lock: for attempt in range(max_attempts): try: return await self._send_command_locked(key, command) except BleakNotFoundError: _LOGGER.error( "%s: device not found, no longer in range, or poor RSSI: %s", self.name, self.rssi, exc_info=True, ) raise except CharacteristicMissingError as ex: if attempt == retry: _LOGGER.error( "%s: characteristic missing: %s; Stopping trying; RSSI: %s", self.name, ex, self.rssi, exc_info=True, ) raise _LOGGER.debug( "%s: characteristic missing: %s; RSSI: %s", self.name, ex, self.rssi, exc_info=True, ) except BLEAK_RETRY_EXCEPTIONS: if attempt == retry: _LOGGER.error( "%s: communication failed; Stopping trying; RSSI: %s", self.name, self.rssi, exc_info=True, ) raise _LOGGER.debug( "%s: communication failed with:", self.name, exc_info=True ) raise RuntimeError("Unreachable") @property def name(self) -> str: """Return device name.""" return f"{self._device.name} ({self._device.address})" @property def data(self) -> dict[str, Any]: """Return device data.""" if self._sb_adv_data: return self._sb_adv_data.data return {} @property def parsed_data(self) -> dict[str, Any]: """Return parsed device data.""" return self.data.get("data") or {} @property def rssi(self) -> int: """Return RSSI of device.""" if self._sb_adv_data: return self._sb_adv_data.rssi return self._device.rssi async def _ensure_connected(self): """Ensure connection to device is established.""" if self._connect_lock.locked(): _LOGGER.debug( "%s: Connection already in progress, waiting for it to complete; RSSI: %s", self.name, self.rssi, ) if self._client and self._client.is_connected: self._reset_disconnect_timer() return async with self._connect_lock: # Check again while holding the lock if self._client and self._client.is_connected: self._reset_disconnect_timer() return _LOGGER.debug("%s: Connecting; RSSI: %s", self.name, self.rssi) client: BleakClientWithServiceCache = await establish_connection( BleakClientWithServiceCache, self._device, self.name, self._disconnected, use_services_cache=True, ble_device_callback=lambda: self._device, ) _LOGGER.debug("%s: Connected; RSSI: %s", self.name, self.rssi) self._client = client try: self._resolve_characteristics(client.services) except CharacteristicMissingError as ex: _LOGGER.debug( "%s: characteristic missing, clearing cache: %s; RSSI: %s", self.name, ex, self.rssi, exc_info=True, ) await client.clear_cache() self._cancel_disconnect_timer() await self._execute_disconnect_with_lock() raise self._reset_disconnect_timer() await self._start_notify() def _resolve_characteristics(self, services: BleakGATTServiceCollection) -> None: """Resolve characteristics.""" self._read_char = services.get_characteristic(READ_CHAR_UUID) if not self._read_char: raise CharacteristicMissingError(READ_CHAR_UUID) self._write_char = services.get_characteristic(WRITE_CHAR_UUID) if not self._write_char: raise CharacteristicMissingError(WRITE_CHAR_UUID) def _reset_disconnect_timer(self): """Reset disconnect timer.""" self._cancel_disconnect_timer() self._expected_disconnect = False self._disconnect_timer = self.loop.call_later( DISCONNECT_DELAY, self._disconnect_from_timer ) def _disconnected(self, client: BleakClientWithServiceCache) -> None: """Disconnected callback.""" if self._expected_disconnect: _LOGGER.debug( "%s: Disconnected from device; RSSI: %s", self.name, self.rssi ) return _LOGGER.warning( "%s: Device unexpectedly disconnected; RSSI: %s", self.name, self.rssi, ) def _disconnect_from_timer(self): """Disconnect from device.""" if self._operation_lock.locked() and self._client.is_connected: _LOGGER.debug( "%s: Operation in progress, resetting disconnect timer; RSSI: %s", self.name, self.rssi, ) self._reset_disconnect_timer() return self._cancel_disconnect_timer() asyncio.create_task(self._execute_timed_disconnect()) def _cancel_disconnect_timer(self): """Cancel disconnect timer.""" if self._disconnect_timer: self._disconnect_timer.cancel() self._disconnect_timer = None async def _execute_forced_disconnect(self) -> None: """Execute forced disconnection.""" self._cancel_disconnect_timer() _LOGGER.debug( "%s: Executing forced disconnect", self.name, ) await self._execute_disconnect() async def _execute_timed_disconnect(self) -> None: """Execute timed disconnection.""" _LOGGER.debug( "%s: Executing timed disconnect after timeout of %s", self.name, DISCONNECT_DELAY, ) await self._execute_disconnect() async def _execute_disconnect(self) -> None: """Execute disconnection.""" _LOGGER.debug("%s: Executing disconnect", self.name) async with self._connect_lock: await self._execute_disconnect_with_lock() async def _execute_disconnect_with_lock(self) -> None: """Execute disconnection while holding the lock.""" assert self._connect_lock.locked(), "Lock not held" _LOGGER.debug("%s: Executing disconnect with lock", self.name) if self._disconnect_timer: # If the timer was reset, don't disconnect _LOGGER.debug("%s: Skipping disconnect as timer reset", self.name) return client = self._client self._expected_disconnect = True self._client = None self._read_char = None self._write_char = None if client and client.is_connected: _LOGGER.debug("%s: Disconnecting", self.name) await client.disconnect() _LOGGER.debug("%s: Disconnect completed", self.name) else: _LOGGER.debug("%s: Already disconnected", self.name) async def _send_command_locked(self, key: str, command: bytes) -> bytes: """Send command to device and read response.""" await self._ensure_connected() try: return await self._execute_command_locked(key, command) except BleakDBusError as ex: # Disconnect so we can reset state and try again await asyncio.sleep(0.25) _LOGGER.debug( "%s: RSSI: %s; Backing off %ss; Disconnecting due to error: %s", self.name, self.rssi, 0.25, ex, ) await self._execute_forced_disconnect() raise except BleakError as ex: # Disconnect so we can reset state and try again _LOGGER.debug( "%s: RSSI: %s; Disconnecting due to error: %s", self.name, self.rssi, ex ) await self._execute_forced_disconnect() raise def _notification_handler(self, _sender: int, data: bytearray) -> None: """Handle notification responses.""" if self._notify_future and not self._notify_future.done(): self._notify_future.set_result(data) return _LOGGER.debug("%s: Received unsolicited notification: %s", self.name, data) async def _start_notify(self) -> None: """Start notification.""" _LOGGER.debug("%s: Subscribe to notifications; RSSI: %s", self.name, self.rssi) await self._client.start_notify(self._read_char, self._notification_handler) async def _execute_command_locked(self, key: str, command: bytes) -> bytes: """Execute command and read response.""" assert self._client is not None assert self._read_char is not None assert self._write_char is not None self._notify_future = asyncio.Future() client = self._client _LOGGER.debug("%s: Sending command: %s", self.name, key) await client.write_gatt_char(self._write_char, command, False) async with async_timeout.timeout(5): notify_msg = await self._notify_future _LOGGER.debug("%s: Notification received: %s", self.name, notify_msg.hex()) self._notify_future = None if notify_msg == b"\x07": _LOGGER.error("Password required") elif notify_msg == b"\t": _LOGGER.error("Password incorrect") return notify_msg def get_address(self) -> str: """Return address of device.""" return self._device.address def _override_state(self, state: dict[str, Any]) -> None: """Override device state.""" if self._override_adv_data is None: self._override_adv_data = {} self._override_adv_data.update(state) self._update_parsed_data(state) def _get_adv_value(self, key: str) -> Any: """Return value from advertisement data.""" if self._override_adv_data and key in self._override_adv_data: _LOGGER.debug( "%s: Using override value for %s: %s", self.name, key, self._override_adv_data[key], ) return self._override_adv_data[key] if not self._sb_adv_data: return None return self._sb_adv_data.data["data"].get(key) def get_battery_percent(self) -> Any: """Return device battery level in percent.""" return self._get_adv_value("battery") def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None: """Update device data from advertisement.""" # Only accept advertisements if the data is not missing # if we already have an advertisement with data self._device = advertisement.device async def get_device_data( self, retry: int | None = None, interface: int | None = None ) -> SwitchBotAdvertisement | None: """Find switchbot devices and their advertisement data.""" if retry is None: retry = self._retry_count if interface: _interface: int = interface else: _interface = int(self._interface.replace("hci", "")) _data = await GetSwitchbotDevices(interface=_interface).discover( retry=retry, scan_timeout=self._scan_timeout ) if self._device.address in _data: self._sb_adv_data = _data[self._device.address] return self._sb_adv_data async def _get_basic_info(self) -> bytes | None: """Return basic info of device.""" _data = await self._send_command( key=DEVICE_GET_BASIC_SETTINGS_KEY, retry=self._retry_count ) if _data in (b"\x07", b"\x00"): _LOGGER.error("Unsuccessful, please try again") return None return _data def _fire_callbacks(self) -> None: """Fire callbacks.""" _LOGGER.debug("%s: Fire callbacks", self.name) for callback in self._callbacks: callback() def subscribe(self, callback: Callable[[], None]) -> Callable[[], None]: """Subscribe to device notifications.""" self._callbacks.append(callback) def _unsub() -> None: """Unsubscribe from device notifications.""" self._callbacks.remove(callback) return _unsub async def update(self, interface: int | None = None) -> None: """Update position, battery percent and light level of device.""" if info := await self.get_basic_info(): self._last_full_update = time.monotonic() self._update_parsed_data(info) self._fire_callbacks() async def get_basic_info(self) -> dict[str, Any] | None: """Get device basic settings.""" if not (_data := await self._get_basic_info()): return None return { "battery": _data[1], "firmware": _data[2] / 10.0, } def _check_command_result( self, result: bytes | None, index: int, values: set[int] ) -> bool: """Check command result.""" if not result or len(result) - 1 < index: result_hex = result.hex() if result else "None" raise SwitchbotOperationError( f"{self.name}: Sending command failed (result={result_hex} index={index} expected={values} rssi={self.rssi})" ) return result[index] in values def _update_parsed_data(self, new_data: dict[str, Any]) -> bool: """Update data. Returns true if data has changed and False if not. """ if not self._sb_adv_data: _LOGGER.exception("No advertisement data to update") return old_data = self._sb_adv_data.data.get("data") or {} merged_data = _merge_data(old_data, new_data) if merged_data == old_data: return False self._set_parsed_data(self._sb_adv_data, merged_data) return True def _set_parsed_data( self, advertisement: SwitchBotAdvertisement, data: dict[str, Any] ) -> None: """Set data.""" self._sb_adv_data = replace( advertisement, data=self._sb_adv_data.data | {"data": data} ) def _set_advertisement_data(self, advertisement: SwitchBotAdvertisement) -> None: """Set advertisement data.""" new_data = advertisement.data.get("data") or {} if advertisement.active: # If we are getting active data, we can assume we are # getting active scans and we do not need to poll self._last_full_update = time.monotonic() if not self._sb_adv_data: self._sb_adv_data = advertisement elif new_data: self._update_parsed_data(new_data) self._override_adv_data = None def switch_mode(self) -> bool | None: """Return true or false from cache.""" # To get actual position call update() first. return self._get_adv_value("switchMode") def poll_needed(self, seconds_since_last_poll: float | None) -> bool: """Return if device needs polling.""" if ( seconds_since_last_poll is not None and seconds_since_last_poll < PASSIVE_POLL_INTERVAL ): return False time_since_last_full_update = time.monotonic() - self._last_full_update if time_since_last_full_update < PASSIVE_POLL_INTERVAL: return False return True class SwitchbotDevice(SwitchbotBaseDevice): """Base Representation of a Switchbot Device. This base class consumes the advertisement data during connection. If the device sends stale advertisement data while connected, use SwitchbotDeviceOverrideStateDuringConnection instead. """ def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None: """Update device data from advertisement.""" super().update_from_advertisement(advertisement) self._set_advertisement_data(advertisement) class SwitchbotDeviceOverrideStateDuringConnection(SwitchbotBaseDevice): """Base Representation of a Switchbot Device. This base class ignores the advertisement data during connection and uses the data from the device instead. """ def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None: super().update_from_advertisement(advertisement) if self._client and self._client.is_connected: # We do not consume the advertisement data if we are connected # to the device. This is because the advertisement data is not # updated when the device is connected for some devices. _LOGGER.debug("%s: Ignore advertisement data during connection", self.name) return self._set_advertisement_data(advertisement) class SwitchbotSequenceDevice(SwitchbotDevice): """A Switchbot sequence device. This class must not use SwitchbotDeviceOverrideStateDuringConnection because it needs to know when the sequence_number has changed. """ def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None: """Update device data from advertisement.""" current_state = self._get_adv_value("sequence_number") super().update_from_advertisement(advertisement) new_state = self._get_adv_value("sequence_number") _LOGGER.debug( "%s: update advertisement: %s (seq before: %s) (seq after: %s)", self.name, advertisement, current_state, new_state, ) if current_state != new_state: asyncio.ensure_future(self.update())