gramps
1abc86358a
Mouth-motion against speech is a statistical proxy and it fails on ordinary talking clips: the per-window estimates swing by 700 ms and the honest answer is "unmeasurable". A clap is not a proxy — it has one unmistakable audio spike and one unmistakable visual spike, a fixed distance apart, with no periodicity to lock onto wrongly. So when the correlation is weak, the tool now detects sharp audio transients (99th percentile, one event per 0.6 s), finds the biggest visual change within ±400 ms of each, and reports the median of the cluster within 60 ms of it — at least three pairs or it still says unmeasurable. The cluster matters: an arm entering the frame moves more than the clap itself and lands early (two pairs on the validation clip came out at +248 and +283 ms against -285). Validated on ty-20261112-1405-621-0000.mp4: the correlation path said +98 ms and the clap path +116 ms (three tight pairs at +117, +115, +107), so the applied 350 ms should come down to about 240. Two methods, same answer, no listening required.
253 lines
10 KiB
Python
253 lines
10 KiB
Python
#!/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 <recording.mp4> [--ffmpeg <path>] [--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 len(strong) >= 2:
|
|
mid = len(strong) // 2
|
|
robust = strong[mid] if len(strong) % 2 else (strong[mid - 1] + strong[mid]) / 2
|
|
spread = strong[-1] - strong[0]
|
|
if spread > 150:
|
|
# Windows that disagree this much cannot be averaged into an answer. Say so.
|
|
return float("nan"), max(corr for _, _, corr in windows), windows
|
|
return robust, max(corr for _, _, corr in windows), windows
|
|
if strong:
|
|
return float("nan"), max(corr for _, _, corr in windows), windows
|
|
return overall_lag, overall_corr, windows
|
|
|
|
|
|
def clap_offset(audio: np.ndarray, motion: np.ndarray, fps: float) -> tuple[float, int]:
|
|
"""Pair sharp audio transients with the nearest big visual change and take the median.
|
|
|
|
A clap is the only event in an ordinary recording with an unmistakable audio spike and an
|
|
unmistakable visual spike a fixed distance apart, which is why it beats correlating a mouth
|
|
against speech: no periodicity to lock onto wrongly. Returns (median_ms, pair_count) where a
|
|
POSITIVE value means the sound arrived later than the picture.
|
|
"""
|
|
if fps <= 0 or len(motion) < 4:
|
|
return float("nan"), 0
|
|
|
|
win = 240 # 5 ms amplitude windows
|
|
n = len(audio) // win
|
|
if n < 8:
|
|
return float("nan"), 0
|
|
amp = np.abs(audio[: n * win].reshape(n, win)).max(axis=1)
|
|
times = np.arange(n) * win / 48000.0
|
|
|
|
threshold = np.percentile(amp, 99.0)
|
|
events: list[float] = []
|
|
last = -9.0
|
|
for t in times[amp >= threshold]:
|
|
if t - last > 0.6: # one event per clap, not one per sample of its decay
|
|
events.append(float(t))
|
|
last = t
|
|
|
|
deltas = []
|
|
for t in events:
|
|
lo, hi = int((t - 0.4) * fps), int((t + 0.4) * fps)
|
|
if lo < 0 or hi >= len(motion):
|
|
continue
|
|
peak = (lo + int(np.argmax(motion[lo:hi]))) / fps
|
|
deltas.append((t - peak) * 1000.0)
|
|
|
|
if len(deltas) < 3:
|
|
return float("nan"), len(deltas)
|
|
|
|
# Keep the cluster: an arm entering the frame moves more than the clap itself and lands early.
|
|
deltas.sort()
|
|
mid = len(deltas) // 2
|
|
median = deltas[mid] if len(deltas) % 2 else (deltas[mid - 1] + deltas[mid]) / 2
|
|
tight = [d for d in deltas if abs(d - median) <= 60]
|
|
if len(tight) < 3:
|
|
return float("nan"), len(tight)
|
|
return float(np.median(tight)), len(tight)
|
|
|
|
|
|
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}")
|
|
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}")
|
|
|
|
if lag != lag or corr < 0.35:
|
|
clap, pairs = clap_offset(samples, motion, fps)
|
|
if clap == clap:
|
|
print(f"offset {clap:+8.0f} ms (clap pairs: {pairs})")
|
|
direction = "sound LATER than the lips" if clap > 0 else "sound EARLIER than the lips"
|
|
if abs(clap) >= 20:
|
|
print(f" {direction} — adjust the mic delay by {-clap:+.0f} ms")
|
|
else:
|
|
print(" within a frame or two — leave it alone")
|
|
return 0
|
|
print("offset unmeasurable — too little agreement between mouth motion and speech")
|
|
print(f" ({pairs} usable clap pairs) record 3 claps, hands in the webcam frame, 2 s apart")
|
|
return 0
|
|
|
|
print(f"offset {lag:+8.0f} ms (correlation {corr:.2f})")
|
|
if abs(lag) >= 20:
|
|
direction = "sound LATER than the lips" if lag > 0 else "sound EARLIER than the lips"
|
|
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())
|