From d716fbaee2d6ec51bd1018613230e7fb5bbdf748 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 31 May 2026 21:59:06 +0100 Subject: [PATCH] Add music-keyed Move chirp planner --- docs/move-controller-stream.md | 55 ++++ state/evidence.jsonl | 2 + tools/music_keyed_move_chirp_sync.py | 458 +++++++++++++++++++++++++++ 3 files changed, 515 insertions(+) create mode 100644 tools/music_keyed_move_chirp_sync.py diff --git a/docs/move-controller-stream.md b/docs/move-controller-stream.md index dd67e64..b3225c4 100644 --- a/docs/move-controller-stream.md +++ b/docs/move-controller-stream.md @@ -135,3 +135,58 @@ controller identity. those histories into the existing Feature evidence path. This is deliberately 2D observation history. 6DoF pose waits for calibrated frustums and the global residual owner. + +## Music-Keyed Calibration Words + +`tools/music_keyed_move_chirp_sync.py` is the first planner for calibration +words that are both musical and machine-decodable. It listens to Scarlett ASIO, +uses the same Perlines-style adaptive FFT delta whitening as +`tools/nightwing_psmove_music_pulse.py`, autocorrelates the whitened broadband +spectral-rise function for tempo, estimates a lightweight root from recent +best-fit fundamentals, then writes one shared plan: + +- beat-aligned chirps quantized to the estimated minor-pentatonic scale; +- de Bruijn-coded event symbols so adjacent timing/frequency/color words remain + distinguishable; +- per-Move RGB pulses using stable controller identities and unique colors; +- a JSON receipt shaped like AquaSynth song analysis vocabulary: + `tempo_bpm`, `beat_seconds`, `tempo_confidence`, `root_note`, + `suggested_scale`, `scale_frequencies_hz`, and + `whitened_spectral_autocorr`. + +The synthetic smoke is: + +```powershell +& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\music_keyed_move_chirp_sync.py --mode synthetic --dry-run --events 8 --out-dir artifacts/runtime/music-keyed-move-chirp-sync-smoke +``` + +The live dry-run against Scarlett is: + +```powershell +& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\music_keyed_move_chirp_sync.py --dry-run --analyze-seconds 5 --events 8 --out-dir artifacts/runtime/music-keyed-move-chirp-sync-live-dry +``` + +For double-time material: + +```powershell +& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' .\tools\music_keyed_move_chirp_sync.py --dry-run --tempo-grid doubletime --tempo-max-bpm 260 --analyze-seconds 5 --events 8 --out-dir artifacts/runtime/music-keyed-move-chirp-sync-dirtyphonics-dry +``` + +On the 2026-05-31 live dry-run, the planner produced a usable but not final +receipt: about `92.31` BPM, beat `0.650s`, tempo confidence `0.343`, root `A`, +key confidence `0.716`. That is enough to prove the dataflow, not enough to +claim final musical tempo authority. + +For dense double-time material, use `--tempo-grid doubletime`; the planner then +selects the double-time member of the autocorrelation tempo family instead of +preferring the slower parent grid. On the Dirtyphonics-style live source this +produced about `171.43` BPM, beat `0.350s`, tempo confidence `0.358`, root `A`, +key confidence `0.703` after widening the search ceiling with +`--tempo-max-bpm 260`. The receipt records `tempo_family` so later decoders can +see the selected grid alongside half-time/double-time alternatives. + +The non-dry run targets the two wireless Nightwing Moves by default: +`00:07:04:a6:be:5f` on `/dev/hidraw2` and `00:06:f5:23:e2:d1` on +`/dev/hidraw3`. Current BlueZ permissions expose those Bluetooth hidraw nodes +as root-only, so multi-Move LED actuation needs a udev rule or privileged +helper before the non-root SSH writer can drive both spheres. diff --git a/state/evidence.jsonl b/state/evidence.jsonl index ae40831..c8ccd2b 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -280,4 +280,6 @@ {"ts": "2026-05-31T21:07:55+01:00", "type": "implementation", "status": "ok", "note": "Nightwing visible ops now drinks typed witness status through Mimir/Odin instead of a local status sidecar: tools/nightwing_typed_witness_publisher.py sends binary MessagePack mimir.eve_sensor_observation.v1 records to Mimir.EveSensorReceiver, Odin projects them as observation-stream nodes, and the Nightwing Eve TUI renders those accepted CultMesh streams. Live deployment showed fresh nightwing-video0..3 and nightwing-leap-tracking streams on the TUI."} {"ts": "2026-05-31T21:25:32+01:00", "type": "implementation", "status": "ok", "note": "Added MimirRoomCalibrationLockSolver as the first room-lock coordinator. It consumes LED spline residuals, camera-rig pose updates, Move controller feature histories, and PS3 Eye sparse feature candidates, reports named missing witness families, and only locks when all families are present. --room-calibration-lock-smoke passes with 3 LED cameras, 5 shared LEDs, 3 pose updates, 2 Moves, 2 Eye sources, 176 stable Eye tracks, confidence 0.778; removing Moves drops to InsufficientWitnesses."} {"ts": "2026-05-31T21:35:10+01:00", "type": "verification", "status": "partial", "note": "Second PS Move was visible on Nightwing USB as Sony 054c:03d5 at /dev/hidraw1. A Linux hidraw feature-report write set its host address to Nightwing Bluetooth adapter 5c:93:a2:9c:a8:a8; controller address read back as 00:07:04:a6:be:5f and host read back correctly. Wireless BlueZ appearance is still pending unplug plus PS-button reconnect near Nightwing."} +{"ts": "2026-05-31T21:48:00+01:00", "type": "verification", "status": "ok", "note": "Nightwing now has two wireless PS Moves connected and trusted: 00:07:04:a6:be:5f appears as bluetooth /dev/hidraw2 plus /dev/input/js1, and 00:06:f5:23:e2:d1 appears as bluetooth /dev/hidraw3 plus /dev/input/js2. A third cabled Move remains visible separately as usb /dev/hidraw1 plus /dev/input/js0. Bluetooth hidraw nodes are root-only on this boot, so multi-Move LED actuation needs a udev permission rule or privileged helper before non-root scripts can drive both lights."} +{"ts": "2026-05-31T21:58:29+01:00", "type": "implementation", "status": "ok", "note": "Added tools/music_keyed_move_chirp_sync.py as the first music-keyed audiovisual calibration word planner. It reuses the Perlines-style whitened broadband spectral-rise onset path, autocorrelates onset for tempo, estimates a lightweight root from recent fundamentals, quantizes chirps to a minor-pentatonic scale, and schedules unique per-Move color pulses from one shared JSON receipt. Synthetic dry-run produced 120 BPM/A with high confidence; live Scarlett dry-run produced a slower 92.31 BPM grid, and Dirtyphonics double-time mode with --tempo-grid doubletime --tempo-max-bpm 260 produced a 171.43 BPM grid. Non-dry multi-Move actuation still needs hidraw permission/root-helper work for bluetooth /dev/hidraw2 and /dev/hidraw3."} {"ts":"2026-05-31T21:48:00+01:00","type":"implementation","status":"ok","note":"Mimir's Eve/CultUI editor surface now uses editor anatomy instead of dashboard columns: Scene Graph left, Scene View center, thin Asset Browser below the view, and Inspector right. The old Program/Sync rail was removed from the inspector side so telemetry no longer steals editor hierarchy. Headless empty-runtime capture verified the shell at 1280x720 with no sibling overlap.","files":["src/Mimir.Runtime/MimirRuntime.cs"],"source":"dotnet build src/Mimir.App/Mimir.App.csproj --no-restore; Mimir.App --headless --headless-width 1280 --headless-height 720 --capture-frame artifacts/runtime/mimir-editor-shell-final-headless.png"} diff --git a/tools/music_keyed_move_chirp_sync.py b/tools/music_keyed_move_chirp_sync.py new file mode 100644 index 0000000..5d7049b --- /dev/null +++ b/tools/music_keyed_move_chirp_sync.py @@ -0,0 +1,458 @@ +#!/usr/bin/env python3 +"""Emit beat/key-aligned chirps and unique PS Move color pulses. + +This is the musical calibration word planner for the current Mimir/Nightwing +setup. It uses the same Perlines-style adaptive whitening onset reader as +``nightwing_psmove_music_pulse.py``, estimates tempo from autocorrelation of +the whitened broadband spectral-rise function, estimates a lightweight root +from recent best-fit fundamentals, then schedules chirps on the beat in that +scale while pulsing each Move with a unique coded color. + +Provenance: +- The onset whitening path is intentionally shared with + ``tools/nightwing_psmove_music_pulse.py`` and its Perlines-derived adaptive + FFT delta whitening. +- The beat/key feature vocabulary mirrors AquaSynth song challenge analysis: + tempo_bpm, beat_seconds, root_note, suggested_scale, scale_frequencies_hz, + and whitened-spectral-autocorr. +- The event codebook reuses the de Bruijn schedule idea already present in + ``tools/move_latency_probe.py``. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import math +import os +import subprocess +import sys +import time +import wave +from dataclasses import dataclass +from pathlib import Path + +import numpy as np + +from nightwing_psmove_music_pulse import AsioOnsetReader + + +REMOTE_MULTI_MOVE = r""" +import os +import sys +import time + +moves = {} +for spec in sys.argv[1].split(","): + if not spec: + continue + name, path = spec.split("=", 1) + moves[name] = path +events = [] +for line in sys.stdin: + line = line.strip() + if not line: + continue + if line == "go": + break + fields = line.split() + if len(fields) != 5: + continue + offset, name, r, g, b = fields + events.append((float(offset), name, int(r), int(g), int(b))) + +log = os.path.expanduser("~/.local/state/gamecult/codex-ssh-activity.log") +os.makedirs(os.path.dirname(log), exist_ok=True) +with open(log, "a", encoding="utf-8") as f: + f.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S%z')} Codex: multi-Move beat/key chirp receiver armed for {len(moves)} Moves.\n") + +def write_rgb(path, r, g, b): + with open(path, "wb", buffering=0) as device: + device.write(bytes([0x06, 0, max(0, min(255, r)), max(0, min(255, g)), max(0, min(255, b)), 0, 0, 0, 0])) + +for path in moves.values(): + write_rgb(path, 0, 0, 0) + +start = time.perf_counter() +for offset, name, r, g, b in events: + wait = start + offset - time.perf_counter() + if wait > 0: + time.sleep(wait) + path = moves.get(name) + if path: + write_rgb(path, r, g, b) + +time.sleep(0.08) +for path in moves.values(): + write_rgb(path, 0, 0, 0) +""" + + +NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] +MINOR_PENTATONIC = [0, 3, 5, 7, 10, 12, 15, 17, 19, 22, 24] + + +@dataclass(frozen=True) +class MoveTarget: + name: str + hidraw: str + color: tuple[int, int, int] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--mode", choices=("live", "synthetic"), default="live") + parser.add_argument("--out-dir", default="artifacts/runtime/music-keyed-move-chirp-sync") + parser.add_argument("--ffplay", default=str(Path(r"C:\Users\Meta\AppData\Local\Microsoft\WinGet\Packages\Gyan.FFmpeg_Microsoft.Winget.Source_8wekyb3d8bbwe\ffmpeg-8.1.1-full_build\bin\ffplay.exe"))) + parser.add_argument("--ssh-target", default="nightwing") + parser.add_argument( + "--move", + action="append", + default=[], + help="Move spec name=/dev/hidrawN:#rrggbb. Defaults to the two wireless Nightwing Moves.", + ) + parser.add_argument("--analyze-seconds", type=float, default=8.0) + parser.add_argument("--events", type=int, default=16) + parser.add_argument("--lead-seconds", type=float, default=0.55) + parser.add_argument("--chirp-ms", type=float, default=55.0) + parser.add_argument("--pulse-ms", type=float, default=90.0) + parser.add_argument("--sample-rate", type=int, default=48000) + parser.add_argument("--fps", type=float, default=60.0) + parser.add_argument("--fft-size", type=int, default=512) + parser.add_argument("--drain-blocks", type=int, default=4096) + parser.add_argument("--tempo-min-bpm", type=float, default=60.0) + parser.add_argument("--tempo-max-bpm", type=float, default=200.0) + parser.add_argument("--tempo-prefer-min-bpm", type=float, default=75.0) + parser.add_argument("--tempo-prefer-max-bpm", type=float, default=165.0) + parser.add_argument( + "--tempo-grid", + choices=("preferred", "strongest", "doubletime"), + default="preferred", + help="Beat grid selection. Use doubletime for dense drum-and-bass/dubstep style calibration pulses.", + ) + parser.add_argument("--dry-run", action="store_true", help="Write the plan and WAV without playing or touching Moves.") + parser.add_argument("--source-gain", type=float, default=0.35) + parser.add_argument("--asio-dll", default=str(Path(__file__).resolve().parents[1] / "native" / "asio_capture" / "build" / "Release" / "mimir_asio_capture.dll")) + parser.add_argument("--asio-clsid", default="{AC4D0455-50D7-4498-B3CD-9A41D130B759}") + parser.add_argument("--asio-channels", default="2,3") + return parser.parse_args() + + +def parse_moves(values: list[str]) -> list[MoveTarget]: + specs = values or [ + "move-00-07-04-a6-be-5f=/dev/hidraw2:#ff2a00", + "move-00-06-f5-23-e2-d1=/dev/hidraw3:#00a8ff", + ] + moves: list[MoveTarget] = [] + for spec in specs: + name_path, color_text = spec.split(":", 1) + name, path = name_path.split("=", 1) + color_text = color_text.lstrip("#") + if len(color_text) != 6: + raise ValueError(f"invalid color in move spec: {spec}") + color = tuple(int(color_text[index : index + 2], 16) for index in (0, 2, 4)) + moves.append(MoveTarget(name, path, color)) # type: ignore[arg-type] + return moves + + +def build_debruijn(alphabet_size: int, order: int) -> list[int]: + sequence: list[int] = [] + a = [0] * (alphabet_size * order) + + def db(t: int, p: int) -> None: + if t > order: + if order % p == 0: + sequence.extend(a[1 : p + 1]) + return + a[t] = a[t - p] + db(t + 1, p) + for value in range(a[t - p] + 1, alphabet_size): + a[t] = value + db(t + 1, t) + + db(1, 1) + return sequence + + +def rotate_to_distinct_opening(symbols: list[int], order: int) -> list[int]: + if len(symbols) <= order: + return symbols + windows: set[tuple[int, ...]] = set() + for index in range(len(symbols)): + window = tuple(symbols[(index + offset) % len(symbols)] for offset in range(order)) + if window in windows: + continue + windows.add(window) + if len(set(window)) == min(order, len(set(symbols))): + return symbols[index:] + symbols[:index] + return symbols + + +def synthetic_analysis(args: argparse.Namespace) -> tuple[np.ndarray, np.ndarray, np.ndarray]: + frame_count = max(32, int(args.analyze_seconds * args.fps)) + times = np.arange(frame_count, dtype=np.float32) / args.fps + beat_seconds = 0.5 + onset = np.zeros(frame_count, dtype=np.float32) + for beat in np.arange(0.25, args.analyze_seconds, beat_seconds): + onset += np.exp(-((times - beat) ** 2) / 0.0009).astype(np.float32) + fundamentals = 220.0 + 8.0 * np.sin(times * 2.0 * math.pi * 0.5) + weights = np.maximum(onset, 0.05).astype(np.float32) + return times, onset, fundamentals.astype(np.float32) * weights + + +def live_analysis(args: argparse.Namespace) -> tuple[np.ndarray, np.ndarray, np.ndarray]: + channels = {int(part.strip()) for part in args.asio_channels.split(",") if part.strip()} + reader = AsioOnsetReader(args.asio_dll, args.asio_clsid, args.sample_rate, channels, args.fft_size) + times: list[float] = [] + onset_scores: list[float] = [] + weighted_fundamentals: list[float] = [] + start = time.monotonic() + try: + while time.monotonic() - start < args.analyze_seconds: + result = reader.read_onset( + floor=0.008, + gain=1.0, + deadline=time.monotonic() + (1.0 / args.fps), + whiten_fast=0.78, + whiten_slow=0.12, + whiten_delta_fast=0.62, + whiten_delta_slow=0.08, + whiten_decay=0.78, + whiten_contrast=3.4, + fundamental_min=55.0, + fundamental_max=880.0, + onset_percentile_threshold=0.0, + onset_exponent=2.0, + onset_history_ms=4500.0, + onset_cooldown_ms=0.0, + warmup_ms=450.0, + drain_blocks=args.drain_blocks, + ) + if result is None: + continue + hit, _body, _balance, fundamental, range_hit, percentile = result + score = max(hit, (max(0.0, min(1.0, percentile)) ** 2) * max(0.2, min(1.5, range_hit))) + times.append(time.monotonic() - start) + onset_scores.append(score) + weighted_fundamentals.append(fundamental * max(score, 0.0)) + finally: + reader.close() + if len(onset_scores) < 8: + raise RuntimeError("not enough live onset frames collected") + return ( + np.asarray(times, dtype=np.float32), + np.asarray(onset_scores, dtype=np.float32), + np.asarray(weighted_fundamentals, dtype=np.float32), + ) + + +def estimate_tempo(args: argparse.Namespace, onset: np.ndarray, fps: float) -> tuple[float, float, float, list[dict[str, float]]]: + centered = np.maximum(onset - float(np.mean(onset)), 0.0) + energy = float(np.dot(centered, centered)) + if energy <= 1e-8: + return 120.0, 0.5, 0.0, [] + min_bpm = max(20.0, min(args.tempo_min_bpm, args.tempo_max_bpm)) + max_bpm = max(min_bpm + 1.0, max(args.tempo_min_bpm, args.tempo_max_bpm)) + prefer_min = max(min_bpm, min(args.tempo_prefer_min_bpm, args.tempo_prefer_max_bpm)) + prefer_max = min(max_bpm, max(args.tempo_prefer_min_bpm, args.tempo_prefer_max_bpm)) + min_lag = max(1, int(math.floor(fps * 60.0 / max_bpm))) + max_lag = min(len(centered) - 1, int(math.ceil(fps * 60.0 / min_bpm))) + candidates: list[dict[str, float]] = [] + for lag in range(min_lag, max_lag + 1): + score = float(np.dot(centered[:-lag], centered[lag:]) / energy) + bpm = 60.0 / (lag / fps) + if prefer_min <= bpm <= prefer_max: + tempo_weight = 1.0 + else: + distance = min(abs(bpm - prefer_min), abs(bpm - prefer_max)) + tempo_weight = max(0.60, 1.0 - distance / 160.0) + candidates.append({"lag": float(lag), "bpm": bpm, "score": max(0.0, score), "weighted_score": max(0.0, score) * tempo_weight}) + if args.tempo_grid == "strongest": + best = max(candidates, key=lambda item: item["score"], default={"lag": fps * 0.5, "bpm": 120.0, "score": 0.0}) + elif args.tempo_grid == "doubletime": + parent = max(candidates, key=lambda item: item["weighted_score"], default={"lag": fps * 0.5, "bpm": 120.0, "score": 0.0}) + target = min(max_bpm, max(min_bpm, parent["bpm"] * 2.0)) + nearby = [ + item for item in candidates + if abs(item["bpm"] - target) <= max(6.0, target * 0.08) and item["score"] >= parent["score"] * 0.45 + ] + best = max(nearby, key=lambda item: item["score"], default=min(candidates, key=lambda item: abs(item["bpm"] - target))) + else: + best = max(candidates, key=lambda item: item["weighted_score"], default={"lag": fps * 0.5, "bpm": 120.0, "score": 0.0}) + beat_seconds = float(best["lag"] / fps) + return float(best["bpm"]), beat_seconds, float(max(0.0, min(1.0, best["score"]))), candidates + + +def estimate_root(weighted_fundamentals: np.ndarray, onset: np.ndarray) -> tuple[int, str, float]: + chroma = np.zeros(12, dtype=np.float32) + for weighted, score in zip(weighted_fundamentals, onset): + if weighted <= 0.0 or score <= 0.0: + continue + hz = float(weighted / max(score, 1e-6)) + if hz < 40.0 or hz > 4000.0: + continue + midi = int(round(69.0 + 12.0 * math.log2(hz / 440.0))) + chroma[midi % 12] += float(score) + if float(np.sum(chroma)) <= 1e-6: + return 57, "A", 0.0 + root_pc = int(np.argmax(chroma)) + root_midi = root_pc + 60 + while root_midi > 69: + root_midi -= 12 + while root_midi < 45: + root_midi += 12 + confidence = float(chroma[root_pc] / max(1e-6, float(np.sum(chroma)))) + return root_midi, NOTE_NAMES[root_pc], max(0.0, min(1.0, confidence)) + + +def midi_to_hz(midi: int) -> float: + return 440.0 * (2.0 ** ((midi - 69) / 12.0)) + + +def render_chirp(start_hz: float, end_hz: float, sample_rate: int, chirp_ms: float) -> np.ndarray: + length = max(16, int(sample_rate * chirp_ms * 0.001)) + t = np.arange(length, dtype=np.float32) / sample_rate + seconds = max(1e-6, chirp_ms * 0.001) + k = (end_hz - start_hz) / seconds + phase = 2.0 * math.pi * (start_hz * t + 0.5 * k * t * t) + chirp = np.sin(phase).astype(np.float32) + chirp *= np.hanning(length).astype(np.float32) + return chirp + + +def write_wav(path: Path, samples: np.ndarray, sample_rate: int) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + pcm = np.clip(samples, -1.0, 1.0) + with wave.open(str(path), "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes((pcm * 32767.0).astype(" dict: + half = bpm * 0.5 + double = bpm * 2.0 + + def nearest(target: float) -> dict[str, float] | None: + if not candidates: + return None + return min(candidates, key=lambda item: abs(item["bpm"] - target)) + + return { + "selected_bpm": bpm, + "half_time": nearest(half), + "double_time": nearest(double), + "strongest": max(candidates, key=lambda item: item["score"], default=None), + } + + +def build_plan(args: argparse.Namespace, moves: list[MoveTarget], bpm: float, beat_seconds: float, tempo_confidence: float, tempo_candidates: list[dict[str, float]], root_midi: int, root_name: str, key_confidence: float) -> dict: + symbols = rotate_to_distinct_opening(build_debruijn(max(2, len(MINOR_PENTATONIC)), 3), 3) + scale = [midi_to_hz(root_midi + interval) for interval in MINOR_PENTATONIC] + events = [] + for index in range(args.events): + symbol = symbols[index % len(symbols)] + note_hz = scale[symbol % len(scale)] + chirp_target_hz = scale[(symbol + 4) % len(scale)] + offset = args.lead_seconds + index * beat_seconds + move_events = [] + for move_index, move in enumerate(moves): + intensity = 1.0 if move_index == symbol % max(1, len(moves)) else 0.55 + color = tuple(int(channel * intensity) for channel in move.color) + move_events.append({"name": move.name, "hidraw": move.hidraw, "rgb": color}) + events.append( + { + "index": index, + "symbol": symbol, + "offset_seconds": offset, + "chirp_start_hz": note_hz, + "chirp_end_hz": chirp_target_hz, + "move_pulses": move_events, + } + ) + return { + "kind": "mimir.music_keyed_move_chirp_plan.v1", + "created_unix": time.time(), + "tempo_bpm": bpm, + "beat_seconds": beat_seconds, + "tempo_confidence": tempo_confidence, + "tempo_grid": args.tempo_grid, + "tempo_family": tempo_family(bpm, tempo_candidates), + "root_note": root_name, + "root_midi": root_midi, + "key_confidence": key_confidence, + "suggested_scale": "minor-pentatonic", + "scale_frequencies_hz": scale, + "moves": [move.__dict__ for move in moves], + "events": events, + } + + +def render_audio(plan: dict, args: argparse.Namespace, path: Path) -> None: + duration = max(event["offset_seconds"] for event in plan["events"]) + 1.0 + audio = np.zeros(max(1, int(duration * args.sample_rate)), dtype=np.float32) + for event in plan["events"]: + chirp = render_chirp(event["chirp_start_hz"], event["chirp_end_hz"], args.sample_rate, args.chirp_ms) + start = int(event["offset_seconds"] * args.sample_rate) + end = min(len(audio), start + len(chirp)) + audio[start:end] += chirp[: end - start] * args.source_gain + write_wav(path, audio, args.sample_rate) + + +def emit(plan: dict, args: argparse.Namespace, moves: list[MoveTarget], wav_path: Path) -> None: + move_specs = ",".join(f"{move.name}={move.hidraw}" for move in moves) + remote_b64 = base64.b64encode(REMOTE_MULTI_MOVE.encode("utf-8")).decode("ascii") + remote_cmd = ( + "tmp=$(mktemp); " + f"printf '%s' '{remote_b64}' | base64 -d > \"$tmp\"; " + f"python3 \"$tmp\" '{move_specs}'; " + "status=$?; rm -f \"$tmp\"; exit $status" + ) + ssh = subprocess.Popen(["ssh", "-o", "BatchMode=yes", args.ssh_target, remote_cmd], stdin=subprocess.PIPE, text=True) + assert ssh.stdin is not None + pulse_seconds = args.pulse_ms * 0.001 + for event in plan["events"]: + for pulse in event["move_pulses"]: + r, g, b = pulse["rgb"] + ssh.stdin.write(f"{event['offset_seconds']:.6f} {pulse['name']} {r} {g} {b}\n") + ssh.stdin.write(f"{event['offset_seconds'] + pulse_seconds:.6f} {pulse['name']} 0 0 0\n") + ssh.stdin.write("go\n") + ssh.stdin.flush() + audio = subprocess.Popen([args.ffplay, "-nodisp", "-autoexit", "-loglevel", "quiet", str(wav_path)]) + audio.wait(timeout=max(event["offset_seconds"] for event in plan["events"]) + 5) + ssh.stdin.close() + ssh.wait(timeout=10) + + +def main() -> int: + args = parse_args() + out_dir = Path(args.out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + moves = parse_moves(args.move) + if args.mode == "synthetic": + _times, onset, weighted_fundamentals = synthetic_analysis(args) + else: + _times, onset, weighted_fundamentals = live_analysis(args) + bpm, beat_seconds, tempo_confidence, autocorr = estimate_tempo(args, onset, args.fps) + root_midi, root_name, key_confidence = estimate_root(weighted_fundamentals, onset) + plan = build_plan(args, moves, bpm, beat_seconds, tempo_confidence, autocorr, root_midi, root_name, key_confidence) + plan["whitened_spectral_autocorr"] = autocorr + plan_path = out_dir / "music-keyed-move-chirp-plan.json" + wav_path = out_dir / "music-keyed-chirps.wav" + plan_path.write_text(json.dumps(plan, indent=2), encoding="utf-8") + render_audio(plan, args, wav_path) + print(f"plan={plan_path}") + print(f"wav={wav_path}") + print( + f"tempo={bpm:.2f}bpm beat={beat_seconds:.3f}s conf={tempo_confidence:.3f} " + f"root={root_name} keyConf={key_confidence:.3f} events={len(plan['events'])}" + ) + if not args.dry_run: + emit(plan, args, moves, wav_path) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())