Rewrite scanner: find by service UUID, scan-then-connect, interactive probe
This commit is contained in:
jessikitty committed
1 parent
686bc7dcea
commit
ef90624ef5
1 file changed
+1
-246
+1
-246
@@ -1,246 +1 @@
|
||||
"""
|
||||
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 # 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
|
||||
import argparse
|
||||
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")
|
||||
|
||||
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, timeout: float = 10.0):
|
||||
"""Scan for device, then immediately connect and dump all services/characteristics."""
|
||||
|
||||
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)
|
||||
|
||||
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
|
||||
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}")
|
||||
|
||||
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, 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
|
||||
]
|
||||
|
||||
test_commands = {
|
||||
"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"),
|
||||
}
|
||||
|
||||
print(f"\nConnecting to {device.address}...")
|
||||
async with BleakClient(device) as client:
|
||||
print(f"Connected!\n")
|
||||
|
||||
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:
|
||||
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:
|
||||
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 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" Press Enter to send '{name}' ({cmd.hex()})...")
|
||||
try:
|
||||
await client.write_gatt_char(writable, cmd, response=False)
|
||||
print(f" Sent OK")
|
||||
except Exception as 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="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, timeout=args.timeout))
|
||||
elif 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 <ADDRESS> to inspect a device")
|
||||
print(" run with --probe <ADDRESS> 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())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
IiIiCmJsZV9zY2FubmVyLnB5IC0gU21hcnQgQkxFIHNjYW5uZXIgZm9yIEtBSVlVIExFRCBjb250cm9sbGVyLgoKS2V5IGltcHJvdmVtZW50czoKICAtIEZpbmRzIGRldmljZSBieSBTRVJWSUNFIFVVSUQgKGhhbmRsZXMgcm90YXRpbmcvcmFuZG9tIE1BQ3MpCiAgLSBTY2FuLXRoZW4tY29ubmVjdCBpbiBvbmUgZmxvdyAoZGV2aWNlIHN0YXlzIGluIGNhY2hlKQogIC0gSW50ZXJhY3RpdmUgcHJvYmUgb2YgQUxMIGtub3duIExFRCBwcm90b2NvbCBmYW1pbGllcwogIC0gQXV0by1kZXRlY3RzIHdyaXRlIGNoYXJhY3RlcmlzdGljLCB0cmllcyBhbGwgaWYgbmVlZGVkCiAgLSBEdW1wcyBmdWxsIEdBVFQgdGFibGUgdG8gSlNPTgoKVXNhZ2U6CiAgICBweXRob24gc2Nhbm5lci9ibGVfc2Nhbm5lci5weSAgICAgICAgICAgICAgICAgICAgICMgc2Nhbiwgc2hvdyBhbGwgZGV2aWNlcyArIHNlcnZpY2UgVVVJRHMKICAgIHB5dGhvbiBzY2FubmVyL2JsZV9zY2FubmVyLnB5IC0tcHJvYmUgICAgICAgICAgICAgIyBhdXRvLWZpbmQgTEVEIGRldmljZSBhbmQgcHJvYmUgaXQKICAgIHB5dGhvbiBzY2FubmVyL2JsZV9zY2FubmVyLnB5IC0tcHJvYmUgQUE6QkI6Q0MgICAgIyBwcm9iZSBhIHNwZWNpZmljIGFkZHJlc3MKICAgIHB5dGhvbiBzY2FubmVyL2JsZV9zY2FubmVyLnB5IC0tY29ubmVjdCBBQTpCQjpDQyAgIyBkdW1wIGZ1bGwgR0FUVCB0YWJsZQoiIiIKCmltcG9ydCBhc3luY2lvCmltcG9ydCBhcmdwYXJzZQppbXBvcnQganNvbgppbXBvcnQgb3MKZnJvbSBibGVhayBpbXBvcnQgQmxlYWtTY2FubmVyLCBCbGVha0NsaWVudAoKIyBLbm93biBzZXJ2aWNlIFVVSURzIGZvciBjaGVhcCBCTEUgTEVEIGNvbnRyb2xsZXJzCkxFRF9TRVJWSUNFX1VVSURTID0gewogICAgIjAwMDBmZmUwLTAwMDAtMTAwMC04MDAwLTAwODA1ZjliMzRmYiI6ICJMRURCTEUgLyBNYWdpY0xpZ2h0IiwKICAgICIwMDAwZmZmMC0wMDAwLTEwMDAtODAwMC0wMDgwNWY5YjM0ZmIiOiAiRUxLLUJMRURPTSIsCiAgICAiMDAwMGZmZDUtMDAwMC0xMDAwLTgwMDAtMDA4MDVmOWIzNGZiIjogIlNvbWUgTEVEIGNvbnRyb2xsZXJzIiwKICAgICIwMDAwZmZhMC0wMDAwLTEwMDAtODAwMC0wMDgwNWY5YjM0ZmIiOiAiQ29sb3JpZmljIiwKICAgICIwMDAwZmYwMC0wMDAwLTEwMDAtODAwMC0wMDgwNWY5YjM0ZmIiOiAiR2VuZXJpYyBMRUQiLAogICAgIjAwMDBmZmIwLTAwMDAtMTAwMC04MDAwLTAwODA1ZjliMzRmYiI6ICJTb21lIGJ1bGJzIiwKfQoKIyBLbm93biB3cml0YWJsZSBjaGFyYWN0ZXJpc3RpYyBVVUlEcwpXUklURV9VVUlEUyA9IFsKICAgICIwMDAwZmZlMS0wMDAwLTEwMDAtODAwMC0wMDgwNWY5YjM0ZmIiLAogICAgIjAwMDBmZmYzLTAwMDAtMTAwMC04MDAwLTAwODA1ZjliMzRmYiIsCiAgICAiMDAwMGZmZDktMDAwMC0xMDAwLTgwMDAtMDA4MDVmOWIzNGZiIiwKICAgICIwMDAwZmZlOS0wMDAwLTEwMDAtODAwMC0wMDgwNWY5YjM0ZmIiLAogICAgIjAwMDBmZmIyLTAwMDAtMTAwMC04MDAwLTAwODA1ZjliMzRmYiIsCl0KCiMgVGVzdCBjb21tYW5kczogKGxhYmVsLCBieXRlcykKVEVTVF9DT01NQU5EUyA9IFsKICAgICgiN0UgIFBvd2VyIE9OIiwgIGJ5dGVzLmZyb21oZXgoIjdlMDAwNDAwMDAwMDAwZmZlZiIpKSwKICAgICgiN0UgIFBvd2VyIE9GRiIsIGJ5dGVzLmZyb21oZXgoIjdlMDAwNDAwMDAwMDAwMDBlZiIpKSwKICAgICgiN0UgIFJlZCIsICAgICAgIGJ5dGVzLmZyb21oZXgoIjdlMDAwNTAzZmYwMDAwMDBlZiIpKSwKICAgICgiN0UgIEdyZWVuIiwgICAgIGJ5dGVzLmZyb21oZXgoIjdlMDAwNTAzMDBmZjAwMDBlZiIpKSwKICAgICgiN0UgIEJsdWUiLCAgICAgIGJ5dGVzLmZyb21oZXgoIjdlMDAwNTAzMDAwMGZmMDBlZiIpKSwKICAgICgiN0UgIFdoaXRlIiwgICAgIGJ5dGVzLmZyb21oZXgoIjdlMDAwNTAzZmZmZmZmZmZlZiIpKSwKICAgICgiN0UgIEJyaSA1MCUiLCAgIGJ5dGVzLmZyb21oZXgoIjdlMDAwMTMyMDAwMDAwZmZlZiIpKSwKICAgICgiNTYgIFBvd2VyIE9OIiwgIGJ5dGVzLmZyb21oZXgoImNjMjMzMyIpKSwKICAgICgiNTYgIFBvd2VyIE9GRiIsIGJ5dGVzLmZyb21oZXgoImNjMjQzMyIpKSwKICAgICgiNTYgIFJlZCIsICAgICAgIGJ5dGVzLmZyb21oZXgoIjU2ZmYwMDAwMDBmMGFhIikpLAogICAgKCI1NiAgR3JlZW4iLCAgICAgYnl0ZXMuZnJvbWhleCgiNTYwMGZmMDAwMGYwYWEiKSksCiAgICAoIjU2ICBCbHVlIiwgICAgICBieXRlcy5mcm9taGV4KCI1NjAwMDBmZjAwZjBhYSIpKSwKXQoKCmFzeW5jIGRlZiBzY2FuX2FsbCh0aW1lb3V0OiBmbG9hdCA9IDEwLjApOgogICAgcHJpbnQoZiJTY2FubmluZyAoe3RpbWVvdXR9cykuLi5cbiIpCiAgICBkZXZpY2VzID0gYXdhaXQgQmxlYWtTY2FubmVyLmRpc2NvdmVyKHRpbWVvdXQ9dGltZW91dCwgcmV0dXJuX2Fkdj1UcnVlKQogICAgZm91bmQgPSBsaXN0KGRldmljZXMudmFsdWVzKCkpCiAgICBwcmludChmInsnIyc6PDR9IHsnTmFtZSc6PDI4fSB7J0FkZHJlc3MnOjwyMn0geydSU1NJJzo8Nn0geydTZXJ2aWNlcyd9IikKICAgIHByaW50KCItIiAqIDkwKQogICAgZm9yIGksIChkZXYsIGFkdikgaW4gZW51bWVyYXRlKGZvdW5kKToKICAgICAgICBuYW1lID0gZGV2Lm5hbWUgb3IgIih1bmtub3duKSIKICAgICAgICBzdmNzID0gW10KICAgICAgICBmb3IgdSBpbiAoYWR2LnNlcnZpY2VfdXVpZHMgb3IgW10pOgogICAgICAgICAgICBsYWJlbCA9IExFRF9TRVJWSUNFX1VVSURTLmdldCh1Lmxvd2VyKCksICIiKQogICAgICAgICAgICBzdmNzLmFwcGVuZChmInt1Wzo4XX0uLi4ge2xhYmVsfSIgaWYgbGFiZWwgZWxzZSB1WzoyMF0pCiAgICAgICAgZXh0cmEgPSAiIHwgIi5qb2luKHN2Y3MpWzo1NV0gaWYgc3ZjcyBlbHNlICIiCiAgICAgICAgaWYgbm90IGV4dHJhIGFuZCBhZHYubWFudWZhY3R1cmVyX2RhdGE6CiAgICAgICAgICAgIGV4dHJhID0gIiAiLmpvaW4oZiIweHtrOjA0eH06e3YuaGV4KClbOjhdfSIgZm9yIGssIHYgaW4gYWR2Lm1hbnVmYWN0dXJlcl9kYXRhLml0ZW1zKCkpWzo1NV0KICAgICAgICBwcmludChmIntpOjw0fSB7bmFtZTo8Mjh9IHtkZXYuYWRkcmVzczo8MjJ9IHthZHYucnNzaTo8Nn0ge2V4dHJhfSIpCiAgICBwcmludCgpCiAgICByZXR1cm4gZm91bmQKCgphc3luYyBkZWYgZmluZF9sZWRfZGV2aWNlKHRpbWVvdXQ6IGZsb2F0ID0gMTIuMCk6CiAgICAiIiJTY2FuIGFuZCByZXR1cm4gZmlyc3QgZGV2aWNlIGFkdmVydGlzaW5nIGEga25vd24gTEVEIHNlcnZpY2UgVVVJRC4iIiIKICAgIHByaW50KGYiTG9va2luZyBmb3IgTEVEIGNvbnRyb2xsZXIgYnkgc2VydmljZSBVVUlEICh7dGltZW91dH1zKS4uLiIpCiAgICBmb3VuZF9kZXYsIGZvdW5kX2FkdiA9IE5vbmUsIE5vbmUKCiAgICBkZWYgY2IoZGV2aWNlLCBhZHYpOgogICAgICAgIG5vbmxvY2FsIGZvdW5kX2RldiwgZm91bmRfYWR2CiAgICAgICAgaWYgZm91bmRfZGV2OgogICAgICAgICAgICByZXR1cm4KICAgICAgICBmb3Igc3ZjIGluIChhZHYuc2VydmljZV91dWlkcyBvciBbXSk6CiAgICAgICAgICAgIGlmIHN2Yy5sb3dlcigpIGluIExFRF9TRVJWSUNFX1VVSURTOgogICAgICAgICAgICAgICAgbGFiZWwgPSBMRURfU0VSVklDRV9VVUlEU1tzdmMubG93ZXIoKV0KICAgICAgICAgICAgICAgIHByaW50KGYiICBGb3VuZCBieSBzZXJ2aWNlIFVVSUQ6IHtkZXZpY2UuYWRkcmVzc30gIFt7bGFiZWx9XSIpCiAgICAgICAgICAgICAgICBmb3VuZF9kZXYsIGZvdW5kX2FLine truncated
|
||||
Reference in new issue
Block a user