__init__.py 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789
  1. """Library to handle connection with Switchbot."""
  2. from __future__ import annotations
  3. import asyncio
  4. import binascii
  5. import logging
  6. from dataclasses import dataclass
  7. from typing import Any
  8. from uuid import UUID
  9. import bleak
  10. from bleak.backends.device import BLEDevice
  11. from bleak.backends.scanner import AdvertisementData
  12. from bleak_retry_connector import BleakClient, establish_connection
  13. DEFAULT_RETRY_COUNT = 3
  14. DEFAULT_RETRY_TIMEOUT = 1
  15. DEFAULT_SCAN_TIMEOUT = 5
  16. # Keys common to all device types
  17. DEVICE_GET_BASIC_SETTINGS_KEY = "5702"
  18. DEVICE_SET_MODE_KEY = "5703"
  19. DEVICE_SET_EXTENDED_KEY = "570f"
  20. # Plug Mini keys
  21. PLUG_ON_KEY = "570f50010180"
  22. PLUG_OFF_KEY = "570f50010100"
  23. # Bot keys
  24. PRESS_KEY = "570100"
  25. ON_KEY = "570101"
  26. OFF_KEY = "570102"
  27. DOWN_KEY = "570103"
  28. UP_KEY = "570104"
  29. # Curtain keys
  30. OPEN_KEY = "570f450105ff00" # 570F4501010100
  31. CLOSE_KEY = "570f450105ff64" # 570F4501010164
  32. POSITION_KEY = "570F450105ff" # +actual_position ex: 570F450105ff32 for 50%
  33. STOP_KEY = "570F450100ff"
  34. CURTAIN_EXT_SUM_KEY = "570f460401"
  35. CURTAIN_EXT_ADV_KEY = "570f460402"
  36. CURTAIN_EXT_CHAIN_INFO_KEY = "570f468101"
  37. # Base key when encryption is set
  38. KEY_PASSWORD_PREFIX = "571"
  39. _LOGGER = logging.getLogger(__name__)
  40. CONNECT_LOCK = asyncio.Lock()
  41. def _sb_uuid(comms_type: str = "service") -> UUID | str:
  42. """Return Switchbot UUID."""
  43. _uuid = {"tx": "002", "rx": "003", "service": "d00"}
  44. if comms_type in _uuid:
  45. return UUID(f"cba20{_uuid[comms_type]}-224d-11e6-9fb8-0002a5d5c51b")
  46. return "Incorrect type, choose between: tx, rx or service"
  47. def _process_wohand(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
  48. """Process woHand/Bot services data."""
  49. _switch_mode = bool(data[1] & 0b10000000)
  50. _bot_data = {
  51. "switchMode": _switch_mode,
  52. "isOn": not bool(data[1] & 0b01000000) if _switch_mode else False,
  53. "battery": data[2] & 0b01111111,
  54. }
  55. return _bot_data
  56. def _process_wocurtain(
  57. data: bytes, mfr_data: bytes | None, reverse: bool = True
  58. ) -> dict[str, bool | int]:
  59. """Process woCurtain/Curtain services data."""
  60. _position = max(min(data[3] & 0b01111111, 100), 0)
  61. _curtain_data = {
  62. "calibration": bool(data[1] & 0b01000000),
  63. "battery": data[2] & 0b01111111,
  64. "inMotion": bool(data[3] & 0b10000000),
  65. "position": (100 - _position) if reverse else _position,
  66. "lightLevel": (data[4] >> 4) & 0b00001111,
  67. "deviceChain": data[4] & 0b00000111,
  68. }
  69. return _curtain_data
  70. def _process_wosensorth(data: bytes, mfr_data: bytes | None) -> dict[str, object]:
  71. """Process woSensorTH/Temp sensor services data."""
  72. _temp_sign = 1 if data[4] & 0b10000000 else -1
  73. _temp_c = _temp_sign * ((data[4] & 0b01111111) + ((data[3] & 0b00001111) / 10))
  74. _temp_f = (_temp_c * 9 / 5) + 32
  75. _temp_f = (_temp_f * 10) / 10
  76. _wosensorth_data = {
  77. "temp": {"c": _temp_c, "f": _temp_f},
  78. "fahrenheit": bool(data[5] & 0b10000000),
  79. "humidity": data[5] & 0b01111111,
  80. "battery": data[2] & 0b01111111,
  81. }
  82. return _wosensorth_data
  83. def _process_wocontact(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
  84. """Process woContact Sensor services data."""
  85. return {
  86. "tested": bool(data[1] & 0b10000000),
  87. "motion_detected": bool(data[1] & 0b01000000),
  88. "battery": data[2] & 0b01111111,
  89. "contact_open": data[3] & 0b00000010 == 0b00000010,
  90. "contact_timeout": data[3] & 0b00000110 == 0b00000110,
  91. "is_light": bool(data[3] & 0b00000001),
  92. "button_count": (data[7] & 0b11110000) >> 4,
  93. }
  94. def _process_wopresence(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
  95. """Process WoPresence Sensor services data."""
  96. return {
  97. "tested": bool(data[1] & 0b10000000),
  98. "motion_detected": bool(data[1] & 0b01000000),
  99. "battery": data[2] & 0b01111111,
  100. "led": (data[5] & 0b00100000) >> 5,
  101. "iot": (data[5] & 0b00010000) >> 4,
  102. "sense_distance": (data[5] & 0b00001100) >> 2,
  103. "light_intensity": data[5] & 0b00000011,
  104. "is_light": bool(data[5] & 0b00000010),
  105. }
  106. def _process_woplugmini(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
  107. """Process plug mini."""
  108. return {
  109. "switchMode": True,
  110. "isOn": mfr_data[7] == 0x80,
  111. "wifi_rssi": mfr_data[9],
  112. }
  113. @dataclass
  114. class SwitchBotAdvertisement:
  115. """Switchbot advertisement."""
  116. address: str
  117. data: dict[str, Any]
  118. device: BLEDevice
  119. def parse_advertisement_data(
  120. device: BLEDevice, advertisement_data: AdvertisementData
  121. ) -> SwitchBotAdvertisement | None:
  122. """Parse advertisement data."""
  123. _services = list(advertisement_data.service_data.values())
  124. _mgr_datas = list(advertisement_data.manufacturer_data.values())
  125. if not _services:
  126. return
  127. _service_data = _services[0]
  128. _mfr_data = _mgr_datas[0] if _mgr_datas else None
  129. _model = chr(_service_data[0] & 0b01111111)
  130. supported_types: dict[str, dict[str, Any]] = {
  131. "d": {"modelName": "WoContact", "func": _process_wocontact},
  132. "H": {"modelName": "WoHand", "func": _process_wohand},
  133. "s": {"modelName": "WoPresence", "func": _process_wopresence},
  134. "c": {"modelName": "WoCurtain", "func": _process_wocurtain},
  135. "T": {"modelName": "WoSensorTH", "func": _process_wosensorth},
  136. "i": {"modelName": "WoSensorTH", "func": _process_wosensorth},
  137. "g": {"modelName": "WoPlug", "func": _process_woplugmini},
  138. }
  139. data = {
  140. "address": device.address, # MacOS uses UUIDs
  141. "rawAdvData": list(advertisement_data.service_data.values())[0],
  142. "data": {
  143. "rssi": device.rssi,
  144. },
  145. }
  146. if _model in supported_types:
  147. data.update(
  148. {
  149. "isEncrypted": bool(_service_data[0] & 0b10000000),
  150. "model": _model,
  151. "modelName": supported_types[_model]["modelName"],
  152. "data": supported_types[_model]["func"](_service_data, _mfr_data),
  153. }
  154. )
  155. data["data"]["rssi"] = device.rssi
  156. return SwitchBotAdvertisement(device.address, data, device)
  157. class GetSwitchbotDevices:
  158. """Scan for all Switchbot devices and return by type."""
  159. def __init__(self, interface: int = 0) -> None:
  160. """Get switchbot devices class constructor."""
  161. self._interface = f"hci{interface}"
  162. self._adv_data: dict[str, SwitchBotAdvertisement] = {}
  163. def detection_callback(
  164. self,
  165. device: BLEDevice,
  166. advertisement_data: AdvertisementData,
  167. ) -> None:
  168. discovery = parse_advertisement_data(device, advertisement_data)
  169. if discovery:
  170. self._adv_data[discovery.address] = discovery
  171. async def discover(
  172. self, retry: int = DEFAULT_RETRY_COUNT, scan_timeout: int = DEFAULT_SCAN_TIMEOUT
  173. ) -> dict:
  174. """Find switchbot devices and their advertisement data."""
  175. devices = None
  176. devices = bleak.BleakScanner(
  177. # TODO: Find new UUIDs to filter on. For example, see
  178. # https://github.com/OpenWonderLabs/SwitchBotAPI-BLE/blob/4ad138bb09f0fbbfa41b152ca327a78c1d0b6ba9/devicetypes/meter.md
  179. adapter=self._interface,
  180. )
  181. devices.register_detection_callback(self.detection_callback)
  182. async with CONNECT_LOCK:
  183. await devices.start()
  184. await asyncio.sleep(scan_timeout)
  185. await devices.stop()
  186. if devices is None:
  187. if retry < 1:
  188. _LOGGER.error(
  189. "Scanning for Switchbot devices failed. Stop trying", exc_info=True
  190. )
  191. return self._adv_data
  192. _LOGGER.warning(
  193. "Error scanning for Switchbot devices. Retrying (remaining: %d)",
  194. retry,
  195. )
  196. await asyncio.sleep(DEFAULT_RETRY_TIMEOUT)
  197. return await self.discover(retry - 1, scan_timeout)
  198. return self._adv_data
  199. async def _get_devices_by_model(
  200. self,
  201. model: str,
  202. ) -> dict:
  203. """Get switchbot devices by type."""
  204. if not self._adv_data:
  205. await self.discover()
  206. return {
  207. address: adv
  208. for address, adv in self._adv_data.items()
  209. if adv.data.get("model") == model
  210. }
  211. async def get_curtains(self) -> dict[str, SwitchBotAdvertisement]:
  212. """Return all WoCurtain/Curtains devices with services data."""
  213. return await self._get_devices_by_model("c")
  214. async def get_bots(self) -> dict[str, SwitchBotAdvertisement]:
  215. """Return all WoHand/Bot devices with services data."""
  216. return await self._get_devices_by_model("H")
  217. async def get_tempsensors(self) -> dict[str, SwitchBotAdvertisement]:
  218. """Return all WoSensorTH/Temp sensor devices with services data."""
  219. base_meters = await self._get_devices_by_model("T")
  220. plus_meters = await self._get_devices_by_model("i")
  221. return {**base_meters, **plus_meters}
  222. async def get_contactsensors(self) -> dict[str, SwitchBotAdvertisement]:
  223. """Return all WoContact/Contact sensor devices with services data."""
  224. return await self._get_devices_by_model("d")
  225. async def get_device_data(
  226. self, address: str
  227. ) -> dict[str, SwitchBotAdvertisement] | None:
  228. """Return data for specific device."""
  229. if not self._adv_data:
  230. await self.discover()
  231. _switchbot_data = {
  232. device: data
  233. for device, data in self._adv_data.items()
  234. # MacOS uses UUIDs instead of MAC addresses
  235. if data.get("address") == address
  236. }
  237. return _switchbot_data
  238. class SwitchbotDevice:
  239. """Base Representation of a Switchbot Device."""
  240. def __init__(
  241. self,
  242. device: BLEDevice,
  243. password: str | None = None,
  244. interface: int = 0,
  245. **kwargs: Any,
  246. ) -> None:
  247. """Switchbot base class constructor."""
  248. self._interface = f"hci{interface}"
  249. self._device = device
  250. self._sb_adv_data: SwitchBotAdvertisement | None = None
  251. self._scan_timeout: int = kwargs.pop("scan_timeout", DEFAULT_SCAN_TIMEOUT)
  252. self._retry_count: int = kwargs.pop("retry_count", DEFAULT_RETRY_COUNT)
  253. if password is None or password == "":
  254. self._password_encoded = None
  255. else:
  256. self._password_encoded = "%x" % (
  257. binascii.crc32(password.encode("ascii")) & 0xFFFFFFFF
  258. )
  259. def _commandkey(self, key: str) -> str:
  260. """Add password to key if set."""
  261. if self._password_encoded is None:
  262. return key
  263. key_action = key[3]
  264. key_suffix = key[4:]
  265. return KEY_PASSWORD_PREFIX + key_action + self._password_encoded + key_suffix
  266. async def _sendcommand(self, key: str, retry: int) -> bytes:
  267. """Send command to device and read response."""
  268. command = bytearray.fromhex(self._commandkey(key))
  269. _LOGGER.debug("Sending command to switchbot %s", command)
  270. max_attempts = retry + 1
  271. async with CONNECT_LOCK:
  272. for attempt in range(max_attempts):
  273. try:
  274. return await self._send_command_locked(key, command)
  275. except (bleak.BleakError, asyncio.exceptions.TimeoutError):
  276. if attempt == retry:
  277. _LOGGER.error(
  278. "Switchbot communication failed. Stopping trying",
  279. exc_info=True,
  280. )
  281. return b"\x00"
  282. _LOGGER.debug("Switchbot communication failed with:", exc_info=True)
  283. raise RuntimeError("Unreachable")
  284. @property
  285. def name(self) -> str:
  286. """Return device name."""
  287. return f"{self._device.name} ({self._device.address})"
  288. async def _send_command_locked(self, key: str, command: bytes) -> bytes:
  289. """Send command to device and read response."""
  290. client: BleakClient | None = None
  291. try:
  292. _LOGGER.debug("%s: Connnecting to switchbot", self.name)
  293. client = await establish_connection(
  294. BleakClient, self._device, self.name, max_attempts=1
  295. )
  296. _LOGGER.debug(
  297. "%s: Connnected to switchbot: %s", self.name, client.is_connected
  298. )
  299. future: asyncio.Future[bytearray] = asyncio.Future()
  300. def _notification_handler(sender: int, data: bytearray) -> None:
  301. """Handle notification responses."""
  302. if future.done():
  303. _LOGGER.debug("%s: Notification handler already done", self.name)
  304. return
  305. future.set_result(data)
  306. _LOGGER.debug("%s: Subscribe to notifications", self.name)
  307. await client.start_notify(_sb_uuid(comms_type="rx"), _notification_handler)
  308. _LOGGER.debug("%s: Sending command, %s", self.name, key)
  309. await client.write_gatt_char(_sb_uuid(comms_type="tx"), command, False)
  310. notify_msg = await asyncio.wait_for(future, timeout=5)
  311. _LOGGER.info("%s: Notification received: %s", self.name, notify_msg)
  312. _LOGGER.debug("%s: UnSubscribe to notifications", self.name)
  313. await client.stop_notify(_sb_uuid(comms_type="rx"))
  314. finally:
  315. if client:
  316. await client.disconnect()
  317. if notify_msg == b"\x07":
  318. _LOGGER.error("Password required")
  319. elif notify_msg == b"\t":
  320. _LOGGER.error("Password incorrect")
  321. return notify_msg
  322. def get_address(self) -> str:
  323. """Return address of device."""
  324. return self._device.address
  325. def _get_adv_value(self, key: str) -> Any:
  326. """Return value from advertisement data."""
  327. if not self._sb_adv_data:
  328. return None
  329. return self._sb_adv_data.data["data"][key]
  330. def get_battery_percent(self) -> Any:
  331. """Return device battery level in percent."""
  332. return self._get_adv_value("battery")
  333. def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
  334. """Update device data from advertisement."""
  335. self._sb_adv_data = advertisement
  336. self._device = advertisement.device
  337. async def get_device_data(
  338. self, retry: int = DEFAULT_RETRY_COUNT, interface: int | None = None
  339. ) -> dict | None:
  340. """Find switchbot devices and their advertisement data."""
  341. if interface:
  342. _interface: int = interface
  343. else:
  344. _interface = int(self._interface.replace("hci", ""))
  345. _data = await GetSwitchbotDevices(interface=_interface).discover(
  346. retry=retry, scan_timeout=self._scan_timeout
  347. )
  348. if self._device.address in _data:
  349. self._sb_adv_data = _data[self._device.address]
  350. return self._sb_adv_data
  351. class Switchbot(SwitchbotDevice):
  352. """Representation of a Switchbot."""
  353. def __init__(self, *args: Any, **kwargs: Any) -> None:
  354. """Switchbot Bot/WoHand constructor."""
  355. super().__init__(*args, **kwargs)
  356. self._inverse: bool = kwargs.pop("inverse_mode", False)
  357. self._settings: dict[str, Any] = {}
  358. async def update(self, interface: int | None = None) -> None:
  359. """Update mode, battery percent and state of device."""
  360. await self.get_device_data(retry=self._retry_count, interface=interface)
  361. async def turn_on(self) -> bool:
  362. """Turn device on."""
  363. result = await self._sendcommand(ON_KEY, self._retry_count)
  364. if result[0] == 1:
  365. return True
  366. if result[0] == 5:
  367. _LOGGER.debug("Bot is in press mode and doesn't have on state")
  368. return True
  369. return False
  370. async def turn_off(self) -> bool:
  371. """Turn device off."""
  372. result = await self._sendcommand(OFF_KEY, self._retry_count)
  373. if result[0] == 1:
  374. return True
  375. if result[0] == 5:
  376. _LOGGER.debug("Bot is in press mode and doesn't have off state")
  377. return True
  378. return False
  379. async def hand_up(self) -> bool:
  380. """Raise device arm."""
  381. result = await self._sendcommand(UP_KEY, self._retry_count)
  382. if result[0] == 1:
  383. return True
  384. if result[0] == 5:
  385. _LOGGER.debug("Bot is in press mode")
  386. return True
  387. return False
  388. async def hand_down(self) -> bool:
  389. """Lower device arm."""
  390. result = await self._sendcommand(DOWN_KEY, self._retry_count)
  391. if result[0] == 1:
  392. return True
  393. if result[0] == 5:
  394. _LOGGER.debug("Bot is in press mode")
  395. return True
  396. return False
  397. async def press(self) -> bool:
  398. """Press command to device."""
  399. result = await self._sendcommand(PRESS_KEY, self._retry_count)
  400. if result[0] == 1:
  401. return True
  402. if result[0] == 5:
  403. _LOGGER.debug("Bot is in switch mode")
  404. return True
  405. return False
  406. async def set_switch_mode(
  407. self, switch_mode: bool = False, strength: int = 100, inverse: bool = False
  408. ) -> bool:
  409. """Change bot mode."""
  410. mode_key = format(switch_mode, "b") + format(inverse, "b")
  411. strength_key = f"{strength:0{2}x}" # to hex with padding to double digit
  412. result = await self._sendcommand(
  413. DEVICE_SET_MODE_KEY + strength_key + mode_key, self._retry_count
  414. )
  415. if result[0] == 1:
  416. return True
  417. return False
  418. async def set_long_press(self, duration: int = 0) -> bool:
  419. """Set bot long press duration."""
  420. duration_key = f"{duration:0{2}x}" # to hex with padding to double digit
  421. result = await self._sendcommand(
  422. DEVICE_SET_EXTENDED_KEY + "08" + duration_key, self._retry_count
  423. )
  424. if result[0] == 1:
  425. return True
  426. return False
  427. async def get_basic_info(self) -> dict[str, Any] | None:
  428. """Get device basic settings."""
  429. _data = await self._sendcommand(
  430. key=DEVICE_GET_BASIC_SETTINGS_KEY, retry=self._retry_count
  431. )
  432. if _data in (b"\x07", b"\x00"):
  433. _LOGGER.error("Unsuccessfull, please try again")
  434. return None
  435. self._settings = {
  436. "battery": _data[1],
  437. "firmware": _data[2] / 10.0,
  438. "strength": _data[3],
  439. "timers": _data[8],
  440. "switchMode": bool(_data[9] & 16),
  441. "inverseDirection": bool(_data[9] & 1),
  442. "holdSeconds": _data[10],
  443. }
  444. return self._settings
  445. def switch_mode(self) -> Any:
  446. """Return true or false from cache."""
  447. # To get actual position call update() first.
  448. return self._get_adv_value("switchMode")
  449. def is_on(self) -> Any:
  450. """Return switch state from cache."""
  451. # To get actual position call update() first.
  452. value = self._get_adv_value("isOn")
  453. if value is None:
  454. return None
  455. if self._inverse:
  456. return not value
  457. return value
  458. class SwitchbotCurtain(SwitchbotDevice):
  459. """Representation of a Switchbot Curtain."""
  460. def __init__(self, *args: Any, **kwargs: Any) -> None:
  461. """Switchbot Curtain/WoCurtain constructor."""
  462. # The position of the curtain is saved returned with 0 = open and 100 = closed.
  463. # This is independent of the calibration of the curtain bot (Open left to right/
  464. # Open right to left/Open from the middle).
  465. # The parameter 'reverse_mode' reverse these values,
  466. # if 'reverse_mode' = True, position = 0 equals close
  467. # and position = 100 equals open. The parameter is default set to True so that
  468. # the definition of position is the same as in Home Assistant.
  469. super().__init__(*args, **kwargs)
  470. self._reverse: bool = kwargs.pop("reverse_mode", True)
  471. self._settings: dict[str, Any] = {}
  472. self.ext_info_sum: dict[str, Any] = {}
  473. self.ext_info_adv: dict[str, Any] = {}
  474. async def open(self) -> bool:
  475. """Send open command."""
  476. result = await self._sendcommand(OPEN_KEY, self._retry_count)
  477. if result[0] == 1:
  478. return True
  479. return False
  480. async def close(self) -> bool:
  481. """Send close command."""
  482. result = await self._sendcommand(CLOSE_KEY, self._retry_count)
  483. if result[0] == 1:
  484. return True
  485. return False
  486. async def stop(self) -> bool:
  487. """Send stop command to device."""
  488. result = await self._sendcommand(STOP_KEY, self._retry_count)
  489. if result[0] == 1:
  490. return True
  491. return False
  492. async def set_position(self, position: int) -> bool:
  493. """Send position command (0-100) to device."""
  494. position = (100 - position) if self._reverse else position
  495. hex_position = "%0.2X" % position
  496. result = await self._sendcommand(POSITION_KEY + hex_position, self._retry_count)
  497. if result[0] == 1:
  498. return True
  499. return False
  500. async def update(self, interface: int | None = None) -> None:
  501. """Update position, battery percent and light level of device."""
  502. await self.get_device_data(retry=self._retry_count, interface=interface)
  503. def get_position(self) -> Any:
  504. """Return cached position (0-100) of Curtain."""
  505. # To get actual position call update() first.
  506. return self._get_adv_value("position")
  507. async def get_basic_info(self) -> dict[str, Any] | None:
  508. """Get device basic settings."""
  509. _data = await self._sendcommand(
  510. key=DEVICE_GET_BASIC_SETTINGS_KEY, retry=self._retry_count
  511. )
  512. if _data in (b"\x07", b"\x00"):
  513. _LOGGER.error("Unsuccessfull, please try again")
  514. return None
  515. _position = max(min(_data[6], 100), 0)
  516. self._settings = {
  517. "battery": _data[1],
  518. "firmware": _data[2] / 10.0,
  519. "chainLength": _data[3],
  520. "openDirection": (
  521. "right_to_left" if _data[4] & 0b10000000 == 128 else "left_to_right"
  522. ),
  523. "touchToOpen": bool(_data[4] & 0b01000000),
  524. "light": bool(_data[4] & 0b00100000),
  525. "fault": bool(_data[4] & 0b00001000),
  526. "solarPanel": bool(_data[5] & 0b00001000),
  527. "calibrated": bool(_data[5] & 0b00000100),
  528. "inMotion": bool(_data[5] & 0b01000011),
  529. "position": (100 - _position) if self._reverse else _position,
  530. "timers": _data[7],
  531. }
  532. return self._settings
  533. async def get_extended_info_summary(self) -> dict[str, Any] | None:
  534. """Get basic info for all devices in chain."""
  535. _data = await self._sendcommand(
  536. key=CURTAIN_EXT_SUM_KEY, retry=self._retry_count
  537. )
  538. if _data in (b"\x07", b"\x00"):
  539. _LOGGER.error("Unsuccessfull, please try again")
  540. return None
  541. self.ext_info_sum["device0"] = {
  542. "openDirectionDefault": not bool(_data[1] & 0b10000000),
  543. "touchToOpen": bool(_data[1] & 0b01000000),
  544. "light": bool(_data[1] & 0b00100000),
  545. "openDirection": (
  546. "left_to_right" if _data[1] & 0b00010000 == 1 else "right_to_left"
  547. ),
  548. }
  549. # if grouped curtain device present.
  550. if _data[2] != 0:
  551. self.ext_info_sum["device1"] = {
  552. "openDirectionDefault": not bool(_data[1] & 0b10000000),
  553. "touchToOpen": bool(_data[1] & 0b01000000),
  554. "light": bool(_data[1] & 0b00100000),
  555. "openDirection": (
  556. "left_to_right" if _data[1] & 0b00010000 else "right_to_left"
  557. ),
  558. }
  559. return self.ext_info_sum
  560. async def get_extended_info_adv(self) -> dict[str, Any] | None:
  561. """Get advance page info for device chain."""
  562. _data = await self._sendcommand(
  563. key=CURTAIN_EXT_ADV_KEY, retry=self._retry_count
  564. )
  565. if _data in (b"\x07", b"\x00"):
  566. _LOGGER.error("Unsuccessfull, please try again")
  567. return None
  568. _state_of_charge = [
  569. "not_charging",
  570. "charging_by_adapter",
  571. "charging_by_solar",
  572. "fully_charged",
  573. "solar_not_charging",
  574. "charging_error",
  575. ]
  576. self.ext_info_adv["device0"] = {
  577. "battery": _data[1],
  578. "firmware": _data[2] / 10.0,
  579. "stateOfCharge": _state_of_charge[_data[3]],
  580. }
  581. # If grouped curtain device present.
  582. if _data[4]:
  583. self.ext_info_adv["device1"] = {
  584. "battery": _data[4],
  585. "firmware": _data[5] / 10.0,
  586. "stateOfCharge": _state_of_charge[_data[6]],
  587. }
  588. return self.ext_info_adv
  589. def get_light_level(self) -> Any:
  590. """Return cached light level."""
  591. # To get actual light level call update() first.
  592. return self._get_adv_value("lightLevel")
  593. def is_reversed(self) -> bool:
  594. """Return True if curtain position is opposite from SB data."""
  595. return self._reverse
  596. def is_calibrated(self) -> Any:
  597. """Return True curtain is calibrated."""
  598. # To get actual light level call update() first.
  599. return self._get_adv_value("calibration")
  600. class SwitchbotPlugMini(SwitchbotDevice):
  601. """Representation of a Switchbot plug mini."""
  602. def __init__(self, *args: Any, **kwargs: Any) -> None:
  603. """Switchbot plug mini constructor."""
  604. super().__init__(*args, **kwargs)
  605. self._settings: dict[str, Any] = {}
  606. async def update(self, interface: int | None = None) -> None:
  607. """Update state of device."""
  608. await self.get_device_data(retry=self._retry_count, interface=interface)
  609. async def turn_on(self) -> bool:
  610. """Turn device on."""
  611. result = await self._sendcommand(PLUG_ON_KEY, self._retry_count)
  612. return result[1] == 0x80
  613. async def turn_off(self) -> bool:
  614. """Turn device off."""
  615. result = await self._sendcommand(PLUG_OFF_KEY, self._retry_count)
  616. return result[1] == 0x00
  617. def is_on(self) -> Any:
  618. """Return switch state from cache."""
  619. # To get actual position call update() first.
  620. value = self._get_adv_value("isOn")
  621. if value is None:
  622. return None
  623. return value