sp656e.py

#!/usr/bin/env python3
"""CLI tool to control SP656E LED controller over BLE."""

import argparse
import asyncio
import sys
from dataclasses import dataclass
from typing import Optional

from bleak import BleakClient, BleakScanner
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData

# BLE UUIDs
SERVICE_UUID = "0000ffe0-0000-1000-8000-00805f9b34fb"
CHAR_UUID = "0000ffe1-0000-1000-8000-00805f9b34fb"

# Vendor-specific UUIDs (some SP6xxE use these instead)
VENDOR_SERVICE_UUID = "5833ff01-9b8b-5191-6142-22a4536ef123"
VENDOR_WRITE_UUID = "5833ff02-9b8b-5191-6142-22a4536ef123"
VENDOR_NOTIFY_UUID = "5833ff03-9b8b-5191-6142-22a4536ef123"

# BanlanX manufacturer ID
MANUFACTURER_ID = 20563  # 0x5053

# Protocol constants
HEADER = 0x53
KEY = 0x00

# Commands
CMD_QUERY = 0x02
CMD_POWER = 0x50
CMD_BRIGHTNESS = 0x51
CMD_COLOR_STATIC = 0x52
CMD_MODE_EFFECT = 0x53
CMD_SPEED = 0x54
CMD_LENGTH = 0x55
CMD_DIRECTION = 0x56
CMD_COLOR_DYNAMIC = 0x57
CMD_LOOP = 0x58
CMD_AUDIO_INPUT = 0x59
CMD_AUDIO_GAIN = 0x5A
CMD_PLAY_PAUSE = 0x5D

# Modes
MODES = {
    0x01: "Static Color",
    0x02: "Static White",
    0x03: "Dynamic Color",
    0x04: "Dynamic White",
    0x05: "Sound Color",
    0x06: "Sound White",
    0x07: "Custom",
}

# Dynamic color effects (SPI)
DYNAMIC_COLOR_EFFECTS = {
    0x01: "Rainbow", 0x02: "Rainbow Meteor", 0x03: "Rainbow Comet",
    0x04: "Rainbow Segment", 0x05: "Rainbow Wave", 0x06: "Rainbow Jump",
    0x07: "Rainbow Stars", 0x08: "Rainbow Spin",
    0x09: "Red/Yellow Fire", 0x0A: "Red/Purple Fire",
    0x0B: "Green/Yellow Fire", 0x0C: "Green/Cyan Fire",
    0x0D: "Blue/Purple Fire", 0x0E: "Blue/Cyan Fire",
    0x0F: "Red Comet", 0x10: "Green Comet", 0x11: "Blue Comet",
    0x12: "Yellow Comet", 0x13: "Cyan Comet", 0x14: "Purple Comet",
    0x15: "White Comet",
    0x16: "Red Meteor", 0x17: "Green Meteor", 0x18: "Blue Meteor",
    0x19: "Yellow Meteor", 0x1A: "Cyan Meteor", 0x1B: "Purple Meteor",
    0x1C: "White Meteor",
    0x4E: "Red Stars", 0x4F: "Green Stars", 0x50: "Blue Stars",
    0x51: "Yellow Stars", 0x52: "Cyan Stars", 0x53: "Purple Stars",
    0x54: "White Stars",
    0x62: "Red Breath", 0x63: "Green Breath", 0x64: "Blue Breath",
    0x65: "Yellow Breath", 0x66: "Cyan Breath", 0x67: "Purple Breath",
    0x68: "White Breath",
    0x92: "Gradient",
}

# Dynamic white effects (SPI)
DYNAMIC_WHITE_EFFECTS = {
    0x01: "White Breath", 0x02: "White Stars", 0x03: "White Meteor",
    0x04: "White Comet Spin", 0x05: "White Dot Spin",
    0x06: "White Segment Spin", 0x07: "White Chasing Dots",
    0x08: "White Comet", 0x09: "White Wave", 0x0A: "White Stacking",
}

