GameCult/Mimir
Watch
9
0
Fork
You've already forked Mimir
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add music-keyed Move chirp planner

Browse source
This commit is contained in:
Meta Vens 2026-05-31 21:59:06 +01:00
commit d716fbaee2
3 changed files with 515 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

55
docs/move-controller-stream.md
View file

@ -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.

2
state/evidence.jsonl
View file

@ -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"}

458
tools/music_keyed_move_chirp_sync.py Normal file
View file

@ -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("<i2").tobytes())
def tempo_family(bpm: float, candidates: list[dict[str, float]]) -> 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())