Add main BLE controller class

controller/kaiyu_controller.py

@@ -0,0 +1,179 @@
+"""
+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())