# Sound color effects
SOUND_COLOR_EFFECTS = {
    0x01: "Full Color Rhythm Spectrum", 0x02: "Single Color Rhythm Spectrum",
    0x03: "Full Color Rhythm Stars", 0x04: "Single Color Rhythm Stars",
    0x05: "Gradient Energy", 0x06: "Single Color Energy",
    0x07: "Gradient Pulse", 0x08: "Single Color Pulse",
    0x09: "Full Color Eject Fwd", 0x0A: "Single Color Eject Fwd",
    0x0B: "Full Color Eject Bwd", 0x0C: "Single Color Eject Bwd",
    0x0D: "Full Color VuMeter", 0x0E: "Single Color VuMeter",
    0x0F: "Love & Peace", 0x10: "Christmas",
    0x11: "Heartbeat", 0x12: "Party",
}


def encode(cmd: int, data: list[int]) -> bytearray:
    """Encode a command in the 0x53 protocol format."""
    length = len(data) & 0xFF
    msg = bytearray([HEADER, cmd & 0xFF, KEY, 0x01, 0x00, length])
    msg.extend(data)
    return msg


def encode_a0(cmd: int, data: list[int]) -> bytearray:
    """Encode a command in the 0xA0 protocol format (older devices)."""
    length = len(data) & 0xFF
    msg = bytearray([0xA0, cmd & 0xFF, length])
    msg.extend(data)
    return msg


@dataclass
class DeviceState:
    firmware: str = ""
    power: bool = False
    mode: int = 0
    effect: int = 0
    color_brightness: int = 0
    white_brightness: int = 0
    r: int = 0
    g: int = 0
    b: int = 0
    speed: int = 0
    length: int = 0
    direction: bool = False
    audio_gain: int = 0
    audio_input: int = 0
    loop: bool = False
    playing: bool = False

    def mode_name(self) -> str:
        return MODES.get(self.mode, f"Unknown(0x{self.mode:02x})")

    def effect_name(self) -> str:
        effects = {
            0x03: DYNAMIC_COLOR_EFFECTS,
            0x04: DYNAMIC_WHITE_EFFECTS,
            0x05: SOUND_COLOR_EFFECTS,
        }
        table = effects.get(self.mode, {})
        return table.get(self.effect, f"#{self.effect}")

    def __str__(self) -> str:
        lines = [
            f"  Power:      {'ON' if self.power else 'OFF'}",
            f"  Mode:       {self.mode_name()}",
            f"  Effect:     {self.effect_name()}",
            f"  Color:      rgb({self.r}, {self.g}, {self.b})",
            f"  Brightness: {self.color_brightness} (color) / {self.white_brightness} (white)",
            f"  Speed:      {self.speed}",
            f"  Length:     {self.length}",
            f"  Direction:  {'Reverse' if self.direction else 'Forward'}",
            f"  Loop:       {'ON' if self.loop else 'OFF'}",
            f"  Playing:    {'Yes' if self.playing else 'Paused'}",
            f"  Audio:      input={self.audio_input} gain={self.audio_gain}",
        ]
        if self.firmware:
            lines.insert(0, f"  Firmware:   {self.firmware}")
        return "\n".join(lines)


def parse_state(data: bytes) -> Optional[DeviceState]:
    """Parse a status notification response (requires at least 47 bytes)."""
    if len(data) < 47:
        return None

    state = DeviceState()

    try:
        state.firmware = bytes(data[11:18]).decode("ascii", errors="replace").strip("\x00")
    except Exception:
        pass

    state.power = data[29] > 0
    state.loop = bool(data[30])
    state.mode = data[32]
    state.effect = data[33]
    state.playing = bool(data[34])
    state.color_brightness = data[35]
    state.white_brightness = data[36]
    state.r = data[37]
    state.g = data[38]
    state.b = data[39]
    state.speed = data[42]
    state.length = data[43]
    state.direction = bool(data[44])
    state.audio_gain = data[45]
    state.audio_input = data[46]

    return state


