From b216cbc56f565ece586ccbc7243ac1f4a0f0c903 Mon Sep 17 00:00:00 2001 From: gramps Date: Thu, 12 Nov 2026 13:57:29 -0800 Subject: [PATCH] tools: scripts/avsync.py measures A/V sync from the file, not from an ear MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three numbers, from the recording itself: each stream's length, the drift implied by a length mismatch (a clock problem), and the residual offset between mouth motion and speech (a constant problem). Per-3s window table with correlations, so a weak reading announces itself instead of reporting a confident wrong number, and a median over only the strongly-correlated windows — a whole-clip search on periodic speech locks onto the wrong beat and returns nonsense (it said -400 ms at the search limit on the clip where the windows said -185). Validated on ty-20261112-1228-457-0000.mp4: lengths 13.271/13.305 s (match, so no drift) and -185 ms (sound early), which agrees with the independent visual read of the same clip. The creator is deaf, so "does it sound right" cannot be the loop. This is. Needs numpy; uses the bundled ffmpeg.exe. --- docs/ai/environment.md | 9 ++ scripts/avsync.py | 185 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+) create mode 100644 scripts/avsync.py diff --git a/docs/ai/environment.md b/docs/ai/environment.md index d2fe7dd..3ee0765 100644 --- a/docs/ai/environment.md +++ b/docs/ai/environment.md @@ -139,6 +139,15 @@ save/load round trips are exact: an element removed in the UI (`RemoveElement` close) does **not** come back after reload (`LayoutStorePersistenceTests` guards this). +## A/V sync measurement + +`scripts/avsync.py [--crop WxH+X+Y]` reports what no ear can: each stream's length, +whether they drift (a length mismatch is a clock problem), and the residual offset between mouth +motion and speech (a constant problem), with a per-3s window table and a correlation so a weak +reading says so. It uses the bundled `ffmpeg.exe` and needs `numpy`. The creator is deaf, so this — +not a listening test — is how sync gets judged: feed it the clip instead of guessing, and quote the +numbers in the commit. + ## Session performance evidence `python3 scripts/session-stats.py` mines every `startup*.log` (primary, rotated, and per-instance) diff --git a/scripts/avsync.py b/scripts/avsync.py new file mode 100644 index 0000000..28f63c4 --- /dev/null +++ b/scripts/avsync.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""Measure a recording's A/V sync without listening to it. + +Prints the three numbers that matter, from the file itself: + 1. how long each stream actually is (a length mismatch is a clock problem), + 2. how far apart they drift per second (a rate problem), + 3. the residual offset between mouth motion and speech (a constant problem). + +Usage: avsync.py [--ffmpeg ] [--crop WxH+X+Y] + +Exit code is 0 when the file is measurable, 1 when it is not. +""" +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys + +import numpy as np + +DEFAULT_FFMPEG = r"C:\Users\gramp\AppData\Roaming\ytLlive\tools\ffmpeg.exe" + + +def host_path(path: str) -> str: + """Accept a Windows path under WSL (and a native path everywhere else).""" + if os.name != "nt" and len(path) > 2 and path[1] == ":": + return f"/mnt/{path[0].lower()}/{path[2:].replace(chr(92), '/').lstrip('/')}" + return path + + +def run(ffmpeg: str, args: list[str]) -> bytes: + proc = subprocess.run([host_path(ffmpeg), "-v", "error", *args], capture_output=True) + if proc.returncode != 0: + sys.stderr.write(proc.stderr.decode("utf-8", "replace")) + raise SystemExit(1) + return proc.stdout + + +def probe(ffprobe: str, path: str) -> dict: + out = run(ffprobe, ["-show_entries", + "stream=index,codec_type,duration", + "-show_entries", "format=duration", + "-of", "json", path]) + return json.loads(out) + + +def stream_durations(meta: dict) -> tuple[float, float]: + video = audio = 0.0 + for s in meta.get("streams", []): + d = float(s.get("duration") or 0) + if s.get("codec_type") == "video" and d: + video = max(video, d) + elif s.get("codec_type") == "audio" and d: + audio = max(audio, d) + return video, audio + + +def envelopes(ffmpeg: str, path: str, crop: str | None): + audio_bytes = run(ffmpeg, ["-i", path, "-map", "0:a", "-ac", "1", "-ar", "48000", + "-f", "s16le", "-"]) + audio = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float64) / 32768.0 + + filters = [] + if crop: + # Accept the everyday "WxH+X+Y" geometry and translate it to the filter's w:h:x:y. + spec = crop.replace("x", ":", 1).replace("+", ":") + if ":" not in spec and crop.count(":") == 3: + spec = crop + filters.append(f"crop={spec}") + filters += ["scale=64:64", "format=gray"] + video_bytes = run(ffmpeg, ["-i", path, "-map", "0:v", "-vf", ",".join(filters), + "-f", "rawvideo", "-"]) + frames = np.frombuffer(video_bytes, dtype=np.uint8).reshape(-1, 64, 64).astype(np.float64) + motion = np.abs(np.diff(frames, axis=0)).mean(axis=(1, 2)) + motion = np.concatenate([[motion[0] if len(motion) else 0.0], motion]) + return audio, motion + + +def offset_ms(audio: np.ndarray, motion: np.ndarray, fps: float, max_ms: float = 400.0): + """Cross-correlate mouth motion against speech energy on a 100 Hz grid. + + Returns (lag_ms, correlation) where a POSITIVE lag means the sound arrives later than the + mouth movement (audio delayed / video ahead). + """ + grid = 100.0 + duration = min(len(audio) / 48000.0, len(motion) / fps) + t = np.arange(0, duration, 1.0 / grid) + + def resample(x: np.ndarray, rate: float) -> np.ndarray: + return x[np.clip((t * rate).astype(int), 0, len(x) - 1)] + + win = max(1, int(48000 / grid)) + n = len(audio) // win + audio_rms = np.sqrt((audio[: n * win].reshape(n, win) ** 2).mean(axis=1)) + audio_env = np.interp(t, np.arange(n) / grid, audio_rms) + + def smooth(x: np.ndarray) -> np.ndarray: + k = max(3, int(grid * 0.05) | 1) + return np.convolve(x, np.ones(k) / k, mode="same") + + def z(x: np.ndarray) -> np.ndarray: + s = x.std() + return (x - x.mean()) / (s if s > 1e-9 else 1.0) + + m = z(smooth(resample(motion, fps))) + a = z(smooth(audio_env)) + + limit = int(grid * max_ms / 1000) + + def best_for(m_seg: np.ndarray, a_seg: np.ndarray) -> tuple[float, float]: + best_corr, best_lag = -9.0, 0 + for k in range(-limit, limit + 1): + a_part = a_seg[max(0, k): len(a_seg) + min(0, k)] + m_part = m_seg[max(0, -k): len(m_seg) - max(0, k)] + corr = float((m_part * a_part).mean()) + if corr > best_corr: + best_corr, best_lag = corr, k + return best_lag / grid * 1000.0, best_corr + + # Whole-clip first; then in 3 s windows, because a periodic voice (a spoken alphabet, a + # song) can lock the global search onto the wrong beat. Only windows that correlate + # strongly are trusted, and their median is the answer. + overall_lag, overall_corr = best_for(m, a) + seg = int(3 * grid) + windows = [] + for start in range(0, max(1, len(m) - seg), seg): + lag, corr = best_for(m[start:start + seg], a[start:start + seg]) + windows.append((start / grid, lag, corr)) + strong = sorted(lag for _, lag, corr in windows if corr >= 0.35) + if strong: + mid = len(strong) // 2 + robust = strong[mid] if len(strong) % 2 else (strong[mid - 1] + strong[mid]) / 2 + return robust, max(corr for _, _, corr in windows), windows + return overall_lag, overall_corr, windows + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("recording") + ap.add_argument("--ffmpeg", default=os.environ.get("YTLLIVE_FFMPEG", DEFAULT_FFMPEG)) + ap.add_argument("--crop", default=None, + help="webcam region, e.g. 764x764+1032+159 (defaults to the whole frame)") + ap.add_argument("--max-ms", type=float, default=400.0) + args = ap.parse_args() + + ffmpeg = args.ffmpeg + ffprobe = ffmpeg.replace("ffmpeg.exe", "ffprobe.exe") + if not os.path.exists(host_path(ffmpeg)): + sys.stderr.write(f"ffmpeg not found: {ffmpeg}\n") + return 1 + + video, audio = stream_durations(probe(ffprobe, args.recording)) + if not video or not audio: + sys.stderr.write("could not read both stream durations\n") + return 1 + + diff = audio - video + pct = diff / video * 100.0 + + samples, motion = envelopes(ffmpeg, args.recording, args.crop) + fps = len(motion) / (len(samples) / 48000.0) if len(motion) else 0.0 + lag, corr, windows = offset_ms(samples, motion, fps, args.max_ms) if fps > 0 else (0.0, 0.0, []) + + print(f"video {video:8.3f} s") + print(f"audio {audio:8.3f} s") + print(f"delta {diff:+8.3f} s ({pct:+.2f}% of the video length)") + verdict = "lengths match" if abs(diff) < 0.05 else "LENGTH MISMATCH — a clock problem" + print(f" {verdict}") + quality = "strong" if corr >= 0.35 else "weak — treat the offset as a hint only" + for start, wlag, wcorr in windows: + mark = "*" if wcorr >= 0.35 else " " + print(f" {start:5.1f}s {wlag:+7.0f} ms corr {wcorr:.2f} {mark}") + print(f"offset {lag:+8.0f} ms (correlation {corr:.2f}, {quality})") + direction = "sound LATER than the lips" if lag > 0 else "sound EARLIER than the lips" + if abs(lag) >= 20: + print(f" {direction} — adjust the mic delay by {-lag:+.0f} ms") + else: + print(" within a frame or two — leave it alone") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())