1
0

curtain.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201
  1. """Library to handle connection with Switchbot."""
  2. from __future__ import annotations
  3. import logging
  4. from typing import Any
  5. from .device import REQ_HEADER, SwitchbotDevice, update_after_operation
  6. # Curtain keys
  7. CURTAIN_COMMAND = "4501"
  8. OPEN_KEYS = [
  9. f"{REQ_HEADER}{CURTAIN_COMMAND}010100",
  10. f"{REQ_HEADER}{CURTAIN_COMMAND}05ff00",
  11. ]
  12. CLOSE_KEYS = [
  13. f"{REQ_HEADER}{CURTAIN_COMMAND}010164",
  14. f"{REQ_HEADER}{CURTAIN_COMMAND}05ff64",
  15. ]
  16. POSITION_KEYS = [
  17. f"{REQ_HEADER}{CURTAIN_COMMAND}0101",
  18. f"{REQ_HEADER}{CURTAIN_COMMAND}05ff",
  19. ] # +actual_position
  20. STOP_KEYS = [f"{REQ_HEADER}{CURTAIN_COMMAND}0001", f"{REQ_HEADER}{CURTAIN_COMMAND}00ff"]
  21. CURTAIN_EXT_SUM_KEY = f"{REQ_HEADER}460401"
  22. CURTAIN_EXT_ADV_KEY = f"{REQ_HEADER}460402"
  23. CURTAIN_EXT_CHAIN_INFO_KEY = f"{REQ_HEADER}468101"
  24. _LOGGER = logging.getLogger(__name__)
  25. class SwitchbotCurtain(SwitchbotDevice):
  26. """Representation of a Switchbot Curtain."""
  27. def __init__(self, *args: Any, **kwargs: Any) -> None:
  28. """Switchbot Curtain/WoCurtain constructor."""
  29. # The position of the curtain is saved returned with 0 = open and 100 = closed.
  30. # This is independent of the calibration of the curtain bot (Open left to right/
  31. # Open right to left/Open from the middle).
  32. # The parameter 'reverse_mode' reverse these values,
  33. # if 'reverse_mode' = True, position = 0 equals close
  34. # and position = 100 equals open. The parameter is default set to True so that
  35. # the definition of position is the same as in Home Assistant.
  36. super().__init__(*args, **kwargs)
  37. self._reverse: bool = kwargs.pop("reverse_mode", True)
  38. self._settings: dict[str, Any] = {}
  39. self.ext_info_sum: dict[str, Any] = {}
  40. self.ext_info_adv: dict[str, Any] = {}
  41. async def _send_multiple_commands(self, keys: list[str]) -> bool:
  42. """Send multiple commands to device.
  43. Since we current have no way to tell which command the device
  44. needs we send both.
  45. """
  46. final_result = False
  47. for key in keys:
  48. result = await self._send_command(key)
  49. final_result |= self._check_command_result(result, 0, {1})
  50. return final_result
  51. @update_after_operation
  52. async def open(self) -> bool:
  53. """Send open command."""
  54. return await self._send_multiple_commands(OPEN_KEYS)
  55. @update_after_operation
  56. async def close(self) -> bool:
  57. """Send close command."""
  58. return await self._send_multiple_commands(CLOSE_KEYS)
  59. @update_after_operation
  60. async def stop(self) -> bool:
  61. """Send stop command to device."""
  62. return await self._send_multiple_commands(STOP_KEYS)
  63. @update_after_operation
  64. async def set_position(self, position: int) -> bool:
  65. """Send position command (0-100) to device."""
  66. position = (100 - position) if self._reverse else position
  67. hex_position = "%0.2X" % position
  68. return await self._send_multiple_commands(
  69. [key + hex_position for key in POSITION_KEYS]
  70. )
  71. def get_position(self) -> Any:
  72. """Return cached position (0-100) of Curtain."""
  73. # To get actual position call update() first.
  74. return self._get_adv_value("position")
  75. async def get_basic_info(self) -> dict[str, Any] | None:
  76. """Get device basic settings."""
  77. if not (_data := await self._get_basic_info()):
  78. return None
  79. _position = max(min(_data[6], 100), 0)
  80. return {
  81. "battery": _data[1],
  82. "firmware": _data[2] / 10.0,
  83. "chainLength": _data[3],
  84. "openDirection": (
  85. "right_to_left" if _data[4] & 0b10000000 == 128 else "left_to_right"
  86. ),
  87. "touchToOpen": bool(_data[4] & 0b01000000),
  88. "light": bool(_data[4] & 0b00100000),
  89. "fault": bool(_data[4] & 0b00001000),
  90. "solarPanel": bool(_data[5] & 0b00001000),
  91. "calibration": bool(_data[5] & 0b00000100),
  92. "calibrated": bool(_data[5] & 0b00000100),
  93. "inMotion": bool(_data[5] & 0b01000011),
  94. "position": (100 - _position) if self._reverse else _position,
  95. "timers": _data[7],
  96. }
  97. async def get_extended_info_summary(self) -> dict[str, Any] | None:
  98. """Get basic info for all devices in chain."""
  99. _data = await self._send_command(key=CURTAIN_EXT_SUM_KEY)
  100. if not _data:
  101. _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
  102. return None
  103. if _data in (b"\x07", b"\x00"):
  104. _LOGGER.error("%s: Unsuccessful, please try again", self.name)
  105. return None
  106. self.ext_info_sum["device0"] = {
  107. "openDirectionDefault": not bool(_data[1] & 0b10000000),
  108. "touchToOpen": bool(_data[1] & 0b01000000),
  109. "light": bool(_data[1] & 0b00100000),
  110. "openDirection": (
  111. "left_to_right" if _data[1] & 0b00010000 == 1 else "right_to_left"
  112. ),
  113. }
  114. # if grouped curtain device present.
  115. if _data[2] != 0:
  116. self.ext_info_sum["device1"] = {
  117. "openDirectionDefault": not bool(_data[1] & 0b10000000),
  118. "touchToOpen": bool(_data[1] & 0b01000000),
  119. "light": bool(_data[1] & 0b00100000),
  120. "openDirection": (
  121. "left_to_right" if _data[1] & 0b00010000 else "right_to_left"
  122. ),
  123. }
  124. return self.ext_info_sum
  125. async def get_extended_info_adv(self) -> dict[str, Any] | None:
  126. """Get advance page info for device chain."""
  127. _data = await self._send_command(key=CURTAIN_EXT_ADV_KEY)
  128. if not _data:
  129. _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
  130. return None
  131. if _data in (b"\x07", b"\x00"):
  132. _LOGGER.error("%s: Unsuccessful, please try again", self.name)
  133. return None
  134. _state_of_charge = [
  135. "not_charging",
  136. "charging_by_adapter",
  137. "charging_by_solar",
  138. "fully_charged",
  139. "solar_not_charging",
  140. "charging_error",
  141. ]
  142. self.ext_info_adv["device0"] = {
  143. "battery": _data[1],
  144. "firmware": _data[2] / 10.0,
  145. "stateOfCharge": _state_of_charge[_data[3]],
  146. }
  147. # If grouped curtain device present.
  148. if _data[4]:
  149. self.ext_info_adv["device1"] = {
  150. "battery": _data[4],
  151. "firmware": _data[5] / 10.0,
  152. "stateOfCharge": _state_of_charge[_data[6]],
  153. }
  154. return self.ext_info_adv
  155. def get_light_level(self) -> Any:
  156. """Return cached light level."""
  157. # To get actual light level call update() first.
  158. return self._get_adv_value("lightLevel")
  159. def is_reversed(self) -> bool:
  160. """Return True if curtain position is opposite from SB data."""
  161. return self._reverse
  162. def is_calibrated(self) -> Any:
  163. """Return True curtain is calibrated."""
  164. # To get actual light level call update() first.
  165. return self._get_adv_value("calibration")