feat: Add DIY style control and music mode via BLE passthrough

Extends BLE-to-MQTT integration with two new features:
- DIY Style select entity for animation styles (Fade, Jumping, Flicker, Marquee, Music)
- Music Mode switch entity to enable/disable audio-reactive microphone mode

Both features require MQTT connection for BLE passthrough (ptReal command).

Changes:
- Add build_diy_style_packet() and build_music_mode_packet() to ble_packet.py
- Add DIYStyle enum and style name mappings
- Add supports_music_mode property to device model
- Add diy_style and music_mode_enabled state fields
- Add async_send_diy_style() and async_send_music_mode() coordinator methods
- Add GoveeDIYStyleSelectEntity and GoveeMusicModeSwitchEntity
- Add translations and orphan cleanup support
- Add comprehensive unit tests (73 BLE packet tests)

Author: Tom Lasswell

.gitignore

@@ -152,4 +152,7 @@ logs/
 
 # Local/temporary documentation (keep govee-protocol-reference.md)
 docs/epic-*.md
-docs/legacy-*.md
+docs/legacy-*.md
+
+# GitHub profile README (not part of this project)
+profile-README.md

custom_components/govee/__init__.py

@@ -41,6 +41,7 @@
 # Order determines entity display order in device view
 PLATFORMS: list[Platform] = [
     Platform.SELECT,  # Scene dropdowns - show first
+    Platform.NUMBER,  # DIY speed controls
     Platform.LIGHT,  # Main light + segments
     Platform.SWITCH,
     Platform.SENSOR,
@@ -241,8 +242,10 @@ async def _async_cleanup_orphaned_entities(
     entity_suffixes = (
         "_scene_select",
         "_diy_scene_select",
+        "_diy_style_select",
         "_refresh_scenes",
         "_night_light",
+        "_music_mode",
     )
 
     # Get all entity entries for this config entry

custom_components/govee/api/ble_packet.py

@@ -0,0 +1,201 @@
+"""BLE packet construction for Govee devices.
+
+Builds 20-byte BLE packets that can be sent via the AWS IoT MQTT
+ptReal (passthrough real) command to control device features not
+exposed via the REST API.
+
+Packet format:
+- Bytes 0-18: Command data (padded with 0x00)
+- Byte 19: XOR checksum of bytes 0-18
+
+DIY Speed packet:
+- Byte 0: 0xA1 (DIY packet identifier)
+- Byte 1: 0x02 (DIY command type)
+- Byte 2: 0x01 (number of segments/modes)
+- Byte 3: 0x00 (style - default)
+- Byte 4: 0x00 (mode - default)
+- Byte 5: speed (0-100, where 0 is static and 100 is fastest)
+
+DIY Style packet:
+- Byte 0: 0xA1 (DIY packet identifier)
+- Byte 1: 0x02 (DIY command type)
+- Byte 2: 0x01 (number of segments/modes)
+- Byte 3: style (0x00=Fade, 0x01=Jumping, 0x02=Flicker, 0x03=Marquee, 0x04=Music)
+- Byte 4: 0x00 (mode - default)
+- Byte 5: speed (0-100, where 0 is static and 100 is fastest)
+
+Music Mode packet:
+- Byte 0: 0x33 (standard command prefix)
+- Byte 1: 0x05 (color/mode command)
+- Byte 2: 0x01 (music mode indicator)
+- Byte 3: enabled (0x01=on, 0x00=off)
+- Byte 4: sensitivity (0-100)
+"""
+
+from __future__ import annotations
+
+import base64
+from enum import IntEnum
+
+# DIY packet constants
+DIY_PACKET_ID = 0xA1
+DIY_COMMAND = 0x02
+
+# Music mode packet constants
+MUSIC_PACKET_PREFIX = 0x33
+MUSIC_MODE_COMMAND = 0x05
+MUSIC_MODE_INDICATOR = 0x01
+
+
+class DIYStyle(IntEnum):
+    """DIY animation style options."""
+
+    FADE = 0x00
+    JUMPING = 0x01
+    FLICKER = 0x02
+    MARQUEE = 0x03
+    MUSIC = 0x04
+
+
+# Style name to enum mapping for select entity
+DIY_STYLE_NAMES = {
+    "Fade": DIYStyle.FADE,
+    "Jumping": DIYStyle.JUMPING,
+    "Flicker": DIYStyle.FLICKER,
+    "Marquee": DIYStyle.MARQUEE,
+    "Music": DIYStyle.MUSIC,
+}
+
+
+def calculate_checksum(data: list[int]) -> int:
+    """Calculate XOR checksum of all bytes.
+
+    Args:
+        data: List of byte values to checksum.
+
+    Returns:
+        XOR of all bytes, masked to 8 bits.
+    """
+    checksum = 0
+    for byte in data:
+        checksum ^= byte
+    return checksum & 0xFF
+
+
+def build_packet(data: list[int]) -> bytes:
+    """Build a 20-byte BLE packet with checksum.
+
+    Pads the data to 19 bytes and appends XOR checksum.
+
+    Args:
+        data: Command bytes (will be padded to 19 bytes).
+
+    Returns:
+        20-byte packet as bytes.
+    """
+    packet = list(data)
+
+    # Pad to 19 bytes
+    while len(packet) < 19:
+        packet.append(0x00)
+
+    # Truncate if too long
+    packet = packet[:19]
+
+    # Append checksum
+    packet.append(calculate_checksum(packet))
+
+    return bytes(packet)
+
+
+def build_diy_speed_packet(speed: int) -> bytes:
+    """Build DIY scene speed control packet.
+
+    Args:
+        speed: Playback speed 0-100, where 0 is static (no animation)
+               and 100 is the fastest playback speed.
+
+    Returns:
+        20-byte BLE packet for DIY speed command.
+    """
+    # Clamp speed to valid range
+    speed = max(0, min(100, speed))
+
+    # Build command data
+    # Packet: A1 02 [NUM] [STYLE] [MODE] [SPEED] ...
+    data = [
+        DIY_PACKET_ID,  # 0xA1 - DIY packet identifier
+        DIY_COMMAND,  # 0x02 - DIY command type
+        0x01,  # Number of segments/modes
+        0x00,  # Style (default)
+        0x00,  # Mode (default)
+        speed,  # Speed value 0-100
+    ]
+
+    return build_packet(data)
+
+
+def build_diy_style_packet(style: int | DIYStyle, speed: int = 50) -> bytes:
+    """Build DIY scene style control packet.
+
+    Args:
+        style: Animation style (0=Fade, 1=Jumping, 2=Flicker, 3=Marquee, 4=Music).
+        speed: Playback speed 0-100, where 0 is static and 100 is fastest.
+
+    Returns:
+        20-byte BLE packet for DIY style command.
+    """
+    # Clamp values to valid ranges
+    style_val = max(0, min(4, int(style)))
+    speed = max(0, min(100, speed))
+
+    # Build command data
+    # Packet: A1 02 [NUM] [STYLE] [MODE] [SPEED] ...
+    data = [
+        DIY_PACKET_ID,  # 0xA1 - DIY packet identifier
+        DIY_COMMAND,  # 0x02 - DIY command type
+        0x01,  # Number of segments/modes
+        style_val,  # Style value
+        0x00,  # Mode (default)
+        speed,  # Speed value 0-100
+    ]
+
+    return build_packet(data)
+
+
+def build_music_mode_packet(enabled: bool, sensitivity: int = 50) -> bytes:
+    """Build music mode control packet.
+
+    Args:
+        enabled: True to enable music mode, False to disable.
+        sensitivity: Microphone sensitivity 0-100.
+
+    Returns:
+        20-byte BLE packet for music mode command.
+    """
+    # Clamp sensitivity to valid range
+    sensitivity = max(0, min(100, sensitivity))
+
+    # Build command data
+    # Packet: 33 05 01 [ENABLED] [SENSITIVITY] ...
+    data = [
+        MUSIC_PACKET_PREFIX,  # 0x33 - Standard command prefix
+        MUSIC_MODE_COMMAND,  # 0x05 - Color/mode command
+        MUSIC_MODE_INDICATOR,  # 0x01 - Music mode indicator
+        0x01 if enabled else 0x00,  # Enabled state
+        sensitivity,  # Sensitivity 0-100
+    ]
+
+    return build_packet(data)
+
+
+def encode_packet_base64(packet: bytes) -> str:
+    """Base64 encode a packet for ptReal command.
+
+    Args:
+        packet: Raw BLE packet bytes.
+
+    Returns:
+        Base64-encoded ASCII string.
+    """
+    return base64.b64encode(packet).decode("ascii")

custom_components/govee/api/mqtt.py

@@ -19,6 +19,7 @@
 import logging
 import ssl
 import tempfile
+import time
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable
 
@@ -106,6 +107,7 @@ def __init__(
         self._task: asyncio.Task[None] | None = None
         self._temp_dir: tempfile.TemporaryDirectory[str] | None = None
         self._max_backoff_count = 0
+        self._client: Any | None = None  # aiomqtt.Client when connected
 
     @property
     def connected(self) -> bool:
@@ -164,6 +166,7 @@ async def async_stop(self) -> None:
             except Exception:
                 pass
 
+        self._client = None
         self._connected = False
         _LOGGER.info("AWS IoT MQTT client stopped")
 
@@ -257,6 +260,7 @@ async def _connection_loop(self) -> None:
                     keepalive=AWS_IOT_KEEPALIVE,
                     timeout=CONNECTION_TIMEOUT,
                 ) as client:
+                    self._client = client
                     self._connected = True
                     self._max_backoff_count = 0
                     reconnect_interval = RECONNECT_BASE
@@ -276,11 +280,14 @@ async def _connection_loop(self) -> None:
                             break  # type: ignore[unreachable]
                         await self._handle_message(message)
 
+                    self._client = None
+
             except asyncio.CancelledError:
                 _LOGGER.debug("AWS IoT connection loop cancelled")
                 raise
 
             except Exception as err:
+                self._client = None
                 self._connected = False
 
                 if self._running:
@@ -346,3 +353,55 @@ async def _handle_message(self, message: Any) -> None:
             _LOGGER.warning("Failed to parse AWS IoT message: %s", err)
         except Exception as err:
             _LOGGER.error("Error handling AWS IoT message: %s", err)
+
+    async def async_publish_ptreal(
+        self,
+        device_id: str,
+        sku: str,
+        ble_packet_base64: str,
+    ) -> bool:
+        """Publish BLE passthrough command via MQTT.
+
+        Sends a ptReal command to the device to execute a BLE packet.
+        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.
+
+        Returns:
+            True if publish succeeded, False otherwise.
+        """
+        if not self._connected or self._client is None:
+            _LOGGER.warning("Cannot publish ptReal: MQTT not connected")
+            return False
+
+        # Build ptReal payload
+        payload = {
+            "msg": {
+                "cmd": "ptReal",
+                "data": {
+                    "command": [ble_packet_base64],
+                },
+                "cmdVersion": 0,
+                "transaction": f"v_{int(time.time() * 1000)}",
+                "type": 1,
+            }
+        }
+
+        # Publish to account topic (same as subscription topic)
+        topic = self._credentials.account_topic
+
+        try:
+            await self._client.publish(topic, json.dumps(payload))
+            _LOGGER.debug(
+                "Published ptReal to %s for device %s (sku=%s)",
+                topic[:30] + "...",
+                device_id,
+                sku,
+            )
+            return True
+        except Exception as err:
+            _LOGGER.error("Failed to publish ptReal: %s", err)
+            return False