From 11a08a3fd6ba4b8f4f7f4f8684f002e5ffe66cc1 Mon Sep 17 00:00:00 2001 From: jessikitty Date: Wed, 1 Jul 2026 19:53:30 +1000 Subject: [PATCH] Add BLE scanner tool --- scanner/ble_scanner.py | 192 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 scanner/ble_scanner.py diff --git a/scanner/ble_scanner.py b/scanner/ble_scanner.py new file mode 100644 index 0000000..e96c853 --- /dev/null +++ b/scanner/ble_scanner.py @@ -0,0 +1,192 @@ +""" +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()