249 lines
9.1 KiB
Python
249 lines
9.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""音色表六弦音高校对：RTT string 回报 + 麦克风基频。
|
||
|
||
对照 Doc/音色文件/自动伴奏新更改和弦规则.xlsx「C 大调」弦音高列（C 调、移调 0）。
|
||
I C / II Dm 与旧表一致；III Em / IV F / V G / VI Am 按新表改为开放把位音高。
|
||
只走 `string N`，不再额外 midi on 期望音，避免麦测被兜底音骗过。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime
|
||
import os
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
import numpy as np
|
||
|
||
TOOLS = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, TOOLS)
|
||
|
||
from k1_harness.transports.rtt_session import open_session # noqa: E402
|
||
from mic_pitch_analyze import ( # noqa: E402
|
||
analyze_audio,
|
||
estimate_f0_acorr,
|
||
hz_to_midi,
|
||
midi_to_name,
|
||
save_wav,
|
||
)
|
||
|
||
try:
|
||
import sounddevice as sd
|
||
except ImportError as exc:
|
||
raise SystemExit("Missing sounddevice: pip install sounddevice") from exc
|
||
|
||
# 六弦→一弦 MIDI。与 midi_tables.c OrganMIDI / MIDI 对应行一致。
|
||
CASES = [
|
||
("C", 1, (43, 48, 52, 55, 60, 64)),
|
||
("Dm", 4, (41, 45, 50, 57, 62, 65)),
|
||
("Em", 7, (40, 47, 52, 55, 59, 64)),
|
||
("F", 10, (41, 48, 53, 57, 60, 65)),
|
||
("G", 13, (43, 47, 50, 55, 59, 67)),
|
||
("Am", 16, (40, 45, 52, 57, 60, 64)),
|
||
]
|
||
|
||
SR = 48000
|
||
HOLD_S = 1.6
|
||
GAP_S = 0.45
|
||
FMIN = 70.0
|
||
FMAX = 520.0
|
||
NOTE_RE = re.compile(r"STRING_OK s=(\d+) note=(\d+)")
|
||
REPORTS = os.path.abspath(os.path.join(TOOLS, "..", "..", "..", "Doc", "reports"))
|
||
|
||
|
||
def record(seconds: float, sr: int, device: int) -> np.ndarray:
|
||
frames = int(seconds * sr)
|
||
audio = sd.rec(frames, samplerate=sr, channels=1, dtype="float32", device=device)
|
||
sd.wait()
|
||
return audio[:, 0]
|
||
|
||
|
||
def fft_f0(audio: np.ndarray, sr: int) -> float | None:
|
||
if len(audio) < sr // 5:
|
||
return None
|
||
mid = len(audio) // 2
|
||
half = int(0.45 * sr)
|
||
seg = audio[max(0, mid - half) : mid + half].astype(np.float64)
|
||
seg = seg - np.mean(seg)
|
||
if float(np.sqrt(np.mean(seg * seg))) < 0.003:
|
||
return None
|
||
win = seg * np.hanning(len(seg))
|
||
n = 1 << int(np.ceil(np.log2(len(win))))
|
||
spec = np.abs(np.fft.rfft(win, n=n))
|
||
freqs = np.fft.rfftfreq(n, 1.0 / sr)
|
||
mask = (freqs >= FMIN) & (freqs <= FMAX)
|
||
if not np.any(mask):
|
||
return None
|
||
peak_i = int(np.where(mask)[0][0] + int(np.argmax(spec[mask])))
|
||
if 1 <= peak_i < len(spec) - 1:
|
||
a, b, c = float(spec[peak_i - 1]), float(spec[peak_i]), float(spec[peak_i + 1])
|
||
denom = a - 2 * b + c
|
||
peak_f = peak_i + (0.5 * (a - c) / denom if abs(denom) > 1e-12 else 0.0)
|
||
else:
|
||
peak_f = float(peak_i)
|
||
return float(peak_f * sr / n)
|
||
|
||
|
||
def detect_pitch(audio: np.ndarray, sr: int):
|
||
frames = analyze_audio(
|
||
audio, sr, fmin=FMIN, fmax=FMAX, conf_min=0.22, rms_min=0.0025, hop_ms=40, win_ms=90
|
||
)
|
||
voiced = [f for f in frames if f.hz > 0 and np.isfinite(f.midi)]
|
||
if len(voiced) >= 3:
|
||
n = len(voiced)
|
||
lo, hi = n // 5, max(n // 5 + 1, (4 * n) // 5)
|
||
med = float(np.median([f.midi for f in voiced[lo:hi]]))
|
||
return med, midi_to_name(med), "acorr"
|
||
hz = fft_f0(audio, sr)
|
||
if hz is None:
|
||
best = (0.0, 0.0)
|
||
win = int(0.09 * sr)
|
||
hop = int(0.04 * sr)
|
||
for i in range(0, max(1, len(audio) - win), hop):
|
||
h, c = estimate_f0_acorr(audio[i : i + win], sr, FMIN, FMAX)
|
||
if c > best[1] and h > 0:
|
||
best = (h, c)
|
||
hz = best[0] if best[1] >= 0.2 else None
|
||
if hz is None or hz <= 0:
|
||
return None, "--", "none"
|
||
midi = float(hz_to_midi(hz))
|
||
return midi, midi_to_name(midi), "fft"
|
||
|
||
|
||
def parse_string_note(text: str, string_n: int) -> int | None:
|
||
found = None
|
||
for m in NOTE_RE.finditer(text or ""):
|
||
if int(m.group(1)) == string_n:
|
||
found = int(m.group(2))
|
||
return found
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--device", type=int, default=1, help="sounddevice input index")
|
||
ap.add_argument("--no-mic", action="store_true", help="只核对 RTT STRING_OK")
|
||
args = ap.parse_args()
|
||
|
||
os.makedirs(REPORTS, exist_ok=True)
|
||
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
out_dir = os.path.join(REPORTS, f"tone_voicing_{ts}")
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
rows = []
|
||
|
||
print("=== 音色表六弦音高：RTT + 麦测 ===", flush=True)
|
||
print(f"mic_device={args.device} sr={SR} wav={out_dir}", flush=True)
|
||
|
||
with open_session(reset=False) as sess:
|
||
sess.cmd("tone pick f", settle=1.0)
|
||
sess.cmd_ack("chord xpose 0", "CHORD_XPOSE_OK", timeout_s=2.0, retries=4)
|
||
sess.cmd("chord key 23", settle=0.4)
|
||
sess.cmd("chord key 0", settle=0.2)
|
||
|
||
for name, key, notes in CASES:
|
||
sess.cmd_ack(f"chord key {key}", "CHORD_KEY_OK", timeout_s=3.0, retries=4)
|
||
time.sleep(0.25)
|
||
print(f"\n-- {name} key={key}", flush=True)
|
||
for s in range(6, 0, -1):
|
||
expect = int(notes[6 - s])
|
||
label = f"{name}_str{s}"
|
||
wav = os.path.join(out_dir, f"{label}.wav")
|
||
audio_box: list = []
|
||
|
||
def _rec(box=audio_box):
|
||
if args.no_mic:
|
||
box.append(None)
|
||
return
|
||
box.append(record(HOLD_S, SR, args.device))
|
||
|
||
th = threading.Thread(target=_rec, daemon=True)
|
||
th.start()
|
||
time.sleep(0.25)
|
||
n0 = len(sess.lines)
|
||
sess.cmd_ack(f"string {s} 110", "STRING_OK", timeout_s=2.5, retries=3)
|
||
th.join()
|
||
sess.cmd(f"string {s} off", settle=0.05)
|
||
|
||
rtt_note = parse_string_note("\n".join(sess.lines[n0:]), s)
|
||
rtt_ok = rtt_note == expect
|
||
heard = heard_name = method = None
|
||
delta = None
|
||
mic_ok = None
|
||
rms = 0.0
|
||
if not args.no_mic:
|
||
audio = audio_box[0] if audio_box else np.zeros(1, dtype=np.float32)
|
||
save_wav(wav, audio, SR)
|
||
rms = float(np.max(np.abs(audio))) if len(audio) else 0.0
|
||
heard, heard_name, method = detect_pitch(audio, SR)
|
||
if heard is not None:
|
||
delta = heard - float(expect)
|
||
mic_ok = abs(delta) <= 0.75
|
||
else:
|
||
mic_ok = False
|
||
|
||
row = {
|
||
"label": label,
|
||
"chord": name,
|
||
"key": key,
|
||
"string": s,
|
||
"expect": expect,
|
||
"expect_name": midi_to_name(float(expect)),
|
||
"rtt": rtt_note,
|
||
"rtt_name": midi_to_name(float(rtt_note)) if rtt_note is not None else "--",
|
||
"rtt_ok": rtt_ok,
|
||
"heard": heard,
|
||
"heard_name": heard_name or "--",
|
||
"delta": delta,
|
||
"mic_ok": mic_ok,
|
||
"method": method,
|
||
"rms": rms,
|
||
}
|
||
rows.append(row)
|
||
mic_s = "skip" if mic_ok is None else ("PASS" if mic_ok else "FAIL")
|
||
print(
|
||
f" {label}: table={row['expect_name']}({expect}) "
|
||
f"rtt={row['rtt_name']}({rtt_note}) [{'PASS' if rtt_ok else 'FAIL'}] "
|
||
f"mic={row['heard_name']}"
|
||
f"({None if heard is None else round(heard, 2)}) "
|
||
f"d={None if delta is None else round(delta, 2)} "
|
||
f"rms={rms:.3f} [{mic_s}]",
|
||
flush=True,
|
||
)
|
||
time.sleep(GAP_S)
|
||
sess.cmd("chord key 0", settle=0.15)
|
||
|
||
rtt_fail = [r for r in rows if not r["rtt_ok"]]
|
||
mic_fail = [r for r in rows if r["mic_ok"] is False]
|
||
md = os.path.join(REPORTS, f"tone_voicing_{ts}.md")
|
||
lines = [
|
||
"# 音色表六弦音高核对",
|
||
"",
|
||
f"- 时间: {datetime.datetime.now():%Y-%m-%d %H:%M:%S}",
|
||
"- 期望: C/Dm 原把位；Em/F/G/Am 为音色改善表上移后的六弦",
|
||
f"- wav: `{out_dir}`",
|
||
"",
|
||
"| 和弦 | 弦 | 表 | RTT | RTT结果 | 麦测 | Δ | 麦测结果 |",
|
||
"|------|----|----|-----|---------|------|---|----------|",
|
||
]
|
||
for r in rows:
|
||
mic_cell = "skip" if r["mic_ok"] is None else ("PASS" if r["mic_ok"] else "FAIL")
|
||
dlt = "" if r["delta"] is None else f"{r['delta']:.2f}"
|
||
lines.append(
|
||
f"| {r['chord']} | {r['string']} | {r['expect_name']} | {r['rtt_name']} | "
|
||
f"{'PASS' if r['rtt_ok'] else 'FAIL'} | {r['heard_name']} | {dlt} | {mic_cell} |"
|
||
)
|
||
lines.append("")
|
||
lines.append(f"- RTT 失败 {len(rtt_fail)} / {len(rows)}")
|
||
if not args.no_mic:
|
||
lines.append(f"- 麦测失败 {len(mic_fail)} / {len(rows)}")
|
||
with open(md, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines) + "\n")
|
||
print(f"\n报告: {md}", flush=True)
|
||
print(f"RTT fail={len(rtt_fail)} mic fail={len(mic_fail)}", flush=True)
|
||
return 1 if rtt_fail or mic_fail else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|