async def find_device(timeout: float = 10.0) -> Optional[BLEDevice]:
    """Scan for SP656E / BanlanX devices."""
    print(f"Scanning for BanlanX devices ({timeout}s)...")
    found: Optional[BLEDevice] = None

    def callback(device: BLEDevice, adv: AdvertisementData) -> None:
        nonlocal found
        if found:
            return  # Already found, skip duplicates
        name = adv.local_name or device.name or ""
        name_lower = name.lower()

        # Check manufacturer data for BanlanX ID
        if MANUFACTURER_ID in adv.manufacturer_data:
            mfr = adv.manufacturer_data[MANUFACTURER_ID]
            print(f"  Found BanlanX device: {name} [{device.address}]")
            print(f"    Manufacturer data: {mfr.hex()}")
            if len(mfr) >= 2:
                print(f"    Model ID: 0x{mfr[0]:02x}, Family byte: 0x{mfr[1]:02x}")
            found = device
            return

        # Also check by name patterns
        if any(kw in name_lower for kw in ["sp656", "sp6", "banlan", "spled"]):
            print(f"  Found by name: {name} [{device.address}]")
            found = device
            return

        # Check for the known service UUID
        if SERVICE_UUID.lower() in [str(s).lower() for s in adv.service_uuids]:
            print(f"  Found by service UUID: {name or 'Unknown'} [{device.address}]")
            found = device

    scanner = BleakScanner(detection_callback=callback)
    await scanner.start()
    # Stop early once device is found
    elapsed = 0.0
    while elapsed < timeout and not found:
        await asyncio.sleep(0.2)
        elapsed += 0.2
    await scanner.stop()

    if not found:
        # List all discovered devices for debugging
        devices = scanner.discovered_devices_and_advertisement_data
        if devices:
            print("\nAll discovered BLE devices:")
            for dev, adv in sorted(devices.values(), key=lambda x: x[1].rssi or -100, reverse=True):
                name = adv.local_name or dev.name or "Unknown"
                rssi = adv.rssi or "?"
                mfr_keys = list(adv.manufacturer_data.keys())
                svcs = [str(s) for s in adv.service_uuids]
                extra = ""
                if mfr_keys:
                    extra += f" mfr_ids={mfr_keys}"
                if svcs:
                    extra += f" svcs={svcs}"
                print(f"  {name:30s} [{dev.address}] RSSI={rssi}{extra}")

    return found


async def connect_and_run(
    address: Optional[str],
    action: str,
    args: argparse.Namespace,
) -> None:
    """Connect to device and execute the requested action."""
    if address:
        print(f"Connecting to {address}...")
        device_or_addr = address
    else:
        device = await find_device(timeout=args.scan_timeout)
        if not device:
            print("No BanlanX device found. Make sure:")
            print("  1. The SP656E is powered on")
            print("  2. BanlanX app is disconnected (only 1 BLE connection allowed)")
            print("  3. Bluetooth is enabled on this Mac")
            print("  4. Try: python3 sp656e.py --scan-timeout 20 scan")
            sys.exit(1)
        device_or_addr = device
        address = device.address

    state_response: Optional[bytearray] = None
    state_event = asyncio.Event()

    def notification_handler(sender: int, data: bytearray) -> None:
        nonlocal state_response
        if len(data) > 1 and data[0] == HEADER:
            state_response = data
            state_event.set()

    async with BleakClient(device_or_addr) as client:
        print(f"Connected to {address}")

        # Discover which characteristic to use
        write_char = None
        notify_char = None

        for service in client.services:
            for char in service.characteristics:
                char_uuid = str(char.uuid).lower()
                if char_uuid == CHAR_UUID.lower():
                    if "write" in char.properties or "write-without-response" in char.properties:
                        write_char = char.uuid
                    if "notify" in char.properties:
                        notify_char = char.uuid
                elif char_uuid == VENDOR_WRITE_UUID.lower():
                    if "write" in char.properties or "write-without-response" in char.properties:
                        write_char = write_char or char.uuid
                elif char_uuid == VENDOR_NOTIFY_UUID.lower():
                    if "notify" in char.properties:
                        notify_char = notify_char or char.uuid

        if not write_char:
            print("Could not find writable characteristic. Available services:")
            for service in client.services:
                print(f"  Service: {service.uuid}")
                for char in service.characteristics:
                    print(f"    Char: {char.uuid} props={char.properties}")
            sys.exit(1)

        print(f"  Write char: {write_char}")
        if notify_char:
            print(f"  Notify char: {notify_char}")
            await client.start_notify(notify_char, notification_handler)

        async def send(cmd: int, data: list[int], use_a0: bool = False) -> None:
            if use_a0:
                msg = encode_a0(cmd, data)
            else:
                msg = encode(cmd, data)
            print(f"  TX: {msg.hex()}")
            try:
                await client.write_gatt_char(write_char, msg, response=True)
            except Exception as e:
                print(f"  Write with response failed ({e}), retrying without response...")
                await client.write_gatt_char(write_char, msg, response=False)

        async def query_state() -> Optional[DeviceState]:
            nonlocal state_response
            state_response = None
            state_event.clear()
            await send(CMD_QUERY, [0x01])
            try:
                await asyncio.wait_for(state_event.wait(), timeout=3.0)
            except asyncio.TimeoutError:
                # Try A0 protocol
                print("  No response from 0x53 protocol, trying 0xA0...")
                state_event.clear()
                await send(0x70, [], use_a0=True)
                try:
                    await asyncio.wait_for(state_event.wait(), timeout=3.0)
                except asyncio.TimeoutError:
                    print("  No status response received")
                    return None
            if state_response:
                print(f"  RX: {state_response.hex()}")
                return parse_state(state_response)
            return None

        # Execute the requested action
        if action == "scan":
            print("Scan complete - device found and connected.")
            print("\nServices and characteristics:")
            for service in client.services:
                print(f"  Service: {service.uuid}")
                for char in service.characteristics:
                    print(f"    {char.uuid} [{', '.join(char.properties)}]")
                    for desc in char.descriptors:
                        print(f"      Descriptor: {desc.uuid}")

            # Read standard GATT info
            for name, uuid in [
                ("Manufacturer", "00002a29-0000-1000-8000-00805f9b34fb"),
                ("Model", "00002a24-0000-1000-8000-00805f9b34fb"),
                ("Firmware", "00002a26-0000-1000-8000-00805f9b34fb"),
                ("Hardware", "00002a27-0000-1000-8000-00805f9b34fb"),
            ]:
                try:
                    val = await client.read_gatt_char(uuid)
                    print(f"  {name}: {val.decode('utf-8', errors='replace')}")
                except Exception:
                    pass

        elif action == "status":
            state = await query_state()
            if state:
                print(f"\nDevice status:\n{state}")
            else:
                print("Could not read device status")

        elif action == "on":
            await send(CMD_POWER, [0x01])
            print("Power ON")

        elif action == "off":
            await send(CMD_POWER, [0x00])
            print("Power OFF")

        elif action == "color":
            r, g, b = args.r, args.g, args.b
            level = args.level if args.level is not None else 255
            # Set static color mode first
            await send(CMD_MODE_EFFECT, [0x01, 0x00])
            await asyncio.sleep(0.4)
            await send(CMD_COLOR_STATIC, [r, g, b, level])
            print(f"Color: rgb({r}, {g}, {b}) brightness={level}")

        elif action == "brightness":
            level = args.level
            # 0x00 = color channel brightness
            await send(CMD_BRIGHTNESS, [0x00, level])
            print(f"Brightness: {level}")

        elif action == "effect":
            mode = args.mode
            effect_id = args.effect_id
            await send(CMD_MODE_EFFECT, [mode, effect_id])
            mode_name = MODES.get(mode, f"0x{mode:02x}")
            effects_table = {
                0x03: DYNAMIC_COLOR_EFFECTS,
                0x04: DYNAMIC_WHITE_EFFECTS,
                0x05: SOUND_COLOR_EFFECTS,
            }
            effect_name = effects_table.get(mode, {}).get(effect_id, f"#{effect_id}")
            print(f"Effect: {mode_name} / {effect_name}")

        elif action == "speed":
            await send(CMD_SPEED, [args.value])
            print(f"Speed: {args.value}")

        elif action == "length":
            await send(CMD_LENGTH, [args.value])
            print(f"Length: {args.value}")

        elif action == "direction":
            d = 0x01 if args.reverse else 0x00
            await send(CMD_DIRECTION, [d])
            print(f"Direction: {'Reverse' if args.reverse else 'Forward'}")

        elif action == "loop":
            val = 0x01 if args.enable else 0x00
            await send(CMD_LOOP, [val])
            print(f"Loop: {'ON' if args.enable else 'OFF'}")

        elif action == "raw":
            # Send raw bytes for experimentation
            raw_bytes = bytes.fromhex(args.hex_data)
            print(f"  TX (raw): {raw_bytes.hex()}")
            await client.write_gatt_char(write_char, raw_bytes, response=False)
            if notify_char:
                await asyncio.sleep(2.0)
                if state_response:
                    print(f"  RX: {state_response.hex()}")

        if notify_char:
            await client.stop_notify(notify_char)


