diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 6e3197c..3ce0f4f 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -31,6 +31,7 @@ - `docs/typed-visual-state.md` documents the CultCache/CultNet visual boundary. - OBS synchronized program surface: - `scripts/setup_obs_synced_program.py` derives OBS-controllable stems from an aligned program audio timeline: host voice, co-streamer voice, ambient, transients, co-streamer loopback, and local loopback. + - `scripts/capture_co_streamer_surfaces.py` captures neighbor Focusrite and neighbor loopback with local loopback ground truth, estimates the late remote-family offset, and writes aligned co-streamer surfaces for the stem packer. - The tool creates/updates local OBS Media Sources for those stems and mutes/disables raw unsynchronized inputs. - Strict mode disables every scene item except the synchronized LocalCastBridge program video and stem controls. @@ -55,4 +56,5 @@ 10. Replace placeholder mic/speaker geometry with measured world coordinates, then build the delay/SRO alignment stage that feeds the bounded field cache and emits aligned six-channel blocks before FOA encoding. 11. Move the render-frame consumer into Aquarium Engine so dense brush/splat rendering replaces the deadline OpenGL point sink behind the same OBS Spout boundary. 12. Replace synthetic live-fusion observations with PS3 Eye detector observations. -13. Replace placeholder silent co-streamer voice/loopback stems with synchronized neighbor Focusrite and loopback captures once those feeds are in the aligned timeline. +13. Fix the neighbor loopback route so `Voicemeeter Out B3` carries real program audio; current remote loopback captures are near-silent and cannot serve as the timing witness. +14. Let the co-streamer surface delay drive the shared presentation buffer horizon for audio stems, AmbiX, and remote video. diff --git a/notes/current-system-map.md b/notes/current-system-map.md index ef12df4..b23c44e 100644 --- a/notes/current-system-map.md +++ b/notes/current-system-map.md @@ -121,7 +121,8 @@ flowchart TD Ownership: - `scripts/setup_obs_synced_program.py` owns packaging aligned program audio into OBS-controllable stems and muting/disabling raw unsynchronized OBS inputs. +- `scripts/capture_co_streamer_surfaces.py` owns the late-arriving neighbor surface import: it captures neighbor Focusrite and neighbor loopback over SSH while recording local loopback, estimates the remote-family offset, and writes aligned co-streamer surfaces for the OBS stem packer. - OBS owns final source volume, track assignment, filters, and stream/record output. - Aquarium/Spout owns synchronized program video, not raw remote desktop capture. -Invariant: raw SRT ingest, Desktop Audio, Mic/Aux, and other non-program scene items may exist for diagnostics, but they must be disabled or muted when broadcasting the synchronized program. Missing synchronized feeds should appear as explicit silent placeholders, not as live unsynchronized substitutes. +Invariant: raw SRT ingest, Desktop Audio, Mic/Aux, and other non-program scene items may exist for diagnostics, but they must be disabled or muted when broadcasting the synchronized program. Missing synchronized feeds should appear as explicit silent placeholders, not as live unsynchronized substitutes. The co-streamer Focusrite, co-streamer loopback, and remote video are the latest-arriving surface family; their measured delay should set the presentation buffer horizon instead of being patched in after the local field. diff --git a/scripts/capture_co_streamer_surfaces.py b/scripts/capture_co_streamer_surfaces.py new file mode 100644 index 0000000..4e8e505 --- /dev/null +++ b/scripts/capture_co_streamer_surfaces.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +import argparse +import json +import math +import subprocess +import time +from datetime import datetime, timezone +from pathlib import Path +from tempfile import NamedTemporaryFile + +import numpy as np +from scipy import signal +from scipy.io import wavfile + + +DEFAULT_RUN = Path("calibration/runs/audio-program-live-20260518-180226") + + +def main() -> int: + parser = argparse.ArgumentParser( + description=( + "Capture the late-arriving co-streamer audio surfaces, align them to " + "local loopback ground truth, and write synced OBS stem inputs." + ) + ) + parser.add_argument("--run", type=Path, default=DEFAULT_RUN) + parser.add_argument("--seconds", type=float, default=24.0) + parser.add_argument("--sample-rate", type=int, default=48000) + parser.add_argument("--ssh-target", default="madman's lullaby@192.168.1.84") + parser.add_argument("--ffmpeg", default=r"C:\Users\Madman's Lullaby\AppData\Local\Microsoft\WinGet\Links\ffmpeg.exe") + parser.add_argument("--remote-dir", default=r"C:\Meta\LocalCastBridge\calibration\remote-captures") + parser.add_argument("--focusrite-device", default="Analogue 1 + 2 (Focusrite USB Audio)") + parser.add_argument("--loopback-device", default="Voicemeeter Out B3 (VB-Audio Voicemeeter VAIO)") + parser.add_argument("--local-loopback-query", default="Scarlett") + parser.add_argument("--max-lag-ms", type=float, default=3000.0) + parser.add_argument("--min-loopback-rms", type=float, default=1e-4) + parser.add_argument("--min-loopback-score", type=float, default=0.01) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + + run = args.run.resolve() + surface_dir = run / "co_streamer_surfaces" + surface_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S") + + remote_focusrite = f"{args.remote_dir.rstrip('\\\\/')}\\co_streamer_focusrite-{stamp}.wav" + remote_loopback = f"{args.remote_dir.rstrip('\\\\/')}\\co_streamer_loopback-{stamp}.wav" + local_focusrite = surface_dir / "raw_focusrite.wav" + local_remote_loopback = surface_dir / "raw_loopback.wav" + local_reference = surface_dir / "local_loopback_reference.wav" + + commands = [ + remote_ffmpeg_command(args, args.focusrite_device, remote_focusrite, channels=1), + remote_ffmpeg_command(args, args.loopback_device, remote_loopback, channels=2), + ] + if args.dry_run: + for command in commands: + print(command) + return 0 + + print("starting remote co-streamer captures") + procs = [subprocess.Popen(["ssh", args.ssh_target, command], text=True) for command in commands] + time.sleep(0.2) + local_loopback = record_local_loopback( + query=args.local_loopback_query, + seconds=float(args.seconds), + sample_rate=int(args.sample_rate), + channels=2, + ) + wavfile.write(local_reference, int(args.sample_rate), local_loopback.astype(np.float32)) + + failures = [] + for proc in procs: + rc = proc.wait() + if rc: + failures.append(rc) + if failures: + raise SystemExit(f"remote capture failed with exit codes {failures}") + + sftp_get(args.ssh_target, remote_focusrite, local_focusrite) + sftp_get(args.ssh_target, remote_loopback, local_remote_loopback) + + sample_rate = int(args.sample_rate) + _, reference = read_wav_float(local_reference, sample_rate) + _, focusrite = read_wav_float(local_focusrite, sample_rate) + _, remote_loop = read_wav_float(local_remote_loopback, sample_rate) + reference_mono = mono(reference) + remote_loop_mono = mono(remote_loop) + max_lag = int(float(args.max_lag_ms) * sample_rate / 1000.0) + remote_loop_rms = rms(remote_loop_mono) + focusrite_mono = mono(focusrite) + focusrite_rms = rms(focusrite_mono) + loopback_delay, loopback_score = normalized_delay(remote_loop_mono, reference_mono, max_lag=max_lag) + focusrite_delay, focusrite_score = normalized_delay(focusrite_mono, reference_mono, max_lag=max_lag) + if remote_loop_rms >= float(args.min_loopback_rms) and loopback_score >= float(args.min_loopback_score): + delay_samples = loopback_delay + score = loopback_score + alignment_witness = "remote-loopback" + else: + delay_samples = focusrite_delay + score = focusrite_score + alignment_witness = "neighbor-focusrite-program-bleed" + + aligned_focusrite = shift_by_delay(focusrite, delay_samples, len(reference)) + aligned_loopback = shift_by_delay(remote_loop, delay_samples, len(reference)) + wavfile.write(surface_dir / "aligned_focusrite.wav", sample_rate, aligned_focusrite.astype(np.float32)) + wavfile.write(surface_dir / "aligned_loopback.wav", sample_rate, aligned_loopback.astype(np.float32)) + + report = { + "schema_version": "gamecult.localcast.co_streamer_surfaces.v1", + "created_utc": datetime.now(timezone.utc).isoformat(), + "sample_rate": sample_rate, + "seconds": float(args.seconds), + "alignment_witness": alignment_witness, + "delay_samples": int(delay_samples), + "delay_ms": 1000.0 * float(delay_samples) / sample_rate, + "correlation_score": float(score), + "remote_loopback": { + "rms": remote_loop_rms, + "delay_samples": int(loopback_delay), + "delay_ms": 1000.0 * float(loopback_delay) / sample_rate, + "correlation_score": float(loopback_score), + }, + "neighbor_focusrite": { + "rms": focusrite_rms, + "delay_samples": int(focusrite_delay), + "delay_ms": 1000.0 * float(focusrite_delay) / sample_rate, + "correlation_score": float(focusrite_score), + }, + "interpretation": ( + "positive delay means the remote loopback arrived later than local loopback " + "and was advanced into the local presentation timeline" + ), + "files": { + "local_loopback_reference": str(local_reference), + "raw_focusrite": str(local_focusrite), + "raw_loopback": str(local_remote_loopback), + "aligned_focusrite": str(surface_dir / "aligned_focusrite.wav"), + "aligned_loopback": str(surface_dir / "aligned_loopback.wav"), + }, + } + (surface_dir / "alignment-report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + print(json.dumps(report, indent=2)) + return 0 + + +def remote_ffmpeg_command(args: argparse.Namespace, device: str, output: str, *, channels: int) -> str: + remote_dir = args.remote_dir.rstrip("\\/") + return ( + f'cmd /c if not exist "{remote_dir}" mkdir "{remote_dir}" && ' + f'"{args.ffmpeg}" -y -hide_banner -nostdin -loglevel warning ' + f'-f dshow -t {float(args.seconds):.3f} -i audio="{device}" ' + f'-vn -ac {channels} -ar {int(args.sample_rate)} -c:a pcm_f32le "{output}"' + ) + + +def sftp_get(ssh_target: str, remote_file: str, local_file: Path) -> None: + remote_sftp_path = "/C:/" + remote_file.replace("\\", "/").split("C:/", 1)[-1] + batch = f'get "{remote_sftp_path}" "{local_file}"\n' + with NamedTemporaryFile("w", encoding="ascii", delete=False, suffix=".sftp") as fh: + fh.write(batch) + batch_path = fh.name + try: + completed = subprocess.run(["sftp", "-b", batch_path, ssh_target], text=True, capture_output=True) + if completed.returncode: + raise SystemExit(completed.stderr or completed.stdout) + finally: + Path(batch_path).unlink(missing_ok=True) + + +def record_local_loopback(query: str, seconds: float, sample_rate: int, channels: int) -> np.ndarray: + try: + import soundcard as sc + except Exception as exc: + raise SystemExit(f"soundcard loopback capture unavailable: {exc!r}") from exc + matches = [ + mic + for mic in sc.all_microphones(include_loopback=True) + if "loopback" in repr(mic).lower() and query.lower() in mic.name.lower() + ] + if not matches: + raise SystemExit(f"No loopback device matching {query!r}") + with matches[0].recorder(samplerate=sample_rate, channels=channels) as recorder: + return recorder.record(numframes=int(seconds * sample_rate)).astype(np.float32) + + +def read_wav_float(path: Path, target_rate: int) -> tuple[int, np.ndarray]: + rate, samples = wavfile.read(path) + samples = pcm_to_float(samples) + if int(rate) != int(target_rate): + gcd = math.gcd(int(rate), int(target_rate)) + samples = signal.resample_poly(samples, int(target_rate) // gcd, int(rate) // gcd, axis=0).astype(np.float32) + rate = target_rate + return int(rate), ensure_2d(samples) + + +def pcm_to_float(samples: np.ndarray) -> np.ndarray: + array = np.asarray(samples) + if array.ndim == 1: + array = array[:, None] + if np.issubdtype(array.dtype, np.floating): + return array.astype(np.float32, copy=False) + if np.issubdtype(array.dtype, np.signedinteger): + scale = float(max(abs(np.iinfo(array.dtype).min), np.iinfo(array.dtype).max)) + return (array.astype(np.float32) / scale).clip(-1.0, 1.0) + raise TypeError(f"unsupported WAV dtype: {array.dtype}") + + +def ensure_2d(samples: np.ndarray) -> np.ndarray: + return samples[:, None] if samples.ndim == 1 else samples + + +def mono(samples: np.ndarray) -> np.ndarray: + samples = ensure_2d(samples) + return np.mean(samples, axis=1, dtype=np.float32) + + +def normalized_delay(signal_a: np.ndarray, signal_b: np.ndarray, max_lag: int) -> tuple[int, float]: + count = min(len(signal_a), len(signal_b)) + a = signal_a[:count].astype(np.float32) + b = signal_b[:count].astype(np.float32) + a = (a - float(np.mean(a))) / (float(np.std(a)) + 1e-9) + b = (b - float(np.mean(b))) / (float(np.std(b)) + 1e-9) + corr = signal.correlate(a, b, mode="full", method="fft") + mid = len(b) - 1 + segment = corr[mid - max_lag : mid + max_lag + 1] + index = int(np.argmax(np.abs(segment))) + lag = index - max_lag + score = float(abs(segment[index]) / max(1, len(a))) + return lag, score + + +def rms(samples: np.ndarray) -> float: + samples = np.asarray(samples, dtype=np.float32) + return float(np.sqrt(np.mean(samples * samples))) if samples.size else 0.0 + + +def shift_by_delay(samples: np.ndarray, delay_samples: int, frame_count: int) -> np.ndarray: + samples = ensure_2d(samples) + out = np.zeros((frame_count, samples.shape[1]), dtype=np.float32) + if delay_samples >= 0: + src_start = delay_samples + dst_start = 0 + else: + src_start = 0 + dst_start = -delay_samples + count = min(frame_count - dst_start, samples.shape[0] - src_start) + if count > 0: + out[dst_start : dst_start + count] = samples[src_start : src_start + count] + return out + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/setup_obs_synced_program.py b/scripts/setup_obs_synced_program.py index a2c1ce8..9284b77 100644 --- a/scripts/setup_obs_synced_program.py +++ b/scripts/setup_obs_synced_program.py @@ -75,11 +75,25 @@ def build_stems(run: Path, stem_dir: Path) -> dict[str, Any]: witness_channels = [channel for channel in (0, 2, 3) if channel < field.shape[1]] witness_mix = mix_channels(field, witness_channels) if witness_channels else np.zeros(frame_count, dtype=np.float32) host_voice = channel_or_silence(field, 4) - co_streamer_voice = channel_or_silence(field, 5) + co_streamer_voice, co_streamer_voice_status = load_co_streamer_surface( + run / "co_streamer_surfaces" / "aligned_focusrite.wav", + field_rate, + frame_count, + channel_fallback=channel_or_silence(field, 5), + placeholder_status="placeholder-silence-until-neighbor-focusrite-is-in-aligned-timeline", + synced_status="synced-from-neighbor-focusrite-surface", + ) ambient = make_ambient_stem(witness_mix, host_voice, co_streamer_voice) transients = make_transient_stem(witness_mix, events_doc, frame_count) local_loopback = load_optional_program_audio(run / "ground_truth_loopback.wav", field_rate, frame_count) - co_streamer_loopback = np.zeros((frame_count, 2), dtype=np.float32) + co_streamer_loopback, co_streamer_loopback_status = load_co_streamer_surface( + run / "co_streamer_surfaces" / "aligned_loopback.wav", + field_rate, + frame_count, + channel_fallback=np.zeros((frame_count, 2), dtype=np.float32), + placeholder_status="placeholder-silence-until-neighbor-loopback-is-captured-and-aligned", + synced_status="synced-from-neighbor-loopback-surface", + ) stems = [ stem_spec( @@ -97,7 +111,7 @@ def build_stems(run: Path, stem_dir: Path) -> dict[str, Any]: co_streamer_voice, field_rate, role="dialogue", - status="placeholder-silence-until-neighbor-focusrite-is-in-aligned-timeline", + status=co_streamer_voice_status, position_m=[0.0, 2.0, 1.2], ), stem_spec( @@ -124,7 +138,7 @@ def build_stems(run: Path, stem_dir: Path) -> dict[str, Any]: co_streamer_loopback, field_rate, role="program-loopback", - status="placeholder-silence-until-neighbor-loopback-is-captured-and-aligned", + status=co_streamer_loopback_status, position_m=None, ), stem_spec( @@ -320,6 +334,23 @@ def load_optional_program_audio(path: Path, sample_rate: int, frame_count: int) return samples[:frame_count] +def load_co_streamer_surface( + path: Path, + sample_rate: int, + frame_count: int, + *, + channel_fallback: np.ndarray, + placeholder_status: str, + synced_status: str, +) -> tuple[np.ndarray, str]: + if not path.exists(): + return channel_fallback, placeholder_status + loaded = load_optional_program_audio(path, sample_rate, frame_count) + if np.asarray(channel_fallback).ndim == 1 and loaded.ndim == 2: + loaded = np.mean(loaded, axis=1, dtype=np.float32) + return loaded, synced_status + + def limit_peak(samples: np.ndarray, target: float = 0.98) -> np.ndarray: peak = float(np.max(np.abs(samples))) if samples.size else 0.0 if peak <= target or peak <= 1e-9: diff --git a/state/map.yaml b/state/map.yaml index 2363423..1177dea 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -209,9 +209,11 @@ current_status: per-channel stems from the aligned program audio timeline, creates/updates OBS Media Sources for host voice, co-streamer voice, ambient, transients, co-streamer loopback, and local loopback, and can disable every scene item - except synchronized LocalCastBridge program surfaces. Missing synchronized - neighbor feeds are explicit silent placeholders, not raw unsynchronized - substitutes. The live producer currently uses synthetic calibrated + except synchronized LocalCastBridge program surfaces. Co-streamer surface + capture now records neighbor Focusrite and neighbor loopback while recording + local loopback ground truth, then writes aligned co-streamer surface WAVs + for the stem packer. Missing synchronized neighbor feeds are explicit + silent placeholders, not raw unsynchronized substitutes. The live producer currently uses synthetic calibrated observations through the real `SensorRig.fuse` path until camera detector observations replace them. Unit tests inject synthetic cameras and frames so the core does not depend on live camera drivers. @@ -232,6 +234,6 @@ current_status: shotgun capture/transport path, replace placeholder mic/speaker geometry with measured world coordinates, then build the delay/SRO alignment stage that feeds the bounded field cache and emits aligned six-channel blocks for - FOA encoding. Replace the current silent co-streamer voice and co-streamer - loopback OBS stems only after those neighbor feeds are captured into the - aligned timeline. + FOA encoding. Fix the neighbor loopback routing so the remote loopback is a + real timing witness, then let the co-streamer surface family set the shared + presentation buffer horizon for audio stems, AmbiX, and remote video.