feat: add extended timer support for 0xB6 devices

The 0xB6 device uses a distinct timer wire format (e0 05/06 wrapped commands,
variable-length 6-slot list, per-slot set):

- const.py: TIMER_ACTION_* and TIMER_EFFECT_* constants
- protocol.py: MSG_TIMERS_EXTENDED and the timer methods on
  ProtocolLEDENETExtendedCustom
- timer.py: LedTimerExtended dataclass (to_bytes/from_bytes, __str__)
- aiodevice.py: timer types widened, async_set_timers sends 0xB6 timers one at a
  time, new async_set_timer

Stacked on the effect-APIs PR.

Author: asxzy

flux_led/aiodevice.py

@@ -49,11 +49,12 @@
     ProtocolLEDENET8Byte,
     ProtocolLEDENETAddressableA3,
     ProtocolLEDENETAddressableChristmas,
+    ProtocolLEDENETExtendedCustom,
     ProtocolLEDENETOriginal,
     RemoteConfig,
 )
 from .scanner import FluxLEDDiscovery
-from .timer import LedTimer
+from .timer import LedTimer, LedTimerExtended
 from .utils import color_temp_to_white_levels, rgbw_brightness, rgbww_brightness
 
 _LOGGER = logging.getLogger(__name__)
@@ -87,7 +88,7 @@ def __init__(
         self._get_time_future: asyncio.Future[bool] | None = None
         self._get_timers_lock: asyncio.Lock = asyncio.Lock()
         self._get_timers_future: asyncio.Future[bool] | None = None
-        self._timers: list[LedTimer] | None = None
+        self._timers: list[LedTimer] | list[LedTimerExtended] | None = None
         self._power_restore_future: asyncio.Future[bool] = loop.create_future()
         self._device_config_lock: asyncio.Lock = asyncio.Lock()
         self._device_config_future: asyncio.Future[bool] = loop.create_future()
@@ -711,7 +712,7 @@ async def async_get_time(self) -> datetime | None:
                 return None
             return self._last_time
 
-    async def async_get_timers(self) -> list[LedTimer] | None:
+    async def async_get_timers(self) -> list[LedTimer] | list[LedTimerExtended] | None:
         """Get the timers."""
         assert self._protocol is not None
         if isinstance(self._protocol, ProtocolLEDENETOriginal):
@@ -728,10 +729,24 @@ async def async_get_timers(self) -> list[LedTimer] | None:
                 return None
             return self._timers
 
-    async def async_set_timers(self, timer_list: list[LedTimer]) -> None:
+    async def async_set_timers(
+        self, timer_list: list[LedTimer] | list[LedTimerExtended]
+    ) -> None:
         """Set the timers."""
         assert self._protocol is not None
-        await self._async_send_msg(self._protocol.construct_set_timers(timer_list))
+        if isinstance(self._protocol, ProtocolLEDENETExtendedCustom):
+            # 0xB6 devices set timers one at a time
+            commands = self._protocol.construct_set_timers(timer_list)  # type: ignore[arg-type]
+            for cmd in commands:
+                await self._async_send_msg(cmd)
+        else:
+            await self._async_send_msg(self._protocol.construct_set_timers(timer_list))  # type: ignore[arg-type]
+
+    async def async_set_timer(self, timer: LedTimerExtended) -> None:
+        """Set a single timer (for 0xB6 devices)."""
+        assert self._protocol is not None
+        if isinstance(self._protocol, ProtocolLEDENETExtendedCustom):
+            await self._async_send_msg(self._protocol.construct_set_timer(timer))
 
     async def async_set_time(self, time: datetime | None = None) -> None:
         """Set the current time."""

flux_led/const.py

@@ -256,3 +256,15 @@ class ScribbleLED(NamedTuple):
 PUSH_UPDATE_INTERVAL = 90  # seconds
 
 NEVER_TIME = -PUSH_UPDATE_INTERVAL
+
+
+# Extended Timer Action Types (0xB6 devices)
+TIMER_ACTION_ON: Final = 0x23
+TIMER_ACTION_OFF: Final = 0x24
+TIMER_ACTION_COLOR: Final = 0xA1
+TIMER_ACTION_SCENE_GRADIENT: Final = 0x29
+TIMER_ACTION_SCENE_SEGMENTS: Final = 0x6B
+
+# Extended Timer Effect Types
+TIMER_EFFECT_GRADIENT: Final = 0x21  # e1 21 - gradient overlay
+TIMER_EFFECT_SEGMENTS: Final = 0x22  # e1 22 - static segments/colorful

flux_led/protocol.py

@@ -28,7 +28,7 @@
     LevelWriteModeData,
     MultiColorEffects,
 )
