feat: Implement multi-packet protocol for regular scene speed control

Previously the scene speed slider for regular (non-DIY) scenes didn't work
because it was sending a simple 0x33 0x05 0x02 packet for manual color mode
instead of the proper scene speed protocol.

This implements the correct multi-packet 0xA3 BLE protocol:
- Get the active scene's scenceParam (base64 animation data) from library
- Decode and modify the speed byte at the speedIndex position
- Build multi-packet sequence (first/middle/last packets)
- Send activation packet 0x33 0x05 0x04 with scene code

Changes:
- api/ble_packet.py: Add modify_scene_speed(), build_multi_packet_sequence(),
  and build_scene_activation_packet() functions
- api/auth.py: Enhance fetch_light_effect_library() to extract scenceParam,
  sceneCode, speedIndex, and speed range from scene data
- api/mqtt.py: Update async_publish_ptreal() to accept list of packets
- coordinator.py: Rewrite async_send_scene_speed() with multi-packet protocol,
  add get_scene_speed_data() helper
- number.py: Make speed range dynamic based on active scene, only show
  available when a speed-capable scene is active
- tests/test_ble_packet.py: Add 28 tests for new functions

Author: Tom Lasswell

custom_components/govee/api/auth.py

@@ -286,6 +286,8 @@ async def fetch_light_effect_library(
         This API returns scene information including speed support:
         - support_speed: Whether the device supports scene speed control
         - scenes with speed_info: Per-scene speed configuration
+        - scenceParam: Base64-encoded animation data for multi-packet protocol
+        - sceneCode: Activation code for BLE scene activation packet
 
         Reference: wez/govee2mqtt light effect library API
 
@@ -297,17 +299,20 @@ async def fetch_light_effect_library(
             Dict with light effect library data:
             {
                 "support_speed": 1,  # 0 or 1
-                "scenes": [
-                    {
+                "categories": [...],  # Raw category data
+                "scenes": {  # Flattened scene lookup by sceneId
+                    "123": {
                         "sceneId": 123,
                         "sceneName": "Aurora",
-                        "speed_info": {
-                            "moveAll": [10, 100],  # [min, max] range
-                            "default": 50
-                        },
-                        ...
-                    }
-                ]
+                        "sceneCode": 10191,
+                        "scenceParam": "base64-encoded-data",
+                        "speedIndex": 5,  # Byte position of speed in scenceParam
+                        "speedMin": 10,
+                        "speedMax": 100,
+                        "speedDefault": 50,
+                    },
+                    ...
+                }
             }
 
         Raises:
@@ -339,14 +344,64 @@ async def fetch_light_effect_library(
                         code=response.status,
                     )
 
-                # Response structure: { "data": { "support_speed": 1, "scenes": [...] } }
-                result = data.get("data", {}) if isinstance(data, dict) else {}
+                # Response structure: { "data": { "support_speed": 1, "categories": [...] } }
+                raw_data = data.get("data", {}) if isinstance(data, dict) else {}
+
+                # Build flattened scene lookup for easy access
+                scenes_lookup: dict[str, dict[str, Any]] = {}
+                categories = raw_data.get("categories", [])
+
+                for category in categories:
+                    for scene in category.get("scenes", []):
+                        scene_id = scene.get("sceneId")
+                        if scene_id is None:
+                            continue
+
+                        scene_data: dict[str, Any] = {
+                            "sceneId": scene_id,
+                            "sceneName": scene.get("sceneName", ""),
+                            "sceneCode": scene.get("sceneCode"),
+                            "sceneType": scene.get("sceneType", 1),
+                        }
+
+                        # Extract scenceParam from lightEffects if available
+                        light_effects = scene.get("lightEffects", [])
+                        if light_effects:
+                            # Use the first light effect (usually the only one)
+                            effect = light_effects[0]
+                            scene_data["scenceParam"] = effect.get("scenceParam")
+                            scene_data["scenceParamId"] = effect.get("scenceParamId")
+
+                        # Extract speed_info if available
+                        speed_info = scene.get("speed_info")
+                        if speed_info:
+                            scene_data["speedIndex"] = speed_info.get("speedIndex")
+
+                            # moveAll contains [min, max] speed range
+                            move_all = speed_info.get("moveAll", [])
+                            if len(move_all) >= 2:
+                                scene_data["speedMin"] = move_all[0]
+                                scene_data["speedMax"] = move_all[1]
+                            else:
+                                # Default range
+                                scene_data["speedMin"] = 1
+                                scene_data["speedMax"] = 100
+
+                            scene_data["speedDefault"] = speed_info.get("default", 50)
+
+                        scenes_lookup[str(scene_id)] = scene_data
+
+                result = {
+                    "support_speed": raw_data.get("support_speed", 0),
+                    "categories": categories,
+                    "scenes": scenes_lookup,
+                }
 
                 _LOGGER.debug(
                     "Light effect library for %s: support_speed=%s, scenes=%d",
                     sku,
                     result.get("support_speed", 0),
-                    len(result.get("scenes", [])),
+                    len(scenes_lookup),
                 )
                 return result
 

custom_components/govee/api/ble_packet.py

@@ -244,3 +244,164 @@ def encode_packet_base64(packet: bytes) -> str:
         Base64-encoded ASCII string.
     """
     return base64.b64encode(packet).decode("ascii")
+
+
+# ==============================================================================
+# Multi-Packet Scene Speed Protocol (0xA3)
+# ==============================================================================
+
+# Multi-packet constants
+MULTI_PACKET_ID = 0xA3
+MULTI_PACKET_FIRST_INDEX = 0x00
+MULTI_PACKET_LAST_INDEX = 0xFF
+
+# Scene activation constants
+SCENE_ACTIVATION_PREFIX = 0x33
+SCENE_ACTIVATION_COMMAND = 0x05
+SCENE_ACTIVATION_INDICATOR = 0x04
+
+
+def modify_scene_speed(scence_param_b64: str, speed_index: int, new_speed: int) -> bytes:
+    """Decode scenceParam and modify the speed byte at speedIndex.
+
+    The scenceParam is base64-encoded animation data from the light effect
+    library. The speedIndex indicates which byte position contains the speed
+    value that needs to be modified.
+
+    Args:
+        scence_param_b64: Base64-encoded scenceParam from light effect library.
+        speed_index: Byte position of the speed value in the decoded data.
+        new_speed: New speed value (will be clamped to valid range).
+
+    Returns:
+        Modified animation data as bytes.
+
+    Raises:
+        ValueError: If speed_index is out of bounds or base64 is invalid.
+    """
+    try:
+        # Decode base64 to get raw animation data
+        data = bytearray(base64.b64decode(scence_param_b64))
+    except Exception as err:
+        raise ValueError(f"Invalid base64 scenceParam: {err}") from err
+
+    # Validate speed_index is within bounds
+    if speed_index < 0 or speed_index >= len(data):
+        raise ValueError(
+            f"speedIndex {speed_index} out of bounds for data length {len(data)}"
+        )
+
+    # Clamp speed to valid range (typically 1-100 for regular scenes)
+    new_speed = max(1, min(100, new_speed))
+
+    # Modify the speed byte
+    data[speed_index] = new_speed
+
+    return bytes(data)
+
+
+def build_multi_packet_sequence(data: bytes, scene_type: int = 2) -> list[bytes]:
+    """Build 0xA3 multi-packet sequence for scene animation data.
+
+    Multi-packet format:
+    - First packet:  [0xA3, 0x00, count, scene_type, data[0:14]...]
+    - Middle packet: [0xA3, index, data[offset:offset+17]...]
+    - Last packet:   [0xA3, 0xFF, remaining_data...]
+
+    Each packet is 20 bytes (19 data + 1 XOR checksum).
+
+    Args:
+        data: Animation data bytes to send.
+        scene_type: Scene type indicator (default 2 for regular scenes).
+
+    Returns:
+        List of 20-byte packets ready for transmission.
+    """
+    packets: list[bytes] = []
+
+    # Calculate how many packets we need
+    # First packet has 14 bytes of data (after 0xA3, 0x00, count, scene_type)
+    # Middle packets have 17 bytes of data (after 0xA3, index)
+    # Last packet has remaining data (after 0xA3, 0xFF)
+
+    if len(data) == 0:
+        # Edge case: empty data, just send a minimal packet
+        packets.append(build_packet([MULTI_PACKET_ID, MULTI_PACKET_FIRST_INDEX, 1, scene_type]))
+        packets.append(build_packet([MULTI_PACKET_ID, MULTI_PACKET_LAST_INDEX]))
+        return packets
+
+    # First packet: 0xA3, 0x00, count, scene_type, data[0:14]
+    first_data_size = 14
+    first_chunk = data[:first_data_size]
+    remaining = data[first_data_size:]
+
+    # Calculate total packet count (including first and last)
+    middle_data_size = 17
+    if len(remaining) == 0:
+        total_packets = 2  # Just first and last
+    else:
+        # Middle packets + last packet for remaining data
+        middle_packet_count = (len(remaining) - 1) // middle_data_size + 1
+        total_packets = 1 + middle_packet_count  # First + middle/last
+
+    # Build first packet
+    first_packet_data = [MULTI_PACKET_ID, MULTI_PACKET_FIRST_INDEX, total_packets, scene_type]
+    first_packet_data.extend(first_chunk)
+    packets.append(build_packet(first_packet_data))
+
+    # Build middle packets (if needed)
+    index = 1
+    offset = 0
+    while offset < len(remaining):
+        chunk = remaining[offset : offset + middle_data_size]
+        is_last = offset + middle_data_size >= len(remaining)
+
+        if is_last:
+            # Last packet uses 0xFF as index
+            packet_data = [MULTI_PACKET_ID, MULTI_PACKET_LAST_INDEX]
+        else:
+            # Middle packet uses sequential index
+            packet_data = [MULTI_PACKET_ID, index]
+
+        packet_data.extend(chunk)
+        packets.append(build_packet(packet_data))
+
+        offset += middle_data_size
+        index += 1
+
+    # If we only had first packet data, add an empty last packet
+    if len(remaining) == 0:
+        packets.append(build_packet([MULTI_PACKET_ID, MULTI_PACKET_LAST_INDEX]))
+
+    return packets
+
+
+def build_scene_activation_packet(scene_code: int) -> bytes:
+    """Build scene activation packet (0x33 0x05 0x04).
+
+    This packet is sent after the multi-packet scene data to activate
+    the scene on the device.
+
+    Packet format:
+    [0x33, 0x05, 0x04, code_lo, code_hi, 0x00...] + checksum
+
+    Args:
+        scene_code: Scene code from light effect library (16-bit value).
+
+    Returns:
+        20-byte BLE packet for scene activation.
+    """
+    # Extract low and high bytes of scene code
+    code_lo = scene_code & 0xFF
+    code_hi = (scene_code >> 8) & 0xFF
+
+    # Build activation packet
+    data = [
+        SCENE_ACTIVATION_PREFIX,  # 0x33
+        SCENE_ACTIVATION_COMMAND,  # 0x05
+        SCENE_ACTIVATION_INDICATOR,  # 0x04
+        code_lo,
+        code_hi,
+    ]
+
+    return build_packet(data)

custom_components/govee/api/mqtt.py

@@ -370,18 +370,20 @@ async def async_publish_ptreal(
         self,
         device_id: str,
         sku: str,
-        ble_packet_base64: str,
+        ble_packet_base64: str | list[str],
         device_topic: str | None = None,
     ) -> bool:
         """Publish BLE passthrough command via MQTT.
 
-        Sends a ptReal command to the device to execute a BLE packet.
+        Sends a ptReal command to the device to execute BLE packet(s).
         This allows controlling device features not exposed via REST API.
 
         Args:
             device_id: Target device identifier.
             sku: Device SKU/model.
-            ble_packet_base64: Base64-encoded BLE packet.
+            ble_packet_base64: Base64-encoded BLE packet or list of packets.
+                               For multi-packet sequences (e.g., scene speed),
+                               pass a list of base64-encoded packets.
             device_topic: Device-specific MQTT topic for publishing commands.
                           Required for AWS IoT - obtained from undocumented API.
 
@@ -400,12 +402,18 @@ async def async_publish_ptreal(
             )
             return False
 
+        # Normalize to list for consistent handling
+        if isinstance(ble_packet_base64, str):
+            packets = [ble_packet_base64]
+        else:
+            packets = ble_packet_base64
+
         # Build ptReal payload with device targeting
         payload = {
             "msg": {
                 "cmd": "ptReal",
                 "data": {
-                    "command": [ble_packet_base64],
+                    "command": packets,
                     "device": device_id,
                     "sku": sku,
                 },
@@ -418,10 +426,11 @@ async def async_publish_ptreal(
         try:
             await self._client.publish(device_topic, json.dumps(payload))
             _LOGGER.debug(
-                "Published ptReal to %s for device %s (sku=%s)",
+                "Published ptReal to %s for device %s (sku=%s, packets=%d)",
                 device_topic[:30] + "...",
                 device_id,
                 sku,
+                len(packets),
             )
             return True
         except Exception as err: