141 lines
4.4 KiB
Python
141 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Systematic BLE channel matrix for Smart Guitar MIDI App SysEx."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from bleak import BleakClient, BleakScanner
|
|
|
|
NAME = "Smart Guitar MIDI"
|
|
MIDI = "7772e5db-3868-4112-a1a9-f2669d106bf3"
|
|
UARTW = "e49a25e0-f69a-11e8-8eb2-f2801f1b9fd1"
|
|
UARTN = "e49a28e1-f69a-11e8-8eb2-f2801f1b9fd1"
|
|
EFF2 = "0000eff2-0000-1000-8000-00805f9b34fb"
|
|
|
|
SYSEX = bytes.fromhex("F0600101F7")
|
|
FRAMED = bytes.fromhex("8080F0600101F7")
|
|
FRAMED_TSF7 = bytes.fromhex("8080F060010180F7")
|
|
|
|
|
|
def hx(b: bytes) -> str:
|
|
return " ".join(f"{x:02X}" for x in b) if b else "(none)"
|
|
|
|
|
|
async def find(timeout=60.0):
|
|
t0 = time.time()
|
|
while time.time() - t0 < timeout:
|
|
d = await BleakScanner.find_device_by_filter(
|
|
lambda d, a: d.name and NAME.lower() in d.name.lower(), timeout=8
|
|
)
|
|
if d:
|
|
return d
|
|
print("waiting advertise...")
|
|
return None
|
|
|
|
|
|
async def unpair_if_needed(addr: str):
|
|
try:
|
|
c = BleakClient(addr, timeout=15)
|
|
await c.connect()
|
|
try:
|
|
await c.unpair()
|
|
print("unpaired")
|
|
except Exception as e:
|
|
print("unpair skip:", e)
|
|
try:
|
|
await c.disconnect()
|
|
except Exception:
|
|
pass
|
|
await asyncio.sleep(1.5)
|
|
except Exception as e:
|
|
print("unpair session:", e)
|
|
|
|
|
|
async def one_session(tag: str, write_uuid: str, payload: bytes, notify_uuids, settle=2.5):
|
|
d = await find(45)
|
|
if not d:
|
|
print(f"[{tag}] NO DEVICE")
|
|
return {"tag": tag, "ok": False, "err": "no device", "notifs": []}
|
|
print(f"\n=== {tag} === write {write_uuid[:8]} {hx(payload)}")
|
|
notifs = []
|
|
err = None
|
|
connected_end = False
|
|
try:
|
|
async with BleakClient(d, timeout=25) as c:
|
|
print("connected", c.is_connected)
|
|
|
|
def cb(sender, data):
|
|
b = bytes(data)
|
|
notifs.append(b)
|
|
print(f" NOTIFY {hx(b)}")
|
|
|
|
for u in notify_uuids:
|
|
try:
|
|
await c.start_notify(u, cb)
|
|
print(" notify on", u[:8])
|
|
except Exception as e:
|
|
print(" notify fail", u[:8], e)
|
|
await asyncio.sleep(0.3)
|
|
try:
|
|
await c.write_gatt_char(write_uuid, payload, response=False)
|
|
print(" write ok")
|
|
except Exception as e:
|
|
err = f"write: {e}"
|
|
print(" write fail", e)
|
|
await asyncio.sleep(settle)
|
|
connected_end = c.is_connected
|
|
print(" end conn=", connected_end, "notifs=", len(notifs))
|
|
except Exception as e:
|
|
err = str(e)
|
|
print(" session err", e)
|
|
return {
|
|
"tag": tag,
|
|
"ok": any(b[:3] == b"\xF0\x60\x01" or b[:4] == b"\x80\x80\xF0\x60" for b in notifs),
|
|
"notifs": [hx(b) for b in notifs],
|
|
"conn_end": connected_end,
|
|
"err": err,
|
|
}
|
|
|
|
|
|
async def main():
|
|
d = await find(60)
|
|
if not d:
|
|
print("FATAL: device not advertising")
|
|
return
|
|
print("found", d.name, d.address)
|
|
await unpair_if_needed(d.address)
|
|
await asyncio.sleep(2)
|
|
|
|
results = []
|
|
# matrix: do NOT put uartw first if it kills radio; do midi first
|
|
cases = [
|
|
("midi-framed", MIDI, FRAMED, [MIDI, UARTN, EFF2]),
|
|
("midi-framed-tsF7", MIDI, FRAMED_TSF7, [MIDI]),
|
|
("midi-raw", MIDI, SYSEX, [MIDI]),
|
|
("eff2-raw", EFF2, SYSEX, [EFF2, MIDI, UARTN]),
|
|
("eff2-framed", EFF2, FRAMED, [EFF2, MIDI]),
|
|
("uartw-raw", UARTW, SYSEX, [UARTN, MIDI, EFF2]),
|
|
("uartw-framed", UARTW, FRAMED, [UARTN, MIDI]),
|
|
]
|
|
for tag, wu, payload, nus in cases:
|
|
r = await one_session(tag, wu, payload, nus)
|
|
results.append(r)
|
|
# if uart killed advertising, wait/reset hint
|
|
if not r.get("conn_end", True):
|
|
print("link dropped; waiting re-advertise...")
|
|
await asyncio.sleep(3)
|
|
if not await find(30):
|
|
print("device gone after drop — stop matrix (need JLink reset)")
|
|
break
|
|
else:
|
|
await asyncio.sleep(1)
|
|
|
|
print("\n===== MATRIX SUMMARY =====")
|
|
for r in results:
|
|
status = "PASS" if r["ok"] else "FAIL"
|
|
print(f"{status:4} {r['tag']:18} conn_end={r.get('conn_end')} notifs={r['notifs'] or ['(none)']} err={r.get('err')}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|