diff --git a/Global/Global.c b/Global/Global.c index c4b9283..449105a 100644 --- a/Global/Global.c +++ b/Global/Global.c @@ -184,6 +184,7 @@ void System_PowerOn(void) BSP_MainPowerEnable(1); BSP_DreamCorePowerEnable(1); BSP_HT7178PowerEnable(1); + /* 蓝牙按 NVM/系统设置开关恢复，勿强制常开 */ BSP_BlueToothPowerEnable((uint8_t)(mGuiData[GUI_BL_SW].Current ? 1 : 0)); powon = true; /* 更新开机标志 */ @@ -196,7 +197,9 @@ void System_PowerOn(void) app_tm1617_init(); XPT2046_Init(); App_Auto_Init(); - LOG_I("PWR", "periph ready"); + /* App 协议走 UART4：须在开机后即解析，不能等到触摸选模式才 StartTask */ + StartTask(); + LOG_I("PWR", "periph ready + BT UART recv started"); } static void PowerOn(void) diff --git a/project/src/usart.c b/project/src/usart.c index 40b2e90..74139b2 100644 --- a/project/src/usart.c +++ b/project/src/usart.c @@ -3,14 +3,14 @@ #define BUFF_SIZE 512 -/* ȽϺ꣺ֲͨ volatile - ͬһʽη volatile Pa082 ߼ */ +/* ?????????????????????????????? volatile???? + ?????????????????? volatile ???? Pa082 ????????? */ #define IS_FULL(head, tail) ((((head) + 1) % BUFF_SIZE) == (tail)) #define IS_EMPTY(head, tail) ((head) == (tail)) -/* ==================== ζнṹ ==================== */ +/* ==================== ???????????? ==================== */ -/* һʵĻζУշһ */ +/* ????????????????????????????? */ typedef struct { uint8_t rx_buff[BUFF_SIZE]; @@ -21,14 +21,14 @@ typedef struct volatile uint16_t tx_head; volatile uint16_t tx_tail; - usart_type *usart; /* Ĵ裨USART2 / UART4 */ + usart_type *usart; /* ??????????????USART2 / UART4?? */ } uart_ring_t; -/* USART2 UART4 һʵ */ +/* USART2 ?? UART4 ???????? */ static uart_ring_t uart2; static uart_ring_t uart4; -/* ʼ裨 wk_usart2_init / wk_uart4_init ã */ +/* ?????????????????????? wk_usart2_init / wk_uart4_init ????? */ void uart_ring_usart2_init(void) { uart2.usart = USART2; @@ -39,19 +39,19 @@ void uart_ring_uart4_init(void) uart4.usart = UART4; } -/* ==================== ͨ/ӣʵ ==================== */ +/* ==================== ??????/???????????????? ==================== */ -/* ӣISR ãȿ volatile ֵٲ Pa082 */ +/* ????ISR ???????????? volatile ???????????? Pa082 */ static void ring_rx_push(uart_ring_t *rb, uint8_t c) { uint16_t h = rb->rx_head; uint16_t t = rb->rx_tail; - if(IS_FULL(h, t)) return; /* 򶪣ֹδ */ + if(IS_FULL(h, t)) return; /* ???????????????????? */ rb->rx_buff[h] = c; rb->rx_head = (h + 1) % BUFF_SIZE; } -/* ӣҵ̵߳ã */ +/* ??????????????? */ static uint8_t ring_rx_pop(uart_ring_t *rb, uint8_t *c) { uint16_t h = rb->rx_head; @@ -62,65 +62,64 @@ static uint8_t ring_rx_pop(uart_ring_t *rb, uint8_t *c) return 1; } -/* д tx УǰӦѹжϣֹ ISR ͬʱ */ +/* ???? tx ??????????????????????? ISR ?????? */ static void ring_tx_write(uart_ring_t *rb, uint8_t *data, uint16_t len) { uint16_t h = rb->tx_head; uint16_t t = rb->tx_tail; for(uint16_t i = 0; i < len; i++) { - if(IS_FULL(h, t)) break; /* tx */ + if(IS_FULL(h, t)) break; /* ?? tx ???? */ rb->tx_buff[h] = data[i]; h = (h + 1) % BUFF_SIZE; } - rb->tx_head = h; /* һд volatile */ + rb->tx_head = h; /* ???????????? volatile */ } -/* ==================== ͽӿڣڲ ==================== */ +/* ==================== ?????????????????? ==================== */ -/* ͨ÷ͣж д TDBE жϴ */ +/* ?????????????? ?? ?????? ?? ?? TDBE ??????????? */ static void uart_ring_send(uart_ring_t *rb, uint8_t *data, uint16_t len) { rt_base_t level = rt_hw_interrupt_disable(); ring_tx_write(rb, data, len); - usart_interrupt_enable(rb->usart, USART_TDBE_INT, TRUE); /* ж */ + usart_interrupt_enable(rb->usart, USART_TDBE_INT, TRUE); /* ?????????? */ rt_hw_interrupt_enable(level); } -/* USART2 ͣ⣬midi_send.c ã */ +/* USART2 ?????????midi_send.c ??????? */ void USART2_SendData(uint8_t *data, uint16_t len) { uart_ring_send(&uart2, data, len); } -/* UART4 ͣ⣩ */ +/* UART4 ????????? */ void USART4_SendData(uint8_t *data, uint16_t len) { uart_ring_send(&uart4, data, len); } -/* USART2 ճӣ⣩ */ +/* USART2 ???????????? */ uint8_t USART2_RxPop(uint8_t *c) { return ring_rx_pop(&uart2, c); } -/* UART4 ճӣ⣩ */ +/* UART4 ???????????? */ uint8_t UART4_RxPop(uint8_t *c) { return ring_rx_pop(&uart4, c); } -/* ==================== ж ==================== */ +/* ==================== ???????? ==================== */ static void uart_ring_isr(uart_ring_t *rb) { - /* գյֽ rx */ if(usart_flag_get(rb->usart, USART_RDBF_FLAG) == SET) { - uint8_t data = usart_data_receive(rb->usart); /* ݣԶ RDBF */ + uint8_t data = usart_data_receive(rb->usart); ring_rx_push(rb, data); if(rb->usart == UART4) { @@ -128,7 +127,6 @@ static void uart_ring_isr(uart_ring_t *rb) } } - /* ͣtx һֽڣ TDBE ж */ if(usart_flag_get(rb->usart, USART_TDBE_FLAG) == SET) { uint16_t h = rb->tx_head; @@ -140,11 +138,10 @@ static void uart_ring_isr(uart_ring_t *rb) } else { - usart_interrupt_enable(rb->usart, USART_TDBE_INT, FALSE); /* ж */ + usart_interrupt_enable(rb->usart, USART_TDBE_INT, FALSE); } } - /* ־ֹжϿ */ if(usart_flag_get(rb->usart, USART_ROERR_FLAG) == SET) usart_flag_clear(rb->usart, USART_ROERR_FLAG); } diff --git a/protocol/bl_uart_parse.c b/protocol/bl_uart_parse.c index d977377..af0eeda 100644 --- a/protocol/bl_uart_parse.c +++ b/protocol/bl_uart_parse.c @@ -162,13 +162,14 @@ static const BLE_SysExCmdItem bleSysExCmdTable[] = void processBLESysEXData(uint8_t* data, uint8_t cnt) { //ResetAutoPowerCount(); - // ����֡У�� + // ֡У�� if (data[0] != FRAME_SYS_HEAD || data[cnt - 1] != FRAME_SYS_TAIL || cnt > UART4_PROCESS_BUFF_SIZE || data[1] != 0x60) { return; } + LOG_I("BLE", "sysex ok len=%u cmd=%02X %02X", (unsigned)cnt, data[2], data[3]); - // ��ȡ ��ָ�� + ��ָ�� + // ȡ��ָ�� + ��ָ�� uint8_t cmd[2] = {data[2], data[3]}; // ����ָ���ƥ�� @@ -198,6 +199,7 @@ void UART4_Data_Process(volatile uint8_t* data) last_tick = now; memset(UART4_Process_Buff, 0, sizeof(UART4_Process_Buff)); UART4_RCV_cnt = 0; + UART4_RCV_Status = UART4_RCV_BUFF_IDLE; } switch (UART4_RCV_Status) @@ -254,7 +256,6 @@ void UART4_Data_Process(volatile uint8_t* data) } else { - // �Ƿ��ֽڣ���λ UART4_RCV_Status = UART4_RCV_BUFF_IDLE; } break; diff --git a/task/task_init.c b/task/task_init.c index a95016a..4ec6af3 100644 --- a/task/task_init.c +++ b/task/task_init.c @@ -127,12 +127,11 @@ void TaskBTRecvThread_entry(void* parameter) { uint8_t c; while(1) - { - if(rt_sem_take(UART4_sem,RT_WAITING_FOREVER) == RT_EOK) + { + if(rt_sem_take(UART4_sem, RT_WAITING_FOREVER) == RT_EOK) { while(UART4_RxPop(&c)) { - //解析函数 UART4_Data_Process(&c); } } diff --git a/tools/_rtt_ble_diagnose.py b/tools/_rtt_ble_diagnose.py new file mode 100644 index 0000000..33c7a39 --- /dev/null +++ b/tools/_rtt_ble_diagnose.py @@ -0,0 +1,193 @@ +# -*- coding: utf-8 -*- +"""Flash is separate; this attaches RTT then writes BLE SysEx and prints MCU logs.""" +from __future__ import annotations + +import asyncio +import sys +import threading +import time + +import pylink +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" + + +def open_jlink(): + j = pylink.JLink() + j.open() + try: + j.exec_command("HideDeviceSelection = 1") + except Exception: + pass + j.set_tif(pylink.enums.JLinkInterfaces.SWD) + last = None + for dev in ("Cortex-M4", "AT32F403AC", "AT32F403A"): + try: + try: + j.exec_command(f"Device = {dev}") + except Exception: + pass + j.connect(dev) + print("JLink OK", dev, flush=True) + try: + j.restart(halt=False) + except Exception: + try: + j.go() + except Exception: + pass + return j + except Exception as e: + last = e + raise SystemExit(f"JLink fail: {last}") + + +async def find_dev(timeout=45.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("wait BLE...", flush=True) + return None + + +async def ble_write(tag, write_uuid, payload, notify_uuids): + d = await find_dev() + if not d: + print("NO BLE", flush=True) + return + print(f"BLE {d.address} write {tag} {payload.hex()}", flush=True) + try: + c = BleakClient(d.address, timeout=20) + await c.connect() + try: + await c.unpair() + except Exception: + pass + await c.disconnect() + except Exception: + pass + await asyncio.sleep(1.0) + d = await find_dev() or d + notifs = [] + async with BleakClient(d, timeout=25) as c: + def cb(_s, data): + notifs.append(bytes(data)) + print("GATT NOTIFY", data.hex(), flush=True) + + for u in notify_uuids: + try: + await c.start_notify(u, cb) + except Exception as e: + print("notify fail", u[:8], e, flush=True) + await asyncio.sleep(0.2) + try: + await c.write_gatt_char(write_uuid, payload, response=False) + print("GATT write ok, conn", c.is_connected, flush=True) + except Exception as e: + print("GATT write fail", e, flush=True) + await asyncio.sleep(3.0) + print("GATT notifs", len(notifs), "conn", c.is_connected, flush=True) + + +async def main(): + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + j = open_jlink() + # Give RTT control block time; try start repeatedly + for i in range(10): + try: + j.rtt_start(block_address=0x20016D68) + st = j.rtt_get_status() + print("RTT status", st, flush=True) + if getattr(st, "NumUpBuffers", 0): + break + except Exception as e: + print("rtt_start", e, flush=True) + time.sleep(0.5) + else: + print("WARN: RTT upbuffers still 0 — will keep reading anyway", flush=True) + + stop = False + lines = [] + + def reader(): + while not stop: + try: + data = j.rtt_read(0, 4096) + if data: + s = bytes(data).decode("utf-8", "replace") + lines.append(s) + print("RTT>", s, end="" if s.endswith("\n") else "\n", flush=True) + except Exception as e: + print("rtt read err", e, flush=True) + break + time.sleep(0.03) + + th = threading.Thread(target=reader, daemon=True) + th.start() + print("collect 4s boot/idle logs...", flush=True) + await asyncio.sleep(4.0) + if not lines: + print("WARNING: no RTT yet — touch screen or wait; continuing BLE test", flush=True) + + # 1) BLE-MIDI framed (stable path historically) + await ble_write( + "midi-framed", + MIDI, + bytes.fromhex("8080F0600101F7"), + [MIDI, UARTN, EFF2], + ) + await asyncio.sleep(2.0) + + # 2) custom uart raw (may drop link) + await ble_write( + "uartw-raw", + UARTW, + bytes.fromhex("F0600101F7"), + [UARTN, MIDI, EFF2], + ) + + await asyncio.sleep(2.0) + stop = True + time.sleep(0.4) + j.close() + + text = "".join(lines) + print("\n===== ANALYSIS =====", flush=True) + print("has U4 rx sniff:", ("rx n=" in text) or ("U4" in text and "rx" in text), flush=True) + print("has sysex ok:", "sysex ok" in text, flush=True) + print("has sysex reject:", "sysex reject" in text, flush=True) + print("has sysex abort:", "sysex abort" in text, flush=True) + print("has U4 tx:", ("U4" in text and "tx" in text) or "tx len=" in text, flush=True) + print("has pin diag:", "edge PC" in text or "pinmap" in text, flush=True) + # Extract last diag line if present + for line in text.splitlines(): + if "edge PC10=" in line: + print("diag:", line.strip(), flush=True) + if "sysex ok" in text and "tx len=" in text: + print("VERDICT: MCU got cmd and replied on UART4 → GATT notify path broken in BLE module", flush=True) + elif "rx n=" in text and "sysex ok" not in text: + print("VERDICT: UART4 got bytes but frame not accepted → protocol/framing", flush=True) + elif "edge PC12=" in text and "isr_rx=0" in text: + # SCH puts B_UART4_RX on PC12; FW UART4 RX is PC11 + print("VERDICT: activity on PC12 while UART4 ISR idle → driver pinmap vs SCH (MCU底层)", flush=True) + elif "edge PC11=" in text and "isr_rx=0" in text: + print("VERDICT: edges on PC11 but no UART ISR → baud/noise or not UART framing", flush=True) + elif "isr_rx=0" in text and "edge PC10=0" in text and "edge PC11=0" in text and "edge PC12=0" in text: + print("VERDICT: no GPIO edges + no UART ISR → ATS2853 did not drive UART (模组固件/桥接)", flush=True) + elif "U4" not in text and "rx" not in text: + print("VERDICT: no UART4 activity → BLE module did not forward GATT write to UART4 (or RTT dead)", flush=True) + else: + print("VERDICT: inconclusive — see RTT dump above", flush=True) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tools/flash_app_cspy.py b/tools/flash_app_cspy.py new file mode 100644 index 0000000..b7a43c7 --- /dev/null +++ b/tools/flash_app_cspy.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +"""Flash APP via cspybat (AT32F403AC) then J-Link reset.""" +from __future__ import annotations + +import shutil +import subprocess +import time +from pathlib import Path + +BASE = Path(r"C:\Users\qjyu\Documents\SoundWalker\一诺国际吉他\Code\YNGJ-GT1-M - AT32F403ARCT7") +STAGE = Path(r"C:\Temp\k1flash_boot") +CSPY = Path(r"C:\Program Files (x86)\IAR Systems\Embedded Workbench 7.3\common\bin\cspybat.exe") +JLINK = Path(r"C:\Program Files\SEGGER\JLink_V818\JLink.exe") + +APP_OUT = BASE / "project" / "IAR_V7.4" / "YNGJ-GT1-M" / "Exe" / "YNGJ-GT1-M.out" +GEN_TMPL = BASE / "project" / "IAR_V7.4" / "settings" / "YNGJ-GT1-M.YNGJ-GT1-M.general.xcl" +DRV_TMPL = BASE / "project" / "IAR_V7.4" / "settings" / "YNGJ-GT1-M.YNGJ-GT1-M.driver.xcl" +RESET_JLINK = BASE / "tools" / "reset_run.jlink" + + +def kill_debuggers() -> None: + for n in ( + "cspybat.exe", "CSpyBat.exe", "JLink.exe", "IarIdePm.exe", + "JLinkGUIServer.exe", "JLinkRTTClient.exe", "JFlash.exe", + ): + subprocess.run(["taskkill", "/F", "/IM", n], capture_output=True) + time.sleep(1.0) + + +def stage_xcl(out_file: Path): + STAGE.mkdir(parents=True, exist_ok=True) + staged_out = STAGE / out_file.name + shutil.copy2(out_file, staged_out) + + gen_lines = GEN_TMPL.read_text(encoding="utf-8", errors="replace").splitlines() + new_gen = [] + for ln in gen_lines: + if ".out" in ln and ("YNGJ" in ln or "BOOT" in ln or "AT32" in ln): + new_gen.append(f'"{staged_out}" ') + else: + new_gen.append(ln) + gen_path = STAGE / "general.xcl" + gen_path.write_text("\n".join(new_gen) + "\n", encoding="utf-8", newline="\n") + + drv_lines = DRV_TMPL.read_text(encoding="utf-8", errors="replace").splitlines() + forced = [ln for ln in drv_lines if not ln.strip().startswith("--jlink_device")] + forced.append("--jlink_device=AT32F403AC") + drv_path = STAGE / "driver.xcl" + drv_path.write_text("\n".join(forced) + "\n", encoding="utf-8", newline="\n") + return gen_path, drv_path, staged_out + + +def cspy_download(out_file: Path, tag: str) -> None: + if not out_file.is_file(): + raise SystemExit(f"missing {out_file}") + gen_path, drv_path, staged_out = stage_xcl(out_file) + logp = STAGE / f"cspy_{tag}.log" + cmd = [ + str(CSPY), "-f", str(gen_path), + f"--debug_file={staged_out}", + "--download_only", "--backend", "-f", str(drv_path), + ] + print(f"cspybat download {tag}: {staged_out.name}", flush=True) + with logp.open("w", encoding="utf-8", errors="replace") as log: + r = subprocess.run(cmd, stdout=log, stderr=subprocess.STDOUT, timeout=240) + text = logp.read_text(encoding="utf-8", errors="replace") + print(text[-1500:], flush=True) + if r.returncode != 0: + raise SystemExit(f"cspybat {tag} failed rc={r.returncode}") + + +def jlink_reset() -> None: + print("J-Link reset/run...", flush=True) + subprocess.run( + [ + str(JLINK), "-Device", "AT32F403AC", "-If", "SWD", + "-Speed", "4000", "-AutoConnect", "1", + "-CommandFile", str(RESET_JLINK), + ], + capture_output=True, timeout=45, + ) + time.sleep(2.5) + + +def main() -> None: + kill_debuggers() + cspy_download(APP_OUT, "app") + jlink_reset() + print("FLASH OK", flush=True) + + +if __name__ == "__main__": + main() diff --git a/tools/test_protocol_ble.py b/tools/test_protocol_ble.py index ee4485c..53c602a 100644 --- a/tools/test_protocol_ble.py +++ b/tools/test_protocol_ble.py @@ -1,25 +1,28 @@ # -*- coding: utf-8 -*- """ -test_protocol_ble.py — 通过笔记本蓝牙(BLE-MIDI)对吉他做全协议测试并出报告 +test_protocol_ble.py — 通过笔记本蓝牙对吉他做全协议测试并出报告 协议来源: Doc/指令测试.docx -链路: 笔记本 BLE <-> "Smart Guitar MIDI" (BLE-MIDI) <-> 吉他 UART4 +链路: 笔记本 BLE <-> "Smart Guitar MIDI" <-> 吉他 UART4 App->设备: F0 60 ... F7 ; 设备主动上报: F0 51 ... F7 -BLE-MIDI: - service 03B80E5A-EDE8-4B33-A751-6CE34EC4C700 - characteristic 7772E5DB-3868-4112-A1A9-F2669D106BF3 (notify + write-no-resp) +GATT: + BLE-MIDI 03B80E5A-... / 7772E5DB-... (framed SysEx) + 自定义串口 e49a25f8-... / e49a25e0(写) + e49a28e1(通知) (raw SysEx) 用法: - python test_protocol_ble.py # 扫描并按名连接 - python test_protocol_ble.py --ble-address AA:BB:CC:... # 按地址连接 - python test_protocol_ble.py --allow-poweroff # 含 05 00 关机用例 - python test_protocol_ble.py --selftest # 无硬件自检(编解码) + python test_protocol_ble.py + python test_protocol_ble.py --transport midi + python test_protocol_ble.py --transport uart + python test_protocol_ble.py --unpair + python test_protocol_ble.py --smoke + python test_protocol_ble.py --selftest 依赖: bleak (pip install bleak) 报告: Doc/reports/ble_sysex_<时间戳>.md -退出码: 0 = 全部通过, 1 = 有 FAIL """ +from __future__ import annotations + import argparse import asyncio import datetime @@ -31,118 +34,173 @@ import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from test_protocol_app_sim import ( # noqa: E402 - HEAD, DEV_ID, build, FrameParser, Results, expect, run_tests, hexs) + HEAD, DEV_ID, build, FrameParser, Results, run_tests, hexs) MIDI_SERVICE = "03B80E5A-EDE8-4B33-A751-6CE34EC4C700" MIDI_CHAR = "7772E5DB-3868-4112-A1A9-F2669D106BF3" +UART_SERVICE = "e49a25f8-f69a-11e8-8eb2-f2801f1b9fd1" +UART_WRITE = "e49a25e0-f69a-11e8-8eb2-f2801f1b9fd1" +UART_NOTIFY = "e49a28e1-f69a-11e8-8eb2-f2801f1b9fd1" DEFAULT_NAME = "Smart Guitar MIDI" -DEFAULT_MTU_PAYLOAD = 20 # ATT 默认 23 -> 应用载荷 20 +DEFAULT_MTU_PAYLOAD = 20 -# ---------------------------------------------------------- BLE-MIDI codec def ble_midi_encode_sysex(frame: bytes, max_payload: int): - """完整 SysEx(F0...F7) -> BLE-MIDI 包列表 (ts=0)""" ts_hi, ts_lo = 0x80, 0x80 pkts = [] - cap = max_payload - 2 # header + timestamp + cap = max_payload - 2 first = frame[:cap] pkts.append(bytes([ts_hi, ts_lo]) + first) rest = frame[len(first):] - while len(rest) > max_payload - 2: # 中间包: header + data - pkts.append(bytes([ts_hi]) + rest[:max_payload - 1]) - rest = rest[max_payload - 1:] - if rest: # 结束包: header + ts + data..F7 + while len(rest) > max_payload - 2: + pkts.append(bytes([ts_hi]) + rest[: max_payload - 1]) + rest = rest[max_payload - 1 :] + if rest: pkts.append(bytes([ts_hi, ts_lo]) + rest) return pkts def ble_midi_decode_packet(payload: bytes) -> bytes: - """单个 BLE-MIDI 通知包 -> MIDI 字节流 (仅 SysEx 场景: 数据均为 7bit)""" if len(payload) < 2: return b"" - i = 1 # 跳过 header - if payload[i] & 0x80: # 时间戳(起始/结束包); 续包无 + i = 1 + if payload[i] & 0x80: i += 1 return bytes(payload[i:]) -# ---------------------------------------------------------- BLE transport class BleMidiSim: """与 AppSim 同接口: send / read_frame / drain / query / close""" - def __init__(self, name=DEFAULT_NAME, address=None, scan_timeout=15.0): + def __init__( + self, + name=DEFAULT_NAME, + address=None, + scan_timeout=20.0, + transport="midi", + unpair=False, + ): from bleak import BleakClient, BleakScanner + self._BleakClient = BleakClient self._BleakScanner = BleakScanner self.parser = FrameParser() self.pending = [] self._rx = queue.Queue() self._client = None - self._char = None + self._write_char = None + self._notify_chars = [] + self._transport = transport + self._unpair = unpair self._loop = asyncio.new_event_loop() - self._thread = threading.Thread(target=self._loop.run_forever, - daemon=True) + self._thread = threading.Thread(target=self._loop.run_forever, daemon=True) self._thread.start() self.info = self._run(self._connect(name, address, scan_timeout)) - # ---- async helpers ---- - def _run(self, coro, timeout=60.0): + def _run(self, coro, timeout=90.0): return asyncio.run_coroutine_threadsafe(coro, self._loop).result(timeout) - async def _connect(self, name, address, scan_timeout): - dev = None + async def _find(self, name, address, scan_timeout): if address: - dev = await self._BleakScanner.find_device_by_address( - address, timeout=scan_timeout) - else: - print(f"扫描 BLE 设备 ({scan_timeout:.0f}s) ...") - devs = await self._BleakScanner.discover( - timeout=scan_timeout, service_uuids=[MIDI_SERVICE]) - for d in devs: + return await self._BleakScanner.find_device_by_address( + address, timeout=scan_timeout + ) + print(f"扫描 BLE 设备 ({scan_timeout:.0f}s) ...") + # name filter is more reliable than service UUID filter on Windows + t0 = time.monotonic() + while time.monotonic() - t0 < scan_timeout: + rem = max(1.0, scan_timeout - (time.monotonic() - t0)) + d = await self._BleakScanner.find_device_by_filter( + lambda d, a: d.name and name.lower() in d.name.lower(), + timeout=min(8.0, rem), + ) + if d: print(f" 发现: {d.name!r} {d.address}") - if d.name and name.lower() in d.name.lower(): - dev = d - if dev is None and devs: - dev = devs[0] - print(f" (未匹配名称 {name!r}, 使用第一个 MIDI 设备)") - if dev is None: - raise RuntimeError(f"未找到 BLE-MIDI 设备 {name!r}") + return d + print(" ...") + return None - client = self._BleakClient(dev) + async def _connect(self, name, address, scan_timeout): + dev = await self._find(name, address, scan_timeout) + if dev is None: + raise RuntimeError(f"未找到设备 {name!r}") + + if self._unpair: + try: + tmp = self._BleakClient(dev.address, timeout=20) + await tmp.connect() + try: + await tmp.unpair() + print("已 unpair Windows 残留配对") + except Exception as e: + print(f"unpair: {e}") + try: + await tmp.disconnect() + except Exception: + pass + await asyncio.sleep(1.5) + dev = await self._find(name, address, scan_timeout) or dev + except Exception as e: + print(f"unpair session: {e}") + + client = self._BleakClient(dev, timeout=30) await client.connect() - char = None + if not client.is_connected: + raise RuntimeError("BLE connect 后立即断开") + + write_char = MIDI_CHAR if self._transport == "midi" else UART_WRITE + notify_list = ( + [MIDI_CHAR] + if self._transport == "midi" + else [UART_NOTIFY, MIDI_CHAR] + ) + + max_payload = DEFAULT_MTU_PAYLOAD for svc in client.services: - if svc.uuid.lower() == MIDI_SERVICE.lower(): - for c in svc.characteristics: - if c.uuid.lower() == MIDI_CHAR.lower(): - char = c - if char is None: - raise RuntimeError("设备无 BLE-MIDI 特征") - try: - max_payload = char.max_write_without_response_size - except Exception: - max_payload = DEFAULT_MTU_PAYLOAD - if not max_payload or max_payload < DEFAULT_MTU_PAYLOAD: - max_payload = DEFAULT_MTU_PAYLOAD + for c in svc.characteristics: + if c.uuid.lower() == write_char.lower(): + try: + max_payload = c.max_write_without_response_size or max_payload + except Exception: + pass def _on_notify(_sender, data): self._rx.put(bytes(data)) - await client.start_notify(char, _on_notify) - self._client, self._char = client, char - self._max_payload = max_payload - return {"name": dev.name, "address": dev.address, - "max_payload": max_payload} + for u in notify_list: + try: + await client.start_notify(u, _on_notify) + self._notify_chars.append(u) + except Exception as e: + print(f"notify fail {u[:8]}: {e}") + + if not self._notify_chars: + raise RuntimeError("无法开启任何 notify（常见原因: Windows 残留配对）") + + self._client = client + self._write_char = write_char + self._max_payload = max(DEFAULT_MTU_PAYLOAD, max_payload or DEFAULT_MTU_PAYLOAD) + return { + "name": getattr(dev, "name", None), + "address": getattr(dev, "address", address), + "max_payload": self._max_payload, + "transport": self._transport, + "notify": list(self._notify_chars), + } async def _disconnect(self): - if self._client is not None: + if self._client is None: + return + for u in self._notify_chars: try: - await self._client.stop_notify(self._char) + await self._client.stop_notify(u) except Exception: pass + try: await self._client.disconnect() + except Exception: + pass - # ---- AppSim-compatible sync API ---- def close(self): try: self._run(self._disconnect(), timeout=10) @@ -157,12 +215,27 @@ class BleMidiSim: data = self._rx.get_nowait() except queue.Empty: break - self.pending.extend(self.parser.feed(ble_midi_decode_packet(data))) + if self._transport == "midi": + midi = ble_midi_decode_packet(data) + else: + # uart notify may be raw SysEx, or occasionally BLE-MIDI wrapped + midi = ( + ble_midi_decode_packet(data) + if data and (data[0] & 0x80) + else data + ) + self.pending.extend(self.parser.feed(midi)) def send(self, frame: bytes): - for pkt in ble_midi_encode_sysex(frame, self._max_payload): - self._run(self._client.write_gatt_char( - self._char, pkt, response=False), timeout=10) + if self._transport == "midi": + pkts = ble_midi_encode_sysex(frame, self._max_payload) + else: + pkts = [frame] + for pkt in pkts: + self._run( + self._client.write_gatt_char(self._write_char, pkt, response=False), + timeout=10, + ) def read_frame(self, timeout=1.0, want_dev=False): deadline = time.monotonic() + timeout @@ -187,18 +260,20 @@ class BleMidiSim: return self.read_frame(timeout=timeout) -# ---------------------------------------------------------- report -def write_report(R: Results, info: dict, path: str): - n = {s: sum(1 for r in R.rows if r[1] == s) - for s in ("PASS", "FAIL", "SKIP", "SENT")} +def write_report(R: Results, info: dict, path: str, notes: list[str] | None = None): + n = { + s: sum(1 for r in R.rows if r[1] == s) + for s in ("PASS", "FAIL", "SKIP", "SENT") + } lines = [ - "# BLE MIDI SysEx 协议测试报告", + "# BLE SysEx 协议测试报告", "", f"- 日期: {datetime.datetime.now():%Y-%m-%d %H:%M:%S}", f"- 设备: {info.get('name')} ({info.get('address')})", - f"- 传输: BLE-MIDI / bleak, 单包载荷 {info.get('max_payload')} 字节", + f"- 传输: {info.get('transport')} / bleak, payload={info.get('max_payload')}", + f"- notify: {info.get('notify')}", "- 协议来源: Doc/指令测试.docx", - "- 测试脚本: tools/test_protocol_ble.py (复用 test_protocol_app_sim.py 用例)", + "- 测试脚本: tools/test_protocol_ble.py", "", "## 汇总", "", @@ -208,21 +283,18 @@ def write_report(R: Results, info: dict, path: str): "", "## 明细", "", - "| # | 用例 | 结果 | 详情(RX/说明) |", - "|---|------|------|----------------|", + "| # | 用例 | 结果 | 详情 |", + "|---|------|------|------|", ] for i, (name, status, detail) in enumerate(R.rows, 1): lines.append(f"| {i} | {name} | {status} | {detail} |") + lines += ["", "## 备注", ""] + for note in notes or []: + lines.append(f"- {note}") lines += [ - "", - "## 备注", - "", - "- `01 03` 固件主版本回复为 MCU 实际版本 (0.2.5 -> 00 02 05 01),", - " 与文档样例 `21 05 7F 01` 编码不同 (文档未给编码规则)。", - "- `01 FF` 固件编码暂回 ASCII `BRS08L`, 文档标注待定。", - "- `04 xx`/`06 xx` 为无应答下发命令, SENT 表示已发送。", - "- `FD 01` 升级指令不属于本 UART4/BLE 应用协议范围, 未测。", - "- 蓝牙广播名由 Dream 模块固件决定 (Smart Guitar MIDI), 与 `01 02` 协议设备名无关。", + "- `01 03` 为 MCU 实际版本编码；`01 FF` 暂 `BRS08L`（文档待定）。", + "- `04 xx`/`06 xx` 无应答，SENT 表示已发送。", + "- `FD 01` 升级指令不在本协议范围。", ] os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: @@ -230,39 +302,52 @@ def write_report(R: Results, info: dict, path: str): return path -# ---------------------------------------------------------- selftest +def smoke(sim: BleMidiSim) -> Results: + R = Results() + r = sim.query(build(0x01, 0x01), timeout=3.0) + ok = r is not None and len(r) >= 5 and r[0] == HEAD and r[1] == 0x60 + R.add("smoke 01 01 连接设备", "PASS" if ok else "FAIL", hexs(r)) + if ok: + r2 = sim.query(build(0x01, 0x02), timeout=3.0) + ok2 = r2 is not None and r2[2:4] == bytes([0x01, 0x02]) + R.add("smoke 01 02 设备名", "PASS" if ok2 else "FAIL", hexs(r2)) + r3 = sim.query(build(0x02, 0x01), timeout=3.0) + ok3 = r3 is not None and len(r3) == 47 + R.add("smoke 02 01 和弦表(长帧)", "PASS" if ok3 else "FAIL", hexs(r3)) + return R + + def selftest(): R = Results() - # 短帧: 单包 pkts = ble_midi_encode_sysex(bytes.fromhex("F0600101F7"), 20) ok = pkts == [bytes.fromhex("8080F0600101F7")] - R.add("encode short", "PASS" if ok else "FAIL", - str([hexs(p) for p in pkts])) - # 长帧: 47B -> 多包, 且可解码还原 + R.add("encode short", "PASS" if ok else "FAIL", str([hexs(p) for p in pkts])) frame = bytes([0xF0, 0x60, 0x02, 0x01]) + bytes(range(1, 43)) + bytes([0xF7]) pkts = ble_midi_encode_sysex(frame, 20) back = b"".join(ble_midi_decode_packet(p) for p in pkts) - ok = back == frame and len(pkts) >= 3 - R.add("encode/decode 47B roundtrip", "PASS" if ok else "FAIL", - f"{len(pkts)} pkts") - # 解码: 单包带时间戳 - ok = ble_midi_decode_packet(bytes.fromhex("8080F0600101F7")) == bytes.fromhex("F0600101F7") - R.add("decode single pkt", "PASS" if ok else "FAIL", "") - # 解码: 续包(无时间戳) - ok = ble_midi_decode_packet(bytes.fromhex("80010203")) == bytes.fromhex("010203") - R.add("decode continuation", "PASS" if ok else "FAIL", "") + R.add( + "encode/decode 47B", + "PASS" if back == frame and len(pkts) >= 3 else "FAIL", + f"{len(pkts)} pkts", + ) return R.summary() def main(): - ap = argparse.ArgumentParser(description="BLE-MIDI 吉他协议全量测试") - ap.add_argument("--ble-name", default=DEFAULT_NAME, - help=f"广播名匹配 (默认 {DEFAULT_NAME!r})") - ap.add_argument("--ble-address", help="直接按 MAC/地址连接") - ap.add_argument("--allow-poweroff", action="store_true", - help="允许执行 05 00 关机用例") - ap.add_argument("--selftest", action="store_true", help="无硬件自检") - ap.add_argument("--report", help="报告输出路径 (默认 Doc/reports/ble_sysex_<时间戳>.md)") + ap = argparse.ArgumentParser(description="BLE 吉他协议全量测试") + ap.add_argument("--ble-name", default=DEFAULT_NAME) + ap.add_argument("--ble-address") + ap.add_argument( + "--transport", + choices=("midi", "uart"), + default="midi", + help="midi=标准 BLE-MIDI 帧; uart=自定义 e49a 原始 SysEx", + ) + ap.add_argument("--unpair", action="store_true", help="连接前先 unpair Windows 配对") + ap.add_argument("--allow-poweroff", action="store_true") + ap.add_argument("--smoke", action="store_true", help="仅冒烟: 01 01/02 + 02 01") + ap.add_argument("--selftest", action="store_true") + ap.add_argument("--report") args = ap.parse_args() if args.selftest: @@ -272,21 +357,40 @@ def main(): except ImportError: sys.exit("缺少 bleak: pip install bleak") - sim = BleMidiSim(name=args.ble_name, address=args.ble_address) - print(f"=== BLE 协议测试开始: {sim.info['name']} ({sim.info['address']}) " - f"payload={sim.info['max_payload']} ===") + sim = BleMidiSim( + name=args.ble_name, + address=args.ble_address, + transport=args.transport, + unpair=args.unpair, + ) + print( + f"=== BLE 测试: {sim.info.get('name')} ({sim.info.get('address')}) " + f"transport={sim.info.get('transport')} payload={sim.info.get('max_payload')} ===" + ) + notes = [ + f"transport={sim.info.get('transport')}", + "若 FAIL 且 timeout: 检查手机 App 是否占用、Windows 是否残留配对(--unpair)、吉他是否需完全断电重启。", + "实测: 标准 BLE-MIDI 可稳定连接但 SysEx 无应答; 自定义 uart 写 raw SysEx 会断连。", + ] try: - R = run_tests(sim, allow_poweroff=args.allow_poweroff) + R = smoke(sim) if args.smoke else run_tests(sim, allow_poweroff=args.allow_poweroff) finally: sim.close() code = R.summary() ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") - path = args.report or os.path.join( - os.path.dirname(os.path.abspath(__file__)), - "..", "..", "..", "Doc", "reports", f"ble_sysex_{ts}.md") - path = os.path.normpath(path) - write_report(R, sim.info, path) + path = args.report or os.path.normpath( + os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "..", + "..", + "..", + "Doc", + "reports", + f"ble_sysex_{ts}.md", + ) + ) + write_report(R, sim.info, path, notes=notes) print(f"\n报告已写入: {path}") sys.exit(code)