dry

switchbot/devices/bulb.py

@@ -1,7 +1,11 @@
 from __future__ import annotations
 
+import asyncio
+from enum import Enum
 from typing import Any
 
+from switchbot.models import SwitchBotAdvertisement
+
 from .device import SwitchbotDevice
 
 REQ_HEADER = "570f"
@@ -19,32 +23,45 @@ RGB_KEY = f"{BULB_COMMAND}16"
 CW_KEY = f"{BULB_COMMAND}17"
 
 
+class ColorMode(Enum):
+
+    OFF = 0
+    COLOR_TEMP = 1
+    RGB = 2
+    EFFECT = 3
+
+
 class SwitchbotBulb(SwitchbotDevice):
     """Representation of a Switchbot bulb."""
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Switchbot bulb constructor."""
         super().__init__(*args, **kwargs)
-        self._settings: dict[str, Any] = {}
+        self._state: dict[str, Any] = {}
+        self._update_task: asyncio.Task = None
 
     async def update(self, interface: int | None = None) -> None:
         """Update state of device."""
         result = await self._sendcommand(BULB_REQUEST)
+        self._update_state(result)
 
     async def turn_on(self) -> bool:
         """Turn device on."""
         result = await self._sendcommand(BULB_ON_KEY)
+        self._update_state(result)
         return result[1] == 0x80
 
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._sendcommand(BULB_OFF_KEY)
+        self._update_state(result)
         return result[1] == 0x00
 
     async def set_brightness(self, brightness: int) -> bool:
         """Set brightness."""
         assert 0 <= brightness <= 100, "Brightness must be between 0 and 100"
         result = await self._sendcommand(f"{BRIGHTNESS_KEY}{brightness:02X}")
+        self._update_state(result)
         return result[1] == 0x80
 
     async def set_color_temp(self, brightness: int, color_temp: int) -> bool:
@@ -54,6 +71,7 @@ class SwitchbotBulb(SwitchbotDevice):
         result = await self._sendcommand(
             f"{CW_BRIGHTNESS_KEY}{brightness:02X}{color_temp:04X}"
         )
+        self._update_state(result)
         return result[1] == 0x80
 
     async def set_rgb(self, brightness: int, r: int, g: int, b: int) -> bool:
@@ -65,13 +83,59 @@ class SwitchbotBulb(SwitchbotDevice):
         result = await self._sendcommand(
             f"{RGB_BRIGHTNESS_KEY}{brightness:02X}{r:02X}{g:02X}{b:02X}"
         )
+        self._update_state(result)
         return result[1] == 0x80
 
     async def turn_off(self) -> bool:
         """Turn device off."""
         result = await self._sendcommand(BULB_OFF_KEY)
+        self._update_state(result)
         return result[1] == 0x00
 
     def is_on(self) -> bool | None:
         """Return blub state from cache."""
         return self._get_adv_value("isOn")
+
+    @property
+    def on(self) -> bool:
+        """Return if bulb 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."""
+        if "cw" not in self._state:
+            return None
+        return self._state["cw"]
+
+    @property
+    def brightness(self) -> int | None:
+        """Return the current brightness value."""
+        return self._get_adv_value("brightness")
+
+    @property
+    def color_mode(self) -> ColorMode:
+        """Return the current color mode."""
+        return ColorMode(self._get_adv_value("color_mode") or 0)
+
+    def _update_state(self, result: bytes) -> None:
+        """Update device state."""
+        self._state["r"] = result[3]
+        self._state["g"] = result[4]
+        self._state["b"] = result[5]
+        self._state["cw"] = int(result[6:7].hex(), 16)
+        self._fire_callbacks()
+
+    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()
+        if current_state != self._get_adv_value("sequence_number"):
+            asyncio.ensure_future(self.update())

switchbot/devices/device.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 import asyncio
 import binascii
 import logging
-from typing import Any
+from typing import Any, Callable
 from uuid import UUID
 
 import async_timeout
@@ -91,6 +91,7 @@ class SwitchbotDevice:
         self._disconnect_timer: asyncio.TimerHandle | None = None
         self._expected_disconnect = False
         self.loop = asyncio.get_event_loop()
+        self._callbacks = list[Callable[[], None]] = []
 
     def _commandkey(self, key: str) -> str:
         """Add password to key if set."""
@@ -376,3 +377,18 @@ class SwitchbotDevice:
             return None
 
         return _data
+
+    def _fire_callbacks(self) -> None:
+        """Fire callbacks."""
+        for callback in self._callbacks:
+            callback()
+
+    def subscribe(self, callback: Callable[[], None]) -> Callable[[], None]:
+        """Subscribe to device notifications."""
+        self._callbacks.append(callback)
+
+        def _unsub() -> None:
+            """Unsubscribe from device notifications."""
+            self._callbacks.remove(callback)
+
+        return _unsub