tools: robot-clap hardware test + live-detector replay harness

robot_clap_test.py drives the antennas on the wire schedule (local Lite
daemon), records the mic, decodes; replay_streaming.mjs feeds a saved recording
through the app's exact StreamingOnsetDetector + wire decoder for offline tuning.

tools/replay_streaming.mjs

@@ -0,0 +1,60 @@
+/**
+ * Replay a recorded float32 mono buffer through the app's EXACT live decode
+ * path: a Web Audio BiquadFilter highpass is emulated here, then blocks are
+ * fed to StreamingOnsetDetector (lib/detector.js) just like mic.js does, and
+ * onsets are decoded with wire.js. This validates the real live path against
+ * real robot-clap audio (no browser, no robot).
+ *
+ *   node tools/replay_streaming.mjs /tmp/robot_clap.f32 --sr 48000 --unit 150 \
+ *        --expect SOS --highpass 2000 --factor 4
+ */
+import { readFileSync } from "node:fs";
+import { StreamingOnsetDetector, highpass } from "../lib/detector.js";
+import { decodeOnsets } from "../lib/wire.js";
+import { makeTiming } from "../lib/timing.js";
+
+const args = process.argv.slice(2);
+const path = args[0];
+const opt = (k, d) => { const i = args.indexOf(`--${k}`); return i >= 0 ? args[i + 1] : d; };
+const sr = parseInt(opt("sr", "48000"), 10);
+const unit = parseInt(opt("unit", "150"), 10);
+const expect = (opt("expect", "") || "").toUpperCase();
+const hp = parseFloat(opt("highpass", "2000"));
+const factor = parseFloat(opt("factor", "4"));
+const refractory = parseFloat(opt("refractory", "120"));
+
+const buf = readFileSync(path);
+const samples = new Float32Array(buf.buffer, buf.byteOffset, Math.floor(buf.byteLength / 4));
+
+// The app applies the highpass via a Web Audio BiquadFilterNode upstream of
+// the ScriptProcessor. Emulate that by pre-filtering, then stream blocks.
+const filtered = highpass(samples, sr, hp);
+
+const onsets = [];
+const det = new StreamingOnsetDetector({
+    sampleRate: sr,
+    thresholdFactor: factor,
+    refractoryMs: refractory,
+    onOnset: (t) => onsets.push(t),
+});
+
+const BLOCK = 2048;
+for (let i = 0; i < filtered.length; i += BLOCK) {
+    const block = filtered.subarray(i, Math.min(i + BLOCK, filtered.length));
+    det.process(block, (i / sr) * 1000);
+}
+
+const t = makeTiming(unit);
+const { morse, text } = decodeOnsets(onsets, t);
+console.log(`samples=${samples.length} (${(samples.length / sr).toFixed(1)}s)`);
+console.log(`onsets detected (live streaming path): ${onsets.length}`);
+console.log(`onset times (ms): ${onsets.map((x) => Math.round(x)).join(", ")}`);
+// inter-onset intervals — reveals rebounds (tiny gaps) vs real spacing
+console.log(`IOIs (ms): ${onsets.slice(1).map((x, i) => Math.round(x - onsets[i])).join(", ")}`);
+console.log(`morse:  ${morse}`);
+console.log(`decoded: ${JSON.stringify(text)}`);
+if (expect) {
+    const ok = text === expect;
+    console.log(`expected ${JSON.stringify(expect)} -> ${ok ? "OK" : "MISMATCH"}`);
+    process.exit(ok ? 0 : 2);
+}

