161 lines
7.3 KiB
Python
Executable File
161 lines
7.3 KiB
Python
Executable File
#!/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:format=duration", "-of", "json", path],
|
|
capture_output=True, text=True, check=True).stdout
|
|
j = json.loads(out); st = j["streams"][0]
|
|
num, den = st["avg_frame_rate"].split("/"); fps = float(num) / float(den or 1)
|
|
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 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 = round(fps)
|
|
ff = min(nominal - 1, int(frac * fps))
|
|
sep = ";" if abs(fps - nominal) > 0.01 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", [])}
|
|
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}")
|
|
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))
|