TTS isvc: check_timing.py
ActivePublic
Actions

Authored by kevinbazira on Wed, Sep 16, 12:30 PM.

Project Tags

Referenced Files

	F103274134: TTS isvc: check_timing.py
	Wed, Sep 16, 12:30 PM

Subscribers

None

	"""
	Verify caption timings against the audio they describe.

	Two independent checks on a (mp3, vtt) pair:

	1. STRUCTURAL (instant, no model): cues monotonic, non-overlapping, word
	count sane, and the last cue's end close to the audio's real duration.
	Catches gross breakage but NOT drift: proportional estimates also end
	where the audio ends, which is exactly why they look fine at both ends
	of a file and wrong in the middle.

	2. DRIFT (needs faster-whisper): transcribe the audio with word-level
	timings, match the recognizer's words to the caption's words, and look
	at how the offset behaves ACROSS the file. Forced alignment gives a
	flat offset (jitter only, no trend). Proportional estimation gives a
	trend: the offset grows steadily, which is the "starts in sync, falls
	behind by the end" symptom. The slope is the number that separates
	them; the recognizer's own jitter is noise with no direction.

	Usage (inside the ASR eval image, which already has faster-whisper):

	python3 check_timing.py --mp3 lead.mp3 --vtt lead.vtt
	python3 check_timing.py --mp3 lead.mp3 --vtt lead.vtt --structural-only
	"""

	import argparse
	import difflib
	import re
	import statistics
	import sys
	from pathlib import Path


	def parse_vtt(text: str) -> list[dict]:
	"""Cue list from WebVTT: [{word, start_ms, end_ms}]."""
	cues: list[dict] = []
	lines = [ln.rstrip("\n") for ln in text.splitlines()]
	for i, line in enumerate(lines):
	if "-->" not in line:
	continue
	left, right = (p.strip() for p in line.split("-->")[:2])
	payload = ""
	for nxt in lines[i + 1 :]:
	if nxt.strip():
	payload = nxt.strip()
	break
	break
	cues.append(
	{
	"word": re.sub(r"<[^>]+>", "", payload),
	"start_ms": _vtt_to_ms(left),
	"end_ms": _vtt_to_ms(right.split()[0] if right else right),
	}
	)
	return cues


	def _vtt_to_ms(stamp: str) -> float:
	parts = stamp.split(":")
	if len(parts) == 3:
	h, m, s = parts
	elif len(parts) == 2:
	h, (m, s) = 0, parts
	else:
	raise ValueError(f"bad timestamp: {stamp!r}")
	return (int(h) * 3600 + int(m) * 60 + float(s)) * 1000.0


	def structural_checks(cues: list[dict], audio_ms: float \| None) -> int:
	"""Return the number of problems found (0 = clean)."""
	problems = 0
	print(f"cues: {len(cues)}")
	if not cues:
	print(" FAIL: no cues parsed")
	return 1

	non_monotonic = [
	(i, cues[i - 1], cues[i])
	for i in range(1, len(cues))
	if cues[i]["start_ms"] < cues[i - 1]["start_ms"]
	]
	overlaps = [
	(i, cues[i - 1], cues[i])
	for i in range(1, len(cues))
	if cues[i]["start_ms"] < cues[i - 1]["end_ms"] - 1.0
	]
	bad_spans = [c for c in cues if c["end_ms"] <= c["start_ms"]]
	empty = [c for c in cues if not c["word"]]

	for label, items in (
	("cues out of order", non_monotonic),
	("overlapping cues", overlaps),
	("zero/negative duration cues", bad_spans),
	("empty cue text", empty),
	):
	if items:
	problems += 1
	print(f" FAIL: {label}: {len(items)}")
	ex = items[0]
	print(f" first example: {ex}")
	else:
	print(f" ok: no {label}")

	span = cues[-1]["end_ms"] - cues[0]["start_ms"]
	durs = [c["end_ms"] - c["start_ms"] for c in cues]
	print(
	f" span {span / 1000:.1f}s; per-word median "
	f"{statistics.median(durs):.0f}ms, max {max(durs):.0f}ms"
	)
	if audio_ms:
	tail = audio_ms - cues[-1]["end_ms"]
	print(f" audio duration {audio_ms / 1000:.1f}s; trailing gap {tail / 1000:.2f}s")
	if tail < -0.5 * 1000:
	problems += 1
	print(" FAIL: captions run past the end of the audio")
	elif tail > 5 * 1000:
	problems += 1
	print(" FAIL: captions end more than 5s before the audio does")
	return problems


	_WORD = re.compile(r"[a-z']+")


	def _norm_words(items: list[str]) -> list[str]:
	out = []
	for w in items:
	m = _WORD.findall(w.lower())
	out.append(m[0] if m else "")
	return out


	def drift_check(mp3: Path, cues: list[dict], model_name: str) -> int:
	from faster_whisper import WhisperModel

	print(f"\ntranscribing with {model_name} (word timings)...")
	model = WhisperModel(model_name, device="cpu", compute_type="int8")
	segments, _ = model.transcribe(
	str(mp3), language="en", beam_size=5, word_timestamps=True, vad_filter=False
	)
	hyp = []
	for seg in segments:
	for w in seg.words or []:
	hyp.append({"word": w.word.strip(), "mid_ms": (w.start + w.end) * 500.0})
	print(f"recognizer words: {len(hyp)}; caption words: {len(cues)}")
	if len(hyp) < 20:
	print(" SKIP: too few recognized words to measure")
	return 0

	ref_w = _norm_words([c["word"] for c in cues])
	hyp_w = _norm_words([h["word"] for h in hyp])
	# Match the two word sequences, then compare timings of matched pairs.
	sm = difflib.SequenceMatcher(a=ref_w, b=hyp_w, autojunk=False)
	pairs = []
	for a, b, size in sm.get_matching_blocks():
	for k in range(size):
	c = cues[a + k]
	h = hyp[b + k]
	ref_mid = (c["start_ms"] + c["end_ms"]) / 2.0
	pairs.append((ref_mid, h["mid_ms"] - ref_mid))
	if len(pairs) < 20:
	print(f" SKIP: only {len(pairs)} matched words")
	return 0
	matched_pct = 100.0 * len(pairs) / len(cues)
	offs = [d for _, d in pairs]
	print(f"matched {len(pairs)} words ({matched_pct:.0f}% of cues)")
	print(
	f"offset (recognizer minus caption): median {statistics.median(offs):+.0f}ms, "
	f"stdev {statistics.pstdev(offs):.0f}ms"
	)

	# Thirds: the shape of the problem, in the terms it was reported.
	pairs.sort()
	n = len(pairs)
	for label, chunk in (
	("first third", pairs[: n // 3]),
	("middle third", pairs[n // 3 : 2 * n // 3]),
	("last third", pairs[2 * n // 3 :]),
	):
	med = statistics.median([d for _, d in chunk])
	print(f" {label:13} median offset {med:+.0f}ms")

	# Per-30s buckets: a LOCAL misalignment (alignment degrading over one
	# stretch, then recovering) does not show up as slope, but a listener
	# hears it as badly as drift. Localising it also says which part of the
	# audio to inspect.
	print("\n30-second buckets (median offset, words matched):")
	worst_bucket, worst_val = None, 0.0
	bucket = 30000.0
	span = max(t for t, _ in pairs)
	for b in range(int(span // bucket) + 1):
	lo, hi = b * bucket, (b + 1) * bucket
	chunk = [d for t, d in pairs if lo <= t < hi]
	if len(chunk) < 5:
	continue
	med = statistics.median(chunk)
	flag = " <-- off by more than 400ms" if abs(med) > 400 else ""
	print(f" {lo / 1000:5.0f}-{hi / 1000:<5.0f}s {med:+6.0f}ms "
	f"({len(chunk):3d} words){flag}")
	if abs(med) > abs(worst_val):
	worst_bucket, worst_val = (lo / 1000, hi / 1000), med

	# Least-squares slope: ms of drift per minute of audio.
	xs = [t / 60000.0 for t, _ in pairs] # minutes
	ys = offs
	mx, my = statistics.mean(xs), statistics.mean(ys)
	denom = sum((x - mx) ** 2 for x in xs)
	slope = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / denom if denom else 0.0
	print(f"\ndrift slope: {slope:+.0f} ms per minute of audio")
	if worst_bucket and abs(worst_val) > 400:
	print(
	f"VERDICT: LOCAL misalignment: captions are {abs(worst_val):.0f}ms "
	f"{'behind' if worst_val < 0 else 'ahead of'} the audio around "
	f"{worst_bucket[0]:.0f}-{worst_bucket[1]:.0f}s. Not a steady drift "
	"(slope is small), but audible as being out of sync. Listen there."
	)
	return 1
	if abs(slope) < 150:
	print("VERDICT: no systematic drift and no local misalignment.")
	return 0
	if abs(slope) < 400:
	print("VERDICT: mild drift; worth a listen, may be recognizer jitter.")
	return 0
	print("VERDICT: systematic drift present (the proportional-estimate signature).")
	return 1


	def main() -> int:
	ap = argparse.ArgumentParser()
	ap.add_argument("--mp3", required=True)
	ap.add_argument("--vtt", required=True)
	ap.add_argument("--audio-ms", type=float, default=None,
	help="duration_ms from the manifest (optional)")
	ap.add_argument("--model", default="medium.en")
	ap.add_argument("--structural-only", action="store_true")
	args = ap.parse_args()

	cues = parse_vtt(Path(args.vtt).read_text(encoding="utf-8"))
	audio_ms = args.audio_ms
	if audio_ms is None:
	try: # PyAV ships with faster-whisper; read the real container duration
	import av

	with av.open(args.mp3) as c:
	audio_ms = c.duration / 1000.0 if c.duration else None
	except Exception:
	pass

	print("=== STRUCTURAL ===")
	problems = structural_checks(cues, audio_ms)
	if not args.structural_only:
	print("\n=== DRIFT ===")
	problems += drift_check(Path(args.mp3), cues, args.model)
	print(f"\n{'ALL CLEAR' if not problems else str(problems) + ' PROBLEM(S)'}")
	return 1 if problems else 0


	if __name__ == "__main__":
	sys.exit(main())

Event Timeline

kevinbazira created this paste.Wed, Sep 16, 12:30 PM

TTS isvc: check_timing.pyActivePublicActions

Event Timeline

TTS isvc: check_timing.py
ActivePublic
Actions