Sharpen PS Move spectral onset colors

docs/implementation-plan.md

@@ -443,8 +443,11 @@ a named invariant that the native runtime cannot protect yet.
   `/dev/hidraw1`.
   `tools/nightwing_psmove_music_pulse.py` is the current controlled
   LED/sphere command proof: Starfire owns the Scarlett ASIO loopback envelope,
-  computes a cheap broadband spectral-rise onset detector with a tiny FFT, and
-  streams RGB report `0x06` frames over SSH to Nightwing's hidraw writer.
+  computes a thresholdable tiny-FFT onset novelty function from
+  adaptive-whitened broadband spectral flux, suppresses duplicate spikes with
+  a short refractory window, fits a provisional fundamental, folds spectral
+  bin energy into one octave around that fundamental, and streams color-balanced
+  RGB report `0x06` frames over SSH to Nightwing's hidraw writer.
   Next useful cut is direct report decoding plus optical sphere observations
   from PS3 Eyes/Kiyos entering the same calibration-evidence path as LED
   spline constraints.

state/evidence.jsonl

@@ -269,3 +269,4 @@
 {"ts": "2026-05-31T17:56:57+01:00", "type": "verification", "status": "partial", "note": "Nightwing BlueZ is installed/enabled with adapter 5c:93:a2:9c:a8:a8 and ClassicBondedOnly=false for PS Move-style classic HID. Mimir.PsMoveProbe --pair-host wrote that host address to the connected PS Move over USB col02 feature report 0x05; controller address read back as 00:06:f5:23:e2:d1. Wireless connection is not yet verified because Starfire still sees the controller on USB; unplug and press PS while Nightwing scans."}
 {"ts": "2026-05-31T18:06:53+01:00", "type": "implementation", "status": "ok", "note": "Mimir editor UI surface is now authored as a nested Eve/CultUI grid: horizontal outer columns with vertical stacks inside, instead of legacy floating debug panels. Fensalir owns weighted Direct2D lowering of the shared surface contract."}
 {"ts": "2026-05-31T18:22:20+01:00", "type": "verification", "status": "ok", "note": "PS Move controller 00:06:f5:23:e2:d1 connects wirelessly to Nightwing after BlueZ PS Move API-style registration and exposes js0/event14/hidraw1; tools/nightwing_psmove_music_pulse.py drives LED report 0x06 from Starfire Scarlett ASIO loopback channels 2/3 using tiny-FFT broadband spectral-rise onset scores."}
+{"ts": "2026-05-31T18:32:23+01:00", "type": "implementation", "status": "ok", "note": "PS Move music pulse now uses a thresholdable onset novelty function: adaptive-whitened spectral flux, baseline/std gating, refractory duplicate suppression, best-fit fundamental search, and one-octave folded spectral color balance for RGB LED report 0x06."}

state/map.yaml

@@ -170,9 +170,11 @@ invariants:
     controller now connects wirelessly as `Motion Controller` and exposes
     `/dev/input/js0`, `/dev/input/event14`, and `/dev/hidraw1`.
     `tools/nightwing_psmove_music_pulse.py` proves the output path by reading
-    Starfire Scarlett ASIO loopback channels 2/3, computing a cheap broadband
-    spectral-rise onset envelope from a tiny FFT, and streaming RGB LED report
-    `0x06` frames over SSH to Nightwing's hidraw writer.
+    Starfire Scarlett ASIO loopback channels 2/3, computing a thresholdable
+    adaptive-whitened spectral-flux onset novelty function, fitting a
+    provisional fundamental, folding spectral bin energy into one octave around
+    that fundamental for RGB balance, and streaming LED report `0x06` frames
+    over SSH to Nightwing's hidraw writer.
   - PS3 Eyes now have a first sparse tracking surface instead of only cadence
     proof. `MimirSparseFeatureTracker` detects Bayer/luma corner features,
     links them into sensor-local tracks with clipspace position, velocity, age,

tools/nightwing_psmove_music_pulse.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import argparse
 import base64
+import colorsys
 import ctypes
 import math
 import os
@@ -101,7 +102,15 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--floor", type=float, default=0.008, help="Noise floor.")
     parser.add_argument("--fft-size", type=int, default=512, help="Tiny FFT size for onset flux.")
     parser.add_argument("--onset-gain", type=float, default=7.0, help="Spectral-rise onset gain.")