-from .timer import LedTimer
+from .timer import LedTimer, LedTimerExtended
 from .utils import (
     scaled_color_temp_to_white_levels,
     utils,
@@ -130,6 +130,7 @@ class PowerRestoreStates:
 MSG_STATE = "state"
 MSG_TIME = "time"
 MSG_TIMERS = "timers"
+MSG_TIMERS_EXTENDED = "timers_extended"
 MSG_MUSIC_MODE_STATE = "music_mode_state"
 MSG_ADDRESSABLE_STATE = "addressable_state"
 MSG_DEVICE_CONFIG = "device_config"
@@ -165,6 +166,7 @@ class PowerRestoreStates:
     (0x63,): MSG_A1_DEVICE_CONFIG,
     (0x72,): MSG_MUSIC_MODE_STATE,
     (0x2B,): MSG_REMOTE_CONFIG,
+    (0xE0, 0x06): MSG_TIMERS_EXTENDED,  # 0xB6 device timer response
 }
 
 MSG_LENGTHS = {
@@ -2033,6 +2035,115 @@ def construct_scribble_paint(
             msg, inner_pre_constructed=True, version=0x02
         )
 
+    # --- Timer Support for 0xB6 devices ---
+    # This protocol uses a different timer format than other protocols:
+    # - Query: e0 06 (wrapped)
+    # - Response: e0 06 + variable length slot data
+    # - Set: e0 05 SS f0 HH MM 00 RR ... (wrapped)
+
+    def construct_get_timers(self) -> bytearray:
+        """Construct a get timers request.
+
+        0xB6 devices use the wrapped e0 06 command for timer queries.
+        """
+        inner = bytearray([0xE0, 0x06])
+        return self.construct_wrapped_message(
+            inner, inner_pre_constructed=True, version=0x01
+        )
+
+    def construct_set_timer(self, timer: LedTimerExtended) -> bytearray:
+        """Construct a set timer command for a single timer.
+
+        0xB6 devices set timers one at a time using e0 05 SS ...
+
+        Args:
+            timer: The LedTimerExtended object to set
+
+        Returns:
+            Wrapped command bytearray
+        """
+        inner = bytearray([0xE0, 0x05, timer.slot])
+        inner.extend(timer.to_bytes())
+        return self.construct_wrapped_message(
+            inner, inner_pre_constructed=True, version=0x01
+        )
+
+    def construct_set_timers(  # type: ignore[override]
+        self, timer_list: list[LedTimerExtended]
+    ) -> list[bytearray]:
+        """Construct set timer commands for multiple timers.
+
+        0xB6 devices set timers one at a time, so this returns a list
+        of commands (one per timer).
+
+        Args:
+            timer_list: List of LedTimerExtended objects to set
+
+        Returns:
+            List of wrapped command bytearrays
+        """
+        return [self.construct_set_timer(timer) for timer in timer_list]
+
+    def is_valid_timers_response(self, msg: bytes) -> bool:
+        """Check if a message is a valid timers response.
+
+        0xB6 timer responses start with e0 06.
+        """
+        if len(msg) < 2:
+            return False
+        return msg[0] == 0xE0 and msg[1] == 0x06
+
+    def parse_get_timers(self, msg: bytes) -> list[LedTimerExtended]:  # type: ignore[override]
+        """Parse timer response into list of LedTimerExtended objects.
+
+        Response format:
+        - Empty: e0 06 (2 bytes)
+        - With timers: e0 06 [slot1] [slot2] ... [slot6]
+
+        Each slot is either:
+        - 7 bytes if empty/inactive
+        - 21 bytes if simple timer (ON/OFF/color)
+        - Variable (48+) bytes if scene timer
+        """
+        if len(msg) <= 2:
+            # Empty response or just header - no timers configured
+            return []
+
+        timers: list[LedTimerExtended] = []
+        offset = 2  # Skip e0 06 header
+
+        while offset < len(msg):
+            # Check if we have enough bytes for minimum slot (7 bytes)
+            if offset + 7 > len(msg):
+                break
+
+            timer, consumed = LedTimerExtended.from_bytes(msg, offset)
+            timers.append(timer)
+            offset += consumed
+
+        return timers
+
+    @property
+    def timer_count(self) -> int:
+        """Number of timer slots supported."""
+        return 6
+
+    def expected_timers_response_length(self, data: bytes) -> int:
+        """Calculate expected timer response length.
+
+        For 0xB6 devices, the timer response has variable length
+        based on the inner message length encoded in the wrapped message.
+        """
+        # Timer responses are wrapped, so we extract from the wrapper
+        if (
+            len(data) >= OUTER_MESSAGE_WRAPPER_START_LEN
+            and data[0] == OUTER_MESSAGE_FIRST_BYTE
+        ):
+            inner_msg_len = (data[8] << 8) + data[9]
+            return OUTER_MESSAGE_WRAPPER_START_LEN + inner_msg_len + CHECKSUM_LEN
+        # Fallback
+        return len(data)
+
 
 class ProtocolLEDENETAddressableBase(ProtocolLEDENET9Byte):
     """Base class for addressable protocols."""