1
0

base_cover.py 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141
  1. """Library to handle connection with Switchbot."""
  2. from __future__ import annotations
  3. import logging
  4. from abc import abstractmethod
  5. from typing import Any
  6. from .device import REQ_HEADER, SwitchbotDevice, update_after_operation
  7. # Cover keys
  8. COVER_COMMAND = "4501"
  9. # For second element of open and close arrs we should add two bytes i.e. ff00
  10. # First byte [ff] stands for speed (00 or ff - normal, 01 - slow) *
  11. # * Only for curtains 3. For other models use ff
  12. # Second byte [00] is a command (00 - open, 64 - close)
  13. POSITION_KEYS = [
  14. f"{REQ_HEADER}{COVER_COMMAND}0101",
  15. f"{REQ_HEADER}{COVER_COMMAND}05", # +speed
  16. ] # +actual_position
  17. STOP_KEYS = [f"{REQ_HEADER}{COVER_COMMAND}0001", f"{REQ_HEADER}{COVER_COMMAND}00ff"]
  18. COVER_EXT_SUM_KEY = f"{REQ_HEADER}460401"
  19. COVER_EXT_ADV_KEY = f"{REQ_HEADER}460402"
  20. _LOGGER = logging.getLogger(__name__)
  21. class SwitchbotBaseCover(SwitchbotDevice):
  22. """Representation of a Switchbot Cover devices for both curtains and tilt blinds."""
  23. def __init__(self, reverse: bool, *args: Any, **kwargs: Any) -> None:
  24. """Switchbot Cover device constructor."""
  25. super().__init__(*args, **kwargs)
  26. self._reverse = reverse
  27. self._settings: dict[str, Any] = {}
  28. self.ext_info_sum: dict[str, Any] = {}
  29. self.ext_info_adv: dict[str, Any] = {}
  30. self._is_opening: bool = False
  31. self._is_closing: bool = False
  32. async def _send_multiple_commands(self, keys: list[str]) -> bool:
  33. """Send multiple commands to device.
  34. Since we current have no way to tell which command the device
  35. needs we send both.
  36. """
  37. final_result = False
  38. for key in keys:
  39. result = await self._send_command(key)
  40. final_result |= self._check_command_result(result, 0, {1})
  41. return final_result
  42. @update_after_operation
  43. async def stop(self) -> bool:
  44. """Send stop command to device."""
  45. return await self._send_multiple_commands(STOP_KEYS)
  46. @update_after_operation
  47. async def set_position(self, position: int, speed: int = 255) -> bool:
  48. """Send position command (0-100) to device. Speed 255 - normal, 1 - slow"""
  49. position = (100 - position) if self._reverse else position
  50. return await self._send_multiple_commands(
  51. [
  52. f"{POSITION_KEYS[0]}{position:02X}",
  53. f"{POSITION_KEYS[1]}{speed:02X}{position:02X}",
  54. ]
  55. )
  56. @abstractmethod
  57. def get_position(self) -> Any:
  58. """Return current device position."""
  59. @abstractmethod
  60. async def get_basic_info(self) -> dict[str, Any] | None:
  61. """Get device basic settings."""
  62. @abstractmethod
  63. async def get_extended_info_summary(self) -> dict[str, Any] | None:
  64. """Get extended info for all devices in chain."""
  65. async def get_extended_info_adv(self) -> dict[str, Any] | None:
  66. """Get advance page info for device chain."""
  67. _data = await self._send_command(key=COVER_EXT_ADV_KEY)
  68. if not _data:
  69. _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
  70. return None
  71. if _data in (b"\x07", b"\x00"):
  72. _LOGGER.error("%s: Unsuccessful, please try again", self.name)
  73. return None
  74. _state_of_charge = [
  75. "not_charging",
  76. "charging_by_adapter",
  77. "charging_by_solar",
  78. "fully_charged",
  79. "solar_not_charging",
  80. "charging_error",
  81. ]
  82. self.ext_info_adv["device0"] = {
  83. "battery": _data[1],
  84. "firmware": _data[2] / 10.0,
  85. "stateOfCharge": _state_of_charge[_data[3]],
  86. }
  87. # If grouped curtain device present.
  88. if _data[4]:
  89. self.ext_info_adv["device1"] = {
  90. "battery": _data[4],
  91. "firmware": _data[5] / 10.0,
  92. "stateOfCharge": _state_of_charge[_data[6]],
  93. }
  94. return self.ext_info_adv
  95. def get_light_level(self) -> Any:
  96. """Return cached light level."""
  97. # To get actual light level call update() first.
  98. return self._get_adv_value("lightLevel")
  99. def is_reversed(self) -> bool:
  100. """Return True if curtain position is opposite from SB data."""
  101. return self._reverse
  102. def is_calibrated(self) -> Any:
  103. """Return True curtain is calibrated."""
  104. # To get actual light level call update() first.
  105. return self._get_adv_value("calibration")
  106. def is_opening(self) -> bool:
  107. """Return True if the curtain is opening."""
  108. return self._is_opening
  109. def is_closing(self) -> bool:
  110. """Return True if the curtain is closing."""
  111. return self._is_closing