lock.py 7.9 KB

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