diff --git a/tools/_ble_channel_matrix.py b/tools/_ble_channel_matrix.py new file mode 100644 index 0000000..d0162df --- /dev/null +++ b/tools/_ble_channel_matrix.py @@ -0,0 +1,140 @@ +# -*- 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()) diff --git a/tools/_probe_ble_rtt.py b/tools/_probe_ble_rtt.py new file mode 100644 index 0000000..60cf7be --- /dev/null +++ b/tools/_probe_ble_rtt.py @@ -0,0 +1,126 @@ +# -*- coding: utf-8 -*- +"""Probe: RTT log + BLE uartw write to see if MCU gets F0 60.""" +import asyncio +import threading +import time +import pylink +from bleak import BleakScanner, BleakClient + +UARTW = "e49a25e0-f69a-11e8-8eb2-f2801f1b9fd1" +UARTN = "e49a28e1-f69a-11e8-8eb2-f2801f1b9fd1" +MIDI = "7772e5db-3868-4112-a1a9-f2669d106bf3" + + +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 connected", 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 connect failed: {last}") + + +async def find_dev(timeout=60): + t0 = time.time() + while time.time() - t0 < timeout: + d = await BleakScanner.find_device_by_filter( + lambda d, a: d.name and "Smart Guitar" in d.name, timeout=8 + ) + if d: + return d + print("waiting BLE...", flush=True) + return None + + +async def main(): + j = open_jlink() + j.rtt_start() + 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 err", e, flush=True) + break + time.sleep(0.03) + + th = threading.Thread(target=reader, daemon=True) + th.start() + print("collect boot logs 3s...", flush=True) + await asyncio.sleep(3) + + d = await find_dev(90) + if not d: + print("NO BLE DEVICE", flush=True) + stop = True + j.close() + return + print("BLE", d.name, d.address, flush=True) + + try: + async with BleakClient(d, timeout=30) as c: + print("connected", flush=True) + notifs = [] + + def cb(_s, data): + notifs.append(bytes(data)) + print("NOTIFY", data.hex(), flush=True) + + for u in (UARTN, MIDI): + try: + await c.start_notify(u, cb) + print("notify ok", u[:8], flush=True) + except Exception as e: + print("notify fail", u[:8], e, flush=True) + + await asyncio.sleep(0.5) + print("WRITE uartw F0 60 01 01 F7", flush=True) + try: + await c.write_gatt_char(UARTW, bytes.fromhex("F0600101F7"), response=False) + print("write returned", flush=True) + except Exception as e: + print("write err", e, flush=True) + await asyncio.sleep(4) + print("notifs", len(notifs), "conn", c.is_connected, flush=True) + except Exception as e: + print("BLE session err", e, flush=True) + + await asyncio.sleep(1) + stop = True + time.sleep(0.3) + j.close() + print("RTT total chunks", len(lines), flush=True) + ble_hits = [x for x in lines if "BLE" in x or "sysex" in x] + print("BLE-related RTT lines:", ble_hits, flush=True) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tools/_test_normal_pick.py b/tools/_test_normal_pick.py new file mode 100644 index 0000000..16353a4 --- /dev/null +++ b/tools/_test_normal_pick.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""RTT verify normal-mode: hold pad + pick sounds; release stops. + +Uses real physical pad hold (inject is cleared by TM1629 scan within one tick). +""" +from __future__ import annotations + +import struct +import sys +import time + +sys.path.insert(0, ".") +from rtt_pitch_reg_test import ( # noqa: E402 + RttSession, + connect_jlink, + find_rtt_control_block, + wait_rtt_ready, +) + +SPEED_TAIL = struct.pack(" int: + return sum(1 for L in lines if "[PITCH]" in L and " on " in L) + + +def parse_tab(lines: list[str]) -> int | None: + for L in reversed(lines): + if "TONE tab=" not in L: + continue + try: + return int(L.split("tab=")[1].split()[0]) + except Exception: + continue + return None + + +def with_halt(jlink, fn): + was = jlink.halted() + if not was: + jlink.halt() + try: + return fn() + finally: + if not was: + if hasattr(jlink, "restart"): + jlink.restart() + else: + jlink.go() + time.sleep(0.02) + + +def find_pattern(jlink, needle: bytes, ram_base=0x20000000, ram_size=0x18000) -> int | None: + def _find(): + chunk = 0x1000 + for off in range(0, ram_size, chunk): + data = bytes(jlink.memory_read8(ram_base + off, min(chunk, ram_size - off))) + idx = data.find(needle) + if idx >= 0: + return ram_base + off + idx + return None + + return with_halt(jlink, _find) + + +def mem_write(jlink, addr: int, data: bytes) -> None: + with_halt(jlink, lambda: jlink.memory_write8(addr, list(data))) + + +def mem_read(jlink, addr: int, n: int) -> bytes: + return with_halt(jlink, lambda: bytes(jlink.memory_read8(addr, n))) + + +def force_tab_normal(jlink) -> bool: + hit = find_pattern(jlink, SPEED_TAIL) + if hit is None: + print("!! GUI_SPEED row not found", flush=True) + return False + base = (hit - 2) - GUI_SWITCH_SIZE + tab_addr = base + TAB_INDEX * GUI_SWITCH_SIZE + mem_write(jlink, tab_addr, struct.pack(" None: + print(f"\n>>> {msg}", flush=True) + print(f" ({seconds:.0f}s countdown)", flush=True) + for left in range(int(seconds), 0, -1): + print(f" {left}...", flush=True) + time.sleep(1.0) + + +def main() -> int: + j = connect_jlink("Cortex-M4") + cb = find_rtt_control_block(j) + if not cb: + raise SystemExit("RTT CB not found") + print(f"RTT CB @ 0x{cb:08X}", flush=True) + j.rtt_start(cb) + wait_rtt_ready(j) + sess = RttSession(j) + + print("\n=== Setup: 1.bin + TAB=普通(2) ===", flush=True) + sess.cmd("log clear", 0.2) + sess.cmd_ack("tone bin1 0", "TONE_BIN1", timeout_s=4.0, retries=4) + if not force_tab_normal(j): + j.rtt_stop() + j.close() + return 3 + sess.cmd_ack("tone status", "TONE tab=", timeout_s=3.0, retries=4) + print(f"tab={parse_tab(sess.lines)}", flush=True) + + # Stop any leftover + sess.cmd_ack("chord key 23", "CHORD_KEY_OK", timeout_s=2.0, retries=3) + force_tab_normal(j) + + print("\n=== TEST1: pick with NO pad hold (expect silence) ===", flush=True) + print("请双手离开指板。", flush=True) + wait_enter("双手离开指板后等待自动拨片测试", 3) + force_tab_normal(j) + sess.cmd("log clear", 0.2) + sess.cmd_ack("tone start", "TONE_START_DONE", timeout_s=3.0, retries=5) + n1 = count_pitch(sess.pump(1.5)) + print(f"TEST1 PITCH on={n1} (expect 0)", flush=True) + + print("\n=== TEST2: hold pad + pick (expect sound) ===", flush=True) + wait_enter("请按住指板任意和弦键不放（建议按住第2排中部）", 5) + force_tab_normal(j) + sess.cmd("log clear", 0.2) + # keep KEY from physical hold; tone start without clearing if we use 'k' + # But we don't know KEY — physical hold already set KEY_ID+PressFlag via scan. + # tone start WITHOUT k clears KEY_ID! Must use tone start k — but then KEY kept. + # If user is holding, KEY_ID is set; PressFlag is set. tone start sets StartFlag=0 + # then clears KEY unless k. Use: don't clear — need tone start k AND physical KEY already set. + sess.cmd_ack("tone start k", "TONE_START_DONE", timeout_s=3.0, retries=5) + play = sess.pump(2.5) + n2 = count_pitch(play) + print(f"TEST2 PITCH on={n2} (expect >0) — keep holding!", flush=True) + for L in play: + if "[PITCH]" in L: + print(" ", L, flush=True) + break + + print("\n=== TEST3: release pad (expect stop) ===", flush=True) + wait_enter("请松开指板（松手）", 3) + force_tab_normal(j) + sess.cmd("log clear", 0.15) + post = sess.pump(2.0) + n3 = count_pitch(post) + print(f"TEST3 PITCH on after release={n3} (expect 0)", flush=True) + + print("\n=== TEST4: hold again + pick (expect sound) ===", flush=True) + wait_enter("请再次按住指板和弦键不放", 5) + force_tab_normal(j) + sess.cmd("log clear", 0.2) + sess.cmd_ack("tone start k", "TONE_START_DONE", timeout_s=3.0, retries=5) + n4 = count_pitch(sess.pump(2.0)) + print(f"TEST4 PITCH on={n4} (expect >0)", flush=True) + print("可松开指板。", flush=True) + time.sleep(1.0) + sess.cmd_ack("chord key 23", "CHORD_KEY_OK", timeout_s=2.0, retries=3) + + t1, t2, t3, t4 = (n1 == 0), (n2 >= 1), (n3 == 0), (n4 >= 1) + print("\n======== RESULT ========", flush=True) + print(f"TEST1 no-hold silence: {'PASS' if t1 else 'FAIL'} (n={n1})", flush=True) + print(f"TEST2 hold+pick sound: {'PASS' if t2 else 'FAIL'} (n={n2})", flush=True) + print(f"TEST3 release stop: {'PASS' if t3 else 'FAIL'} (n={n3})", flush=True) + print(f"TEST4 re-trigger: {'PASS' if t4 else 'FAIL'} (n={n4})", flush=True) + ok_all = t1 and t2 and t3 and t4 + print("OVERALL:", "PASS" if ok_all else "FAIL", flush=True) + j.rtt_stop() + j.close() + return 0 if ok_all else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/_usb_midi_sysex_probe.py b/tools/_usb_midi_sysex_probe.py new file mode 100644 index 0000000..1f2c4f9 --- /dev/null +++ b/tools/_usb_midi_sysex_probe.py @@ -0,0 +1,158 @@ +# -*- coding: utf-8 -*- +"""Quick USB MIDI (SAM5704) SysEx probe for App F0 60 frames.""" +from __future__ import annotations + +import ctypes +import ctypes.wintypes as w +import sys +import time + +winmm = ctypes.WinDLL("winmm") +CALLBACK_FUNCTION = 0x30000 +MIM_DATA, MIM_LONGDATA = 0x3C3, 0x3C4 +MHDR_DONE = 0x01 + + +class MIDIINCAPS(ctypes.Structure): + _fields_ = [ + ("wMid", w.WORD), ("wPid", w.WORD), ("vDriverVersion", w.DWORD), + ("szPname", ctypes.c_char * 32), ("dwSupport", w.DWORD), + ] + + +class MIDIOUTCAPS(ctypes.Structure): + _fields_ = [ + ("wMid", w.WORD), ("wPid", w.WORD), ("vDriverVersion", w.DWORD), + ("szPname", ctypes.c_char * 32), ("wTechnology", w.WORD), + ("wVoices", w.WORD), ("wNotes", w.WORD), ("wChannelMask", w.WORD), + ("dwSupport", w.DWORD), + ] + + +class MIDIHDR(ctypes.Structure): + _fields_ = [ + ("lpData", ctypes.c_void_p), + ("dwBufferLength", w.DWORD), + ("dwBytesRecorded", w.DWORD), + ("dwUser", ctypes.c_void_p), + ("dwFlags", w.DWORD), + ("lpNext", ctypes.c_void_p), + ("reserved", ctypes.c_void_p), + ("dwOffset", w.DWORD), + ("dwReserved", ctypes.c_size_t * 8), + ] + + +def find_sam(kind: str): + n = winmm.midiInGetNumDevs() if kind == "in" else winmm.midiOutGetNumDevs() + for i in range(n): + if kind == "in": + c = MIDIINCAPS() + winmm.midiInGetDevCapsA(i, ctypes.byref(c), ctypes.sizeof(c)) + else: + c = MIDIOUTCAPS() + winmm.midiOutGetDevCapsA(i, ctypes.byref(c), ctypes.sizeof(c)) + name = c.szPname.decode("mbcs", "replace") + if "SAM5704" in name: + return i, name + return None, None + + +def hx(b: bytes) -> str: + return " ".join(f"{x:02X}" for x in b) + + +def main() -> int: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + in_i, in_n = find_sam("in") + out_i, out_n = find_sam("out") + print(f"IN={in_i} {in_n} OUT={out_i} {out_n}", flush=True) + if in_i is None or out_i is None: + print("SAM5704 not found") + return 1 + + received: list[bytes] = [] + keep = [] # keep buffer refs alive + + MidiInProc = ctypes.WINFUNCTYPE( + None, w.HANDLE, w.UINT, ctypes.c_void_p, ctypes.c_size_t, ctypes.c_size_t + ) + + @MidiInProc + def on_midi_in(h, msg, _inst, p1, p2): + if msg == MIM_LONGDATA: + hdr = ctypes.cast(p1, ctypes.POINTER(MIDIHDR)).contents + n = hdr.dwBytesRecorded + if n and hdr.lpData: + data = ctypes.string_at(hdr.lpData, n) + received.append(data) + print("RX", hx(data), flush=True) + winmm.midiInAddBuffer(h, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + elif msg == MIM_DATA: + print( + f"RX SHORT {(p1 & 0xFF):02X} {((p1 >> 8) & 0xFF):02X} {((p1 >> 16) & 0xFF):02X}", + flush=True, + ) + + h_in = w.HANDLE() + r = winmm.midiInOpen(ctypes.byref(h_in), in_i, on_midi_in, 0, CALLBACK_FUNCTION) + print("midiInOpen", r, flush=True) + if r != 0: + return r + + for _ in range(4): + buf = ctypes.create_string_buffer(1024) + hdr = MIDIHDR() + hdr.lpData = ctypes.cast(buf, ctypes.c_void_p) + hdr.dwBufferLength = 1024 + keep.append((buf, hdr)) + winmm.midiInPrepareHeader(h_in, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + winmm.midiInAddBuffer(h_in, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + winmm.midiInStart(h_in) + + h_out = w.HANDLE() + r = winmm.midiOutOpen(ctypes.byref(h_out), out_i, 0, 0, 0) + print("midiOutOpen", r, flush=True) + if r != 0: + return r + + def send_sysex(msg: bytes): + buf = ctypes.create_string_buffer(msg) + hdr = MIDIHDR() + hdr.lpData = ctypes.cast(buf, ctypes.c_void_p) + hdr.dwBufferLength = len(msg) + keep.append((buf, hdr)) + winmm.midiOutPrepareHeader(h_out, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + print("TX", hx(msg), flush=True) + winmm.midiOutLongMsg(h_out, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + t0 = time.time() + while not (hdr.dwFlags & MHDR_DONE) and time.time() - t0 < 2: + time.sleep(0.01) + winmm.midiOutUnprepareHeader(h_out, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + + for label, msg in [ + ("01 01", bytes.fromhex("F0600101F7")), + ("01 02", bytes.fromhex("F0600102F7")), + ("01 03", bytes.fromhex("F0600103F7")), + ]: + print("---", label, flush=True) + n0 = len(received) + send_sysex(msg) + time.sleep(1.2) + print(" new rx", len(received) - n0, flush=True) + + print("TOTAL RX", len(received), flush=True) + app_like = [x for x in received if len(x) >= 2 and x[0] == 0xF0 and x[1] == 0x60] + print("F0 60 replies", len(app_like), flush=True) + + winmm.midiInStop(h_in) + winmm.midiInReset(h_in) + for buf, hdr in keep[:4]: + winmm.midiInUnprepareHeader(h_in, ctypes.byref(hdr), ctypes.sizeof(MIDIHDR)) + winmm.midiInClose(h_in) + winmm.midiOutClose(h_out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/ble_log_pull.py b/tools/ble_log_pull.py new file mode 100644 index 0000000..cbd3373 --- /dev/null +++ b/tools/ble_log_pull.py @@ -0,0 +1,343 @@ +# -*- coding: utf-8 -*- +""" +ble_log_pull.py — 通过 BLE 拉取 K1 设备现场日志分区 (LOG.BIN) + +协议 (SysEx, App->设备 F0 60 ... F7): + 07 00 查询 meta + 07 01 按 offset 读取分片 (每片最多 20 原始字节, nibble 编码) + 07 02 清空 (需确认码 55 2A) + +注意: 本模组上 UART GATT 裸写容易导致断连；默认优先 BLE-MIDI。 + +依赖: bleak +用法: + python ble_log_pull.py --out LOG.BIN --decode crash.log -v + python ble_log_pull.py --transport midi --address CB:4E:FD:F1:C0:79 --out LOG.BIN +""" +from __future__ import annotations + +import argparse +import os +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from test_protocol_ble import BleMidiSim, DEFAULT_NAME # noqa: E402 +from test_protocol_app_sim import HEAD, APP_ID, build, hexs # noqa: E402 + +CHUNK = 20 + + +def u32_to_7bit(v: int) -> bytes: + v &= 0xFFFFFFFF + return bytes( + [ + (v >> 28) & 0x7F, + (v >> 21) & 0x7F, + (v >> 14) & 0x7F, + (v >> 7) & 0x7F, + v & 0x7F, + ] + ) + + +def u32_from_7bit(b: bytes, i: int = 0) -> int: + return ( + ((b[i] & 0x7F) << 28) + | ((b[i + 1] & 0x7F) << 21) + | ((b[i + 2] & 0x7F) << 14) + | ((b[i + 3] & 0x7F) << 7) + | (b[i + 4] & 0x7F) + ) + + +def nibbles_to_bytes(nibs: bytes) -> bytes: + out = bytearray() + for i in range(0, len(nibs) - 1, 2): + out.append(((nibs[i] & 0x0F) << 4) | (nibs[i + 1] & 0x0F)) + return bytes(out) + + +def parse_meta(frame: bytes) -> dict: + if len(frame) < 36 or frame[2] != 0x07 or frame[3] != 0x00: + raise ValueError(f"bad meta frame len={len(frame)}: {hexs(frame)}") + i = 4 + valid = frame[i] + i += 1 + ver = frame[i] + i += 1 + size = u32_from_7bit(frame, i) + i += 5 + data_size = u32_from_7bit(frame, i) + i += 5 + write_off = u32_from_7bit(frame, i) + i += 5 + wrap = u32_from_7bit(frame, i) + i += 5 + boot = u32_from_7bit(frame, i) + i += 5 + lines = u32_from_7bit(frame, i) + return { + "valid": valid, + "ver": ver, + "partition_size": size, + "data_size": data_size, + "write_off": write_off, + "wrap_count": wrap, + "boot_count": boot, + "line_count": lines, + } + + +def parse_chunk(frame: bytes) -> tuple[int, bytes]: + if len(frame) < 11 or frame[2] != 0x07 or frame[3] != 0x01: + raise ValueError(f"bad chunk frame: {hexs(frame)}") + off = u32_from_7bit(frame, 4) + n = frame[9] + nibs = frame[10 : 10 + n] + if len(nibs) < n: + raise ValueError("truncated chunk") + return off, nibbles_to_bytes(nibs) + + +def safe_send(sim: BleMidiSim, frame: bytes) -> None: + if not sim.is_connected(): + raise RuntimeError("Not connected") + sim.send(frame) + + +def wait_cmd(sim: BleMidiSim, cmd1: int, cmd2: int, timeout: float, verbose: bool): + t0 = time.monotonic() + others = [] + while time.monotonic() - t0 < timeout: + if not sim.is_connected(): + if verbose: + print(" (wait) link dropped") + return None + sim._pump() + i = 0 + while i < len(sim.pending): + fr = sim.pending[i] + if ( + len(fr) >= 4 + and fr[0] == HEAD + and fr[1] == APP_ID + and fr[2] == cmd1 + and fr[3] == cmd2 + ): + sim.pending.pop(i) + return fr + others.append(sim.pending.pop(i)) + continue + time.sleep(0.01) + if verbose and others: + print(f" (timeout) saw {len(others)} other frame(s), last={hexs(others[-1][:32])}") + elif verbose: + print(" (timeout) no frames received at all") + return None + + +def query_cmd( + sim: BleMidiSim, + frame: bytes, + cmd1: int, + cmd2: int, + timeout: float, + verbose: bool, + retries: int = 2, +): + # 轻量 drain，避免 read_frame 清空 pending 时误伤 + t_drain = time.monotonic() + 0.15 + while time.monotonic() < t_drain: + sim._pump() + sim.pending.clear() + time.sleep(0.02) + + for attempt in range(retries + 1): + if verbose: + print(f" TX[{attempt}]: {hexs(frame)}") + try: + safe_send(sim, frame) + except Exception as e: + if verbose: + print(f" TX fail: {e}") + return None + fr = wait_cmd(sim, cmd1, cmd2, timeout=timeout, verbose=verbose) + if fr is not None: + if verbose: + print(f" RX: {hexs(fr[:48])}{'...' if len(fr) > 48 else ''}") + return fr + time.sleep(0.25) + return None + + +def soft_probe(sim: BleMidiSim, verbose: bool = False) -> bool: + """只用 01 01 探活，失败时不立刻 unpair（Windows 上 unpair 易把模组打哑巴）。""" + time.sleep(0.6) # 连接后给模组 settle + r = query_cmd(sim, build(0x01, 0x01), 0x01, 0x01, timeout=3.0, verbose=verbose, retries=2) + if r is not None and len(r) >= 5 and r[0] == HEAD and r[1] == APP_ID: + return True + if verbose: + print(" soft probe failed; try probe_or_rebind once") + try: + return bool( + sim.probe_or_rebind( + name=sim.info.get("name") or DEFAULT_NAME, + address=sim.info.get("address"), + ) + ) + except Exception as e: + if verbose: + print(f" probe_or_rebind exception: {e}") + return False + + +def pull(sim: BleMidiSim, out_path: str, timeout: float = 4.0, verbose: bool = False) -> dict: + if not soft_probe(sim, verbose=verbose): + raise RuntimeError( + "BLE 协议无应答(01 01)。请确认琴已开机、蓝牙开关为开；" + "若刚用过 --transport uart 导致断连，请断电复位后再用 --transport midi。" + ) + + meta_frame = query_cmd( + sim, build(0x07, 0x00), 0x07, 0x00, timeout=timeout, verbose=verbose, retries=3 + ) + if meta_frame is None: + ver = query_cmd( + sim, build(0x01, 0x03), 0x01, 0x03, timeout=2.0, verbose=verbose, retries=1 + ) + tip = "" + if ver is not None: + tip = ( + " 普通指令(01 03)正常，但 07 00 无应答 → 固件可能未含日志协议，请重新烧录。" + ) + elif not sim.is_connected(): + tip = " 链路已断开（uart 传输常见）。请改用: python ble_log_pull.py --transport midi ..." + raise RuntimeError("timeout waiting log meta (07 00)." + tip) + + meta = parse_meta(meta_frame) + print( + f"meta valid={meta['valid']} ver={meta['ver']} size={meta['partition_size']} " + f"write={meta['write_off']} wrap={meta['wrap_count']} " + f"boot={meta['boot_count']} lines={meta['line_count']}" + ) + if not meta["valid"] or meta["partition_size"] == 0: + raise RuntimeError("device log partition invalid") + + total = meta["partition_size"] + buf = bytearray(b"\xff" * total) + got = 0 + off = 0 + while off < total: + want = min(CHUNK, total - off) + req = bytes([HEAD, APP_ID, 0x07, 0x01]) + u32_to_7bit(off) + bytes([want, 0xF7]) + chunk = None + for attempt in range(4): + if not sim.is_connected(): + raise RuntimeError(f"link dropped at offset 0x{off:X}") + try: + safe_send(sim, req) + except Exception as e: + time.sleep(0.1 * (attempt + 1)) + if verbose: + print(f" chunk TX fail @0x{off:X}: {e}") + continue + fr = wait_cmd(sim, 0x07, 0x01, timeout=timeout, verbose=False) + if fr is None: + time.sleep(0.05 * (attempt + 1)) + continue + try: + roff, data = parse_chunk(fr) + except ValueError: + continue + if roff == off and data: + chunk = data + break + if chunk is None: + raise RuntimeError(f"timeout at offset 0x{off:X} (after retries)") + buf[off : off + len(chunk)] = chunk + got += len(chunk) + off += len(chunk) + if off % 4096 == 0 or off >= total: + print(f" {off}/{total} ({100.0 * off / total:.1f}%)") + + with open(out_path, "wb") as f: + f.write(buf) + print(f"wrote {out_path} ({got} bytes)") + return meta + + +def main(): + ap = argparse.ArgumentParser(description="Pull K1 field log over BLE") + ap.add_argument("--out", default="LOG.BIN") + ap.add_argument("--name", default=DEFAULT_NAME) + ap.add_argument("--address", default=None) + ap.add_argument( + "--transport", + choices=("midi", "uart", "auto"), + default="midi", + help="默认 midi（uart 裸写在本模组上容易 Not connected）", + ) + ap.add_argument("--decode", default=None, help="also decode to this text path") + ap.add_argument("--clear", action="store_true", help="clear flash log after pull") + ap.add_argument("-v", "--verbose", action="store_true") + ap.add_argument("--timeout", type=float, default=4.0) + args = ap.parse_args() + + transports = [args.transport] + if args.transport == "auto": + # midi 优先：uart GATT 写可能导致模组断连 + transports = ["midi", "uart"] + elif args.transport == "uart": + print( + "警告: --transport uart 在本机模组上常导致 Not connected；" + "若失败请改用 --transport midi" + ) + + last_err = None + for tr in transports: + print(f"=== transport={tr} ===") + sim = None + try: + sim = BleMidiSim(name=args.name, address=args.address, transport=tr) + print(f"connected: {sim.info}") + time.sleep(0.8) + if not sim.is_connected(): + raise RuntimeError("connected then immediately dropped") + meta = pull(sim, args.out, timeout=args.timeout, verbose=args.verbose) + if args.decode: + from log_decode import decode_file + + decode_file( + args.out, + args.decode, + meta.get("write_off", 0), + meta.get("wrap_count", 0), + ) + if args.clear: + req = build(0x07, 0x02, 0x55, 0x2A) + safe_send(sim, req) + time.sleep(0.3) + sim._pump() + print("clear requested") + return + except Exception as e: + last_err = e + print(f"FAILED ({tr}): {e}") + finally: + if sim is not None: + try: + sim.close() + except Exception: + pass + time.sleep(1.0) + + raise SystemExit( + f"all transports failed: {last_err}\n" + "建议: 1) 断电复位吉他 2) python ble_log_pull.py --transport midi -v --out LOG.BIN --decode crash.log\n" + "若有 J-Link: python rtt_flash_log_dump.py --out crash.log --scan" + ) + + +if __name__ == "__main__": + main() diff --git a/tools/build_artist_kit.py b/tools/build_artist_kit.py new file mode 100644 index 0000000..2447512 --- /dev/null +++ b/tools/build_artist_kit.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""Assemble portable K1 artist tone-update kit + zip for handoff.""" +from __future__ import annotations + +import shutil +import zipfile +from datetime import datetime +from pathlib import Path + +REPO = Path(__file__).resolve().parents[3] # 一诺国际吉他 (tools->proj->Code->repo) +# __file__ = .../YNGJ.../tools/build_artist_kit.py → parents[0]=tools [1]=proj [2]=Code [3]=repo +PROJ = Path(__file__).resolve().parents[1] +TOOLS = Path(__file__).resolve().parent + + +def main() -> None: + # Fix REPO: tools -> project -> Code -> 一诺国际吉他 + repo = TOOLS.parents[1] # Code's parent? TOOLS.parent=proj, TOOLS.parents[1]=Code, [2]=repo + repo = TOOLS.parents[2] + proj = TOOLS.parent + day = datetime.now().strftime("%Y%m%d") + kit_name = f"K1_音师音色更新工具_{day}" + out_root = repo / "tools" / "out" + out_root.mkdir(parents=True, exist_ok=True) + kit = out_root / kit_name + if kit.exists(): + shutil.rmtree(kit) + kit.mkdir(parents=True) + + baseline = kit / "基线" + drop = kit / "投放" + outdir = kit / "输出" + baseline.mkdir() + drop.mkdir() + outdir.mkdir() + + shutil.copy2(TOOLS / "tone_artist_pack.py", kit / "tone_artist_pack.py") + + bat = """@echo off +chcp 65001 >nul +setlocal +cd /d "%~dp0" + +echo. +echo ======================================== +echo K1 音师音色更新工具 (USB / SoundWalkerIAP) +echo 依赖: 仅 Python 3，无需 pip / 无需 J-Link +echo ======================================== +echo. + +where python >nul 2>&1 +if errorlevel 1 ( + echo [错误] 未找到 python。请安装 Python 3，安装时勾选 Add python.exe to PATH。 + echo 下载: https://www.python.org/downloads/ + pause + exit /b 1 +) + +echo 默认读取「投放」目录中的 1.bin 2.bin 3.bin +echo 也可传参，例如: 音师更新音色.bat --bin1 .\\1.bin --bin2 .\\2.bin --bin3 .\\3.bin +echo. + +python "%~dp0tone_artist_pack.py" --open %* +set ERR=%ERRORLEVEL% +echo. +if not "%ERR%"=="0" ( + echo 打包失败。查看: python "%~dp0tone_artist_pack.py" -h + pause + exit /b %ERR% +) +echo 完成。请打开「输出\\音师音色包_日期」按 请这样刷.txt 用 SoundWalkerIAP 刷机。 +pause +exit /b 0 +""" + (kit / "音师更新音色.bat").write_text(bat, encoding="utf-8", newline="\r\n") + + ziliao = next( + p + for p in repo.iterdir() + if p.is_dir() and (p / "logo.bin").is_file() and (p / "Charg.bin").is_file() + ) + shutil.copy2(ziliao / "logo.bin", baseline / "logo.bin") + shutil.copy2(ziliao / "Charg.bin", baseline / "Charg.bin") + shutil.copy2(proj / "tools" / "out" / "ui0902_res.bin", baseline / "ui0902_res.bin") + + iap = repo / "升级" / "MCU主控升级" + shutil.copy2(iap / "SoundWalkerIAP.exe", kit / "SoundWalkerIAP.exe") + doc = iap / "升级步骤.docx" + if doc.is_file(): + shutil.copy2(doc, kit / "升级步骤.docx") + + src0909 = repo / "Doc" / "音色文件" / "0909" + for n in ("1.bin", "2.bin", "3.bin"): + shutil.copy2(src0909 / n, drop / n) + (drop / "说明.txt").write_text( + "请把新的 1.bin / 2.bin / 3.bin 放到本目录（覆盖即可），然后双击上一级「音师更新音色.bat」。\n" + "\n" + "也可用命令指定任意路径:\n" + " 音师更新音色.bat --bin1 路径\\1.bin --bin2 路径\\2.bin --bin3 路径\\3.bin\n", + encoding="utf-8", + newline="\n", + ) + + (kit / "使用说明.txt").write_text( + "\n".join( + [ + "K1 音师音色更新工具 — 使用说明", + "================================", + "", + "【需要准备】", + " 1. 安装 Python 3（https://www.python.org/downloads/ ，勾选 Add to PATH）", + " 只需标准库，不用 pip 装任何包", + " 2. 本工具包（解压到任意目录，如桌面）", + " 3. 吉他 + USB 线（SoundWalkerIAP 升级，不需要 J-Link）", + "", + "【目录说明】", + " 投放\\ ← 把新的 1.bin 2.bin 3.bin 放这里", + " 基线\\ ← logo / Charg / UI 底图（勿改）", + " 输出\\ ← 打包结果（自动生成）", + " 音师更新音色.bat", + " tone_artist_pack.py", + " SoundWalkerIAP.exe", + " 升级步骤.docx", + " 使用说明.txt ← 本文件", + "", + "【日常步骤】", + " 1. 用新文件覆盖 投放\\1.bin、2.bin、3.bin", + " 2. 双击「音师更新音色.bat」", + " 3. 在弹出的「输出\\音师音色包_日期」里:", + " - 打开 SoundWalkerIAP.exe", + " - 吉他进入 USB 升级模式（见 升级步骤.docx）", + " - 擦除外部 Flash", + " - 把 extflash_ALL_artist_*.res 烧到地址 0x00000000", + " 4. 退出升级模式，重启试听", + "", + "【命令参数示例】", + " 音师更新音色.bat", + " 音师更新音色.bat --drop 投放", + " 音师更新音色.bat --bin1 .\\投放\\1.bin --bin2 .\\投放\\2.bin --bin3 .\\投放\\3.bin", + " 音师更新音色.bat --bin1 D:\\我的音色\\1.bin --bin2 D:\\我的音色\\2.bin --bin3 D:\\我的音色\\3.bin", + " python tone_artist_pack.py -h", + "", + "【生成的包格式】（与开发正式发布的 ALL.res 相同）", + " 0x00000000 logo.bin", + " 0x0000CB70 Charg.bin", + " 0x0001B8F0 1.bin（节奏）", + " 0x0009D07D 2.bin（本地曲，≤41KB）", + " 0x000A71AC 3.bin（万能）", + " 0x00100000 UI 界面图", + "", + "【注意】", + " - 必须刷 ALL.res，不要只刷单独的 1/2/3.bin（否则模式选择花屏）", + " - 一般只需刷外部 Flash，不用重刷 MCU 固件", + " - 若增删本地曲目或改曲名，需联系开发同步固件后再测", + " - 「投放」里已带示例 1/2/3.bin，可先双击 bat 试跑流程", + "", + "【出问题】", + " - 提示找不到 python → 安装 Python 3 并勾选 PATH，重开窗口", + " - DAB/大小校验失败 → 检查 bin 是否完整，2.bin 是否超过 41KB", + " - 其它问题把黑色窗口全文截图发给开发", + "", + ] + ), + encoding="utf-8", + newline="\n", + ) + + zip_path = out_root / f"{kit_name}.zip" + if zip_path.exists(): + zip_path.unlink() + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: + for f in kit.rglob("*"): + if f.is_file(): + zf.write(f, f.relative_to(out_root).as_posix()) + + print(f"KIT {kit}") + print(f"ZIP {zip_path}") + print(f"ZIP size = {zip_path.stat().st_size} bytes") + for p in sorted(kit.iterdir()): + print(f" {p.name}") + + +if __name__ == "__main__": + main() diff --git a/tools/flash_boot_app_v024.py b/tools/flash_boot_app_v024.py new file mode 100644 index 0000000..1064e2c --- /dev/null +++ b/tools/flash_boot_app_v024.py @@ -0,0 +1,169 @@ +# -*- coding: utf-8 -*- +"""Flash Boot then APP via cspybat (AT32F403AC), reset, verify Boot strings.""" +from __future__ import annotations + +import os +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") + +BOOT_OUT = BASE / "AT32F403ARCT7_BOOT" / "project" / "IAR_V7.4" / "AT32F403ARCT7_BOOT" / "Exe" / "AT32F403ARCT7_BOOT.out" +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" + +NEW_GBK = bytes([0xC9, 0xD5, 0xC2, 0xBC, 0xC4, 0xA3, 0xCA, 0xBD]) # 鐑у綍妯″紡 +OLD_GBK = bytes([0xC9, 0xFD, 0xBC, 0xB6, 0xC4, 0xA3, 0xCA, 0xBD]) # 鍗囩骇妯″紡 + + +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) -> tuple[Path, 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 verify_boot_in_mcu() -> None: + import pylink + + j = pylink.JLink() + j.open() + try: + try: + j.exec_command("HideDeviceSelection = 1") + except Exception: + pass + j.set_tif(pylink.enums.JLinkInterfaces.SWD) + last = None + for dev in ("AT32F403AC", "Cortex-M4", "AT32F403A"): + try: + try: + j.exec_command(f"Device = {dev}") + except Exception: + pass + j.connect(dev) + print(f"verify connect: {dev}", flush=True) + break + except Exception as exc: + last = exc + else: + raise SystemExit(f"verify connect failed: {last}") + + j.halt() + # Boot image is in first ~32KB; search GBK markers + chunk = bytes(j.memory_read8(0x08000000, 0x8000)) + has_new = NEW_GBK in chunk + has_old = OLD_GBK in chunk + print(f"MCU@0x08000000 contains 鐑у綍妯″紡={has_new} 鍗囩骇妯″紡={has_old}", flush=True) + if not has_new: + raise SystemExit("Boot flash verify FAILED: new 鐑у綍妯″紡 string missing") + if has_old: + print("WARN: old 鍗囩骇妯″紡 string still present somewhere in Boot region", flush=True) + else: + print("Boot flash verify OK", flush=True) + j.reset(halt=False) + finally: + j.close() + + +def main() -> None: + boot_bin = BOOT_OUT.with_suffix(".bin") + data = boot_bin.read_bytes() + print( + f"local BOOT.bin: new={NEW_GBK in data} old={OLD_GBK in data} size={len(data)}", + flush=True, + ) + kill_debuggers() + cspy_download(BOOT_OUT, "boot") + kill_debuggers() + cspy_download(APP_OUT, "app") + jlink_reset() + verify_boot_in_mcu() + print("Done. Keep KEY for burn mode; screen should use UI0902_FLASH_MODE if ExtFlash has art.", flush=True) + + +if __name__ == "__main__": + main() diff --git a/tools/log_decode.py b/tools/log_decode.py new file mode 100644 index 0000000..c5511ac --- /dev/null +++ b/tools/log_decode.py @@ -0,0 +1,95 @@ +# -*- coding: utf-8 -*- +""" +log_decode.py — 将 K1 LOG.BIN（W25Q 日志分区镜像）解码为可读时间线 + +布局: + [0x0000 .. 0x0FFF] header (magic K1LG ...) + [0x1000 .. end] 文本环形区，行以 \\n 结束 + +用法: + python log_decode.py LOG.BIN -o problem.log + python log_decode.py LOG.BIN --write-off 1234 --wrap 1 -o problem.log +""" +from __future__ import annotations + +import argparse +import struct + +MAGIC = 0x4B314C47 # 'K1LG' +HDR_SIZE = 0x1000 + + +def unpack_header(blob: bytes) -> dict: + if len(blob) < 36: + return {"valid": False} + magic = struct.unpack_from(" str: + """Linearize ring: if wrapped, [write_off..end) + [0..write_off); else [0..write_off).""" + if not data: + return "" + if write_off > len(data): + write_off = len(data) + if wrapped and write_off < len(data): + raw = data[write_off:] + data[:write_off] + else: + raw = data[:write_off] if write_off else data + text = raw.decode("utf-8", errors="replace") + text = text.replace("\xff", "") + return text + + +def decode_file(bin_path: str, out_path: str, write_off=None, wrap_count=None) -> None: + with open(bin_path, "rb") as f: + blob = f.read() + hdr = unpack_header(blob) + data = blob[HDR_SIZE:] if len(blob) > HDR_SIZE else b"" + + wo = write_off if write_off is not None else hdr.get("write_off", 0) + wrap = wrap_count if wrap_count is not None else hdr.get("wrap_count", 0) + + text = extract_text(data, wo, wrap > 0) + lines = [ln for ln in text.splitlines() if ln.strip()] + + with open(out_path, "w", encoding="utf-8", newline="\n") as f: + f.write( + f"# K1 log decode magic_ok={hdr.get('valid')} ver={hdr.get('ver')} " + f"boot={hdr.get('boot_count')} write_off={wo} wrap={wrap} " + f"line_count={hdr.get('line_count')} decoded_lines={len(lines)}\n" + ) + for ln in lines: + f.write(ln.rstrip("\r") + "\n") + print(f"decoded {len(lines)} lines -> {out_path}") + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("bin") + ap.add_argument("-o", "--out", default="problem.log") + ap.add_argument("--write-off", type=int, default=None) + ap.add_argument("--wrap", type=int, default=None) + args = ap.parse_args() + decode_file(args.bin, args.out, args.write_off, args.wrap) + + +if __name__ == "__main__": + main() diff --git a/tools/rtt_flash_log_dump.py b/tools/rtt_flash_log_dump.py new file mode 100644 index 0000000..98ac402 --- /dev/null +++ b/tools/rtt_flash_log_dump.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Dump K1 persisted W25Q field log via J-Link RTT (`log flash dump` / `log flash scan`).""" + +from __future__ import annotations + +import argparse +import os +import sys +import time +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from rtt_log_dump import ( # noqa: E402 + connect_jlink, + find_rtt_control_block, + import_deps, + send_command, + wait_rtt_ready, +) + +DUMP_CMD = b"log flash dump\n" +SCAN_CMD = b"log flash scan 65536\n" + + +def dump_flash_log(out_path: str, device: str, timeout_s: float, cmd: bytes = DUMP_CMD) -> None: + jlink = connect_jlink(device) + lines: list[str] = [] + try: + cb = find_rtt_control_block(jlink) + if cb is None: + raise SystemExit("SEGGER RTT control block not found in SRAM") + print(f"RTT CB @ 0x{cb:08X}") + + jlink.rtt_start(cb) + wait_rtt_ready(jlink) + + for _ in range(5): + stale = jlink.rtt_read(0, 4096) + if stale: + text = bytes(stale).decode("utf-8", errors="replace") + if text.strip(): + print("RTT0 stale:", text.strip()[:200]) + time.sleep(0.05) + + print(f"Sending: {cmd!r}") + send_command(jlink, cmd) + + buffer = b"" + deadline = time.time() + timeout_s + while time.time() < deadline: + chunk = jlink.rtt_read(0, 8192) + if chunk: + buffer += bytes(chunk) + if len(buffer) >= 4096 and (len(buffer) % 8192) < 300: + print(f" received {len(buffer)} bytes...") + if b"LOG_FLASH_DUMP_END" in buffer: + break + else: + time.sleep(0.01) + + if b"LOG_FLASH_DUMP_END" not in buffer: + preview = buffer[-500:].decode("utf-8", errors="replace") if buffer else "" + raise SystemExit( + "Timeout waiting for LOG_FLASH_DUMP_END.\n" + f"got {len(buffer)} bytes. tail={preview!r}" + ) + + text = buffer.decode("utf-8", errors="replace") + capture = False + for line in text.splitlines(): + if line.strip() == "LOG_FLASH_DUMP_BEGIN": + capture = True + continue + if line.strip() == "LOG_FLASH_DUMP_END": + break + if capture: + lines.append(line) + + header = ( + f"# K1 flash log dump {datetime.now().isoformat(timespec='seconds')}\n" + f"# device={device}\n" + f"# cmd={cmd.decode('ascii', errors='replace').strip()}\n" + ) + body = "\n".join(lines) + ("\n" if lines else "") + with open(out_path, "w", encoding="utf-8", newline="\n") as f: + f.write(header) + f.write(body) + print(f"Saved {len(lines)} lines to {out_path}") + print(f"Full path: {os.path.abspath(out_path)}") + finally: + try: + jlink.rtt_stop() + except Exception: + pass + jlink.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description="Dump K1 Flash field log via RTT") + parser.add_argument("--out", default="crash.log") + parser.add_argument("--device", default="AT32F403AC") + parser.add_argument("--timeout", type=float, default=300.0) + parser.add_argument( + "--scan", + action="store_true", + help="Ignore write pointer; scan first 64KB data for recoverable lines", + ) + args = parser.parse_args() + import_deps() + cmd = SCAN_CMD if args.scan else DUMP_CMD + dump_flash_log(args.out, args.device, args.timeout, cmd=cmd) + + +if __name__ == "__main__": + main()