123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181 |
- """Library to handle connection with Switchbot."""
- from __future__ import annotations
- import logging
- from typing import Any
- from ..models import SwitchBotAdvertisement
- from .base_cover import COVER_COMMAND, COVER_EXT_SUM_KEY, SwitchbotBaseCover
- from .device import REQ_HEADER, update_after_operation
- OPEN_KEYS = [
- f"{REQ_HEADER}{COVER_COMMAND}010100",
- f"{REQ_HEADER}{COVER_COMMAND}05",
- ]
- CLOSE_KEYS = [
- f"{REQ_HEADER}{COVER_COMMAND}010164",
- f"{REQ_HEADER}{COVER_COMMAND}05",
- ]
- POSITION_KEYS = [
- f"{REQ_HEADER}{COVER_COMMAND}0101",
- f"{REQ_HEADER}{COVER_COMMAND}05",
- ]
- STOP_KEYS = [f"{REQ_HEADER}{COVER_COMMAND}0001", f"{REQ_HEADER}{COVER_COMMAND}00ff"]
- CURTAIN_EXT_CHAIN_INFO_KEY = f"{REQ_HEADER}468101"
- _LOGGER = logging.getLogger(__name__)
- class SwitchbotCurtain(SwitchbotBaseCover):
- """Representation of a Switchbot Curtain."""
- def __init__(self, *args: Any, **kwargs: Any) -> None:
- """Switchbot Curtain/WoCurtain constructor."""
-
-
-
-
-
-
-
- self._reverse: bool = kwargs.pop("reverse_mode", True)
- super().__init__(self._reverse, *args, **kwargs)
- self._settings: dict[str, Any] = {}
- self.ext_info_sum: dict[str, Any] = {}
- self.ext_info_adv: dict[str, Any] = {}
- def _set_parsed_data(
- self, advertisement: SwitchBotAdvertisement, data: dict[str, Any]
- ) -> None:
- """Set data."""
- in_motion = data["inMotion"]
- previous_position = self._get_adv_value("position")
- new_position = data["position"]
- self._update_motion_direction(in_motion, previous_position, new_position)
- super()._set_parsed_data(advertisement, data)
- @update_after_operation
- async def open(self, speed: int = 255) -> bool:
- """Send open command. Speed 255 - normal, 1 - slow"""
- self._is_opening = True
- self._is_closing = False
- return await self._send_multiple_commands(
- [OPEN_KEYS[0], f"{OPEN_KEYS[1]}{speed:02X}00"]
- )
- @update_after_operation
- async def close(self, speed: int = 255) -> bool:
- """Send close command. Speed 255 - normal, 1 - slow"""
- self._is_closing = True
- self._is_opening = False
- return await self._send_multiple_commands(
- [CLOSE_KEYS[0], f"{CLOSE_KEYS[1]}{speed:02X}64"]
- )
- @update_after_operation
- async def stop(self) -> bool:
- """Send stop command to device."""
- self._is_opening = self._is_closing = False
- return await super().stop()
- @update_after_operation
- async def set_position(self, position: int, speed: int = 255) -> bool:
- """Send position command (0-100) to device. Speed 255 - normal, 1 - slow"""
- direction_adjusted_position = (100 - position) if self._reverse else position
- self._update_motion_direction(
- True, self._get_adv_value("position"), direction_adjusted_position
- )
- return await super().set_position(position, speed)
- def get_position(self) -> Any:
- """Return cached position (0-100) of Curtain."""
-
- return self._get_adv_value("position")
- async def get_basic_info(self) -> dict[str, Any] | None:
- """Get device basic settings."""
- if not (_data := await self._get_basic_info()):
- return None
- _position = max(min(_data[6], 100), 0)
- _direction_adjusted_position = (100 - _position) if self._reverse else _position
- _previous_position = self._get_adv_value("position")
- _in_motion = bool(_data[5] & 0b01000011)
- self._update_motion_direction(
- _in_motion, _previous_position, _direction_adjusted_position
- )
- return {
- "battery": _data[1],
- "firmware": _data[2] / 10.0,
- "chainLength": _data[3],
- "openDirection": (
- "right_to_left" if _data[4] & 0b10000000 == 128 else "left_to_right"
- ),
- "touchToOpen": bool(_data[4] & 0b01000000),
- "light": bool(_data[4] & 0b00100000),
- "fault": bool(_data[4] & 0b00001000),
- "solarPanel": bool(_data[5] & 0b00001000),
- "calibration": bool(_data[5] & 0b00000100),
- "calibrated": bool(_data[5] & 0b00000100),
- "inMotion": _in_motion,
- "position": _direction_adjusted_position,
- "timers": _data[7],
- }
- def _update_motion_direction(
- self, in_motion: bool, previous_position: int | None, new_position: int
- ) -> None:
- """Update opening/closing status based on movement."""
- if previous_position is None:
- return
- if in_motion is False:
- self._is_closing = self._is_opening = False
- return
- if new_position != previous_position:
- self._is_opening = new_position > previous_position
- self._is_closing = new_position < previous_position
- async def get_extended_info_summary(self) -> dict[str, Any] | None:
- """Get extended info for all devices in chain."""
- _data = await self._send_command(key=COVER_EXT_SUM_KEY)
- if not _data:
- _LOGGER.error("%s: Unsuccessful, no result from device", self.name)
- return None
- if _data in (b"\x07", b"\x00"):
- _LOGGER.error("%s: Unsuccessful, please try again", self.name)
- return None
- self.ext_info_sum["device0"] = {
- "openDirectionDefault": not bool(_data[1] & 0b10000000),
- "touchToOpen": bool(_data[1] & 0b01000000),
- "light": bool(_data[1] & 0b00100000),
- "openDirection": (
- "left_to_right" if _data[1] & 0b00010000 else "right_to_left"
- ),
- }
-
- if _data[2] != 0:
- self.ext_info_sum["device1"] = {
- "openDirectionDefault": not bool(_data[2] & 0b10000000),
- "touchToOpen": bool(_data[2] & 0b01000000),
- "light": bool(_data[2] & 0b00100000),
- "openDirection": (
- "left_to_right" if _data[2] & 0b00010000 else "right_to_left"
- ),
- }
- return self.ext_info_sum
|