diff --git a/APP/app_log.c b/APP/app_log.c index 62384ee..9be7ba1 100644 --- a/APP/app_log.c +++ b/APP/app_log.c @@ -11,7 +11,7 @@ static uint32_t s_overflow; static uint32_t s_boot_count; static char s_ui_page[16] = "boot"; static char s_debug_level = 'D'; -static char s_cmd_buf[48]; +static char s_cmd_buf[64]; static uint8_t s_cmd_len; /* RTT 二进制烧录:写到 W25Q128(ui0902 或本地曲目) */ @@ -398,6 +398,73 @@ uint8_t app_log_try_command(const char *cmd) SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, line); return 1U; } + /* tone bin1 [idx] → 1.bin 节奏(专业+节奏类型) */ + if (strncmp(cmd, "tone bin1", 9) == 0) { + char line[160]; + int ret; + unsigned idx = 0; + int count; + if (cmd[9] == ' ' && cmd[10] != '\0') + idx = (unsigned)strtoul(cmd + 10, NULL, 10); + mGuiData[GUI_TAB_INDEX].Current = 1; /* 专业 */ + mGuiData[GUI_AUTOBAND_SW].Current = 0; /* 节奏类型→1.bin */ + ParamGuiData[SONG_MODE_PARAM].Current = (uint8_t)idx; + UI_ApplyToneAddress(); + AutoBandTop1_Stop(); + StartFlag = 0; + count = AutoBandTop1_GetPresetItemCount(); + if (count > 0 && (int)idx >= count) + idx = (unsigned)(count - 1); + ParamGuiData[SONG_MODE_PARAM].Current = (uint8_t)idx; + ret = AutoBandTop1_LoadPresetItemFromFlash((int)idx); + snprintf(line, sizeof(line), + "TONE_BIN1 ret=%d idx=%u addr=0x%08lX map=BIN1@0x%08lX name=%s count=%d %s\n", + ret, idx, (unsigned long)ADDRESS, + (unsigned long)EXTFLASH_BIN1_RHYTHM_ADDR, + AutoBandTop1_GetPresetName() ? AutoBandTop1_GetPresetName() : "(null)", + AutoBandTop1_GetPresetItemCount(), + (ADDRESS == EXTFLASH_BIN1_RHYTHM_ADDR) ? "OK" : "MISMATCH"); + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, line); + return 1U; + } + /* tone bin3 [idx] → 3.bin 万能和弦走向 */ + if (strncmp(cmd, "tone bin3", 9) == 0) { + char line[160]; + int ret; + unsigned idx = 0; + int count; + if (cmd[9] == ' ' && cmd[10] != '\0') + idx = (unsigned)strtoul(cmd + 10, NULL, 10); + mGuiData[GUI_TAB_INDEX].Current = 0; /* 万能 */ + mGuiData[GUI_AUTOBAND_SW].Current = 0; + ParamGuiData[ALL_MODE_PARAM].Current = (uint8_t)idx; + UI_ApplyToneAddress(); + AutoBandTop1_Stop(); + StartFlag = 0; + count = AutoBandTop1_GetPresetItemCount(); + if (count > 0 && (int)idx >= count) + idx = (unsigned)(count - 1); + ParamGuiData[ALL_MODE_PARAM].Current = (uint8_t)idx; + ret = AutoBandTop1_LoadPresetItemFromFlash((int)idx); + snprintf(line, sizeof(line), + "TONE_BIN3 ret=%d idx=%u addr=0x%08lX map=BIN3@0x%08lX name=%s count=%d %s\n", + ret, idx, (unsigned long)ADDRESS, + (unsigned long)EXTFLASH_BIN3_UNIVERSAL_ADDR, + AutoBandTop1_GetPresetName() ? AutoBandTop1_GetPresetName() : "(null)", + AutoBandTop1_GetPresetItemCount(), + (ADDRESS == EXTFLASH_BIN3_UNIVERSAL_ADDR) ? "OK" : "MISMATCH"); + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, line); + return 1U; + } + /* tone start:在当前已加载音色上拨片起奏(不改 TAB/库) */ + if (strncmp(cmd, "tone start", 10) == 0) { + StartFlag = 0; + if (!(cmd[10] == ' ' && (cmd[11] == 'k' || cmd[11] == 'K'))) + KEY_ID_1629 = 0; + Pick_Handle(); + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, "TONE_START_DONE\n"); + return 1U; + } if (strncmp(cmd, "tone pick", 9) == 0) { /* tone pick uni → 万能;默认仍测专业+本地曲目 */ if (cmd[9] == ' ' && (cmd[10] == 'u' || cmd[10] == 'U')) { @@ -413,15 +480,44 @@ uint8_t app_log_try_command(const char *cmd) SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, "TONE_PICK_UNI_DONE\n"); return 1U; } - /* 模拟专业+本地曲目拨片(无和弦板) */ + /* 模拟专业+本地曲目拨片;chord pick keep → 保留已注入的 KEY_ID */ mGuiData[GUI_TAB_INDEX].Current = 1; mGuiData[GUI_AUTOBAND_SW].Current = 1; - KEY_ID_1629 = 0; /* 走默认和弦兜底 */ + if (!(cmd[9] == ' ' && (cmd[10] == 'k' || cmd[10] == 'K'))) { + KEY_ID_1629 = 0; /* 默认走一级和弦兜底 */ + } StartFlag = 0; Pick_Handle(); SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, "TONE_PICK_DONE\n"); return 1U; } + /* 测试注入:chord key N(0释放 / 1~21和弦 / 22拍速 / 23停止) */ + if (strncmp(cmd, "chord key ", 10) == 0 && cmd[10] != '\0') { + unsigned key = (unsigned)strtoul(cmd + 10, NULL, 10); + char line[48]; + if (key > 23U) { + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, "CHORD_KEY_BAD\n"); + return 1U; + } + app_tm1629_inject_key((uint8_t)key); + snprintf(line, sizeof(line), "CHORD_KEY_OK key=%u\n", key); + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, line); + return 1U; + } + /* 测试注入:chord xpose N(0~11,C=0) */ + if (strncmp(cmd, "chord xpose ", 12) == 0 && cmd[12] != '\0') { + unsigned xp = (unsigned)strtoul(cmd + 12, NULL, 10); + char line[48]; + if (xp > 11U) { + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, "CHORD_XPOSE_BAD\n"); + return 1U; + } + mGuiData[GUI_TRANSPOSE].Current = (uint8_t)xp; + snprintf(line, sizeof(line), "CHORD_XPOSE_OK xp=%u\n", xp); + SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, line); + LOG_I("REG", "xpose set %u", xp); + return 1U; + } if (strncmp(cmd, "flash erase chip", 16) == 0) { SEGGER_RTT_WriteString(APP_LOG_RTT_CHANNEL, "FLASH_ERASE_BEGIN\n"); LOG_I("FLASH", "W25Q128 chip erase start"); diff --git a/APP/app_tm1629.c b/APP/app_tm1629.c index 309af2a..7d63b68 100644 --- a/APP/app_tm1629.c +++ b/APP/app_tm1629.c @@ -38,6 +38,13 @@ uint8_t app_tm1629_Scan_Key(void) return KEY_NONE; } +void app_tm1629_inject_key(uint8_t key) +{ + key_last = key; + LOG_I("KEY", "inject key=%u", (unsigned)key); + TM1629_Handle(key); +} + static void app_tm1629_led_set(LedNum_TypeDef led, LedColor_TypeDef color, uint8_t on) { diff --git a/APP/app_tm1629.h b/APP/app_tm1629.h index 29c85ac..0798980 100644 --- a/APP/app_tm1629.h +++ b/APP/app_tm1629.h @@ -4,6 +4,7 @@ void app_tm1629_init(void); uint8_t app_tm1629_Scan_Key(void); +void app_tm1629_inject_key(uint8_t key); /* RTT/测试:绕过扫描直接注入 */ void app_tm1629_set_chord_led(uint8_t key_idx, LedColor_TypeDef color); void app_tm1629_set_led(LedNum_TypeDef led, LedColor_TypeDef color); void app_tm1629_all_off(void); diff --git a/MyCode/App_Auto.c b/MyCode/App_Auto.c index 63befd7..fbce757 100644 --- a/MyCode/App_Auto.c +++ b/MyCode/App_Auto.c @@ -19,10 +19,318 @@ static uint8_t pStoreBuffer[PRESET_BUFFER_SIZE]; //static uint8_t* pStoreBuffer; #define DEFAULT_TEMPO 100 +/* ======== Bass/和弦分通道八度(无 AutoBand 源码时的输出侧映射) ======== + * 代码通道(0 起,与 MIDI status 低 4 位一致):8=bass,9=鼓;其余=和弦。 + * 日志打印同号(ch8=bass)。I/II(键1~6)原样;III+:bass -12 地板 E1; + * 和弦 -12 夹在 E2~#G4,超上限就近和弦内音。移调≥#F(6) 时再 -12。 + * 对照 Doc/吉他和弦表最终版.xlsx */ +#define CHORD_REG_FIX_EN 1 +#define BASS_CH 8 /* 日志 ch8 */ +#define DRUM_CH 9 /* 日志 ch9 */ +#define MIDI_E1 28 +#define MIDI_E2 40 +#define MIDI_GSHARP4 68 +#define XPOSE_FS 6 /* C=0 时 #F=6 */ + +/* 映射策略指纹:变化时 CC123 清旧音,防挂音(不建大音符表) */ +static uint8_t s_reg_fp = 0xFFu; +/* 每个指纹周期内 NoteOn 映射日志限额(和弦与 bass 分开,避免和弦占满看不到 bass) */ +static uint8_t s_pitch_log_left = 0; +static uint8_t s_bass_log_left = 0; +static uint16_t s_ch_ever_mask = 0; /* 会话内各通道是否出现过 NoteOn */ +#define PITCH_LOG_PER_FP 24 +#define BASS_LOG_PER_FP 16 + +static int App_Auto_ClampMidi(int n) +{ + if (n < 0) return 0; + if (n > 127) return 127; + return n; +} + +static int App_Auto_ChordDegree(void) +{ + if (KEY_ID_1629 < 1 || KEY_ID_1629 > 21) + return 0; + return (int)((KEY_ID_1629 - 1) / 3) + 1; /* 1=I .. 7=VII */ +} + +static int App_Auto_NeedRegisterFix(void) +{ + /* 键 1~6 = I/II:不动;7~21 = III~VII */ + return (KEY_ID_1629 >= 7 && KEY_ID_1629 <= 21) ? 1 : 0; +} + +static int App_Auto_XposeExtra(void) +{ + return (mGuiData[GUI_TRANSPOSE].Current >= XPOSE_FS) ? 1 : 0; +} + +static uint8_t App_Auto_RegFingerprint(void) +{ + /* bit0=need_fix, bit1=xpose_extra, bit2.. = 粗粒度级数 (key/3) */ + uint8_t fp = 0; + if (App_Auto_NeedRegisterFix()) fp |= 0x01u; + if (App_Auto_XposeExtra()) fp |= 0x02u; + if (KEY_ID_1629 >= 1 && KEY_ID_1629 <= 21) + fp |= (uint8_t)(((KEY_ID_1629 - 1) / 3) << 2); + return fp; +} + +static void App_Auto_QueueAllNotesOffMelodic(void) +{ + MidiFifoItem_t m; + int ch; + for (ch = 0; ch < 16; ch++) + { + if (ch == DRUM_CH) continue; + m.msg[0] = (uint8_t)(0xB0 | ch); + m.msg[1] = 123; /* All Notes Off */ + m.msg[2] = 0; + m.msglen = 3; + mymidififo_InQueue(&m_MidiSendFifo, &m); + } +} + +/* 超出 #G4:在当前和弦内音(GetChordNotesByType,对照和弦表「和弦伴奏」列)中就近 */ +static int App_Auto_NearestChordTone(int target) +{ + uint8_t buff[4] = {0, 0, 0, 0}; + uint8_t n; + uint8_t pcs[4]; + int i, oct, best = -1, best_dist = 9999; + uint8_t type; + + if (KEY_ID_1629 < 1 || KEY_ID_1629 > 21) + return App_Auto_ClampMidi(target); + + type = chord_type_index_map[KEY_ID_1629].type; + n = GetChordNotesByType(KEY_ID_1629, type, buff); + if (n == 0) + return App_Auto_ClampMidi(target > MIDI_GSHARP4 ? MIDI_GSHARP4 : target); + + for (i = 0; i < (int)n; i++) + pcs[i] = (uint8_t)(buff[i] % 12); + + for (i = 0; i < (int)n; i++) + { + for (oct = 0; oct < 11; oct++) + { + int cand = (int)pcs[i] + oct * 12; + int dist; + if (cand < MIDI_E2 || cand > MIDI_GSHARP4) + continue; + dist = cand - target; + if (dist < 0) dist = -dist; + /* 并列:优先较低者(不超过上限) */ + if (dist < best_dist || (dist == best_dist && (best < 0 || cand < best))) + { + best_dist = dist; + best = cand; + } + } + } + + if (best < 0) + return MIDI_GSHARP4; + return best; +} + +static int App_Auto_MapBassNote(int key, uint8_t *flags) +{ + int out = key - 12; + uint8_t f = 0x01u; /* bit0: -12 applied */ + while (out < MIDI_E1) + { + out += 12; + f |= 0x02u; /* bit1: floored up to E1 */ + } + if (App_Auto_XposeExtra()) + { + out -= 12; + f |= 0x04u; /* bit2: xpose extra -12 */ + while (out < MIDI_E1) + { + out += 12; + f |= 0x02u; + } + } + if (flags) *flags = f; + return App_Auto_ClampMidi(out); +} + +static int App_Auto_MapChordNote(int key, uint8_t *flags) +{ + int out = key - 12; + uint8_t f = 0x01u; /* bit0: -12 */ + + while (out < MIDI_E2) + { + out += 12; + f |= 0x08u; /* bit3: raised to E2 */ + } + + if (out > MIDI_GSHARP4) + { + out = App_Auto_NearestChordTone(out); + f |= 0x10u; /* bit4: nearest chord tone */ + } + + if (App_Auto_XposeExtra()) + { + out -= 12; + f |= 0x04u; + while (out < MIDI_E2) + { + out += 12; + f |= 0x08u; + } + if (out > MIDI_GSHARP4) + { + out = App_Auto_NearestChordTone(out); + f |= 0x10u; + } + } + if (flags) *flags = f; + return App_Auto_ClampMidi(out); +} + +static int App_Auto_MapNote(int channel, int key, uint8_t *flags) +{ + int out = key; + if (flags) *flags = 0; + +#if CHORD_REG_FIX_EN + if (channel == DRUM_CH || key < 0 || key > 127) + return key; + + if (!App_Auto_NeedRegisterFix()) + return key; + + if (channel == BASS_CH) + out = App_Auto_MapBassNote(key, flags); + else + out = App_Auto_MapChordNote(key, flags); +#else + (void)channel; +#endif + return out; +} + +static const char *App_Auto_RoleName(int channel) +{ + if (channel == DRUM_CH) return "drum"; + if (channel == BASS_CH) return "bass"; + return "chord"; +} + +static void App_Auto_LogPitch(int channel, int key, int out, uint8_t flags, int is_on) +{ + const char *role; + int deg; + int fix; + int xf; + int pass; + + if (channel == DRUM_CH) + return; + + if (channel == BASS_CH) + { + if (s_bass_log_left == 0) + return; + } + else if (s_pitch_log_left == 0) + { + return; + } + + role = App_Auto_RoleName(channel); + deg = App_Auto_ChordDegree(); + fix = App_Auto_NeedRegisterFix(); + xf = App_Auto_XposeExtra(); + pass = (!fix || channel == DRUM_CH) ? 1 : 0; + + LOG_I("PITCH", "%s ch%u(%s) %d->%d d=%d deg=%d chord=%u xp=%u xf=%d fl=0x%02X %s%s%s%s", + is_on ? "on" : "off", + (unsigned)channel, role, /* 0 基,与 BASS_CH=8 一致 */ + key, out, out - key, + deg, (unsigned)KEY_ID_1629, + (unsigned)mGuiData[GUI_TRANSPOSE].Current, xf, + (unsigned)flags, + pass ? "PASS " : "", + (flags & 0x10u) ? "NEAR " : "", + (flags & 0x02u) ? "E1UP " : "", + (flags & 0x08u) ? "E2UP " : ""); + + if (channel == BASS_CH) + { + if (s_bass_log_left > 0) + s_bass_log_left--; + } + else if (s_pitch_log_left > 0) + { + s_pitch_log_left--; + } +} + +/* NoteOn 路径:策略变化时清旧音;返回映射后音高 */ +static int App_Auto_MapNoteOn(int channel, int key) +{ +#if CHORD_REG_FIX_EN + uint8_t fp = App_Auto_RegFingerprint(); + uint8_t flags = 0; + int out; + + if (fp != s_reg_fp) + { + if (s_reg_fp != 0xFFu) + { + LOG_I("REG", "fp %u->%u chord=%u deg=%d xp=%u xf=%d", + (unsigned)s_reg_fp, (unsigned)fp, + (unsigned)KEY_ID_1629, + App_Auto_ChordDegree(), + (unsigned)mGuiData[GUI_TRANSPOSE].Current, + App_Auto_XposeExtra()); + App_Auto_QueueAllNotesOffMelodic(); + } + else + { + LOG_I("REG", "fp init %u chord=%u deg=%d xp=%u", + (unsigned)fp, (unsigned)KEY_ID_1629, + App_Auto_ChordDegree(), + (unsigned)mGuiData[GUI_TRANSPOSE].Current); + } + s_reg_fp = fp; + s_pitch_log_left = PITCH_LOG_PER_FP; + s_bass_log_left = BASS_LOG_PER_FP; + s_ch_ever_mask = 0; /* 新策略周期重新报到通道 */ + } + + out = App_Auto_MapNote(channel, key, &flags); + if (channel >= 0 && channel < 16) + { + uint16_t bit = (uint16_t)(1u << channel); + if ((s_ch_ever_mask & bit) == 0u) + { + s_ch_ever_mask |= bit; + LOG_I("REG", "ch-seen ch%u role=%s key=%d", + (unsigned)channel, App_Auto_RoleName(channel), key); + } + } + App_Auto_LogPitch(channel, key, out, flags, 1); + return out; +#else + (void)channel; + return key; +#endif +} + static void Func_CallBack_ProgramChange(int channel, int program) { MidiFifoItem_t midimsg; - midimsg.msg[0] = 0xc0 | channel; + LOG_I("REG", "pc ch%u prog=%d", (unsigned)channel, program); + midimsg.msg[0] = 0xc0 | (channel & 0x0F); midimsg.msg[1] = program; midimsg.msg[2] = 0; midimsg.msglen = 2; @@ -32,8 +340,10 @@ static void Func_CallBack_ProgramChange(int channel, int program) static void Func_CallBack_NoteOff(int channel,int key) { MidiFifoItem_t midimsg; - midimsg.msg[0] = 0x80 | channel; - midimsg.msg[1] = key; + uint8_t flags = 0; + int out_key = App_Auto_MapNote(channel, key, &flags); + midimsg.msg[0] = 0x80 | (channel & 0x0F); + midimsg.msg[1] = (uint8_t)out_key; midimsg.msg[2] = 0; midimsg.msglen = 3; mymidififo_InQueue(&m_MidiSendFifo,&midimsg); @@ -43,9 +353,19 @@ static void Func_CallBack_NoteOff(int channel,int key) static void Func_CallBack_NoteOn(int channel,int key,int vel) { MidiFifoItem_t midimsg; - midimsg.msg[0] = 0x90 | channel; - midimsg.msg[1] = key; - midimsg.msg[2] = vel; + int out_key; + + if (vel == 0) + { + uint8_t flags = 0; + out_key = App_Auto_MapNote(channel, key, &flags); /* 视同 NoteOff */ + } + else + out_key = App_Auto_MapNoteOn(channel, key); + + midimsg.msg[0] = 0x90 | (channel & 0x0F); + midimsg.msg[1] = (uint8_t)out_key; + midimsg.msg[2] = (uint8_t)vel; midimsg.msglen = 3; mymidififo_InQueue(&m_MidiSendFifo,&midimsg); @@ -135,6 +455,9 @@ static int Func_CallBack_ReadFlash(int address,int length,uint8_t* pOutput) // void App_Auto_Init(void) { + LOG_I("REG", "ch map bass=%d drum=%d fix=%d E1=%d E2=%d Gs4=%d", + BASS_CH, DRUM_CH, CHORD_REG_FIX_EN, MIDI_E1, MIDI_E2, MIDI_GSHARP4); + LOG_I("REG", "PITCH log: chN(role) in->out d=delta deg=I..VII xp=xpose xf=#Fextra fl=flags"); AutoBandTop1_SetPresetStoreBuffer(pStoreBuffer,PRESET_BUFFER_SIZE); AutoBandTop1_Init(); AutoBandTop1_RegisterCallBack_NoteOn(Func_CallBack_NoteOn); diff --git a/tools/mic_pitch_analyze.py b/tools/mic_pitch_analyze.py new file mode 100644 index 0000000..cc4c08e --- /dev/null +++ b/tools/mic_pitch_analyze.py @@ -0,0 +1,387 @@ +#!/usr/bin/env python3 +"""Laptop-mic pitch analyzer for K1 accompaniment register checks. + +Uses autocorrelation F0 estimate (numpy only + sounddevice). +Helps verify bass/chord octave when MIDI channel logs lack ch8(bass). + +Examples: + python tools/mic_pitch_analyze.py --list + python tools/mic_pitch_analyze.py --seconds 8 --out mic_i.log + python tools/mic_pitch_analyze.py --live + python tools/mic_pitch_analyze.py --compare mic_i.wav mic_iii.wav +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +import wave +from collections import Counter +from dataclasses import dataclass +from datetime import datetime + +import numpy as np + +try: + import sounddevice as sd +except ImportError as exc: + raise SystemExit("Missing sounddevice: pip install sounddevice") from exc + +A4_HZ = 440.0 +A4_MIDI = 69 + + +@dataclass +class PitchFrame: + t: float + hz: float + midi: float + note: str + conf: float + rms: float + + +NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] + + +def hz_to_midi(hz: float) -> float: + if hz <= 0: + return float("nan") + return A4_MIDI + 12.0 * np.log2(hz / A4_HZ) + + +def midi_to_name(midi: float) -> str: + if not np.isfinite(midi): + return "--" + n = int(round(midi)) + return f"{NOTE_NAMES[n % 12]}{n // 12 - 1}" + + +def list_devices() -> None: + if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass + for i, d in enumerate(sd.query_devices()): + name = str(d.get("name", "")).encode("utf-8", "replace").decode("utf-8", "replace") + print( + f"{i}: in={d['max_input_channels']} out={d['max_output_channels']} " + f"sr={d['default_samplerate']} {name}", + flush=True, + ) + try: + print("default device pair:", sd.default.device, flush=True) + except Exception: + pass + + +def record(seconds: float, sr: int, device: int | None) -> np.ndarray: + frames = int(seconds * sr) + print(f"Recording {seconds:.1f}s @ {sr} Hz (Ctrl+C to abort)...", flush=True) + audio = sd.rec(frames, samplerate=sr, channels=1, dtype="float32", device=device) + sd.wait() + return audio[:, 0] + + +def save_wav(path: str, audio: np.ndarray, sr: int) -> None: + pcm = np.clip(audio, -1.0, 1.0) + pcm16 = (pcm * 32767.0).astype(np.int16) + with wave.open(path, "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(sr) + w.writeframes(pcm16.tobytes()) + + +def load_wav(path: str) -> tuple[np.ndarray, int]: + with wave.open(path, "rb") as w: + sr = w.getframerate() + nch = w.getnchannels() + raw = w.readframes(w.getnframes()) + pcm = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0 + if nch > 1: + pcm = pcm.reshape(-1, nch).mean(axis=1) + return pcm, sr + + +def estimate_f0_acorr( + frame: np.ndarray, + sr: int, + fmin: float, + fmax: float, +) -> tuple[float, float]: + """Return (hz, confidence). confidence in [0,1] from normalized peak.""" + x = frame.astype(np.float64) + x = x - np.mean(x) + rms = float(np.sqrt(np.mean(x * x)) + 1e-12) + if rms < 1e-4: + return 0.0, 0.0 + + # Hamming window + x = x * np.hamming(len(x)) + corr = np.correlate(x, x, mode="full") + corr = corr[len(corr) // 2 :] + + i_min = max(1, int(sr / fmax)) + i_max = min(len(corr) - 1, int(sr / fmin)) + if i_max <= i_min: + return 0.0, 0.0 + + seg = corr[i_min : i_max + 1] + peak_rel = int(np.argmax(seg)) + peak = peak_rel + i_min + if corr[0] <= 1e-12: + return 0.0, 0.0 + conf = float(corr[peak] / corr[0]) + if conf < 0.25: + return 0.0, conf + + # parabolic interpolation around peak + if 1 <= peak < len(corr) - 1: + a, b, c = corr[peak - 1], corr[peak], corr[peak + 1] + denom = a - 2 * b + c + if abs(denom) > 1e-12: + peak = peak + 0.5 * (a - c) / denom + + hz = float(sr / peak) + if hz < fmin or hz > fmax: + return 0.0, conf + return hz, conf + + +def analyze_audio( + audio: np.ndarray, + sr: int, + hop_ms: float = 50.0, + win_ms: float = 80.0, + fmin: float = 40.0, + fmax: float = 600.0, + conf_min: float = 0.35, + rms_min: float = 0.01, +) -> list[PitchFrame]: + hop = max(1, int(sr * hop_ms / 1000.0)) + win = max(hop, int(sr * win_ms / 1000.0)) + out: list[PitchFrame] = [] + if len(audio) < win: + return out + + for start in range(0, len(audio) - win, hop): + frame = audio[start : start + win] + rms = float(np.sqrt(np.mean(frame.astype(np.float64) ** 2))) + hz, conf = estimate_f0_acorr(frame, sr, fmin, fmax) + t = start / sr + if hz <= 0 or conf < conf_min or rms < rms_min: + out.append(PitchFrame(t, 0.0, float("nan"), "--", conf, rms)) + continue + midi = hz_to_midi(hz) + out.append(PitchFrame(t, hz, midi, midi_to_name(midi), conf, rms)) + return out + + +def summarize(frames: list[PitchFrame], label: str = "") -> dict: + voiced = [f for f in frames if f.hz > 0 and np.isfinite(f.midi)] + if not voiced: + return {"label": label, "voiced": 0, "total": len(frames)} + + midis = np.array([f.midi for f in voiced], dtype=np.float64) + hz = np.array([f.hz for f in voiced], dtype=np.float64) + notes = [f.note for f in voiced] + # round to nearest MIDI for histogram + rounded = [int(round(m)) for m in midis] + top = Counter(rounded).most_common(8) + # bass-ish: MIDI <= 48 (C3) + bass_ratio = float(np.mean(midis <= 48.0)) + low_ratio = float(np.mean(midis <= 40.0)) # <= E2 + return { + "label": label, + "voiced": len(voiced), + "total": len(frames), + "hz_median": float(np.median(hz)), + "hz_p10": float(np.percentile(hz, 10)), + "hz_p90": float(np.percentile(hz, 90)), + "midi_median": float(np.median(midis)), + "midi_p10": float(np.percentile(midis, 10)), + "midi_p90": float(np.percentile(midis, 90)), + "note_median": midi_to_name(float(np.median(midis))), + "top_notes": [(midi_to_name(float(n)), c) for n, c in top], + "bass_le_C3_ratio": bass_ratio, + "low_le_E2_ratio": low_ratio, + } + + +def print_summary(s: dict) -> None: + if s.get("voiced", 0) == 0: + print(f"[{s.get('label','')}] no pitched frames", flush=True) + return + print( + f"[{s.get('label','')}] voiced={s['voiced']}/{s['total']} " + f"median={s['note_median']} ({s['midi_median']:.1f} / {s['hz_median']:.1f}Hz) " + f"p10={s['midi_p10']:.1f} p90={s['midi_p90']:.1f} " + f"<=E2={s['low_le_E2_ratio']*100:.0f}% <=C3={s['bass_le_C3_ratio']*100:.0f}%", + flush=True, + ) + tops = ", ".join(f"{n}×{c}" for n, c in s["top_notes"][:5]) + print(f" top: {tops}", flush=True) + + +def write_frame_log(path: str, frames: list[PitchFrame], summary: dict) -> None: + with open(path, "w", encoding="utf-8", newline="\n") as f: + f.write(f"# mic pitch {datetime.now().isoformat(timespec='seconds')}\n") + f.write("# " + json.dumps(summary, ensure_ascii=False) + "\n") + f.write("t_s,hz,midi,note,conf,rms\n") + for fr in frames: + midi = "" if not np.isfinite(fr.midi) else f"{fr.midi:.2f}" + f.write( + f"{fr.t:.3f},{fr.hz:.2f},{midi},{fr.note},{fr.conf:.3f},{fr.rms:.4f}\n" + ) + + +def live_monitor( + sr: int, + device: int | None, + fmin: float, + fmax: float, + seconds: float, +) -> None: + win = int(sr * 0.08) + hop = int(sr * 0.05) + print("Live pitch (Ctrl+C stop)...", flush=True) + buf = np.zeros(0, dtype=np.float32) + t0 = time.time() + + def callback(indata, frames, time_info, status): # noqa: ARG001 + nonlocal buf + if status: + print(status, flush=True) + buf = np.concatenate([buf, indata[:, 0].copy()]) + while len(buf) >= win: + frame = buf[:win] + buf = buf[hop:] + hz, conf = estimate_f0_acorr(frame, sr, fmin, fmax) + rms = float(np.sqrt(np.mean(frame.astype(np.float64) ** 2))) + if hz > 0 and conf >= 0.35 and rms >= 0.01: + midi = hz_to_midi(hz) + print( + f"{time.time()-t0:6.1f}s {midi_to_name(midi):4s} " + f"midi={midi:5.1f} {hz:6.1f}Hz conf={conf:.2f} rms={rms:.3f}", + flush=True, + ) + + with sd.InputStream(samplerate=sr, channels=1, dtype="float32", device=device, callback=callback): + if seconds > 0: + sd.sleep(int(seconds * 1000)) + else: + while True: + sd.sleep(200) + + +def compare_summaries(a: dict, b: dict) -> None: + print("\n======== COMPARE ========", flush=True) + print_summary(a) + print_summary(b) + if a.get("voiced", 0) == 0 or b.get("voiced", 0) == 0: + print("Need pitched content in both takes.", flush=True) + return + d_midi = b["midi_median"] - a["midi_median"] + print(f"median delta (B-A): {d_midi:+.2f} semitones", flush=True) + if d_midi <= -9: + print("PASS-ish: B is about an octave lower than A (expected III+ vs I/II).", flush=True) + elif d_midi <= -5: + print("PARTIAL: B lower than A but less than full octave.", flush=True) + elif abs(d_midi) < 2: + print("FAIL-ish: medians similar — register fix may not be audible on mic mix.", flush=True) + else: + print("CHECK: unexpected direction/amount; inspect top notes / low_le_E2 ratios.", flush=True) + print( + f"low<=E2: A={a['low_le_E2_ratio']*100:.0f}% B={b['low_le_E2_ratio']*100:.0f}%", + flush=True, + ) + + +def main() -> None: + if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass + p = argparse.ArgumentParser(description="Mic pitch analyzer for K1 register tests") + p.add_argument("--list", action="store_true", help="List audio devices") + p.add_argument("--device", type=int, default=None, help="Input device index") + p.add_argument("--sr", type=int, default=16000) + p.add_argument("--seconds", type=float, default=8.0) + p.add_argument("--live", action="store_true") + p.add_argument("--wav", default="", help="Analyze existing wav instead of recording") + p.add_argument("--out", default="", help="Prefix for wav/csv outputs") + p.add_argument("--fmin", type=float, default=40.0, help="Min F0 Hz (bass ~41=E1)") + p.add_argument("--fmax", type=float, default=600.0, help="Max F0 Hz") + p.add_argument( + "--compare", + nargs=2, + metavar=("A", "B"), + help="Compare two wav/csv summary sources (wav preferred)", + ) + p.add_argument("--label", default="") + args = p.parse_args() + + if args.list: + list_devices() + return + + if args.compare: + summaries = [] + for i, path in enumerate(args.compare): + label = "A" if i == 0 else "B" + if path.lower().endswith(".wav"): + audio, sr = load_wav(path) + frames = analyze_audio(audio, sr, fmin=args.fmin, fmax=args.fmax) + s = summarize(frames, label=f"{label}:{os.path.basename(path)}") + else: + raise SystemExit("compare expects .wav files") + summaries.append(s) + compare_summaries(summaries[0], summaries[1]) + return + + if args.live: + live_monitor(args.sr, args.device, args.fmin, args.fmax, args.seconds if args.seconds > 0 else 0) + return + + prefix = args.out or f"mic_pitch_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + if args.wav: + audio, sr = load_wav(args.wav) + label = args.label or os.path.basename(args.wav) + else: + audio = record(args.seconds, args.sr, args.device) + sr = args.sr + label = args.label or "rec" + wav_path = prefix if prefix.lower().endswith(".wav") else prefix + ".wav" + save_wav(wav_path, audio, sr) + print(f"Saved wav: {os.path.abspath(wav_path)}", flush=True) + + frames = analyze_audio(audio, sr, fmin=args.fmin, fmax=args.fmax) + summary = summarize(frames, label=label) + print_summary(summary) + log_path = prefix if prefix.lower().endswith(".csv") else prefix + ".csv" + # if prefix was .wav, still write .csv alongside + if log_path.lower().endswith(".wav.csv"): + log_path = log_path[:-8] + ".csv" + elif prefix.lower().endswith(".wav"): + log_path = prefix[:-4] + ".csv" + write_frame_log(log_path, frames, summary) + print(f"Saved log: {os.path.abspath(log_path)}", flush=True) + print( + "Tip: record I/II then III+ separately, then:\n" + " python tools/mic_pitch_analyze.py --compare mic_i.wav mic_iii.wav", + flush=True, + ) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\nStopped.", flush=True) + sys.exit(130) diff --git a/tools/rtt_bin13_reg_test.py b/tools/rtt_bin13_reg_test.py new file mode 100644 index 0000000..1c746d5 --- /dev/null +++ b/tools/rtt_bin13_reg_test.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +"""Probe 1.bin (rhythm) and 3.bin (universal) for bass ch8 + plan compliance. + +Requires FW with: tone bin1 / tone bin3 / tone start / chord key / chord xpose / PITCH ch8(bass) +""" + +from __future__ import annotations + +import argparse +import os +import re +import sys +import time +from collections import Counter, defaultdict +from dataclasses import dataclass, field +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from rtt_pitch_reg_test import ( # noqa: E402 + RttSession, + analyze_lines, + connect_jlink, + expect_for_event, + find_rtt_control_block, + import_deps, + parse_pitch_line, + wait_rtt_ready, +) + +PITCH_RE_CH = re.compile(r"\[PITCH\].*?ch(\d+)\((\w+)\)") +CH_SEEN_RE = re.compile(r"ch-seen ch(\d+) role=(\w+) key=(-?\d+)") +TONE_OK_RE = re.compile(r"TONE_BIN([13]).*name=(\S+).*count=(\d+)") + + +@dataclass +class PresetResult: + bin_id: str + idx: int + name: str = "" + channels: Counter = field(default_factory=Counter) + roles: Counter = field(default_factory=Counter) + bass_n: int = 0 + drum_n: int = 0 + chord_n: int = 0 + pass_ok: int = 0 + pass_fail: int = 0 + iii_ok: int = 0 + iii_fail: int = 0 + xf_ok: int = 0 + xf_fail: int = 0 + bass_ok: int = 0 + bass_fail: int = 0 + samples: list[str] = field(default_factory=list) + + +def run_preset(sess: RttSession, bin_id: str, idx: int, hold: float) -> PresetResult: + res = PresetResult(bin_id=bin_id, idx=idx) + mark = len(sess.lines) + + if bin_id == "1": + ok = sess.cmd_ack(f"tone bin1 {idx}", "TONE_BIN1", timeout_s=3.0, retries=4) + else: + ok = sess.cmd_ack(f"tone bin3 {idx}", "TONE_BIN3", timeout_s=3.0, retries=4) + if not ok: + res.samples.append("FAIL load ack") + return res + + # parse name from recent lines + for line in sess.lines[-8:]: + m = TONE_OK_RE.search(line) + if m and m.group(1) == bin_id: + res.name = m.group(2) + break + + sess.set_xpose(2) + sess.cmd("tone start", settle=0.7) + + # I/II + sess.set_key(0) + sess.set_key(2) + sess.pump(hold) + + # III + sess.set_key(0) + sess.set_key(8) + sess.pump(hold) + + # VII + #F + sess.set_xpose(6) + sess.set_key(0) + sess.set_key(20) + sess.pump(hold) + + sess.set_key(23) # stop + sess.pump(0.3) + + chunk = sess.lines[mark:] + for line in chunk: + m = CH_SEEN_RE.search(line) + if m: + ch, role = int(m.group(1)), m.group(2) + res.channels[ch] += 1 + res.roles[role] += 1 + res.samples.append(line.strip()) + ev = parse_pitch_line(line) + if not ev or not ev.is_on: + continue + res.channels[ev.ch] += 1 + res.roles[ev.role] += 1 + if ev.role == "bass" or ev.ch == 8: + res.bass_n += 1 + elif ev.role == "drum" or ev.ch == 9: + res.drum_n += 1 + else: + res.chord_n += 1 + + detail, ok = expect_for_event(ev) + tag = f"ch{ev.ch}({ev.role}) {ev.key_in}->{ev.key_out} chord={ev.chord} | {detail}" + if ev.chord <= 6: + if ok: + res.pass_ok += 1 + else: + res.pass_fail += 1 + res.samples.append("FAIL " + tag) + elif ev.xf: + if ok: + res.xf_ok += 1 + else: + res.xf_fail += 1 + res.samples.append("FAIL " + tag) + elif ev.role == "bass" or ev.ch == 8: + if ok: + res.bass_ok += 1 + else: + res.bass_fail += 1 + res.samples.append("FAIL " + tag) + else: + if ok: + res.iii_ok += 1 + else: + res.iii_fail += 1 + res.samples.append("FAIL " + tag) + + return res + + +def main() -> int: + if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass + + ap = argparse.ArgumentParser() + ap.add_argument("--hold", type=float, default=2.2) + ap.add_argument("--bin1", default="0,1,2,3,4,5,8,12,16,20,24,28", help="1.bin indices") + ap.add_argument("--bin3", default="0,1,2", help="3.bin indices") + ap.add_argument("--out", default="") + args = ap.parse_args() + + bin1_idxs = [int(x) for x in args.bin1.split(",") if x.strip() != ""] + bin3_idxs = [int(x) for x in args.bin3.split(",") if x.strip() != ""] + + import_deps() + jlink = connect_jlink("AT32F403AC") + results: list[PresetResult] = [] + try: + cb = find_rtt_control_block(jlink) + if cb is None: + raise SystemExit("RTT CB not found") + print(f"RTT CB @ 0x{cb:08X}", flush=True) + jlink.rtt_start(cb) + wait_rtt_ready(jlink) + sess = RttSession(jlink) + sess.pump(0.3) + sess.cmd("log clear", 0.3) + + print("\n######## BIN1 / 1.bin rhythms ########", flush=True) + for idx in bin1_idxs: + print(f"\n--- BIN1 idx={idx} ---", flush=True) + r = run_preset(sess, "1", idx, args.hold) + results.append(r) + print( + f"name={r.name} ch={dict(r.channels)} roles={dict(r.roles)} " + f"bass={r.bass_n} drum={r.drum_n} chord={r.chord_n} " + f"I/II {r.pass_ok}/{r.pass_fail} III {r.iii_ok}/{r.iii_fail} " + f"xf {r.xf_ok}/{r.xf_fail} bassMap {r.bass_ok}/{r.bass_fail}", + flush=True, + ) + + print("\n######## BIN3 / 3.bin universal ########", flush=True) + for idx in bin3_idxs: + print(f"\n--- BIN3 idx={idx} ---", flush=True) + r = run_preset(sess, "3", idx, args.hold) + results.append(r) + print( + f"name={r.name} ch={dict(r.channels)} roles={dict(r.roles)} " + f"bass={r.bass_n} drum={r.drum_n} chord={r.chord_n} " + f"I/II {r.pass_ok}/{r.pass_fail} III {r.iii_ok}/{r.iii_fail} " + f"xf {r.xf_ok}/{r.xf_fail} bassMap {r.bass_ok}/{r.bass_fail}", + flush=True, + ) + + # Summary vs plan + print("\n======== PLAN CHECK (bass和弦分通道八度) ========", flush=True) + with_bass = [r for r in results if r.bass_n > 0] + print(f"Presets tested: {len(results)}; with ch8(bass) NoteOn: {len(with_bass)}", flush=True) + if with_bass: + for r in with_bass: + print( + f" BASS HIT {r.bin_id}.bin[{r.idx}] {r.name}: " + f"bass_n={r.bass_n} map_ok={r.bass_ok} fail={r.bass_fail}", + flush=True, + ) + else: + print(" NO preset produced ch8(bass). Plan bass rule NOT verified on-device.", flush=True) + + # Aggregate chord rules + pass_ok = sum(r.pass_ok for r in results) + pass_fail = sum(r.pass_fail for r in results) + iii_ok = sum(r.iii_ok for r in results) + iii_fail = sum(r.iii_fail for r in results) + xf_ok = sum(r.xf_ok for r in results) + xf_fail = sum(r.xf_fail for r in results) + bass_ok = sum(r.bass_ok for r in results) + bass_fail = sum(r.bass_fail for r in results) + + def gate(name, ok, fail, need=True): + status = "PASS" if fail == 0 and (ok > 0 or not need) else ("FAIL" if fail else "SKIP") + print(f" [{status}] {name}: ok={ok} fail={fail}", flush=True) + return status != "FAIL" + + all_ok = True + all_ok &= gate("I/II PASS (keys1-6)", pass_ok, pass_fail) + all_ok &= gate("III+ chord map", iii_ok, iii_fail) + all_ok &= gate("xpose>=#F extra-12", xf_ok, xf_fail) + all_ok &= gate("bass ch8 map", bass_ok, bass_fail, need=False) + if not with_bass: + all_ok = False + print(" [FAIL] plan requires bass on code ch8 — never seen across 1.bin/3.bin samples", flush=True) + + # channel histogram + ch_all: Counter = Counter() + for r in results: + ch_all.update(r.channels) + print(f"Channel histogram: {dict(sorted(ch_all.items()))}", flush=True) + + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + out = args.out or os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + f"bin13_reg_{stamp}.log", + ) + with open(out, "w", encoding="utf-8", newline="\n") as f: + f.write(f"# bin1/bin3 plan check {datetime.now().isoformat(timespec='seconds')}\n") + for line in sess.lines: + f.write(line + "\n") + f.write("\n# SUMMARY\n") + for r in results: + f.write( + f"# {r.bin_id}[{r.idx}] {r.name} ch={dict(r.channels)} " + f"bass={r.bass_n} I={r.pass_ok}/{r.pass_fail} " + f"III={r.iii_ok}/{r.iii_fail} xf={r.xf_ok}/{r.xf_fail} " + f"bassMap={r.bass_ok}/{r.bass_fail}\n" + ) + print(f"Log: {out}", flush=True) + return 0 if all_ok else 1 + finally: + try: + jlink.rtt_stop() + except Exception: + pass + jlink.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/rtt_mic_reg_test.py b/tools/rtt_mic_reg_test.py new file mode 100644 index 0000000..6190591 --- /dev/null +++ b/tools/rtt_mic_reg_test.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +"""Drive K1 via RTT while recording laptop mic; compare I/II vs III+ pitch.""" + +from __future__ import annotations + +import os +import sys +import time +from datetime import datetime + +# Reuse helpers from sibling tools +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from mic_pitch_analyze import ( # noqa: E402 + analyze_audio, + compare_summaries, + print_summary, + record, + save_wav, + summarize, + write_frame_log, +) +from rtt_pitch_reg_test import ( # noqa: E402 + RttSession, + connect_jlink, + find_rtt_control_block, + import_deps, + wait_rtt_ready, +) + + +def main() -> int: + if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except Exception: + pass + + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + out_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + hold = 8.0 + sr = 16000 + device = None # default mic + + import_deps() + jlink = connect_jlink("AT32F403AC") + try: + cb = find_rtt_control_block(jlink) + if cb is None: + raise SystemExit("RTT CB not found") + print(f"RTT CB @ 0x{cb:08X}", flush=True) + jlink.rtt_start(cb) + wait_rtt_ready(jlink) + sess = RttSession(jlink) + sess.pump(0.3) + + sess.cmd("log clear", 0.2) + sess.cmd("tone local", 1.0) + sess.set_xpose(2) + sess.cmd("tone pick", 0.8) + + # ---- Take A: I/II ---- + print("\n=== TAKE A: I/II (key2) — put guitar near mic ===", flush=True) + sess.set_key(0) + sess.set_key(2) + sess.pump(0.8) + print("MIC record A starting...", flush=True) + audio_a = record(hold, sr, device) + wav_a = os.path.join(out_dir, f"mic_A_I_{stamp}.wav") + save_wav(wav_a, audio_a, sr) + frames_a = analyze_audio(audio_a, sr, fmin=40, fmax=500) + sum_a = summarize(frames_a, label="A:I/II") + print_summary(sum_a) + write_frame_log(os.path.join(out_dir, f"mic_A_I_{stamp}.csv"), frames_a, sum_a) + + # ---- Take B: III ---- + print("\n=== TAKE B: III (key8) ===", flush=True) + sess.set_key(0) + sess.set_key(8) + sess.pump(0.8) + print("MIC record B starting...", flush=True) + audio_b = record(hold, sr, device) + wav_b = os.path.join(out_dir, f"mic_B_III_{stamp}.wav") + save_wav(wav_b, audio_b, sr) + frames_b = analyze_audio(audio_b, sr, fmin=40, fmax=500) + sum_b = summarize(frames_b, label="B:III") + print_summary(sum_b) + write_frame_log(os.path.join(out_dir, f"mic_B_III_{stamp}.csv"), frames_b, sum_b) + + # ---- Take C: VII + xpose #F (extra -12) ---- + print("\n=== TAKE C: VII key20 + xpose=#F ===", flush=True) + sess.set_xpose(6) + sess.set_key(0) + sess.set_key(20) + sess.pump(0.8) + print("MIC record C starting...", flush=True) + audio_c = record(hold, sr, device) + wav_c = os.path.join(out_dir, f"mic_C_VII_xf_{stamp}.wav") + save_wav(wav_c, audio_c, sr) + frames_c = analyze_audio(audio_c, sr, fmin=40, fmax=500) + sum_c = summarize(frames_c, label="C:VII+#F") + print_summary(sum_c) + write_frame_log(os.path.join(out_dir, f"mic_C_VII_xf_{stamp}.csv"), frames_c, sum_c) + + # Also bass-biased analysis (prefer low F0) + print("\n=== Bass-biased (fmax=180Hz) ===", flush=True) + for label, audio in (("A", audio_a), ("B", audio_b), ("C", audio_c)): + fr = analyze_audio(audio, sr, fmin=40, fmax=180) + print_summary(summarize(fr, label=f"{label}-bassband")) + + print("\n=== A vs B (I/II vs III) ===", flush=True) + compare_summaries(sum_a, sum_b) + print("\n=== A vs C (I/II vs VII+#F) ===", flush=True) + compare_summaries(sum_a, sum_c) + + print("\n=== stop ===", flush=True) + sess.stop_band() + + # Save RTT lines + rtt_path = os.path.join(out_dir, f"mic_rtt_{stamp}.log") + with open(rtt_path, "w", encoding="utf-8", newline="\n") as f: + f.write(f"# rtt+mic test {datetime.now().isoformat(timespec='seconds')}\n") + for line in sess.lines: + f.write(line + "\n") + print(f"RTT log: {rtt_path}", flush=True) + print(f"WAV A/B/C: {wav_a}\n {wav_b}\n {wav_c}", flush=True) + return 0 + finally: + try: + jlink.rtt_stop() + except Exception: + pass + jlink.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/rtt_pitch_reg_test.py b/tools/rtt_pitch_reg_test.py new file mode 100644 index 0000000..15c12b7 --- /dev/null +++ b/tools/rtt_pitch_reg_test.py @@ -0,0 +1,589 @@ +#!/usr/bin/env python3 +"""K1 bass/chord register auto-test via J-Link RTT. + +Requires firmware with: + - [PITCH]/[REG] logs (App_Auto.c) + - RTT cmds: chord key N / chord xpose N / tone pick / log clear + +Modes: + python tools/rtt_pitch_reg_test.py # inject + live assert + python tools/rtt_pitch_reg_test.py --listen # you press keys; we assert + python tools/rtt_pitch_reg_test.py --analyze oct_reg3.log +""" + +from __future__ import annotations + +import argparse +import os +import re +import sys +import time +from dataclasses import dataclass, field +from datetime import datetime +from typing import Iterable + +DEFAULT_DEVICES = ("Cortex-M4", "AT32F403AC", "AT32F403A") + +MIDI_E1 = 28 +MIDI_E2 = 40 +MIDI_GSHARP4 = 68 +XPOSE_FS = 6 +BASS_CH_DISP = 8 # 代码/日志通道;BASS_CH=8 +DRUM_CH_DISP = 9 + +PITCH_RE = re.compile( + r"\[PITCH\]\s+(on|off)\s+ch(\d+)\((\w+)\)\s+" + r"(\d+)->(\d+)\s+d=(-?\d+)\s+deg=(\d+)\s+chord=(\d+)\s+" + r"xp=(\d+)\s+xf=(-?\d+)\s+fl=0x([0-9A-Fa-f]+)\s*(.*)$" +) +REG_RE = re.compile( + r"\[REG\s*\]\s+fp\s+(\d+)->(\d+)\s+chord=(\d+)\s+deg=(\d+)\s+xp=(\d+)\s+xf=(-?\d+)" +) + + +@dataclass +class PitchEvent: + is_on: bool + ch: int + role: str + key_in: int + key_out: int + delta: int + deg: int + chord: int + xp: int + xf: int + flags: int + tags: str + raw: str + + +@dataclass +class CheckResult: + ok: int = 0 + fail: int = 0 + skip: int = 0 + messages: list[str] = field(default_factory=list) + + def add_ok(self, msg: str) -> None: + self.ok += 1 + self.messages.append(f"OK {msg}") + + def add_fail(self, msg: str) -> None: + self.fail += 1 + self.messages.append(f"FAIL {msg}") + + def add_skip(self, msg: str) -> None: + self.skip += 1 + self.messages.append(f"SKIP {msg}") + + +def import_deps() -> None: + try: + import pylink # noqa: F401 + except ImportError as exc: + raise SystemExit("Missing dependency: pip install pylink-square") from exc + + +def connect_jlink(device: str): + import pylink + + jlink = pylink.JLink() + jlink.open() + try: + jlink.exec_command("HideDeviceSelection = 1") + except Exception: + pass + jlink.set_tif(pylink.enums.JLinkInterfaces.SWD) + + last_error = None + for candidate in (device, *DEFAULT_DEVICES): + try: + try: + jlink.exec_command(f"Device = {candidate}") + except Exception: + pass + jlink.connect(candidate) + print(f"Connected as {candidate}", flush=True) + return jlink + except pylink.errors.JLinkException as exc: + last_error = exc + jlink.close() + raise SystemExit(f"Failed to connect J-Link: {last_error}") + + +def find_rtt_control_block(jlink, ram_base: int = 0x20000000, ram_size: int = 0x18000) -> int | None: + needle = b"SEGGER RTT" + chunk = 0x1000 + was_halted = jlink.halted() + if not was_halted: + jlink.halt() + try: + 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 + finally: + if not was_halted: + if hasattr(jlink, "restart"): + jlink.restart() + else: + jlink.go() + time.sleep(0.1) + return None + + +def wait_rtt_ready(jlink, timeout_s: float = 15.0) -> None: + deadline = time.time() + timeout_s + while time.time() < deadline: + try: + if jlink.rtt_get_num_up_buffers() > 0: + return + except Exception: + pass + time.sleep(0.2) + raise SystemExit("RTT control block found but buffers not ready") + + +def send_command(jlink, payload: bytes, retries: int = 30) -> None: + if isinstance(payload, str): + payload = payload.encode("ascii") + if not payload.endswith(b"\n"): + payload += b"\n" + for _ in range(retries): + wrote = jlink.rtt_write(0, list(payload)) + if wrote > 0: + print(f">> {payload.decode('ascii', errors='replace').strip()}", flush=True) + return + time.sleep(0.2) + raise SystemExit(f"Failed to send RTT cmd: {payload!r}") + + +def drain_rtt(jlink, seconds: float = 0.15) -> str: + buf = b"" + deadline = time.time() + seconds + while time.time() < deadline: + chunk = jlink.rtt_read(0, 4096) + if chunk: + buf += bytes(chunk) + else: + time.sleep(0.01) + return buf.decode("utf-8", errors="replace") + + +def parse_pitch_line(line: str) -> PitchEvent | None: + m = PITCH_RE.search(line) + if not m: + return None + return PitchEvent( + is_on=m.group(1) == "on", + ch=int(m.group(2)), + role=m.group(3), + key_in=int(m.group(4)), + key_out=int(m.group(5)), + delta=int(m.group(6)), + deg=int(m.group(7)), + chord=int(m.group(8)), + xp=int(m.group(9)), + xf=int(m.group(10)), + flags=int(m.group(11), 16), + tags=m.group(12).strip(), + raw=line.strip(), + ) + + +def chord_degree(chord: int) -> int: + if chord < 1 or chord > 21: + return 0 + return (chord - 1) // 3 + 1 + + +def need_register_fix(chord: int) -> bool: + return 7 <= chord <= 21 + + +def map_bass(key: int, xp: int) -> tuple[int, int]: + """Return (out, flags) mirroring App_Auto_MapBassNote.""" + out = key - 12 + flags = 0x01 + while out < MIDI_E1: + out += 12 + flags |= 0x02 + if xp >= XPOSE_FS: + out -= 12 + flags |= 0x04 + while out < MIDI_E1: + out += 12 + flags |= 0x02 + return max(0, min(127, out)), flags + + +def map_chord(key: int, xp: int) -> tuple[int, int, bool]: + """Return (out_or_clamp_hint, flags, need_near). + + When need_near is True, exact out is unknown without chord table; + caller should only require out in [E2, #G4]. + """ + out = key - 12 + flags = 0x01 + while out < MIDI_E2: + out += 12 + flags |= 0x08 + need_near = out > MIDI_GSHARP4 + if need_near: + flags |= 0x10 + out = MIDI_GSHARP4 # placeholder; exact nearest checked soft + if xp >= XPOSE_FS: + out -= 12 + flags |= 0x04 + while out < MIDI_E2: + out += 12 + flags |= 0x08 + if out > MIDI_GSHARP4: + flags |= 0x10 + need_near = True + out = MIDI_GSHARP4 + return max(0, min(127, out)), flags, need_near + + +def expect_for_event(ev: PitchEvent) -> tuple[str, bool]: + """Return (detail, ok).""" + if not ev.is_on: + return "note-off ignored", True + + if ev.delta != ev.key_out - ev.key_in: + return f"d mismatch {ev.delta} != {ev.key_out - ev.key_in}", False + + deg_exp = chord_degree(ev.chord) + if deg_exp and ev.deg != deg_exp: + return f"deg {ev.deg} != expected {deg_exp}", False + + xf_exp = 1 if ev.xp >= XPOSE_FS else 0 + if ev.xf != xf_exp: + return f"xf {ev.xf} != expected {xf_exp} (xp={ev.xp})", False + + if ev.ch == DRUM_CH_DISP or ev.role == "drum": + return "drum should not appear in PITCH", False + + fix = need_register_fix(ev.chord) + if not fix: + if ev.key_out != ev.key_in or ev.delta != 0: + return f"I/II should PASS {ev.key_in}->{ev.key_in}, got {ev.key_out}", False + if "PASS" not in ev.tags and ev.flags != 0: + # flags may be 0 on pass path + pass + return f"PASS I/II {ev.key_in}", True + + if ev.ch == BASS_CH_DISP or ev.role == "bass": + exp, exp_fl = map_bass(ev.key_in, ev.xp) + if ev.key_out != exp: + return f"bass expect {ev.key_in}->{exp}, got {ev.key_out}", False + if ev.key_out < MIDI_E1: + return f"bass below E1: {ev.key_out}", False + # flag bits that must be present + if (exp_fl & 0x01) and not (ev.flags & 0x01): + return f"bass missing -12 flag fl=0x{ev.flags:02X}", False + return f"bass {ev.key_in}->{ev.key_out}", True + + # chord path + exp, exp_fl, need_near = map_chord(ev.key_in, ev.xp) + if not (MIDI_E2 <= ev.key_out <= MIDI_GSHARP4): + return f"chord out {ev.key_out} not in [{MIDI_E2},{MIDI_GSHARP4}]", False + if need_near: + if not (ev.flags & 0x10) and "NEAR" not in ev.tags: + # soft: range ok is enough if nearest not flagged but clamped somehow + return f"chord NEAR expected (in={ev.key_in} out={ev.key_out})", True + return f"chord NEAR {ev.key_in}->{ev.key_out}", True + if ev.key_out != exp: + return f"chord expect {ev.key_in}->{exp}, got {ev.key_out}", False + return f"chord {ev.key_in}->{ev.key_out}", True + + +def analyze_lines(lines: Iterable[str], result: CheckResult | None = None) -> CheckResult: + result = result or CheckResult() + pitch_n = 0 + roles: set[str] = set() + chords: set[int] = set() + for line in lines: + line = line.rstrip("\n") + if "[REG" in line and "fp " in line: + m = REG_RE.search(line) + if m: + chord = int(m.group(3)) + deg = int(m.group(4)) + xp = int(m.group(5)) + xf = int(m.group(6)) + deg_exp = chord_degree(chord) + xf_exp = 1 if xp >= XPOSE_FS else 0 + if deg_exp and deg != deg_exp: + result.add_fail(f"REG deg {deg}!={deg_exp} chord={chord}") + elif xf != xf_exp: + result.add_fail(f"REG xf {xf}!={xf_exp} xp={xp}") + else: + result.add_ok(f"REG fp chord={chord} deg={deg} xp={xp} xf={xf}") + + ev = parse_pitch_line(line) + if not ev: + continue + pitch_n += 1 + roles.add(ev.role) + chords.add(ev.chord) + detail, ok = expect_for_event(ev) + msg = f"ch{ev.ch}({ev.role}) chord={ev.chord} {ev.key_in}->{ev.key_out} | {detail}" + if ok: + result.add_ok(msg) + else: + result.add_fail(msg) + + if pitch_n == 0: + result.add_skip("no [PITCH] lines found") + else: + if "bass" not in roles and not any( + f"ch{BASS_CH_DISP}(" in m for m in result.messages + ): + result.add_skip(f"no bass PITCH on ch{BASS_CH_DISP} (style may omit bass, or old FW)") + if not any(need_register_fix(c) for c in chords): + result.add_skip("no III+ chord keys observed") + if not any(not need_register_fix(c) and 1 <= c <= 6 for c in chords): + result.add_skip("no I/II chord keys observed") + return result + + +def print_report(result: CheckResult, out_path: str | None = None) -> int: + print("\n======== PITCH REG TEST REPORT ========", flush=True) + for msg in result.messages: + # only print fails + summary skips loudly; OK compacted + if msg.startswith("FAIL") or msg.startswith("SKIP"): + print(msg, flush=True) + ok_short = [m for m in result.messages if m.startswith("OK")] + print(f"OK={result.ok} FAIL={result.fail} SKIP={result.skip}", flush=True) + if ok_short: + print(f"(first OK samples: {len(ok_short)} total)", flush=True) + for m in ok_short[:8]: + print(m, flush=True) + if len(ok_short) > 8: + print(f"... +{len(ok_short) - 8} more OK", flush=True) + if out_path: + with open(out_path, "w", encoding="utf-8", newline="\n") as f: + f.write(f"# pitch reg report {datetime.now().isoformat(timespec='seconds')}\n") + f.write(f"OK={result.ok} FAIL={result.fail} SKIP={result.skip}\n") + for m in result.messages: + f.write(m + "\n") + print(f"Report saved: {os.path.abspath(out_path)}", flush=True) + return 0 if result.fail == 0 and result.ok > 0 else 1 + + +class RttSession: + def __init__(self, jlink): + self.jlink = jlink + self.buf = b"" + self.lines: list[str] = [] + + def cmd(self, text: str, settle: float = 0.2) -> None: + send_command(self.jlink, text.encode("ascii")) + self.pump(settle) + + def pump(self, seconds: float) -> list[str]: + new_lines: list[str] = [] + deadline = time.time() + seconds + while time.time() < deadline: + chunk = self.jlink.rtt_read(0, 4096) + if chunk: + self.buf += bytes(chunk) + while b"\n" in self.buf: + raw, self.buf = self.buf.split(b"\n", 1) + line = raw.decode("utf-8", errors="replace").rstrip("\r") + if line: + self.lines.append(line) + new_lines.append(line) + if ( + "[PITCH]" in line + or "[REG" in line + or "[KEY" in line + or "CHORD_" in line + or "TONE_" in line + ): + print(line, flush=True) + else: + time.sleep(0.01) + return new_lines + + def cmd_ack(self, text: str, ack: str, timeout_s: float = 3.0, retries: int = 6) -> bool: + """Send command until ack substring appears (handles RTT down loss under load).""" + for attempt in range(1, retries + 1): + # brief quiet drain then send + self.pump(0.15) + send_command(self.jlink, text.encode("ascii")) + deadline = time.time() + timeout_s + while time.time() < deadline: + for line in self.pump(0.1): + if ack in line: + return True + print(f"!! no ack '{ack}' for '{text}' (try {attempt}/{retries})", flush=True) + time.sleep(0.25) + return False + + def stop_band(self) -> None: + self.cmd_ack("chord key 23", "CHORD_KEY_OK", timeout_s=2.0, retries=4) + self.pump(0.4) + + def set_xpose(self, xp: int) -> bool: + return self.cmd_ack(f"chord xpose {xp}", "CHORD_XPOSE_OK", timeout_s=2.0, retries=5) + + def set_key(self, key: int) -> bool: + return self.cmd_ack(f"chord key {key}", "CHORD_KEY_OK", timeout_s=2.0, retries=5) + + def phase_switch(self, name: str, key: int, xpose: int | None, hold_s: float) -> None: + """Switch chord/xpose while accompaniment is already running. + Note: plain `tone pick` clears KEY_ID to 1 — do not call it after inject. + """ + print(f"\n=== {name}: key={key} xpose={xpose} ===", flush=True) + if xpose is not None: + if not self.set_xpose(xpose): + print(f"FAIL setup xpose={xpose}", flush=True) + self.set_key(0) + if not self.set_key(key): + print(f"FAIL setup key={key}", flush=True) + self.pump(hold_s) + + +def run_auto(device: str, hold_s: float, report: str | None, log_path: str | None) -> int: + import_deps() + jlink = connect_jlink(device) + try: + cb = find_rtt_control_block(jlink) + if cb is None: + raise SystemExit("SEGGER RTT CB not found") + print(f"RTT CB @ 0x{cb:08X}", flush=True) + jlink.rtt_start(cb) + wait_rtt_ready(jlink) + sess = RttSession(jlink) + sess.pump(0.3) + + sess.cmd("log clear", 0.3) + sess.cmd("tone local", 1.0) + # Start band first (pick forces chord=1), then inject keys while playing + if not sess.set_xpose(2): + print("FAIL initial xpose=2", flush=True) + sess.cmd("tone pick", settle=0.8) + print("\n=== I (default after pick) ===", flush=True) + sess.pump(hold_s) + + sess.phase_switch("I/II key2 PASS", key=2, xpose=None, hold_s=hold_s) + sess.phase_switch("III key8 -12", key=8, xpose=None, hold_s=hold_s) + sess.phase_switch("VII key20 -12", key=20, xpose=None, hold_s=hold_s) + sess.phase_switch("VII key20 xpose=#F", key=20, xpose=6, hold_s=hold_s) + + print("\n=== stop ===", flush=True) + sess.stop_band() + sess.pump(0.4) + + if log_path: + with open(log_path, "w", encoding="utf-8", newline="\n") as f: + f.write(f"# auto pitch test {datetime.now().isoformat(timespec='seconds')}\n") + for line in sess.lines: + f.write(line + "\n") + print(f"Log saved: {os.path.abspath(log_path)}", flush=True) + + joined = "\n".join(sess.lines) + if "CHORD_KEY_OK" not in joined and "CHORD_KEY_BAD" not in joined: + print( + "WARNING: no CHORD_KEY_OK — firmware may lack 'chord key' RTT cmd. " + "Rebuild/flash, or use --listen / --analyze.", + flush=True, + ) + + result = analyze_lines(sess.lines) + has_pass = any("PASS I/II" in m for m in result.messages) + has_iii_raw = any( + "[PITCH]" in line and ("chord=8" in line or "chord=20" in line or "deg=3" in line or "deg=7" in line) + for line in sess.lines + ) + has_iii_ok = any( + m.startswith("OK") and "PASS" not in m and "(chord)" in m + for m in result.messages + ) + has_xf = any("[PITCH]" in line and "xf=1" in line and "deg=7" in line for line in sess.lines) + if not has_pass: + result.add_fail("coverage: no I/II PASS samples") + if not has_iii_raw: + result.add_fail("coverage: no III+/VII chord in PITCH") + elif not has_iii_ok: + result.add_fail("coverage: III+ present but mapping asserts failed") + if not has_xf: + result.add_skip("coverage: no VII+xf=1 samples") + + return print_report(result, report) + finally: + try: + jlink.rtt_stop() + except Exception: + pass + jlink.close() + + +def run_listen(device: str, seconds: float, report: str | None, log_path: str | None) -> int: + import_deps() + jlink = connect_jlink(device) + try: + cb = find_rtt_control_block(jlink) + if cb is None: + raise SystemExit("SEGGER RTT CB not found") + print(f"RTT CB @ 0x{cb:08X}", flush=True) + jlink.rtt_start(cb) + wait_rtt_ready(jlink) + sess = RttSession(jlink) + sess.pump(0.2) + sess.cmd("log clear", 0.2) + print( + f"Listening {seconds:.0f}s — press I/II then III+ chords; set transpose>=#F if possible.", + flush=True, + ) + sess.pump(seconds) + if log_path: + with open(log_path, "w", encoding="utf-8", newline="\n") as f: + f.write(f"# listen pitch test {datetime.now().isoformat(timespec='seconds')}\n") + for line in sess.lines: + f.write(line + "\n") + print(f"Log saved: {os.path.abspath(log_path)}", flush=True) + result = analyze_lines(sess.lines) + return print_report(result, report) + finally: + try: + jlink.rtt_stop() + except Exception: + pass + jlink.close() + + +def run_analyze(path: str, report: str | None) -> int: + with open(path, encoding="utf-8", errors="replace") as f: + lines = f.readlines() + result = analyze_lines(lines) + return print_report(result, report) + + +def main() -> None: + parser = argparse.ArgumentParser(description="K1 pitch/register RTT auto-test") + parser.add_argument("--device", default="AT32F403AC") + parser.add_argument("--hold", type=float, default=2.5, help="Seconds to hold each injected key") + parser.add_argument("--seconds", type=float, default=45.0, help="Listen duration") + parser.add_argument("--listen", action="store_true", help="Do not inject; monitor only") + parser.add_argument("--analyze", metavar="LOG", help="Offline analyze a dump log") + parser.add_argument("--out-log", default="", help="Save captured RTT lines") + parser.add_argument("--report", default="", help="Save pass/fail report") + args = parser.parse_args() + + report = args.report or None + log_path = args.out_log or None + + if args.analyze: + raise SystemExit(run_analyze(args.analyze, report)) + if args.listen: + raise SystemExit(run_listen(args.device, args.seconds, report, log_path)) + raise SystemExit(run_auto(args.device, args.hold, report, log_path)) + + +if __name__ == "__main__": + main()