""" ble_scanner.py - Discover and dump BLE services/characteristics from nearby devices. Run this first to find your KAIYU LED controller and identify its BLE services. Usage: python scanner/ble_scanner.py python scanner/ble_scanner.py --filter KAIYU python scanner/ble_scanner.py --connect AA:BB:CC:DD:EE:FF """ import asyncio import argparse import json from bleak import BleakScanner, BleakClient async def scan_devices(filter_name: str = None, timeout: float = 10.0): """Scan for nearby BLE devices and print them.""" print(f"Scanning for BLE devices ({timeout}s)...\n") devices = await BleakScanner.discover(timeout=timeout, return_adv=True) found = [] for device, adv in devices.values(): name = device.name or "(unknown)" if filter_name and filter_name.lower() not in name.lower(): continue found.append((device, adv)) if not found: print("No devices found." if not filter_name else f"No devices matching '{filter_name}' found.") return [] print(f"{'#':<4} {'Name':<30} {'Address':<20} {'RSSI':<8} {'Manufacturer'}") print("-" * 80) for i, (device, adv) in enumerate(found): name = device.name or "(unknown)" mfr = "" if adv.manufacturer_data: mfr = " ".join(f"{k:#06x}:{v.hex()}" for k, v in adv.manufacturer_data.items()) print(f"{i:<4} {name:<30} {device.address:<20} {adv.rssi:<8} {mfr}") print() return found async def inspect_device(address: str): """Connect to a device and dump all its services and characteristics.""" print(f"Connecting to {address}...") async with BleakClient(address) as client: print(f"Connected: {client.is_connected}\n") print("=" * 60) print("SERVICES & CHARACTERISTICS") print("=" * 60) dump = {} for service in client.services: print(f"\nSERVICE: {service.uuid}") print(f" Description: {service.description}") dump[service.uuid] = { "description": service.description, "characteristics": {} } for char in service.characteristics: props = ", ".join(char.properties) print(f" CHAR: {char.uuid}") print(f" Properties : {props}") print(f" Handle : {char.handle}") value_hex = None value_str = None if "read" in char.properties: try: value = await client.read_gatt_char(char.uuid) value_hex = value.hex() try: value_str = value.decode("utf-8") except Exception: value_str = None print(f" Value (hex): {value_hex}") if value_str: print(f" Value (str): {value_str}") except Exception as e: print(f" Value : (read error: {e})") dump[service.uuid]["characteristics"][char.uuid] = { "properties": list(char.properties), "handle": char.handle, "value_hex": value_hex, "value_str": value_str, } for descriptor in char.descriptors: print(f" DESC: {descriptor.uuid} handle={descriptor.handle}") # Save dump to JSON for reference dump_file = f"scanner/device_{address.replace(':', '-')}.json" with open(dump_file, "w") as f: json.dump(dump, f, indent=2) print(f"\nDump saved to {dump_file}") # Identify likely write characteristic print("\n" + "=" * 60) print("LIKELY COMMAND CHARACTERISTICS (writable):") print("=" * 60) for svc_uuid, svc in dump.items(): for char_uuid, char in svc["characteristics"].items(): props = char["properties"] if "write" in props or "write-without-response" in props: print(f" -> {char_uuid} [{', '.join(props)}]") async def try_known_protocols(address: str): """Try sending known protocol commands to see what sticks.""" KNOWN_WRITE_UUIDS = [ "0000ffe1-0000-1000-8000-00805f9b34fb", # LEDBLE / MagicLight "0000fff3-0000-1000-8000-00805f9b34fb", # ELK-BLEDOM "0000ffd9-0000-1000-8000-00805f9b34fb", # Some controllers ] # Try the 0x7e command format (common in cheap controllers) test_commands = { "Turn ON": bytes.fromhex("7e000400000000ff ef".replace(" ", "")), "Turn OFF": bytes.fromhex("7e000400000000 00 ef".replace(" ", "")), "Red": bytes.fromhex("7e0005 03 ff 00 00 00 ef".replace(" ", "")), "Green": bytes.fromhex("7e0005 03 00 ff 00 00 ef".replace(" ", "")), "Blue": bytes.fromhex("7e0005 03 00 00 ff 00 ef".replace(" ", "")), "White": bytes.fromhex("7e0005 03 ff ff ff 00 ef".replace(" ", "")), "Warm Hellfire": bytes.fromhex("7e0005 03 ff 40 00 00 ef".replace(" ", "")), } async with BleakClient(address) as client: print(f"Connected to {address}") writable = None for svc in client.services: for char in svc.characteristics: if char.uuid in KNOWN_WRITE_UUIDS: writable = char.uuid print(f"Found known write characteristic: {char.uuid}") break if not writable: # Try any writable characteristic for svc in client.services: for char in svc.characteristics: if "write" in char.properties or "write-without-response" in char.properties: writable = char.uuid print(f"Using writable characteristic: {char.uuid}") break if not writable: print("No writable characteristic found!") return for name, cmd in test_commands.items(): input(f"\nPress Enter to send '{name}' ({cmd.hex()})...") try: await client.write_gatt_char(writable, cmd, response=False) print(f" Sent OK") except Exception as e: print(f" Error: {e}") def main(): parser = argparse.ArgumentParser(description="KAIYU BLE Scanner & Inspector") parser.add_argument("--filter", "-f", help="Filter devices by name") parser.add_argument("--connect", "-c", help="Connect to device by MAC address and dump services") parser.add_argument("--probe", "-p", help="Probe device with known commands") parser.add_argument("--timeout", "-t", type=float, default=10.0, help="Scan timeout (default 10s)") args = parser.parse_args() if args.connect: asyncio.run(inspect_device(args.connect)) elif args.probe: asyncio.run(try_known_protocols(args.probe)) else: async def scan_and_pick(): found = await scan_devices(filter_name=args.filter, timeout=args.timeout) if found: print("Tip: run with --connect
to inspect a device") print(" run with --probe
to test known protocols") asyncio.run(scan_and_pick()) if __name__ == "__main__": main()