def parse_color(s: str) -> tuple[int, int, int]:
    """Parse color from hex (#ff0000) or r,g,b format."""
    s = s.strip().lstrip("#")
    if "," in s:
        parts = [int(x.strip()) for x in s.split(",")]
        if len(parts) != 3:
            raise ValueError(f"Expected 3 color components, got {len(parts)}")
        r, g, b = parts[0], parts[1], parts[2]
    elif len(s) == 6:
        r, g, b = int(s[0:2], 16), int(s[2:4], 16), int(s[4:6], 16)
    else:
        raise ValueError(f"Invalid color: {s}. Use #rrggbb or r,g,b")
    if not all(0 <= v <= 255 for v in (r, g, b)):
        raise ValueError(f"Color components must be 0-255, got ({r}, {g}, {b})")
    return (r, g, b)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Control SP656E LED controller over BLE",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""Examples:
  %(prog)s scan                    # Find and inspect device
  %(prog)s status                  # Read current state
  %(prog)s on                      # Turn on
  %(prog)s off                     # Turn off
  %(prog)s color ff0000            # Set red
  %(prog)s color 0,255,0           # Set green (r,g,b)
  %(prog)s color 0000ff --level 128  # Blue at 50%% brightness
  %(prog)s brightness 200          # Set brightness (0-255)
  %(prog)s effect 3 1              # Dynamic color: Rainbow
  %(prog)s effect 5 18             # Sound: Party
  %(prog)s speed 5                 # Effect speed (1-10)
  %(prog)s effects                 # List all effects
  %(prog)s raw "53 50 00 01 00 01 01"  # Send raw hex bytes
""",
    )
    parser.add_argument("--address", "-a", help="BLE MAC address (skip scan)")
    parser.add_argument("--scan-timeout", type=float, default=10.0, help="Scan timeout in seconds")

    sub = parser.add_subparsers(dest="action", required=True)

    sub.add_parser("scan", help="Scan for device and show info")
    sub.add_parser("status", help="Read device status")
    sub.add_parser("on", help="Turn on")
    sub.add_parser("off", help="Turn off")

    color_p = sub.add_parser("color", help="Set static color")
    color_p.add_argument("value", help="Color: #rrggbb or r,g,b")
    color_p.add_argument("--level", type=int, default=None, help="Brightness 0-255")

    bright_p = sub.add_parser("brightness", help="Set brightness")
    bright_p.add_argument("level", type=int, help="Brightness 0-255")

    effect_p = sub.add_parser("effect", help="Set effect (mode + effect_id)")
    effect_p.add_argument("mode", type=int, help="Mode: 3=dynamic color, 4=dynamic white, 5=sound color")
    effect_p.add_argument("effect_id", type=int, help="Effect ID (use 'effects' to list)")

    speed_p = sub.add_parser("speed", help="Set effect speed")
    speed_p.add_argument("value", type=int, choices=range(1, 11), help="Speed 1-10")

    length_p = sub.add_parser("length", help="Set effect length")
    length_p.add_argument("value", type=int, choices=range(1, 151), help="Length 1-150", metavar="1-150")

    dir_p = sub.add_parser("direction", help="Set effect direction")
    dir_p.add_argument("--reverse", action="store_true", help="Reverse direction")

    loop_p = sub.add_parser("loop", help="Toggle effect loop")
    loop_p.add_argument("--enable", action="store_true", help="Enable loop")

    sub.add_parser("effects", help="List all available effects")

    raw_p = sub.add_parser("raw", help="Send raw hex bytes")
    raw_p.add_argument("hex_data", help='Hex string, e.g. "53 50 00 01 00 01 01"')

    args = parser.parse_args()

    # Parse color arg
    if args.action == "color":
        r, g, b = parse_color(args.value)
        args.r, args.g, args.b = r, g, b

    if args.action == "effects":
        print("\nDynamic Color Effects (mode=3):")
        for eid, name in sorted(DYNAMIC_COLOR_EFFECTS.items()):
            print(f"  {eid:3d} (0x{eid:02x})  {name}")
        print("\nDynamic White Effects (mode=4):")
        for eid, name in sorted(DYNAMIC_WHITE_EFFECTS.items()):
            print(f"  {eid:3d} (0x{eid:02x})  {name}")
        print("\nSound Color Effects (mode=5):")
        for eid, name in sorted(SOUND_COLOR_EFFECTS.items()):
            print(f"  {eid:3d} (0x{eid:02x})  {name}")
        return

    asyncio.run(connect_and_run(args.address, args.action, args))


if __name__ == "__main__":
    main()