From 46d2eec5b64b2ec5fc6219d84c8ee870ef909697 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 31 May 2026 18:32:57 +0100 Subject: [PATCH] Sharpen PS Move spectral onset colors --- docs/implementation-plan.md | 7 +- state/evidence.jsonl | 1 + state/map.yaml | 8 +- tools/nightwing_psmove_music_pulse.py | 182 +++++++++++++++++++++++--- 4 files changed, 174 insertions(+), 24 deletions(-) diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 3724a9e..2e94d1c 100644 --- a/docs/implementation-plan.md +++ b/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. diff --git a/state/evidence.jsonl b/state/evidence.jsonl index ba498a2..828dd01 100644 --- a/state/evidence.jsonl +++ b/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."} diff --git a/state/map.yaml b/state/map.yaml index b3a6a0b..ae19f28 100644 --- a/state/map.yaml +++ b/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, diff --git a/tools/nightwing_psmove_music_pulse.py b/tools/nightwing_psmove_music_pulse.py index dcce58f..0a171df 100644 --- a/tools/nightwing_psmove_music_pulse.py +++ b/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")