1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
|
from __future__ import annotations
import logging
from abc import abstractmethod
from typing import Any
from ..helpers import create_background_task
from ..models import SwitchBotAdvertisement
from .device import SwitchbotDevice, SwitchbotOperationError, update_after_operation
_LOGGER = logging.getLogger(__name__)
class SwitchbotBaseLight(SwitchbotDevice):
"""Representation of a Switchbot light."""
_effect_dict: dict[str, list[str]] = {}
_set_brightness_command: str = ""
_set_color_temp_command: str = ""
_set_rgb_command: str = ""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Switchbot base light constructor."""
super().__init__(*args, **kwargs)
self._state: dict[str, Any] = {}
@property
def on(self) -> bool | None:
"""Return if light is on."""
return self.is_on()
@property
def rgb(self) -> tuple[int, int, int] | None:
"""Return the current rgb value."""
if "r" not in self._state or "g" not in self._state or "b" not in self._state:
return None
return self._state["r"], self._state["g"], self._state["b"]
@property
def color_temp(self) -> int | None:
"""Return the current color temp value."""
return self._state.get("cw") or self.min_temp
@property
def brightness(self) -> int | None:
"""Return the current brightness value."""
return self._get_adv_value("brightness") or 0
@property
@abstractmethod
def color_mode(self) -> Any:
"""Return the current color mode."""
raise NotImplementedError("Subclasses must implement color mode")
@property
def min_temp(self) -> int:
"""Return minimum color temp."""
return 2700
@property
def max_temp(self) -> int:
"""Return maximum color temp."""
return 6500
@property
def get_effect_list(self) -> list[str] | None:
"""Return the list of supported effects."""
return list(self._effect_dict) if self._effect_dict else None
def is_on(self) -> bool | None:
"""Return bulb state from cache."""
return self._get_adv_value("isOn")
def get_effect(self):
"""Return the current effect."""
return self._get_adv_value("effect")
@update_after_operation
async def set_brightness(self, brightness: int) -> bool:
"""Set brightness."""
assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
hex_brightness = f"{brightness:02X}"
self._check_function_support(self._set_brightness_command)
result = await self._send_command(
self._set_brightness_command.format(hex_brightness)
)
return self._check_command_result(result, 0, {1})
@update_after_operation
async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
"""Set color temp."""
assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
assert 2700 <= color_temp <= 6500, "Color Temp must be between 2700 and 6500"
hex_data = f"{brightness:02X}{color_temp:04X}"
self._check_function_support(self._set_color_temp_command)
result = await self._send_command(self._set_color_temp_command.format(hex_data))
return self._check_command_result(result, 0, {1})
@update_after_operation
async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
"""Set rgb."""
assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
assert 0 <= r <= 255, "r must be between 0 and 255"
assert 0 <= g <= 255, "g must be between 0 and 255"
assert 0 <= b <= 255, "b must be between 0 and 255"
self._check_function_support(self._set_rgb_command)
hex_data = f"{brightness:02X}{r:02X}{g:02X}{b:02X}"
result = await self._send_command(self._set_rgb_command.format(hex_data))
return self._check_command_result(result, 0, {1})
@update_after_operation
async def set_effect(self, effect: str) -> bool:
"""Set effect."""
effect_template = self._effect_dict.get(effect.lower())
if not effect_template:
raise SwitchbotOperationError(f"Effect {effect} not supported")
result = await self._send_multiple_commands(effect_template)
if result:
self._override_state({"effect": effect})
return result
async def _get_multi_commands_results(
self, commands: list[str]
) -> tuple[bytes, bytes] | None:
"""Check results after sending multiple commands."""
if not (results := await self._get_basic_info_by_multi_commands(commands)):
return None
_version_info, _data = results[0], results[1]
_LOGGER.debug(
"version info: %s, data: %s, address: %s",
_version_info,
_data,
self._device.address,
)
return _version_info, _data
async def _get_basic_info_by_multi_commands(
self, commands: list[str]
) -> list[bytes] | None:
"""Get device basic settings by sending multiple commands."""
results = []
for command in commands:
if not (result := await self._get_basic_info(command)):
return None
results.append(result)
return results
class SwitchbotSequenceBaseLight(SwitchbotBaseLight):
"""Representation of a Switchbot light."""
def update_from_advertisement(self, advertisement: SwitchBotAdvertisement) -> None:
"""Update device data from advertisement."""
current_state = self._get_adv_value("sequence_number")
super().update_from_advertisement(advertisement)
new_state = self._get_adv_value("sequence_number")
_LOGGER.debug(
"%s: update advertisement: %s (seq before: %s) (seq after: %s)",
self.name,
advertisement,
current_state,
new_state,
)
if current_state != new_state:
create_background_task(self.update())
|