hh-bt-controller/controller/kaiyu_controller.py

"""
kaiyu_controller.py - Python BLE controller for the KAIYU LED hub.

Wraps bleak for async BLE communication with colour, brightness,
and effect helpers tailored to the Hazbin Hotel LEGO display.

Usage (standalone test):
    python controller/kaiyu_controller.py
"""

import asyncio
import sys
from bleak import BleakClient, BleakError

from config import DEVICE_ADDRESS, WRITE_CHARACTERISTIC, WRITE_WITH_RESPONSE, COLORS
from protocol import ActiveProtocol as P


class KaiyuController:
    def __init__(
        self,
        address: str = DEVICE_ADDRESS,
        write_uuid: str = WRITE_CHARACTERISTIC,
        write_response: bool = WRITE_WITH_RESPONSE,
    ):
        self.address = address
        self.write_uuid = write_uuid
        self.write_response = write_response
        self._client: BleakClient | None = None

    # ── Connection ─────────────────────────────────────────────────────────

    async def connect(self) -> bool:
        print(f"Connecting to {self.address}...")
        self._client = BleakClient(self.address)
        try:
            await self._client.connect()
            print(f"Connected. Services: {[s.uuid for s in self._client.services]}")
            # Auto-detect write characteristic if default doesn't exist
            await self._auto_detect_write_uuid()
            return True
        except BleakError as e:
            print(f"Connection failed: {e}")
            return False

    async def disconnect(self):
        if self._client and self._client.is_connected:
            await self._client.disconnect()
            print("Disconnected.")

    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, *_):
        await self.disconnect()

    # ── Internal ───────────────────────────────────────────────────────────

    async def _auto_detect_write_uuid(self):
        """If configured UUID isn't found, pick the first writable characteristic."""
        found_uuids = [
            c.uuid
            for s in self._client.services
            for c in s.characteristics
        ]
        if self.write_uuid in found_uuids:
            return  # Already good

        print(f"Write UUID {self.write_uuid} not found. Searching...")
        for s in self._client.services:
            for c in s.characteristics:
                if "write" in c.properties or "write-without-response" in c.properties:
                    self.write_uuid = c.uuid
                    print(f"Using: {c.uuid}  [{', '.join(c.properties)}]")
                    return
        print("WARNING: No writable characteristic found!")

    async def _send(self, data: bytes):
        if not self._client or not self._client.is_connected:
            raise RuntimeError("Not connected")
        await self._client.write_gatt_char(
            self.write_uuid, data, response=self.write_response
        )

    # ── Controls ───────────────────────────────────────────────────────────

    async def on(self):
        await self._send(P.turn_on())

    async def off(self):
        await self._send(P.turn_off())

    async def set_color(self, r: int, g: int, b: int):
        await self._send(P.set_color(r, g, b))

    async def set_color_name(self, name: str):
        if name not in COLORS:
            raise ValueError(f"Unknown colour '{name}'. Available: {list(COLORS.keys())}")
        await self.set_color(*COLORS[name])

    async def set_brightness(self, level: int):
        """0–100"""
        await self._send(P.set_brightness(level))

    async def set_effect(self, effect_id: int, speed: int = 3):
        await self._send(P.set_effect(effect_id, speed))

    async def fade_to(
        self,
        r2: int, g2: int, b2: int,
        r1: int = 0, g1: int = 0, b1: int = 0,
        steps: int = 20,
        duration: float = 1.0,
    ):
        """Fade from (r1,g1,b1) to (r2,g2,b2) over `duration` seconds."""
        delay = duration / steps
        for i in range(steps + 1):
            t = i / steps
            r = int(r1 + (r2 - r1) * t)
            g = int(g1 + (g2 - g1) * t)
            b = int(b1 + (b2 - b1) * t)
            await self.set_color(r, g, b)
            await asyncio.sleep(delay)

    async def flash(
        self,
        r: int, g: int, b: int,
        times: int = 3,
        on_time: float = 0.2,
        off_time: float = 0.1,
    ):
        """Flash a colour on/off."""
        for _ in range(times):
            await self.set_color(r, g, b)
            await asyncio.sleep(on_time)
            await self.set_color(0, 0, 0)
            await asyncio.sleep(off_time)

    async def pulse(
        self,
        r: int, g: int, b: int,
        cycles: int = 2,
        period: float = 1.0,
        steps: int = 20,
    ):
        """Pulse (breathe) a colour."""
        half = period / 2
        for _ in range(cycles):
            await self.fade_to(r, g, b, 0, 0, 0, steps=steps, duration=half)
            await self.fade_to(0, 0, 0, r, g, b, steps=steps, duration=half)


# ── Standalone test ────────────────────────────────────────────────────────────

async def _test():
    async with KaiyuController() as ctrl:
        print("\n--- Testing basic colours ---")
        for name in ["hellfire_red", "angel_gold", "heaven_blue", "hope_pink", "warm_white", "off"]:
            print(f"  -> {name}")
            await ctrl.set_color_name(name)
            await asyncio.sleep(1.5)

        print("\n--- Fade: off → hope pink ---")
        await ctrl.fade_to(*COLORS["hope_pink"], duration=2.0)
        await asyncio.sleep(1.0)

        print("\n--- Flash: angel gold ---")
        await ctrl.flash(*COLORS["angel_gold"], times=5)

        print("\n--- Pulse: heaven blue ---")
        await ctrl.pulse(*COLORS["heaven_blue"], cycles=2, period=1.5)

        await ctrl.off()
        print("\nTest complete.")


if __name__ == "__main__":
    asyncio.run(_test())