From 35707b88d457aa27ce955f9ad11e345cc860c6d0 Mon Sep 17 00:00:00 2001 From: jessikitty Date: Wed, 1 Jul 2026 20:09:09 +1000 Subject: [PATCH] Fix: scan-then-connect in one flow to avoid device going stale --- scanner/ble_scanner.py | 124 +++++++++++++++++++++++++++++------------ 1 file changed, 89 insertions(+), 35 deletions(-) diff --git a/scanner/ble_scanner.py b/scanner/ble_scanner.py index e96c853..e6cf524 100644 --- a/scanner/ble_scanner.py +++ b/scanner/ble_scanner.py @@ -4,9 +4,13 @@ ble_scanner.py - Discover and dump BLE services/characteristics from nearby devi 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 + python scanner/ble_scanner.py # scan only + python scanner/ble_scanner.py --connect 7C:77:6C:6B:8E:C8 # scan then connect immediately + python scanner/ble_scanner.py --probe 7C:77:6C:6B:8E:C8 # scan then probe with test commands + python scanner/ble_scanner.py --filter KAIYU # scan with name filter + +IMPORTANT: --connect and --probe now re-scan first so the device is still + advertising when we try to connect. Do not scan separately then connect. """ import asyncio @@ -15,6 +19,28 @@ import json from bleak import BleakScanner, BleakClient +async def scan_and_get_device(address: str, timeout: float = 10.0): + """ + Scan for a specific address and return the BLEDevice object while it's + still in the scanner's cache (i.e. still advertising). + Returns None if not found. + """ + print(f"Scanning for {address} ({timeout}s)...") + target = None + + def detection_callback(device, adv): + nonlocal target + if device.address.upper() == address.upper(): + target = device + print(f"Found: {device.address} name={device.name or '(unknown)'} rssi={adv.rssi}") + + scanner = BleakScanner(detection_callback=detection_callback) + await scanner.start() + await asyncio.sleep(timeout) + await scanner.stop() + return target + + 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") @@ -45,12 +71,19 @@ async def scan_devices(filter_name: str = None, timeout: float = 10.0): 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 def inspect_device(address: str, timeout: float = 10.0): + """Scan for device, then immediately connect and dump all services/characteristics.""" - async with BleakClient(address) as client: - print(f"Connected: {client.is_connected}\n") + device = await scan_and_get_device(address, timeout=timeout) + if device is None: + print(f"\nDevice {address} not found during scan.") + print("Make sure it is powered on and not connected to another device (e.g. your phone).") + return + + print(f"\nConnecting to {device.address}...") + + async with BleakClient(device) as client: + print(f"Connected!\n") print("=" * 60) print("SERVICES & CHARACTERISTICS") print("=" * 60) @@ -98,13 +131,12 @@ async def inspect_device(address: str): for descriptor in char.descriptors: print(f" DESC: {descriptor.uuid} handle={descriptor.handle}") - # Save dump to JSON for reference + # Save dump to JSON 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) @@ -115,27 +147,38 @@ async def inspect_device(address: str): print(f" -> {char_uuid} [{', '.join(props)}]") -async def try_known_protocols(address: str): - """Try sending known protocol commands to see what sticks.""" +async def try_known_protocols(address: str, timeout: float = 10.0): + """Scan for device then probe it with known protocol commands.""" + + device = await scan_and_get_device(address, timeout=timeout) + if device is None: + print(f"\nDevice {address} not found during scan.") + print("Make sure it is powered on and not connected to your phone.") + return + 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(" ", "")), + "Turn ON (7e)": bytes.fromhex("7e000400000000ffef"), + "Turn OFF (7e)": bytes.fromhex("7e000400000000 00ef".replace(" ", "")), + "Red (7e)": bytes.fromhex("7e000503ff000000ef"), + "Green (7e)": bytes.fromhex("7e00050300ff0000ef"), + "Blue (7e)": bytes.fromhex("7e0005030000ff00ef"), + "White (7e)": bytes.fromhex("7e000503ffffffff ef".replace(" ", "")), + "Turn ON (56/cc)": bytes.fromhex("cc2333"), + "Turn OFF (56/cc)": bytes.fromhex("cc2433"), + "Red (56)": bytes.fromhex("56ff000000f0aa"), + "Green (56)": bytes.fromhex("5600ff000000f0aa".replace(" ", "")), + "Blue (56)": bytes.fromhex("560000ff00f0aa"), } - async with BleakClient(address) as client: - print(f"Connected to {address}") + print(f"\nConnecting to {device.address}...") + async with BleakClient(device) as client: + print(f"Connected!\n") writable = None for svc in client.services: @@ -146,45 +189,56 @@ async def try_known_protocols(address: str): break if not writable: - # Try any writable characteristic + print("No known UUID found. Available writable characteristics:") 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 + print(f" {char.uuid} [{', '.join(char.properties)}]") + if not writable: + writable = char.uuid + print(f" ^ Using this one") if not writable: - print("No writable characteristic found!") + print("No writable characteristic found at all!") return + print(f"\nSending test commands to: {writable}") + print("Watch the lights and press Enter for each one.\n") + for name, cmd in test_commands.items(): - input(f"\nPress Enter to send '{name}' ({cmd.hex()})...") + input(f" Press Enter to send '{name}' ({cmd.hex()})...") try: await client.write_gatt_char(writable, cmd, response=False) - print(f" Sent OK") + print(f" Sent OK") except Exception as e: - print(f" Error: {e}") + # Try with response=True as fallback + try: + await client.write_gatt_char(writable, cmd, response=True) + print(f" Sent OK (with response)") + except Exception as e2: + print(f" Error: {e2}") 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)") + parser.add_argument("--connect", "-c", help="Scan then connect to address and dump services") + parser.add_argument("--probe", "-p", help="Scan then probe address with known commands") + parser.add_argument("--timeout", "-t", type=float, default=10.0, help="Scan timeout in seconds (default 10)") args = parser.parse_args() if args.connect: - asyncio.run(inspect_device(args.connect)) + asyncio.run(inspect_device(args.connect, timeout=args.timeout)) elif args.probe: - asyncio.run(try_known_protocols(args.probe)) + asyncio.run(try_known_protocols(args.probe, timeout=args.timeout)) 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") + print("\nIMPORTANT: run --connect immediately after the device appears") + print(" in a scan, or it may stop advertising.") asyncio.run(scan_and_pick())