#!/usr/bin/env python3 """Vanguard timecode strip — shared format (encoder for tests, decoder for the sweep). The published canvas is the game image plus an "overscan band" of extra rows at the bottom. Everything is expressed relative to the canvas WIDTH so any uniform rescale keeps the geometry: cell = W / 160 (16 px at 2560, 12 px at 1920) band = round_even(cell) rows at the bottom of the frame strip = 64 cells starting at x = cell, vertically centred in the band: [1,0,1,0] framing · 48-bit server time in ms (MSB first) · 8-bit CRC-8 (poly 0x07) · [0,1,0,1] framing 1 = white, 0 = black text = human-readable server clock to the right of the strip (not decoded) Decode: sample the centre of each cell on the row y = H - band/2; threshold at the midpoint of the framing cells' brightness; verify framing + CRC. The decoded value is SERVER time at render; frame-0 capture time = median(decoded_k - pts_k) over several frames. """ import sys, json, struct, subprocess, statistics CELLS = 64 FRAME_A = [1, 0, 1, 0] FRAME_B = [0, 1, 0, 1] def crc8(data: bytes, poly=0x07) -> int: c = 0 for b in data: c ^= b for _ in range(8): c = ((c << 1) ^ poly) & 0xFF if c & 0x80 else (c << 1) & 0xFF return c def encode_bits(ms: int): payload = ms.to_bytes(6, "big") bits = [(payload[i // 8] >> (7 - i % 8)) & 1 for i in range(48)] c = crc8(payload) bits += [(c >> (7 - i)) & 1 for i in range(8)] return FRAME_A + bits + FRAME_B def decode_bits(bits): if bits[:4] != FRAME_A or bits[-4:] != FRAME_B: return None body = bits[4:-4] payload = bytes(int("".join(map(str, body[i:i + 8])), 2) for i in range(0, 48, 8)) c = int("".join(map(str, body[48:56])), 2) if crc8(payload) != c: return None return int.from_bytes(payload, "big") def geometry(w): cell = w / 160.0 band = int(round(cell)); band += band % 2 return cell, band def cell_centres(w): cell, band = geometry(w) return [int(round(cell * (1 + i + 0.5))) for i in range(CELLS)], band # ------------------------------------------------------------------ decoder over a raw gray band def decode_band_row(row: bytes, w: int): """row = one grayscale row (w bytes) taken from the middle of the band.""" xs, _ = cell_centres(w) vals = [] for x in xs: lo, hi = max(0, x - 1), min(w, x + 2) vals.append(sum(row[lo:hi]) / max(1, hi - lo)) fr = vals[:4] + vals[-4:] thr = (min(fr) + max(fr)) / 2.0 if max(fr) - min(fr) < 40: return None # no contrast: strip absent return decode_bits([1 if v > thr else 0 for v in vals]) def probe(path): out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,avg_frame_rate,r_frame_rate:format=duration", "-of", "json", path], capture_output=True, text=True, check=True).stdout j = json.loads(out); st = j["streams"][0] # nominal rate (r_frame_rate) — avg_frame_rate is skewed by dropped frames / gaps def rate(x): try: num, den = x.split("/"); return float(num) / float(den or 1) except Exception: return 0.0 fps = rate(st.get("r_frame_rate", "")) or rate(st.get("avg_frame_rate", "")) or 60.0 if fps > 240: fps = rate(st.get("avg_frame_rate", "")) or 60.0 # bogus r_frame_rate on some files return int(st["width"]), int(st["height"]), fps, float(j["format"]["duration"]) def sample_window(path, w, h, fps, start_s, dur_s): """Decode every frame in [start, start+dur). Returns [(pts_s, server_ms)].""" import re _, band = geometry(w) y0 = h - band # -copyts keeps the file's own timestamps; showinfo prints the exact pts of every emitted frame cmd = ["ffmpeg", "-nostdin", "-v", "info", "-copyts", "-ss", f"{start_s:.3f}", "-t", f"{dur_s:.3f}", "-i", path, "-an", "-vf", f"crop={w}:{band}:0:{y0},format=gray,showinfo", "-f", "rawvideo", "-"] r = subprocess.run(cmd, capture_output=True, check=True) raw = r.stdout pts = [float(m) for m in re.findall(r"\bpts_time:\s*([0-9.]+)", r.stderr.decode(errors="replace"))] frame_len = w * band res = [] for k in range(min(len(raw) // frame_len, len(pts))): row = raw[k * frame_len + (band // 2) * w: k * frame_len + (band // 2) * w + w] ms = decode_band_row(row, w) if ms is not None: res.append((pts[k], ms)) return res def first_pts(path): """pts of the first video frame (fMP4 from x264 with B-frames starts ~2 frames after 0).""" out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "frame=pts_time", "-read_intervals", "%+#1", "-of", "csv=p=0", path], capture_output=True, text=True).stdout.strip() try: return float(out.splitlines()[0].strip(",").split(",")[0]) except Exception: return 0.0 def analyse(path, win_s=4.0): w, h, fps, dur = probe(path) p0 = first_pts(path) # timecode start is the FIRST FRAME, not pts 0 windows = [0.0] if dur > 60: windows.append(max(0.0, dur / 2 - win_s / 2)) if dur > 20: windows.append(max(0.0, dur - win_s - 0.5)) samples = [] for ws in windows: pts = sample_window(path, w, h, fps, ws, win_s) if pts: offs = [ms - (pts_s - p0) * 1000.0 for pts_s, ms in pts] samples.append({"at_s": round(ws, 3), "frames": len(pts), "t0_ms": int(statistics.median(offs))}) if not samples: return {"source": "none", "fps": fps, "width": w, "height": h, "duration_s": dur} t0 = samples[0]["t0_ms"] drift = samples[-1]["t0_ms"] - t0 if len(samples) > 1 else 0 return {"source": "strip", "t0_ms": t0, "drift_ms": drift, "fps": fps, "width": w, "height": h, "duration_s": dur, "samples": samples} def find_gaps(path, fps, max_list=50): """Holes in the video timeline (dropped frames): deltas > 1.5 frame periods. Demux only, fast.""" out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "packet=pts_time", "-of", "csv=p=0", path], capture_output=True, text=True).stdout pts = sorted(float(x.strip(",").split(",")[0]) for x in out.split() if x.strip(",") not in ("", "N/A")) period = 1.0 / (fps or 60.0) gaps, total = [], 0.0 for a, b in zip(pts, pts[1:]): d = b - a if d > period * 1.5: total += d - period if len(gaps) < max_list: gaps.append({"at_s": round(a, 3), "missing_ms": int(round((d - period) * 1000))}) return {"count": sum(1 for a, b in zip(pts, pts[1:]) if b - a > period * 1.5), "total_missing_ms": int(round(total * 1000)), "first": gaps} def timecode_string(t0_ms, fps, tz="America/New_York"): """Time-of-day timecode for the tmcd track: HH:MM:SS:FF (drop-frame ';' for 29.97/59.94).""" from datetime import datetime, timezone from zoneinfo import ZoneInfo local = datetime.fromtimestamp(t0_ms / 1000.0, timezone.utc).astimezone(ZoneInfo(tz)) frac = (t0_ms % 1000) / 1000.0 nominal = max(1, round(fps)) ff = min(nominal - 1, int(frac * fps)) drop = any(abs(fps - r) < 0.05 for r in (29.97, 59.94, 23.976)) # true NTSC rates only sep = ";" if drop else ":" return f"{local:%H:%M:%S}{sep}{ff:02d}", local def plan(path, arrival_ms=None, tz="America/New_York"): """Decide the timecode for a recording: strip if present, else server arrival time. Prints shell vars.""" from datetime import datetime, timezone r = analyse(path) if r.get("source") == "strip": t0 = r["t0_ms"]; src = "strip" elif arrival_ms: t0 = int(arrival_ms); src = "arrival" else: t0 = None; src = "none" out = {"source": src, "fps": r.get("fps"), "drift_ms": r.get("drift_ms", 0), "samples": r.get("samples", [])} try: out["gaps"] = find_gaps(path, r.get("fps") or 60.0) except Exception: pass if t0 is not None: tc, local = timecode_string(t0, r.get("fps") or 60.0, tz) utc = datetime.fromtimestamp(t0 / 1000.0, timezone.utc) out.update({"t0_ms": t0, "timestamp": utc.strftime("%Y-%m-%dT%H:%M:%S.") + f"{t0 % 1000:03d}Z", "timecode": tc, "timecode_tz": tz, "local_start": local.isoformat()}) print(f"TC_TIMECODE='{tc}'") # quoted: drop-frame uses ';' which the shell would split on print(f"TC_CREATION='{utc.strftime('%Y-%m-%dT%H:%M:%S')}.{t0 % 1000:03d}Z'") print(f"TC_SOURCE='{src}'") print("TC_JSON=" + json.dumps(json.dumps(out))) # double-encoded: shell-safe single token return out if __name__ == "__main__": if len(sys.argv) < 2: print(__doc__); sys.exit(2) if sys.argv[1] == "plan": plan(sys.argv[2], int(sys.argv[3]) if len(sys.argv) > 3 and sys.argv[3] else None); sys.exit(0) print(json.dumps(analyse(sys.argv[1]), indent=2))