From 2b868e4ad6d608cc5c0686ff8c5ac00628708217 Mon Sep 17 00:00:00 2001 From: jessikitty Date: Wed, 1 Jul 2026 19:54:31 +1000 Subject: [PATCH] Add main BLE controller class --- controller/kaiyu_controller.py | 179 +++++++++++++++++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 controller/kaiyu_controller.py diff --git a/controller/kaiyu_controller.py b/controller/kaiyu_controller.py new file mode 100644 index 0000000..e5dc952 --- /dev/null +++ b/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())