-    parser.add_argument("--onset-decay", type=float, default=0.72, help="Per-frame onset decay.")
+    parser.add_argument("--onset-decay", type=float, default=0.38, help="Per-frame onset decay.")
+    parser.add_argument("--onset-threshold", type=float, default=1.45, help="Adaptive onset threshold in standard deviations.")
+    parser.add_argument("--onset-baseline-rate", type=float, default=0.015, help="Adaptive novelty baseline update rate.")
+    parser.add_argument("--onset-cooldown-ms", type=float, default=70.0, help="Suppress duplicate onset spikes inside this window.")
+    parser.add_argument("--whiten-rate", type=float, default=0.035, help="Adaptive spectral-floor update rate.")
+    parser.add_argument("--whiten-contrast", type=float, default=2.4, help="Adaptive whitening contrast exponent.")
+    parser.add_argument("--color-contrast", type=float, default=1.8, help="Folded octave color balance contrast.")
+    parser.add_argument("--fundamental-min", type=float, default=55.0, help="Minimum best-fit fundamental in Hz.")
+    parser.add_argument("--fundamental-max", type=float, default=880.0, help="Maximum best-fit fundamental in Hz.")
     return parser.parse_args()
 
 
@@ -111,7 +120,14 @@ class AsioOnsetReader:
         self.fft_size = max(64, int(fft_size))
         self.pending: list[float] = []
         self.previous = np.zeros(self.fft_size // 2 + 1, dtype=np.float32)
+        self.floor = np.zeros(self.fft_size // 2 + 1, dtype=np.float32)
         self.window = np.hanning(self.fft_size).astype(np.float32)
+        self.color_balance = np.array([0.25, 0.2, 0.55], dtype=np.float32)
+        self.flux_mean = 0.0
+        self.flux_var = 1e-6
+        self.frame_index = 0
+        self.last_onset_frame = -1_000_000
+        self.last_onset_score = 0.0
         self.dll = ctypes.WinDLL(dll_path)
         self.dll.mimir_asio_create.argtypes = [
             ctypes.c_char_p,
@@ -151,12 +167,25 @@ class AsioOnsetReader:
         self.sample_rate = int(actual_rate.value)
         self.input_count = int(input_count.value)
         self.max_frames = max(1, int(max_frames.value))
+        self.freqs = np.fft.rfftfreq(self.fft_size, 1.0 / self.sample_rate)
         if not self.dll.mimir_asio_start(self.handle):
             self.close()
             raise RuntimeError("Could not start Mimir ASIO capture source")
         self.buffer = (ctypes.c_float * self.max_frames)()
 
-    def read_onset(self, floor: float, gain: float, deadline: float) -> tuple[float, float] | None:
+    def read_onset(
+        self,
+        floor: float,
+        gain: float,
+        deadline: float,
+        whiten_rate: float,
+        whiten_contrast: float,
+        fundamental_min: float,
+        fundamental_max: float,
+        onset_threshold: float,
+        onset_baseline_rate: float,
+        onset_cooldown_ms: float,
+    ) -> tuple[float, float, tuple[float, float, float], float] | None:
         channel = ctypes.c_int()
         timestamp_ns = ctypes.c_longlong()
         sequence = ctypes.c_ulonglong()
@@ -185,15 +214,99 @@ class AsioOnsetReader:
                 rms = float(np.sqrt(np.mean(frame * frame)))
                 spectrum = np.abs(np.fft.rfft(frame * self.window)).astype(np.float32)
                 spectrum = np.log1p(spectrum)
-                rise = np.maximum(spectrum - self.previous, 0.0)
-                self.previous = spectrum
-                flux = float(np.mean(rise[2:]))
-                onset = max(0.0, (flux - floor) * gain)
-                body = max(0.0, (rms - floor) * 2.0)
-                return onset, body
+                if not np.any(self.floor):
+                    self.floor = spectrum.copy()
+                floor_alpha = max(0.001, min(0.5, whiten_rate))
+                self.floor = np.minimum(
+                    spectrum,
+                    (1.0 - floor_alpha) * self.floor + floor_alpha * spectrum,
+                )
+                whitened = np.maximum(spectrum - self.floor, 0.0)
+                if whitened.size > 4:
+                    whitened -= np.quantile(whitened[2:], 0.35)
+                whitened = np.maximum(whitened, 0.0)
+                if whitened.size > 4:
+                    peak = float(np.quantile(whitened[2:], 0.97))
+                    if peak > 1e-6:
+                        whitened = np.clip(whitened / peak, 0.0, 2.0)
+                whitened = np.power(whitened, max(0.5, whiten_contrast)).astype(np.float32)
+
+                rise = np.maximum(whitened - self.previous, 0.0)
+                self.previous = whitened
+                active = rise[2:]
+                flux = float(np.mean(active) + 0.35 * np.percentile(active, 92))
+                baseline_rate = max(0.001, min(0.2, onset_baseline_rate))
+                baseline = self.flux_mean
+                flux_std = math.sqrt(max(1e-9, self.flux_var))
+                novelty = max(0.0, flux - baseline - max(0.0, onset_threshold) * flux_std)
+                delta = flux - self.flux_mean
+                adapt_rate = baseline_rate * (0.25 if novelty > 0.0 else 1.0)
+                self.flux_mean += adapt_rate * delta
+                self.flux_var = (1.0 - adapt_rate) * self.flux_var + adapt_rate * delta * delta
+                hop_seconds = (self.fft_size // 2) / max(1, self.sample_rate)
+                cooldown_frames = max(1, int((onset_cooldown_ms / 1000.0) / hop_seconds))
+                if self.frame_index - self.last_onset_frame < cooldown_frames:
+                    if novelty <= self.last_onset_score * 1.35:
+                        novelty = 0.0
+                if novelty > 0.0:
+                    self.last_onset_frame = self.frame_index
+                    self.last_onset_score = novelty
+                self.frame_index += 1
+                fundamental = self._estimate_fundamental(whitened, fundamental_min, fundamental_max)
+                color = self._fold_color(whitened + rise * 1.8, fundamental)
+                onset = max(0.0, novelty * gain)
+                body = max(0.0, (rms - floor) * 0.45)
+                return onset, body, color, fundamental
         if not self.pending:
             return None
-        return 0.0, 0.0
+        return 0.0, 0.0, tuple(float(v) for v in self.color_balance), 0.0
+
+    def _estimate_fundamental(self, spectrum: np.ndarray, min_hz: float, max_hz: float) -> float:
+        nyquist = self.sample_rate * 0.5
+        min_hz = max(20.0, min_hz)
+        max_hz = min(max_hz, nyquist * 0.45)
+        candidates = np.geomspace(min_hz, max_hz, 48)
+        best_hz = min_hz
+        best_score = 0.0
+        for candidate in candidates:
+            score = 0.0
+            harmonic = 1
+            while harmonic * candidate < nyquist:
+                freq = harmonic * candidate
+                bin_index = int(round(freq * self.fft_size / self.sample_rate))
+                if 1 <= bin_index < spectrum.size:
+                    score += float(spectrum[bin_index]) / math.sqrt(harmonic)
+                harmonic += 1
+            score /= max(1.0, math.sqrt(harmonic - 1))
+            if score > best_score:
+                best_score = score
+                best_hz = float(candidate)
+        return best_hz if best_score > 1e-4 else 0.0
+
+    def _fold_color(self, spectrum: np.ndarray, fundamental: float) -> tuple[float, float, float]:
+        if fundamental <= 0.0:
+            return tuple(float(v) for v in self.color_balance)
+        accum = np.zeros(3, dtype=np.float32)
+        for bin_index in range(2, spectrum.size):
+            energy = float(spectrum[bin_index])
+            if energy <= 0.0:
+                continue
+            freq = float(self.freqs[bin_index])
+            if freq < fundamental * 0.5:
+                continue
+            octave_phase = math.log2(freq / fundamental) % 1.0
+            r, g, b = colorsys.hsv_to_rgb(octave_phase, 0.95, 1.0)
+            accum += energy * np.array([r, g, b], dtype=np.float32)
+        total = float(np.sum(accum))
+        if total <= 1e-6:
+            target = self.color_balance
+        else:
+            target = accum / total
+        self.color_balance = 0.72 * self.color_balance + 0.28 * target
+        color_total = float(np.sum(self.color_balance))
+        if color_total > 1e-6:
+            self.color_balance /= color_total
+        return tuple(float(v) for v in self.color_balance)
 
     def close(self) -> None:
         if getattr(self, "handle", None):
@@ -201,13 +314,16 @@ class AsioOnsetReader:
             self.handle = None
 
 
-def rgb_from_level(level: float) -> tuple[int, int, int]:
+def rgb_from_level(level: float, balance: tuple[float, float, float], color_contrast: float) -> tuple[int, int, int]:
     level = max(0.0, min(1.0, level))
     glow = level ** 0.55
-    # Blue-white idle, hot magenta/white peaks. Useful for camera tracking too.
-    r = int(18 + 237 * (glow ** 1.35))
-    g = int(28 + 170 * max(0.0, glow - 0.35) / 0.65)
-    b = int(80 + 175 * glow)
+    color = np.asarray(balance, dtype=np.float32)
+    color = np.power(np.maximum(color, 0.0), max(0.5, color_contrast))
+    color /= max(1e-6, float(np.max(color)))
+    white = max(0.0, glow - 0.86) / 0.14
+    r = int(8 + 232 * glow * (0.12 + 0.88 * float(color[0])) + 14 * white)
+    g = int(8 + 232 * glow * (0.12 + 0.88 * float(color[1])) + 14 * white)
+    b = int(12 + 232 * glow * (0.12 + 0.88 * float(color[2])) + 14 * white)
     return (min(255, r), min(255, g), min(255, b))
 
 
@@ -277,10 +393,21 @@ def main() -> int:
     try:
         while not stopping:
             if asio is not None:
-                onset = asio.read_onset(args.floor, args.onset_gain, time.monotonic() + (1.0 / args.fps))
+                onset = asio.read_onset(
+                    args.floor,
+                    args.onset_gain,
+                    time.monotonic() + (1.0 / args.fps),
+                    args.whiten_rate,
+                    args.whiten_contrast,
+                    args.fundamental_min,
+                    args.fundamental_max,
+                    args.onset_threshold,
+                    args.onset_baseline_rate,
+                    args.onset_cooldown_ms,
+                )
                 if onset is None:
                     continue
-                hit, body = onset
+                hit, body, balance, fundamental = onset
                 onset_env = max(hit, onset_env * args.onset_decay)
                 level = max(body * args.gain, onset_env)
             else:
@@ -292,15 +419,32 @@ def main() -> int:
                 samples = struct.unpack("<" + "h" * count, data)
                 rms = math.sqrt(sum(sample * sample for sample in samples) / count) / 32768.0
                 level = max(0.0, (rms - args.floor) * args.gain)
-            env = max(level, env * 0.82)
-            r, g, b = rgb_from_level(env)
+                balance = (0.25, 0.2, 0.55)
+                fundamental = 0.0
+                hit = 0.0
+            env_decay = args.onset_decay if asio is not None else 0.82
+            env = max(level, env * env_decay)
+            r, g, b = rgb_from_level(env, balance, args.color_contrast)
             if ssh.poll() is not None:
                 raise RuntimeError(f"SSH LED receiver exited with {ssh.returncode}")
             ssh.stdin.write(f"{r} {g} {b}\n".encode("ascii"))
             ssh.stdin.flush()
             frames += 1
+            if hit > 0.02:
+                print(
+                    f"peak onset={hit:.3f} f0={fundamental:.1f} "
+                    f"balance={balance[0]:.2f},{balance[1]:.2f},{balance[2]:.2f} "
+                    f"rgb={r},{g},{b}",
+                    flush=True,
+                )
             if frames % int(args.fps * 2) == 0:
-                print(f"level={env:.3f} onset={onset_env:.3f} rgb={r},{g},{b}", flush=True)
+                print(
+                    "level="
+                    f"{env:.3f} onset={onset_env:.3f} f0={fundamental:.1f} "
+                    f"balance={balance[0]:.2f},{balance[1]:.2f},{balance[2]:.2f} "
+                    f"rgb={r},{g},{b}",
+                    flush=True,
+                )
     finally:
         try:
             ssh.stdin.write(b"0 0 32\n")