From 8febbd77428280efe8785adc4edb651141e76337 Mon Sep 17 00:00:00 2001 From: yuquanjun Date: Tue, 8 Sep 2026 18:05:04 +0800 Subject: [PATCH] Align BLE SysEx style/transpose replies with Doc protocol and add BLE test harness. Drop ineffective Dream BT rename SysEx; fix 03 04 boot/list/total and 03 07 read length. Co-authored-by: Cursor --- Global/Global.c | 3 - midi/midi_send.c | 29 ---- midi/midi_send.h | 3 - protocol/bl_uart_parse.c | 61 ++++++- tools/test_protocol_app_sim.py | 30 +++- tools/test_protocol_ble.py | 295 +++++++++++++++++++++++++++++++++ 6 files changed, 369 insertions(+), 52 deletions(-) create mode 100644 tools/test_protocol_ble.py diff --git a/Global/Global.c b/Global/Global.c index b156b35..ad17158 100644 --- a/Global/Global.c +++ b/Global/Global.c @@ -196,9 +196,6 @@ void System_PowerOn(void) app_tm1617_init(); XPT2046_Init(); App_Auto_Init(); - /* Dream boot settle, then set unique BLE/BT Audio ADV name via SysEx */ - rt_thread_mdelay(500); - Dream_ApplyUniqueBtNames(); LOG_I("PWR", "periph ready"); } diff --git a/midi/midi_send.c b/midi/midi_send.c index ef32c28..16016d8 100644 --- a/midi/midi_send.c +++ b/midi/midi_send.c @@ -56,32 +56,3 @@ void BleBuildDeviceName(char *out) out[5 + i] = hex[(tail >> (4 * (5 - i))) & 0xFu]; out[BLE_DEVICE_NAME_LEN] = '\0'; } - -/* - * Dream BT rename SysEx — reconstructed from factory tool - * 升级/.../bluetooth_set/set_bl_name.exe - * prefixes: f0554c45 (BLE), f0555441 (BT Audio) - * Format: F0 55 4C 45 F7 / F0 55 54 41 F7 - * Needs hardware confirmation after first flash. - */ -static void dream_send_name_sysex(uint8_t b1, uint8_t b2, uint8_t b3, const char *name) -{ - uint8_t msg[4 + BLE_DEVICE_NAME_LEN + 1]; - uint8_t n = BLE_DEVICE_NAME_LEN; - msg[0] = 0xF0; - msg[1] = b1; - msg[2] = b2; - msg[3] = b3; - memcpy(&msg[4], name, n); - msg[4 + n] = 0xF7; - SendMidiDataToDreamDSP(msg, (uint16_t)(5 + n)); -} - -void Dream_ApplyUniqueBtNames(void) -{ - char name[BLE_DEVICE_NAME_LEN + 1]; - BleBuildDeviceName(name); - dream_send_name_sysex(0x55, 0x4C, 0x45, name); /* BLE */ - dream_send_name_sysex(0x55, 0x54, 0x41, name); /* BT Audio */ - LOG_I("BT", "dream name set %s", name); -} diff --git a/midi/midi_send.h b/midi/midi_send.h index ef12d7b..df1646a 100644 --- a/midi/midi_send.h +++ b/midi/midi_send.h @@ -15,7 +15,4 @@ void DefaultTask_SendMsg(uint16_t Data1, uint16_t Data2, uint16_t Data3, uint16_ #define BLE_DEVICE_NAME_LEN 11 void BleBuildDeviceName(char *out); -/* Push name to Dream (BLE + BT Audio) via proprietary SysEx on USART2. */ -void Dream_ApplyUniqueBtNames(void); - #endif /* __MIDI_SEND_H__ */ diff --git a/protocol/bl_uart_parse.c b/protocol/bl_uart_parse.c index 4334e51..d977377 100644 --- a/protocol/bl_uart_parse.c +++ b/protocol/bl_uart_parse.c @@ -349,8 +349,9 @@ static void handleTranspose(uint8_t *data) { case 0: { - uint8_t ReturnTranspose[8] = {0xF0,0x60,0x03,0x07,0x00,0x00,0x00,0xF7}; - ReturnTranspose[5] = mGuiData[GUI_TRANSPOSE].Current; + /* doc: F0 60 03 07 00 F7 (7 bytes total) */ + uint8_t ReturnTranspose[7] = {0xF0,0x60,0x03,0x07,0x00,0x00,0xF7}; + ReturnTranspose[5] = (uint8_t)mGuiData[GUI_TRANSPOSE].Current; USART4_SendData(ReturnTranspose,sizeof(ReturnTranspose)); } break; @@ -740,17 +741,59 @@ static void handleRhythmStyle(uint8_t *data) StartFlag = 0; AutoBandTop1_Stop(); UI_ReloadTonePreset(); - resp[5] = (uint8_t)(mGuiData[GUI_SPEED].Current & 0x7F); /* doc: single-byte bpm */ + /* doc: single-byte bpm; clamp >127 (App should read exact bpm via 03 06) */ + resp[5] = (mGuiData[GUI_SPEED].Current > 0x7F) + ? 0x7F : (uint8_t)mGuiData[GUI_SPEED].Current; USART4_SendData(resp, sizeof(resp)); + + /* boot / song-entry special: F0 60 03 04 01 00 00 00 F7 + doc: also push current BPM as active report F0 51 03 F7 */ + if (data[5] == 0 && data[6] == 0 && data[7] == 0) + { + uint8_t bpmr[6] = {0xF0, 0x51, 0x03, 0x00, 0x00, 0xF7}; + bpmr[3] = (uint8_t)(mGuiData[GUI_SPEED].Current / 128); + bpmr[4] = (uint8_t)(mGuiData[GUI_SPEED].Current % 128); + USART4_SendData(bpmr, sizeof(bpmr)); + } } break; - case 2: /* user style list page: reply total count */ - case 3: /* user style total */ + case 2: /* user style list page: F0 60 03 04 02 F7 + reply: F0 60 03 04 02 [ ...] F7 */ { - uint8_t resp[8] = {0xF0, 0x60, 0x03, 0x04, 0x00, 0x00, 0x00, 0xF7}; - resp[4] = data[4]; - resp[5] = (uint8_t)(LOCAL_SONG_COUNT / 128); - resp[6] = (uint8_t)(LOCAL_SONG_COUNT % 128); + uint8_t resp[64]; + uint32_t from = 0, to = LOCAL_SONG_COUNT; + uint16_t i, n = 0, len; + /* from/to: 24-bit big-endian, optional (doc example: 00 00 00 07) */ + if (data[5] || data[6] || data[7]) + { + from = ((uint32_t)data[5] << 16) | ((uint32_t)data[6] << 8) | data[7]; + to = ((uint32_t)data[8] << 16) | ((uint32_t)data[9] << 8) | data[10]; + } + if (from > LOCAL_SONG_COUNT) from = LOCAL_SONG_COUNT; + if (to > LOCAL_SONG_COUNT || to <= from) to = LOCAL_SONG_COUNT; + resp[0] = 0xF0; resp[1] = 0x60; resp[2] = 0x03; resp[3] = 0x04; resp[4] = 0x02; + len = 7; /* count filled after loop */ + for (i = (uint16_t)from; i < to; i++) + { + if (len + 6 + 1 > sizeof(resp)) break; + resp[len++] = (uint8_t)(i / 128); + resp[len++] = (uint8_t)(i % 128); + memcpy(&resp[len], LocalSongCode[i], 4); + len += 4; + n++; + } + resp[5] = (uint8_t)(n / 128); + resp[6] = (uint8_t)(n % 128); + resp[len++] = 0xF7; + USART4_SendData(resp, len); + } + break; + case 3: /* user style total: reply F0 60 03 04 00 F7 (doc: 00 00 1E) */ + { + uint8_t resp[9] = {0xF0, 0x60, 0x03, 0x04, 0x00, 0x00, 0x00, 0x00, 0xF7}; + resp[5] = (uint8_t)((LOCAL_SONG_COUNT >> 16) & 0xFF); + resp[6] = (uint8_t)((LOCAL_SONG_COUNT >> 8) & 0xFF); + resp[7] = (uint8_t)(LOCAL_SONG_COUNT & 0xFF); USART4_SendData(resp, sizeof(resp)); } break; diff --git a/tools/test_protocol_app_sim.py b/tools/test_protocol_app_sim.py index 57fb2b9..cc85683 100644 --- a/tools/test_protocol_app_sim.py +++ b/tools/test_protocol_app_sim.py @@ -231,20 +231,34 @@ def run_tests(sim, allow_poweroff=False): R.add("03 04 读当前节奏风格", "PASS" if ok else "FAIL", d) saved_style = bytes(r[5:8]) if ok else None + # 开机特例: F0 60 03 04 01 00 00 00 F7 -> F0 60 03 04 01 F7 + F0 51 03 F7 r = sim.query(build(0x03, 0x04, 0x01, 0x00, 0x00, 0x00)) ok, d = expect(r, (0x60, 0x03, 0x04, 0x01), 7) - R.add("03 04 设系统风格#0", "PASS" if ok else "FAIL", d) + R.add("03 04 开机特例(读风格#0)", "PASS" if ok else "FAIL", d) + r51 = sim.read_frame(timeout=1.0, want_dev=True) + ok51 = (r51 is not None and len(r51) == 6 + and tuple(r51[:3]) == (HEAD, DEV_ID, 0x03)) + R.add("03 04 开机特例(F0 51 BPM上报)", + "PASS" if ok51 else "FAIL", hexs(r51)) r = sim.query(build(0x03, 0x04, 0x01, 0x01, 0x00, 0x00)) ok, d = expect(r, (0x60, 0x03, 0x04, 0x01), 7) R.add("03 04 设用户风格#0", "PASS" if ok else "FAIL", d) + # 用户风格分页: 回复 F0 60 03 04 02 [ ...] F7 r = sim.query(build(0x03, 0x04, 0x02, 0x00, 0x00, 0x00, 0x07)) - ok, d = expect(r, (0x60, 0x03, 0x04, 0x02), 8) + ok, d = expect(r, (0x60, 0x03, 0x04, 0x02)) + if ok: + cnt = r[5] * 128 + r[6] + ok = len(r) == 7 + cnt * 6 + 1 + d += f" cnt={cnt}" R.add("03 04 用户风格分页", "PASS" if ok else "FAIL", d) + # 用户风格总数: 回复 F0 60 03 04 00 F7 r = sim.query(build(0x03, 0x04, 0x03)) - ok, d = expect(r, (0x60, 0x03, 0x04, 0x03), 8) + ok, d = expect(r, (0x60, 0x03, 0x04, 0x00), 9) + if ok: + d += f" total={(r[5] << 16) | (r[6] << 8) | r[7]}" R.add("03 04 用户风格总数", "PASS" if ok else "FAIL", d) if saved_style: @@ -277,9 +291,9 @@ def run_tests(sim, allow_poweroff=False): R.add("03 06 写BPM=120(写+读回)", "PASS" if ok else "FAIL", hexs(r)) sim.send(build(0x03, 0x06, 0x01, orig // 128, orig % 128)) # restore - # 移调: 读 -> 写 -> 读 -> 恢复 + # 移调: 读 -> 写 -> 读 -> 恢复 (doc: F0 60 03 07 00 F7, 7 bytes) r0 = sim.query(build(0x03, 0x07, 0x00)) - ok, d = expect(r0, (0x60, 0x03, 0x07, 0x00), 8) + ok, d = expect(r0, (0x60, 0x03, 0x07, 0x00), 7) R.add("03 07 读移调", "PASS" if ok else "FAIL", d) if ok: orig = r0[5] @@ -287,7 +301,7 @@ def run_tests(sim, allow_poweroff=False): sim.send(build(0x03, 0x07, 0x01, new)) time.sleep(0.1) r = sim.query(build(0x03, 0x07, 0x00)) - ok = r is not None and len(r) == 8 and r[5] == new + ok = r is not None and len(r) == 7 and r[5] == new R.add("03 07 写移调(写+读回)", "PASS" if ok else "FAIL", hexs(r)) sim.send(build(0x03, 0x07, 0x01, orig)) # restore @@ -320,7 +334,7 @@ def run_tests(sim, allow_poweroff=False): else: R.add("05 00 复位/关机", "SKIP", "需 --allow-poweroff") - return R.summary() + return R # ---------------------------------------------------------------- selftest @@ -359,7 +373,7 @@ def main(): sim = AppSim(args.port, args.baud) print(f"=== 协议测试开始 {args.port}@{args.baud} ===") try: - code = run_tests(sim, allow_poweroff=args.allow_poweroff) + code = run_tests(sim, allow_poweroff=args.allow_poweroff).summary() finally: sim.close() sys.exit(code) diff --git a/tools/test_protocol_ble.py b/tools/test_protocol_ble.py new file mode 100644 index 0000000..ee4485c --- /dev/null +++ b/tools/test_protocol_ble.py @@ -0,0 +1,295 @@ +# -*- coding: utf-8 -*- +""" +test_protocol_ble.py — 通过笔记本蓝牙(BLE-MIDI)对吉他做全协议测试并出报告 + +协议来源: Doc/指令测试.docx +链路: 笔记本 BLE <-> "Smart Guitar MIDI" (BLE-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) + +用法: + 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 # 无硬件自检(编解码) + +依赖: bleak (pip install bleak) +报告: Doc/reports/ble_sysex_<时间戳>.md +退出码: 0 = 全部通过, 1 = 有 FAIL +""" +import argparse +import asyncio +import datetime +import os +import queue +import sys +import threading +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) + +MIDI_SERVICE = "03B80E5A-EDE8-4B33-A751-6CE34EC4C700" +MIDI_CHAR = "7772E5DB-3868-4112-A1A9-F2669D106BF3" +DEFAULT_NAME = "Smart Guitar MIDI" +DEFAULT_MTU_PAYLOAD = 20 # ATT 默认 23 -> 应用载荷 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 + 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 + 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 + 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): + 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._loop = asyncio.new_event_loop() + 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): + return asyncio.run_coroutine_threadsafe(coro, self._loop).result(timeout) + + async def _connect(self, name, address, scan_timeout): + dev = None + 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: + 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}") + + client = self._BleakClient(dev) + await client.connect() + char = None + 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 + + 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} + + async def _disconnect(self): + if self._client is not None: + try: + await self._client.stop_notify(self._char) + except Exception: + pass + await self._client.disconnect() + + # ---- AppSim-compatible sync API ---- + def close(self): + try: + self._run(self._disconnect(), timeout=10) + except Exception: + pass + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(timeout=5) + + def _pump(self): + while True: + try: + data = self._rx.get_nowait() + except queue.Empty: + break + self.pending.extend(self.parser.feed(ble_midi_decode_packet(data))) + + 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) + + def read_frame(self, timeout=1.0, want_dev=False): + deadline = time.monotonic() + timeout + while True: + self._pump() + for i, f in enumerate(self.pending): + is_dev = len(f) > 1 and f[1] == DEV_ID + if is_dev == want_dev: + return self.pending.pop(i) + self.pending.clear() + if time.monotonic() >= deadline: + return None + time.sleep(0.01) + + def drain(self, quiet=0.2): + while self.read_frame(timeout=quiet) is not None: + pass + + def query(self, frame: bytes, timeout=1.0): + self.drain() + self.send(frame) + 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")} + lines = [ + "# BLE MIDI 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')} 字节", + "- 协议来源: Doc/指令测试.docx", + "- 测试脚本: tools/test_protocol_ble.py (复用 test_protocol_app_sim.py 用例)", + "", + "## 汇总", + "", + "| PASS | FAIL | SKIP | SENT |", + "|------|------|------|------|", + f"| {n['PASS']} | {n['FAIL']} | {n['SKIP']} | {n['SENT']} |", + "", + "## 明细", + "", + "| # | 用例 | 结果 | 详情(RX/说明) |", + "|---|------|------|----------------|", + ] + for i, (name, status, detail) in enumerate(R.rows, 1): + lines.append(f"| {i} | {name} | {status} | {detail} |") + 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` 协议设备名无关。", + ] + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(lines) + "\n") + return path + + +# ---------------------------------------------------------- selftest +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 -> 多包, 且可解码还原 + 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", "") + 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)") + args = ap.parse_args() + + if args.selftest: + sys.exit(selftest()) + try: + import bleak # noqa: F401 + 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']} ===") + try: + R = 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) + print(f"\n报告已写入: {path}") + sys.exit(code) + + +if __name__ == "__main__": + main()