|
| 1 | +"""Library to handle connection with Switchbot Lock.""" |
1 | 2 | from __future__ import annotations |
| 3 | + |
| 4 | +import asyncio |
| 5 | +import logging |
| 6 | +from typing import Any |
| 7 | + |
| 8 | +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes |
| 9 | + |
| 10 | +from ..const import LockStatus |
| 11 | +from ..models import SwitchBotAdvertisement |
| 12 | +from .device import SwitchbotDevice |
| 13 | + |
| 14 | +COMMAND_HEADER = "57" |
| 15 | +COMMAND_GET_CK_IV = f"{COMMAND_HEADER}0f2103" |
| 16 | +COMMAND_LOCK_INFO = f"{COMMAND_HEADER}0f4f8101" |
| 17 | +COMMAND_UNLOCK = f"{COMMAND_HEADER}0f4e01011080" |
| 18 | +COMMAND_LOCK = f"{COMMAND_HEADER}0f4e01011000" |
| 19 | +COMMAND_ENABLE_NOTIFICATIONS = f"{COMMAND_HEADER}0e01001e00008101" |
| 20 | +COMMAND_DISABLE_NOTIFICATIONS = f"{COMMAND_HEADER}0e00" |
| 21 | + |
| 22 | +MOVING_STATUSES = {LockStatus.LOCKING, LockStatus.UNLOCKING} |
| 23 | +BLOCKED_STATUSES = {LockStatus.LOCKING_STOP, LockStatus.UNLOCKING_STOP} |
| 24 | +REST_STATUSES = {LockStatus.LOCKED, LockStatus.UNLOCKED, LockStatus.NOT_FULLY_LOCKED} |
| 25 | + |
| 26 | +_LOGGER = logging.getLogger(__name__) |
| 27 | + |
| 28 | + |
| 29 | +class SwitchbotLock(SwitchbotDevice): |
| 30 | + """Representation of a Switchbot Lock.""" |
| 31 | + |
| 32 | + def __init__( |
| 33 | + self, |
| 34 | + advertisement: SwitchBotAdvertisement, |
| 35 | + key_id: str, |
| 36 | + encryption_key: str, |
| 37 | + interface: int = 0, |
| 38 | + **kwargs: Any, |
| 39 | + ) -> None: |
| 40 | + if len(key_id) == 0: |
| 41 | + raise ValueError("key_id is missing") |
| 42 | + elif len(key_id) != 2: |
| 43 | + raise ValueError("key_id is invalid") |
| 44 | + if len(encryption_key) == 0: |
| 45 | + raise ValueError("encryption_key is missing") |
| 46 | + elif len(encryption_key) != 32: |
| 47 | + raise ValueError("encryption_key is invalid") |
| 48 | + self._iv = None |
| 49 | + self._cipher = None |
| 50 | + self._key_id = key_id |
| 51 | + self._encryption_key = bytearray.fromhex(encryption_key) |
| 52 | + self._notifications_enabled: bool = False |
| 53 | + super().__init__(advertisement.device, None, interface, **kwargs) |
| 54 | + self.update_from_advertisement(advertisement) |
| 55 | + |
| 56 | + async def lock(self) -> bool: |
| 57 | + """Send lock command.""" |
| 58 | + return await self._lock_unlock( |
| 59 | + COMMAND_LOCK, {LockStatus.LOCKED, LockStatus.LOCKING} |
| 60 | + ) |
| 61 | + |
| 62 | + async def unlock(self) -> bool: |
| 63 | + """Send unlock command.""" |
| 64 | + return await self._lock_unlock( |
| 65 | + COMMAND_UNLOCK, {LockStatus.UNLOCKED, LockStatus.UNLOCKING} |
| 66 | + ) |
| 67 | + |
| 68 | + async def _lock_unlock( |
| 69 | + self, command: str, ignore_statuses: set[LockStatus] |
| 70 | + ) -> bool: |
| 71 | + status = self.get_lock_status() |
| 72 | + if status is None: |
| 73 | + await self.update() |
| 74 | + status = self.get_lock_status() |
| 75 | + if status in ignore_statuses: |
| 76 | + return True |
| 77 | + |
| 78 | + await self._enable_notifications() |
| 79 | + result = await self._send_command(command) |
| 80 | + if not self._check_command_result(result, 0, {1}): |
| 81 | + return False |
| 82 | + return True |
| 83 | + |
| 84 | + async def get_basic_info(self) -> dict[str, Any] | None: |
| 85 | + """Get device basic status.""" |
| 86 | + lock_raw_data = await self._get_lock_info() |
| 87 | + if not lock_raw_data: |
| 88 | + return None |
| 89 | + |
| 90 | + basic_data = await self._get_basic_info() |
| 91 | + if not basic_data: |
| 92 | + return None |
| 93 | + |
| 94 | + lock_data = self._parse_lock_data(lock_raw_data[1:]) |
| 95 | + lock_data.update(battery=basic_data[1], firmware=basic_data[2] / 10.0) |
| 96 | + |
| 97 | + return lock_data |
| 98 | + |
| 99 | + def is_calibrated(self) -> Any: |
| 100 | + """Return True if lock is calibrated.""" |
| 101 | + return self._get_adv_value("calibration") |
| 102 | + |
| 103 | + def get_lock_status(self) -> LockStatus: |
| 104 | + """Return lock status.""" |
| 105 | + return self._get_adv_value("status") |
| 106 | + |
| 107 | + def is_door_open(self) -> bool: |
| 108 | + """Return True if door is open.""" |
| 109 | + return self._get_adv_value("door_open") |
| 110 | + |
| 111 | + def is_unclosed_alarm_on(self) -> bool: |
| 112 | + """Return True if unclosed door alarm is on.""" |
| 113 | + return self._get_adv_value("unclosed_alarm") |
| 114 | + |
| 115 | + def is_unlocked_alarm_on(self) -> bool: |
| 116 | + """Return True if lock unlocked alarm is on.""" |
| 117 | + return self._get_adv_value("unlocked_alarm") |
| 118 | + |
| 119 | + def is_auto_lock_paused(self) -> bool: |
| 120 | + """Return True if auto lock is paused.""" |
| 121 | + return self._get_adv_value("auto_lock_paused") |
| 122 | + |
| 123 | + async def _get_lock_info(self) -> bytes | None: |
| 124 | + """Return lock info of device.""" |
| 125 | + _data = await self._send_command(key=COMMAND_LOCK_INFO, retry=self._retry_count) |
| 126 | + |
| 127 | + if not self._check_command_result(_data, 0, {1}): |
| 128 | + _LOGGER.error("Unsuccessful, please try again") |
| 129 | + return None |
| 130 | + |
| 131 | + return _data |
| 132 | + |
| 133 | + async def _enable_notifications(self) -> bool: |
| 134 | + if self._notifications_enabled: |
| 135 | + return True |
| 136 | + result = await self._send_command(COMMAND_ENABLE_NOTIFICATIONS) |
| 137 | + if self._check_command_result(result, 0, {1}): |
| 138 | + self._notifications_enabled = True |
| 139 | + return self._notifications_enabled |
| 140 | + |
| 141 | + async def _disable_notifications(self) -> bool: |
| 142 | + if not self._notifications_enabled: |
| 143 | + return True |
| 144 | + result = await self._send_command(COMMAND_DISABLE_NOTIFICATIONS) |
| 145 | + if self._check_command_result(result, 0, {1}): |
| 146 | + self._notifications_enabled = False |
| 147 | + return not self._notifications_enabled |
| 148 | + |
| 149 | + def _notification_handler(self, _sender: int, data: bytearray) -> None: |
| 150 | + if self._notifications_enabled and self._check_command_result(data, 0, {0xF}): |
| 151 | + self._update_lock_status(data) |
| 152 | + else: |
| 153 | + super()._notification_handler(_sender, data) |
| 154 | + |
| 155 | + def _update_lock_status(self, data: bytearray) -> None: |
| 156 | + data = self._decrypt(data[4:]) |
| 157 | + lock_data = self._parse_lock_data(data) |
| 158 | + current_status = self.get_lock_status() |
| 159 | + if ( |
| 160 | + lock_data["status"] != current_status or current_status not in REST_STATUSES |
| 161 | + ) and ( |
| 162 | + lock_data["status"] in REST_STATUSES |
| 163 | + or lock_data["status"] in BLOCKED_STATUSES |
| 164 | + ): |
| 165 | + asyncio.create_task(self._disable_notifications()) |
| 166 | + |
| 167 | + self._update_parsed_data(lock_data) |
| 168 | + |
| 169 | + @staticmethod |
| 170 | + def _parse_lock_data(data: bytes) -> dict[str, Any]: |
| 171 | + return { |
| 172 | + "calibration": bool(data[0] & 0b10000000), |
| 173 | + "status": LockStatus((data[0] & 0b01110000) >> 4), |
| 174 | + "door_open": bool(data[0] & 0b00000100), |
| 175 | + "unclosed_alarm": bool(data[1] & 0b00100000), |
| 176 | + "unlocked_alarm": bool(data[1] & 0b00010000), |
| 177 | + } |
| 178 | + |
| 179 | + async def _send_command( |
| 180 | + self, key: str, retry: int | None = None, encrypt: bool = True |
| 181 | + ) -> bytes | None: |
| 182 | + if not encrypt: |
| 183 | + return await super()._send_command(key[:2] + "000000" + key[2:], retry) |
| 184 | + |
| 185 | + result = await self._ensure_encryption_initialized() |
| 186 | + if not result: |
| 187 | + _LOGGER.error("Failed to initialize encryption") |
| 188 | + return None |
| 189 | + |
| 190 | + encrypted = ( |
| 191 | + key[:2] + self._key_id + self._iv[0:2].hex() + self._encrypt(key[2:]) |
| 192 | + ) |
| 193 | + result = await super()._send_command(encrypted, retry) |
| 194 | + return result[:1] + self._decrypt(result[4:]) |
| 195 | + |
| 196 | + async def _ensure_encryption_initialized(self) -> bool: |
| 197 | + if self._iv is not None: |
| 198 | + return True |
| 199 | + |
| 200 | + result = await self._send_command( |
| 201 | + COMMAND_GET_CK_IV + self._key_id, encrypt=False |
| 202 | + ) |
| 203 | + ok = self._check_command_result(result, 0, {0x01}) |
| 204 | + if ok: |
| 205 | + self._iv = result[4:] |
| 206 | + |
| 207 | + return ok |
| 208 | + |
| 209 | + async def _execute_disconnect(self) -> None: |
| 210 | + await super()._execute_disconnect() |
| 211 | + self._iv = None |
| 212 | + self._cipher = None |
| 213 | + self._notifications_enabled = False |
| 214 | + |
| 215 | + def _get_cipher(self) -> Cipher: |
| 216 | + if self._cipher is None: |
| 217 | + self._cipher = Cipher( |
| 218 | + algorithms.AES128(self._encryption_key), modes.CTR(self._iv) |
| 219 | + ) |
| 220 | + return self._cipher |
| 221 | + |
| 222 | + def _encrypt(self, data: str) -> str: |
| 223 | + if len(data) == 0: |
| 224 | + return "" |
| 225 | + encryptor = self._get_cipher().encryptor() |
| 226 | + return (encryptor.update(bytearray.fromhex(data)) + encryptor.finalize()).hex() |
| 227 | + |
| 228 | + def _decrypt(self, data: bytearray) -> bytes: |
| 229 | + if len(data) == 0: |
| 230 | + return b"" |
| 231 | + decryptor = self._get_cipher().decryptor() |
| 232 | + return decryptor.update(data) + decryptor.finalize() |
0 commit comments