"""
ble_gatt_test.py (v10) - repeat connect + GATT discovery against a Pybricks hub
and tally failures. Built to test the "Characteristic 2a26 was not found"
failure seen with pybricksdev on Windows, and to compare Windows cached vs
uncached service discovery. Optionally also times write-with-response round
trips (the operation that stalls in Pybricks Code downloads).
Usage:
python ble_gatt_test.py -n 40 --modes default uncached --csv run1.csv
python ble_gatt_test.py -n 20 --pause 30 --csv gap30.csv
python ble_gatt_test.py -n 10 --prompt-between --read-fw --csv cold.csv
Modes are used round-robin per iteration (A, B, A, B, ...) so slow drift in the
radio environment hits every mode equally.
default : no winrt option, Bleak/Windows default behaviour
cached : winrt use_cached_services=True
uncached : winrt use_cached_services=False (what Chrome does: allow_cache:0)
v10: sub_writes result -- subscribed run's hub kept a steady (still
"connected") light and vanished from scanning afterward, unlike every prior
disconnect without a subscription. Working theory: disconnecting a --fresh
child process that never unsubscribes/never pauses before exit may not let
the BLE teardown complete over the air, leaving the hub believing it is still
connected. Fix: explicitly stop_notify before disconnect, plus a short pause
(--teardown-pause, default 1.0s) after disconnect before the process exits.
If the hub still gets stuck even with this, that's a stronger, cleaner
finding (points at the hub/adapter, not just script hygiene) worth testing
against the real UI/pybricksdev too.
v9: --subscribe (usable with --writes, any mode except pdev*) subscribes to
Pybricks notifications right after connecting, before the write test -- the
--hold result shows a bare connect+writes (what v7's --writes test did) is
NOT what the real download does; the beta UI subscribes first. This makes
--writes --subscribe a much closer reproduction of an actual download.
v8: --hold SECONDS runs a dedicated, separate test instead of the normal
connect/disconnect loop: connect ONCE, subscribe to Pybricks notifications
(what pybricksdev/Pybricks Code always do right after connecting, and what
every earlier test in this script never did), then just hold the connection
open and poll is_connected every --hold-poll seconds (default 10) for the
full duration, printing when/if it drops. Pass --hold-no-subscribe to run the
same test WITHOUT subscribing, as a control -- that reproduces exactly what
the rest of this script has been doing (bare connect, no subscribe).
Example, run back to back to compare (needs one hub power-on covering both):
python ble_gatt_test.py --hold 360 (subscribed)
python ble_gatt_test.py --hold 360 --hold-no-subscribe (bare, control)
If "subscribed" survives past where "bare" drops (around the hub's own
power-off time, roughly 3 min per earlier tests), that CONFIRMS the earlier
"deafness" was this script never subscribing, not an adapter/Windows fault --
important since it changes what the real problem is: whether the beta UI's
downloads sometimes fail to establish that same subscription.
v7: cache theory is dead (m5: uncached fails at the same rate as default, same
fast-connect/0-services signature either way) -- this is not a GATT services
cache, more likely a stale/phantom Windows-level connection to the address.
--immediate-retry N: on MISSING_2A26, reconnect to the SAME BLEDevice object
right away (no rescan) up to N times, without disconnecting first -- cheap
test of whether a stale connection just needs kicking. Retry count is recorded
in the detail column as retries=K.
v6: mode "find-raw" = pybricksdev's find_device() scan + plain Bleak connect
(fills the scan-vs-connect matrix), a scan_s column (seconds until the hub was
seen), CHILD_FAIL rows now carry the child's last error line, and
--retry-services: an EXPERIMENT that makes Bleak's Windows client restart
service discovery when Windows reports a GATT services-changed event during
discovery (Bleak 1.1.1 and 3.0.2 have this hard-wired off) and counts how often
that happens (sc_events=N in the detail column). Script-local monkeypatch only;
nothing on the system is changed.
v5: --fresh runs every iteration in a NEW python process (like every pybricksdev
CLI call is); mode "pdev-find" scans with pybricksdev's own find_device()
(waits for the scan response, service filter); --prompt-every K asks for a hub
power-cycle only every K iterations; --append keeps appending to one CSV.
Why: v4 showed 24/24 OK in one long-lived process, while pybricksdev (a fresh
process per run) failed 7 of 20 connects. Example without a single hub button
press: python ble_gatt_test.py --fresh -n 6 --modes default --csv fresh.csv --append
(6 iterations fit inside the hub's ~2 min auto power-off; power-cycle and repeat).
v4: --delays S1 S2 ... waits S seconds between first seeing the hub advertising
and connecting (tests "connect right after the hub booted"), and the mode
"pdev" connects with pybricksdev's own PybricksHubBLE.connect(), i.e. the
code path that shows the 2a26 failure, for a side-by-side A/B. Every
(mode, delay) combination is used round-robin. Example:
python ble_gatt_test.py -n 24 --prompt-between --modes default pdev --delays 0 6 --csv ab.csv
(power-cycle the hub and press Enter at once each time; the script varies
the delay, so your reaction time is out of the equation).
v3: after the first hit the hub is also matched by address, and every
NOT_FOUND is followed by an unfiltered 5 s scan (--no-diag to skip) that
reports how many devices the adapter hears at all and whether the hub is
among them.
v2 changes: CSV is written row by row (survives Ctrl+C), the tally is printed
even when interrupted, the run stops after --max-missing consecutive
NOT_FOUND (hub powered off or not advertising), an elapsed_s column shows when
things went wrong, --read-fw mimics pybricksdev (actual read of 2a26), and
--prompt-between lets you power-cycle the hub between iterations (cold start).
Before running: hub powered on and advertising, Pybricks Code closed (the hub
accepts only one connection), no pybricksdev running.
--writes N sends N x "stop user program" (0x00, write with response) to the
Pybricks command characteristic after connecting. That is a no-op if no program
runs, but it WILL stop a program that is running on the hub. Each write has a
5 s timeout (the ATT timeout in Windows is 30 s).
"""
import argparse
import asyncio
import csv
import importlib.metadata
import itertools
import logging
import os
import subprocess
import tempfile
import platform
import sys
import time
from collections import Counter, defaultdict
from bleak import BleakClient, BleakScanner
PYBRICKS_SERVICE = "c5f50001-8280-46da-89f4-6d8051e4aeef"
PYBRICKS_COMMAND = "c5f50002-8280-46da-89f4-6d8051e4aeef"
FW_REV = "00002a26-0000-1000-8000-00805f9b34fb"
FIELDS = ["iter", "time", "elapsed_s", "mode", "delay_s", "scan_s", "result", "connect_s",
"services", "chars", "fw_rev_after_2s", "writes_ok",
"write_max_ms", "detail"]
async def hold_test(args):
"""Dedicated single-connection test: connect once, optionally subscribe
to Pybricks notifications, then hold the connection open and poll
is_connected. See --hold in --help for why."""
subscribe = not args.hold_no_subscribe
print("# hold test: %ss, subscribe=%s, poll every %ss" % (
args.hold, subscribe, args.hold_poll))
dev = await find_hub(args.name, args.scan_timeout)
if dev is None:
print("Hub not found.")
return
notifications = [0]
def on_notify(_handle, data):
notifications[0] += 1
client = BleakClient(dev, timeout=15.0)
t0 = time.monotonic()
await client.connect()
print("+%5.1fs connected" % (time.monotonic() - t0))
if subscribe:
try:
await client.start_notify(PYBRICKS_COMMAND, on_notify)
print("+%5.1fs subscribed to %s" % (
time.monotonic() - t0, PYBRICKS_COMMAND))
except Exception as e: # noqa: BLE001
print("subscribe failed: %s %s" % (type(e).__name__, e))
try:
while time.monotonic() - t0 < args.hold:
await asyncio.sleep(args.hold_poll)
elapsed = time.monotonic() - t0
connected = client.is_connected
print("+%5.1fs is_connected=%s notifications=%d" % (
elapsed, connected, notifications[0]))
if not connected:
print("Dropped at +%.1fs." % elapsed)
return
print("+%5.1fs held for the full %ss, still connected." % (
time.monotonic() - t0, args.hold))
finally:
try:
await client.disconnect()
except Exception: # noqa: BLE001
pass
async def find_hub(name, scan_timeout, addr=None):
"""Match the remembered address, or the name, or the Pybricks service UUID."""
def match(dev, adv):
if addr and dev.address.upper() == addr.upper():
return True
if name:
return name in (dev.name, adv.local_name)
return PYBRICKS_SERVICE in [u.lower() for u in adv.service_uuids]
return await BleakScanner.find_device_by_filter(match, timeout=scan_timeout)
async def diag_scan(addr, secs=5.0):
"""Unfiltered scan: is the adapter hearing anything at all, and the hub?"""
found = await BleakScanner.discover(timeout=secs, return_adv=True)
by_addr = bool(addr) and any(a.upper() == addr.upper() for a in found)
by_uuid = [a for a, (d, adv) in found.items()
if PYBRICKS_SERVICE in [u.lower() for u in adv.service_uuids]]
names = ", ".join("%s(%s)" % (adv.local_name or d.name or "?", adv.rssi)
for d, adv in list(found.values())[:6])
return "diag %.0fs: %d devices heard, hub by address=%s, by uuid=%d [%s]" % (
secs, len(found), by_addr, len(by_uuid), names)
async def maybe_subscribe(client, do_it):
if not do_it:
return
try:
await client.start_notify(PYBRICKS_COMMAND, lambda *_: None)
except Exception as e: # noqa: BLE001
print("# subscribe failed: %s %s" % (type(e).__name__, e))
async def write_test(client, count, timeout=5.0):
"""Returns (ok_count, worst_ms, error_text)."""
ok, worst = 0, 0.0
for i in range(count):
t = time.monotonic()
try:
await asyncio.wait_for(
client.write_gatt_char(PYBRICKS_COMMAND, b"\x00", response=True),
timeout)
except Exception as e: # noqa: BLE001 - we want every failure type
return ok, worst, "write %d: %s %s" % (i, type(e).__name__, str(e)[:60])
worst = max(worst, (time.monotonic() - t) * 1000)
ok += 1
return ok, worst, ""
async def attempt_pdev(device):
"""Connect the way pybricksdev does (same reads right after connect)."""
row = {k: "" for k in FIELDS}
try:
from pybricksdev.connections.pybricks import PybricksHubBLE
except ImportError as e:
row["result"] = "NO_PYBRICKSDEV"
row["detail"] = str(e)[:80]
return row
hub = None
t0 = time.monotonic()
try:
hub = PybricksHubBLE(device)
await hub.connect()
row["connect_s"] = round(time.monotonic() - t0, 2)
row["result"] = "OK"
row["detail"] = "fw=%s" % getattr(hub, "fw_version", "?")
except Exception as e: # noqa: BLE001
msg = str(e)
row["result"] = "MISSING_2A26" if "2a26" in msg.lower() else type(e).__name__
row["detail"] = msg[:100]
finally:
if hub is not None:
try:
await hub.disconnect()
except Exception: # noqa: BLE001
pass
return row
SC_EVENTS = [0]
class _ScCounter(logging.Handler):
def emit(self, record):
if "services changed" in record.getMessage().lower():
SC_EVENTS[0] += 1
def enable_retry_services():
"""Experiment, see docstring. Windows only."""
if sys.platform != "win32":
print("# --retry-services ignored (not Windows)")
return
from bleak.backends.winrt.client import BleakClientWinRT
orig = BleakClientWinRT.__init__
def patched(self, *a, **k):
orig(self, *a, **k)
self._retry_on_services_changed = True
BleakClientWinRT.__init__ = patched
lg = logging.getLogger("bleak.backends.winrt.client")
lg.setLevel(logging.DEBUG)
lg.addHandler(_ScCounter())
async def pdev_find(timeout):
"""Scan the way pybricksdev's CLI does; None if not found."""
from pybricksdev.ble import find_device
try:
return await find_device(timeout=timeout)
except asyncio.TimeoutError:
return None
async def run_child(args, mode, delay):
"""Run one iteration in a fresh python process and return its CSV row."""
fd, tmp = tempfile.mkstemp(suffix=".csv")
os.close(fd)
cmd = [sys.executable, os.path.abspath(__file__), "-n", "1", "--modes", mode,
"--delays", str(delay), "--no-diag", "--csv", tmp,
"--scan-timeout", str(args.scan_timeout)]
if args.read_fw:
cmd.append("--read-fw")
if args.retry_services:
cmd.append("--retry-services")
if args.writes:
cmd += ["--writes", str(args.writes)]
if args.immediate_retry:
cmd += ["--immediate-retry", str(args.immediate_retry)]
if args.subscribe:
cmd.append("--subscribe")
if args.teardown_pause != 1.0:
cmd += ["--teardown-pause", str(args.teardown_pause)]
if args.name:
cmd += ["--name", args.name]
row, err = None, ""
try:
cp = await asyncio.to_thread(subprocess.run, cmd, capture_output=True,
text=True, timeout=120)
with open(tmp, newline="") as f:
rows = list(csv.DictReader(f))
row = rows[0] if rows else None
if row is None:
lines = (cp.stderr or cp.stdout or "").strip().splitlines()
err = ("rc=%s %s" % (cp.returncode, lines[-1] if lines else ""))[:110]
except Exception as e: # noqa: BLE001
err = "%s %s" % (type(e).__name__, str(e)[:60])
finally:
try:
os.unlink(tmp)
except OSError:
pass
if row is None:
row = {k: "" for k in FIELDS}
row["result"] = "CHILD_FAIL"
row["detail"] = err
return {k: row.get(k, "") for k in FIELDS}
async def attempt(device, mode, writes, read_fw, immediate_retry=0, subscribe=False,
teardown_pause=1.0):
if mode in ("pdev", "pdev-find"):
return await attempt_pdev(device)
row = {k: "" for k in FIELDS}
kwargs = {}
if sys.platform == "win32" and mode in ("cached", "uncached"):
kwargs["winrt"] = dict(use_cached_services=(mode == "cached"))
client = BleakClient(device, timeout=15.0, **kwargs)
t0 = time.monotonic()
try:
await client.connect()
row["connect_s"] = round(time.monotonic() - t0, 2)
svcs = list(client.services)
row["services"] = len(svcs)
row["chars"] = sum(len(s.characteristics) for s in svcs)
retries_used = 0
while (client.services.get_characteristic(FW_REV) is None
and retries_used < immediate_retry):
retries_used += 1
try:
await client.disconnect()
except Exception: # noqa: BLE001
pass
await client.connect()
svcs = list(client.services)
row["services"] = len(svcs)
row["chars"] = sum(len(sv.characteristics) for sv in svcs)
await maybe_subscribe(client, subscribe)
if client.services.get_characteristic(FW_REV) is not None:
row["result"] = "OK"
if retries_used:
row["detail"] = "retries=%d" % retries_used
if read_fw: # what pybricksdev does right after connecting
try:
v = await asyncio.wait_for(client.read_gatt_char(FW_REV), 5.0)
row["detail"] = ((row["detail"] + " ") if row["detail"] else "") \
+ "fw=" + v.decode(errors="replace")
except Exception as e: # noqa: BLE001
row["result"] = "READ_" + type(e).__name__
row["detail"] = str(e)[:80]
else:
row["result"] = "MISSING_2A26"
row["detail"] = "retries=%d services: %s" % (
retries_used, " ".join(sv.uuid[:8] for sv in svcs))
await asyncio.sleep(2.0)
row["fw_rev_after_2s"] = str(
client.services.get_characteristic(FW_REV) is not None)
if writes and client.services.get_characteristic(PYBRICKS_COMMAND):
ok, worst, err = await write_test(client, writes)
row["writes_ok"] = "%d/%d" % (ok, writes)
row["write_max_ms"] = round(worst)
if err:
row["result"] = (row["result"] + "+" if row["result"] != "OK" else "") + "WRITE_FAIL"
row["detail"] = (row["detail"] + " " + err).strip()
except Exception as e: # noqa: BLE001
row["result"] = type(e).__name__
row["detail"] = str(e)[:100]
finally:
if subscribe:
try:
await client.stop_notify(PYBRICKS_COMMAND)
except Exception: # noqa: BLE001
pass
try:
await client.disconnect()
except Exception: # noqa: BLE001
pass
if subscribe and teardown_pause:
await asyncio.sleep(teardown_pause)
return row
def print_tally(tally):
print("\n== tally per mode/delay (NOT_FOUND = hub not seen, not a link fault) ==")
for key, counts in tally.items():
print("%-18s n=%d %s" % (key, sum(counts.values()), dict(counts)))
async def main(args):
print("# python %s, bleak %s, %s" % (
platform.python_version(), importlib.metadata.version("bleak"),
platform.platform()))
if args.hold:
await hold_test(args)
return
if args.retry_services:
enable_retry_services()
tally = defaultdict(Counter)
combos = list(itertools.product(args.modes, args.delays))
start = time.monotonic()
missing_streak = 0
hub_addr = None
out = writer = None
if args.csv:
fresh_file = not (args.append and os.path.exists(args.csv)
and os.path.getsize(args.csv) > 0)
out = open(args.csv, "a" if args.append else "w", newline="")
writer = csv.DictWriter(out, fieldnames=FIELDS)
if fresh_file:
writer.writeheader()
out.flush()
prompt_every = 1 if args.prompt_between else args.prompt_every
try:
for i in range(1, args.n + 1):
if prompt_every and i > 1 and (i - 1) % prompt_every == 0:
await asyncio.to_thread(
input, "Power-cycle the hub, wait until it advertises, "
"then press Enter... ")
mode, delay = combos[(i - 1) % len(combos)]
scan_s = ""
if args.fresh:
row = await run_child(args, mode, delay)
seen = row["result"] != "NOT_FOUND"
if not seen and not args.no_diag:
try:
row["detail"] = await diag_scan(hub_addr)
except Exception as e: # noqa: BLE001
row["detail"] = "diag failed: " + type(e).__name__
else:
scan_err = ""
t_scan = time.monotonic()
try:
if mode in ("pdev-find", "find-raw"):
dev = await pdev_find(args.scan_timeout)
else:
dev = await find_hub(args.name, args.scan_timeout, hub_addr)
except Exception as e: # noqa: BLE001
dev, scan_err = None, "%s %s" % (type(e).__name__, str(e)[:70])
scan_s = round(time.monotonic() - t_scan, 1)
seen = dev is not None
if not seen:
row = {k: "" for k in FIELDS}
row["result"] = "SCAN_ERROR" if scan_err else "NOT_FOUND"
row["detail"] = scan_err
if not scan_err and not args.no_diag:
try:
row["detail"] = await diag_scan(hub_addr)
except Exception as e: # noqa: BLE001
row["detail"] = "diag failed: " + type(e).__name__
else:
hub_addr = dev.address
if delay:
await asyncio.sleep(delay)
SC_EVENTS[0] = 0
row = await attempt(dev, mode, args.writes, args.read_fw,
args.immediate_retry, args.subscribe,
args.teardown_pause)
if SC_EVENTS[0]:
row["detail"] = ("sc_events=%d %s" % (
SC_EVENTS[0], row["detail"])).strip()
if scan_s != "":
row["scan_s"] = scan_s
row.update(iter=i, mode=mode, delay_s=delay,
time=time.strftime("%H:%M:%S"),
elapsed_s=round(time.monotonic() - start))
tally["%s/delay=%g" % (mode, delay)][row["result"]] += 1
print("%3d %s +%4ss %-8s d=%-3g %-14s connect=%ss svc=%s chr=%s w=%s %s" % (
i, row["time"], row["elapsed_s"], mode, delay, row["result"],
row["connect_s"], row["services"], row["chars"],
row["writes_ok"], row["detail"]))
if writer:
writer.writerow(row)
out.flush()
missing_streak = 0 if seen else missing_streak + 1
if missing_streak >= args.max_missing:
print("Scanner did not see the hub for %d iterations in a row "
"(hub off, not advertising, or adapter not hearing it). "
"Stopping." % missing_streak)
break
await asyncio.sleep(args.pause)
finally:
print_tally(tally)
if out:
out.close()
print("CSV:", args.csv)
if __name__ == "__main__":
p = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
p.add_argument("-n", type=int, default=30, help="iterations (default 30)")
p.add_argument("--modes", nargs="+", default=["default", "uncached"],
choices=["default", "cached", "uncached", "pdev", "pdev-find",
"find-raw"])
p.add_argument("--delays", nargs="+", type=float, default=[0.0],
help="seconds to wait after first seeing the hub, before "
"connecting (default 0); combined with --modes round-robin")
p.add_argument("--name", help="match this advertised hub name instead of "
"the Pybricks service UUID")
p.add_argument("--writes", type=int, default=0,
help="write round trips per connection (default 0 = off)")
p.add_argument("--subscribe", action="store_true",
help="subscribe to notifications right after connecting, "
"before writes -- matches what a real download does "
"(see --hold); ignored for pdev/pdev-find modes")
p.add_argument("--teardown-pause", type=float, default=1.0,
help="seconds to wait after disconnect before the process "
"exits, only when --subscribe was used (default 1.0)")
p.add_argument("--read-fw", action="store_true",
help="also read 2a26 after connecting, like pybricksdev")
p.add_argument("--pause", type=float, default=2.0,
help="seconds between iterations (default 2)")
p.add_argument("--prompt-between", action="store_true",
help="wait for Enter between iterations (power-cycle hub)")
p.add_argument("--prompt-every", type=int, default=0, metavar="K",
help="wait for Enter every K iterations (0 = never)")
p.add_argument("--immediate-retry", type=int, default=0, metavar="N",
help="on MISSING_2A26, reconnect to the same device up to "
"N times without rescanning (default 0 = off)")
p.add_argument("--retry-services", action="store_true",
help="experiment: retry Bleak service discovery on Windows "
"services-changed events and count them")
p.add_argument("--fresh", action="store_true",
help="run each iteration in a new python process")
p.add_argument("--append", action="store_true",
help="append to --csv instead of overwriting it")
p.add_argument("--max-missing", type=int, default=3,
help="stop after this many consecutive NOT_FOUND (default 3)")
p.add_argument("--scan-timeout", type=float, default=10.0)
p.add_argument("--no-diag", action="store_true",
help="skip the unfiltered 5 s diagnostic scan on NOT_FOUND")
p.add_argument("--csv", help="write per-iteration results here as it runs")
p.add_argument("--hold", type=float, default=0.0, metavar="SECONDS",
help="run the dedicated hold/subscribe test for this many "
"seconds instead of the normal loop (0 = off)")
p.add_argument("--hold-poll", type=float, default=10.0,
help="seconds between is_connected checks in --hold "
"(default 10)")
p.add_argument("--hold-no-subscribe", action="store_true",
help="in --hold, do NOT subscribe (control test)")
try:
asyncio.run(main(p.parse_args()))
except KeyboardInterrupt:
pass
Describe the bug
At Pybricks Beta v3.1.0-beta.5 with firmware v4.1.0b4:
On BLE-connected hubs, pressing F5 sometimes never starts the program:
the F5/Run button disables as normal, but the progress indicator doesn't move, or only reaches a low percentage, and then nothing happens.
The hub light stays a steady blue rather than showing the program running.
This happens intermittently, I don't have a reliable trigger, it just happens sometimes.
To reproduce
I don't have reliable steps.
What I can say:
Expected behavior
The program downloads and starts reliably, or if the hub is in a stale/stuck connection state, the UI detects this and either reconnects cleanly or shows a clear error, rather than silently stalling.
Screenshots
None; I could not capture a Bluetooth trace of the original failing session, so this report is a script-based reconstruction of the same symptom rather than a trace of the exact UI failure.
Investigation notes (hypothesis, not confirmed against the actual UI failure)
I couldn't capture a Bluetooth trace of the real failing session and can't reliably reproduce it on demand, so I built a script to probe possible causes instead.
One thing found: if a BLE connection with an active hub-notification subscription is disconnected without first unsubscribing, and the client exits right away, the hub can be left believing it's still connected steady blue, and unreachable to a new connection attempt for a while afterward, until it recovers on its own (or manual).
Adding an explicit unsubscribe before disconnect fixed this in my testing, every time.
I don't know whether the beta UI's own connect/disconnect handling does this cleanly or not, so I can't say this is the cause of the symptom above, only that it's a real BLE-level issue I found nearby, and it seemed worth mentioning in case it's related.
extra info
Running win11 25H2 and pybricks as chrome app.
Noticed that unplug/replug the bluetooth adapter clearer the problem at least once. Better than a reboot.
Probably power-off / power-on would do the same but takes longer 😄
The latest version of the test script:
ble_gatt_test.py for me C:\home\bert\py\antonsmindstorms\EV3\v4.1.0b4_test\ble_gatt_test.py