|
| 1 | +"""Library to handle connection with Switchbot.""" |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +import logging |
| 5 | +from typing import Any |
| 6 | + |
| 7 | +from switchbot.devices.device import REQ_HEADER, update_after_operation |
| 8 | + |
| 9 | +from .curtain import CURTAIN_EXT_SUM_KEY, SwitchbotCurtain |
| 10 | + |
| 11 | +_LOGGER = logging.getLogger(__name__) |
| 12 | + |
| 13 | + |
| 14 | +BLIND_COMMAND = "4501" |
| 15 | +OPEN_KEYS = [ |
| 16 | + f"{REQ_HEADER}{BLIND_COMMAND}010132", |
| 17 | + f"{REQ_HEADER}{BLIND_COMMAND}05ff32", |
| 18 | +] |
| 19 | +CLOSE_DOWN_KEYS = [ |
| 20 | + f"{REQ_HEADER}{BLIND_COMMAND}010100", |
| 21 | + f"{REQ_HEADER}{BLIND_COMMAND}05ff00", |
| 22 | +] |
| 23 | +CLOSE_UP_KEYS = [ |
| 24 | + f"{REQ_HEADER}{BLIND_COMMAND}010164", |
| 25 | + f"{REQ_HEADER}{BLIND_COMMAND}05ff64", |
| 26 | +] |
| 27 | + |
| 28 | + |
| 29 | +class SwitchbotBlindTilt(SwitchbotCurtain): |
| 30 | + """Representation of a Switchbot Blind Tilt.""" |
| 31 | + |
| 32 | + # The position of the blind is saved returned with 0 = closed down, 50 = open and 100 = closed up. |
| 33 | + # This is independent of the calibration of the blind. |
| 34 | + # The parameter 'reverse_mode' reverse these values, |
| 35 | + # if 'reverse_mode' = True, position = 0 equals closed up |
| 36 | + # and position = 100 equals closed down. The parameter is default set to False so that |
| 37 | + # the definition of position is the same as in Home Assistant. |
| 38 | + # This is opposite to the base class so needs to be overwritten. |
| 39 | + |
| 40 | + def __init__(self, *args: Any, **kwargs: Any) -> None: |
| 41 | + """Switchbot Blind Tilt/woBlindTilt constructor.""" |
| 42 | + super().__init__(*args, **kwargs) |
| 43 | + |
| 44 | + self._reverse: bool = kwargs.pop("reverse_mode", False) |
| 45 | + |
| 46 | + @update_after_operation |
| 47 | + async def open(self) -> bool: |
| 48 | + """Send open command.""" |
| 49 | + return await self._send_multiple_commands(OPEN_KEYS) |
| 50 | + |
| 51 | + @update_after_operation |
| 52 | + async def close_up(self) -> bool: |
| 53 | + """Send close up command.""" |
| 54 | + return await self._send_multiple_commands(CLOSE_UP_KEYS) |
| 55 | + |
| 56 | + @update_after_operation |
| 57 | + async def close_down(self) -> bool: |
| 58 | + """Send close down command.""" |
| 59 | + return await self._send_multiple_commands(CLOSE_DOWN_KEYS) |
| 60 | + |
| 61 | + # The aim of this is to close to the nearest endpoint. |
| 62 | + # If we're open upwards we close up, if we're open downwards we close down. |
| 63 | + # If we're in the middle we default to close down as that seems to be the app's preference. |
| 64 | + @update_after_operation |
| 65 | + async def close(self) -> bool: |
| 66 | + """Send close command.""" |
| 67 | + if self.get_position() > 50: |
| 68 | + return await self.close_up() |
| 69 | + else: |
| 70 | + return await self.close_down() |
| 71 | + |
| 72 | + def get_position(self) -> Any: |
| 73 | + """Return cached tilt (0-100) of Blind Tilt.""" |
| 74 | + # To get actual tilt call update() first. |
| 75 | + return self._get_adv_value("tilt") |
| 76 | + |
| 77 | + async def get_basic_info(self) -> dict[str, Any] | None: |
| 78 | + """Get device basic settings.""" |
| 79 | + if not (_data := await self._get_basic_info()): |
| 80 | + return None |
| 81 | + |
| 82 | + _tilt = max(min(_data[6], 100), 0) |
| 83 | + return { |
| 84 | + "battery": _data[1], |
| 85 | + "firmware": _data[2] / 10.0, |
| 86 | + "light": bool(_data[4] & 0b00100000), |
| 87 | + "fault": bool(_data[4] & 0b00001000), |
| 88 | + "solarPanel": bool(_data[5] & 0b00001000), |
| 89 | + "calibration": bool(_data[5] & 0b00000100), |
| 90 | + "calibrated": bool(_data[5] & 0b00000100), |
| 91 | + "inMotion": bool(_data[5] & 0b00000011), |
| 92 | + "motionDirection": { |
| 93 | + "up": bool(_data[5] & (0b00000010 if self._reverse else 0b00000001)), |
| 94 | + "down": bool(_data[5] & (0b00000001 if self._reverse else 0b00000010)), |
| 95 | + }, |
| 96 | + "tilt": (100 - _tilt) if self._reverse else _tilt, |
| 97 | + "timers": _data[7], |
| 98 | + } |
| 99 | + |
| 100 | + async def get_extended_info_summary(self) -> dict[str, Any] | None: |
| 101 | + """Get basic info for all devices in chain.""" |
| 102 | + _data = await self._send_command(key=CURTAIN_EXT_SUM_KEY) |
| 103 | + |
| 104 | + if not _data: |
| 105 | + _LOGGER.error("%s: Unsuccessful, no result from device", self.name) |
| 106 | + return None |
| 107 | + |
| 108 | + if _data in (b"\x07", b"\x00"): |
| 109 | + _LOGGER.error("%s: Unsuccessful, please try again", self.name) |
| 110 | + return None |
| 111 | + |
| 112 | + self.ext_info_sum["device0"] = { |
| 113 | + "light": bool(_data[1] & 0b00100000), |
| 114 | + } |
| 115 | + |
| 116 | + return self.ext_info_sum |
0 commit comments