tools/robot_clap_test.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Robot antenna-clap test for the Morse Code app (Lite / local daemon).
+
+Drives the antennas to clap on the app's impulse wire-code schedule (same code
+as lib/robot-tapper.js), records the laptop mic, runs the app's detector, and
+decodes — confirming the clap is audible + detectable and letting us tune the
+contact angle / hold / latency lead. Mirrors the proven, safe collision recipe
+from marionette/tests/test_antenna_collision.py (low-PID antennas stall at
+contact; gentle).
+
+MOVES THE ROBOT and MAKES NOISE. Run only when ready.
+
+    /Users/remi/.virtualenvs/mini/bin/python tools/robot_clap_test.py --once
+    /Users/remi/.virtualenvs/mini/bin/python tools/robot_clap_test.py --phrase "SOS" --unit 150
+"""
+from __future__ import annotations
+
+import argparse
+import math
+import sys
+import time
+from pathlib import Path
+
+import numpy as np
+import requests
+import sounddevice as sd
+
+TOOLS = Path(__file__).resolve().parent
+sys.path.insert(0, str(TOOLS))
+from calibrate_audio import (  # noqa: E402
+    timing, text_to_onsets, onsets_to_text, detect_transient_onsets, SR,
+)
+
+DEG = math.pi / 180.0
+
+# Defaults mirror lib/robot-tapper.js DEFAULT_TAP_PROFILE.
+RIGHT_REST_DEG = -39.0
+LEFT_REST_DEG = 0.0
+COLLISION_DEG = 40.0
+APPROACH_MS = 80.0
+HOLD_MS = 50.0
+RETURN_MS = 80.0
+RATE_HZ = 50.0
+LEAD_MS = 0.0
+
+
+def left_angle_at(t_ms, onsets, collision, approach, hold, ret, rest):
+    for T in onsets:
+        a0, h_end, r_end = T - approach, T + hold, T + hold + ret
+        if t_ms < a0 or t_ms > r_end:
+            continue
+        if t_ms <= T:
+            f = (t_ms - a0) / approach if approach > 0 else 1.0
+            f = min(1.0, max(0.0, f))
+            return rest + (collision - rest) * (f * f)
+        if t_ms <= h_end:
+            return collision
+        f = (t_ms - h_end) / ret
+        f = min(1.0, max(0.0, f))
+        return collision + (rest - collision) * (1 - (1 - f) * (1 - f))
+    return rest
+
+
+def stop_current_app():
+    try:
+        requests.post("http://127.0.0.1:8000/api/apps/stop-current-app", timeout=4)
+    except Exception:
+        pass
+
+
+def run(args):
+    from reachy_mini import ReachyMini
+    from reachy_mini.utils import create_head_pose
+
+    t = timing(args.unit)
+    onsets = [0.0] if args.once else text_to_onsets(args.phrase, t)
+    onsets = [o - args.lead for o in onsets]
+    collision = args.collision
+    end_ms = (onsets[-1] if onsets else 0) + HOLD_MS + RETURN_MS + 300
+    rec_secs = (args.leadin_ms + end_ms + args.settle_ms + 1200) / 1000.0
+
+    print(f"Phrase {args.phrase!r}  unit={args.unit}ms  contact={collision}°  "
+          f"hold={args.hold}ms  lead={args.lead}ms")
+    print(f"Recording mic for {rec_secs:.1f}s; robot will clap {len(onsets)} time(s).")
+
+    stop_current_app()
+    time.sleep(0.5)
+
+    r = ReachyMini(media_backend="no_media")
+    try:
+        r.goto_target(create_head_pose(),
+                      antennas=[LEFT_REST_DEG * DEG, RIGHT_REST_DEG * DEG], duration=1.0)
+        time.sleep(1.2)
+
+        rec = sd.rec(int(rec_secs * SR), samplerate=SR, channels=1, dtype="float32")
+        leadin = args.leadin_ms / 1000.0
+        time.sleep(leadin)  # silence head so the first onset has a quiet window
+
+        period = max(0.01, 1.0 / RATE_HZ)
+        t0 = time.monotonic()
+        while True:
+            tm = (time.monotonic() - t0) * 1000.0
+            if tm > end_ms:
+                break
+            left = left_angle_at(tm, onsets, collision, APPROACH_MS, args.hold, RETURN_MS, LEFT_REST_DEG)
+            r.set_target(head=np.eye(4), body_yaw=0.0,
+                         antennas=np.array([left * DEG, RIGHT_REST_DEG * DEG]))
+            nxt = t0 + (math.floor((tm / 1000.0) / period) + 1) * period
+            dt = nxt - time.monotonic()
+            if dt > 0:
+                time.sleep(dt)
+
+        # Let the final clap fully settle BEFORE any return motion — otherwise
+        # the return goto jerks the antenna and rings out extra clicks right
+        # after the last real onset (corrupting the last symbol).
+        time.sleep(args.settle_ms / 1000.0)
+        r.goto_target(create_head_pose(),
+                      antennas=[LEFT_REST_DEG * DEG, RIGHT_REST_DEG * DEG], duration=0.8)
+        sd.wait()
+    finally:
+        try:
+            r.goto_target(create_head_pose(), antennas=[0.0, 0.0], duration=0.8)
+            time.sleep(1.0)
+        except Exception:
+            pass
+
+    audio = rec.flatten()
+    if args.save:
+        audio.astype(np.float32).tofile(args.save)
+        print(f"Saved raw float32 mono @{SR}Hz to {args.save} ({audio.size} samples)")
+    detected = detect_transient_onsets(audio, SR, highpass_freq=args.highpass,
+                                       threshold_db=20 * math.log10(args.ratio))
+    detected_ms = [d * 1000.0 for d in detected]
+    peak = float(np.max(np.abs(audio))) if audio.size else 0.0
+
+    print(f"\nMic peak amplitude: {peak:.3f}  (want clearly > the room floor)")
+    print(f"Detected {len(detected_ms)} click(s); expected {len(onsets)}.")
+    if not args.once:
+        decoded = onsets_to_text(detected_ms, t)
+        ok = decoded == args.phrase.upper()
+        print(f"Decoded: {decoded!r}  [{'OK' if ok else 'MISMATCH'}]")
+        return 0 if ok else 2
+    print("Single-clap check: a click should be clearly visible above the floor.")
+    return 0 if len(detected_ms) >= 1 else 2
+
+
+def main():
+    ap = argparse.ArgumentParser(description="Robot antenna-clap test")
+    ap.add_argument("--once", action="store_true", help="one gentle clap (safety check)")
+    ap.add_argument("--phrase", default="SOS")
+    ap.add_argument("--unit", type=int, default=150)
+    ap.add_argument("--collision", type=float, default=COLLISION_DEG)
+    ap.add_argument("--hold", type=float, default=HOLD_MS)
+    ap.add_argument("--lead", type=float, default=LEAD_MS, help="advance timeline (ms)")
+    ap.add_argument("--leadin-ms", type=float, default=600.0)
+    ap.add_argument("--settle-ms", type=float, default=600.0,
+                    help="quiet settle after the last clap before returning to rest")
+    ap.add_argument("--highpass", type=float, default=2000.0)
+    ap.add_argument("--ratio", type=float, default=0.1)
+    ap.add_argument("--save", default=None, help="save raw float32 mono recording to this path")
+    return run(ap.parse_args())
+
+
+if __name__ == "__main__":
+    sys.exit(main())