ClosedCaptioning
Broadcast QC Automation

Building a Compliant Caption Ingestion Gateway

A caption ingestion gateway fails at scale for one specific reason: it reads the whole payload into RAM before it validates anything, so a single 500 MB Scenarist Closed Caption (SCC) archive triggers an OOM kill long before a compliance gate ever runs. The fix is a worker that never materializes the file — it maps the bytes, walks them as a generator, batches the cues under a fixed backpressure unit, and runs every admission check inside a memory-bounded, filesystem-restricted process. This page implements that worker: a gateway that holds resident memory flat regardless of archive size while enforcing the exact 29.97 fps drop-frame timecode arithmetic of SMPTE ST 12-1 and the sustained 20 characters-per-second (cps) readability ceiling that 47 CFR § 79.1 makes operationally mandatory. It is the runtime that executes the admission gates defined in secure caption pipeline design, but at archive scale and under a hard OOM ceiling.

Minimal working gateway

import hashlib
import json
import mmap
import re
from concurrent.futures import ProcessPoolExecutor, as_completed
from fractions import Fraction
from pathlib import Path
from typing import Iterator

NTSC_FPS  = Fraction(30000, 1001)            # SMPTE ST 12-1 — exact 29.97 rational
FRAME_MS  = float(1 / NTSC_FPS * 1000)       # 33.3667 ms/frame; never a rounded literal
CPS_LIMIT = 20.0                             # sustained 608 ceiling; hard max ~30 (CTA-608-E)
MAX_BYTES = 256 * 1024 * 1024                # ingest policy — bound memory before any read
CHUNK_LINES = 5000                           # backpressure unit; caps per-worker resident set
_SCC_LINE = re.compile(r"^(\d{2}:\d{2}:\d{2}[:;]\d{2})\s+([0-9a-fA-F ]+)$")

def stream_scc(path: str) -> Iterator[tuple[int, int, str, str]]:
    """Zero-copy SCC reader; yields (line_idx, byte_offset, timecode, hex_payload)."""
    p = Path(path)
    size = p.stat().st_size
    if size == 0:                            # truncated upload -> quarantine, never mmap empty
        raise ValueError("empty payload")
    if size > MAX_BYTES:                      # OOM guard: reject before a single byte is read
        raise ValueError(f"payload {size} B exceeds {MAX_BYTES} B ingest ceiling")
    with open(p, "rb") as fh, mmap.mmap(fh.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        offset = idx = 0
        while offset < mm.size():
            end = mm.find(b"\n", offset)
            if end == -1:
                end = mm.size()
            line = mm[offset:end].decode("ascii", "replace").strip()
            m = _SCC_LINE.match(line)
            if m:
                yield idx, offset, m.group(1), m.group(2).strip()
                idx += 1
            offset = end + 1

def tc_to_ms(tc: str) -> float:
    h, m, s, f = map(int, re.split(r"[:;]", tc))
    frames = (h * 3600 + m * 60 + s) * 30 + f
    if ";" in tc:                            # SMPTE ST 12-1 — drop 2 frames/min, not every 10th
        mins = h * 60 + m
        frames -= 2 * (mins - mins // 10)
    return frames * FRAME_MS

def admit_line(rec: tuple[int, int, str, str]) -> dict:
    idx, offset, tc, hexpairs = rec
    start = tc_to_ms(tc)
    text = "".join(
        chr(b) for w in hexpairs.split() if len(w) == 4
        for b in ((int(w, 16) >> 8) & 0x7F, int(w, 16) & 0x7F) if b >= 0x20
    ).strip()
    cps = len(text.replace(" ", "")) / 1.0   # onset-only SCC; end inferred downstream (1 s window)
    digest = hashlib.sha256(f"{offset}:{idx}:{start}:{text}".encode()).hexdigest()  # FIPS 180-4
    return {"idx": idx, "offset": offset, "start_ms": round(start, 3), "text": text,
            "peak_cps": round(cps, 1), "compliant": cps <= CPS_LIMIT, "audit_hash": digest}

def process_chunk(chunk: list) -> list[dict]:
    return [admit_line(r) for r in chunk]    # runs in an isolated, FS-restricted worker

def run_gateway(src: str, manifest_path: str, workers: int = 4) -> dict:
    pending, results, batch = [], [], []
    with ProcessPoolExecutor(max_workers=workers) as pool:
        for rec in stream_scc(src):          # constant memory: the file is never materialized
            batch.append(rec)
            if len(batch) >= CHUNK_LINES:
                pending.append(pool.submit(process_chunk, batch)); batch = []
        if batch:
            pending.append(pool.submit(process_chunk, batch))
        for fut in as_completed(pending):
            results.extend(fut.result())
    admitted = [r for r in results if r["compliant"]]
    with open(manifest_path, "w", encoding="utf-8") as out:
        json.dump(admitted, out, indent=2, ensure_ascii=False)   # append-only / WORM target
    return {"lines": len(results), "admitted": len(admitted)}

Code walkthrough

The constants are the contract. Every threshold the gateway enforces is a named module-level value, so re-pointing the worker at a different timebase or jurisdiction is a one-line edit rather than a code change. FRAME_MS is derived from Fraction(30000, 1001) and never from a rounded 33.37 literal — that exact NTSC rational is what keeps drop-frame arithmetic from drifting roughly 3.6 seconds per hour over a feature-length program, exactly as SMPTE ST 12-1 specifies. MAX_BYTES and CHUNK_LINES are the two levers that hold memory flat: the first caps the file the gateway will accept at all, the second caps how many cue records any single worker holds at once.

stream_scc is the zero-copy front door. It checks st_size before it maps anything, so a zero-byte truncated upload is quarantined as malformed instead of raising ValueError: cannot mmap an empty file, and an oversized payload is rejected before a single byte is read — the OOM guard runs first, not after the damage. The mmap mapping lets the kernel page bytes in on demand; mm.find(b"\n", offset) walks line boundaries by offset without ever building a list of lines, and the function yields one record at a time. Resident memory for a 5 GB archive is therefore identical to a 5 KB one. Each record carries its own byte_offset, which becomes the provenance key in the audit trail.

tc_to_ms is the only place frame math lives. It splits on either separator and branches on the ; drop-frame marker required by SMPTE ST 12-1, subtracting two frames per minute except every tenth minute. Because it returns milliseconds from the exact FRAME_MS, every downstream consumer inherits frame-accurate onsets without re-deriving the arithmetic. The deeper end-to-end conversion variant of this is covered in converting SCC to SRT without timing loss.

admit_line is the per-cue gate. It decodes each 4-hex-digit word into its two 7-bit characters (stripping the parity bit and dropping control bytes below 0x20), measures displayed density excluding whitespace, and stamps the result compliant only if it stays at or under the 20 cps ceiling — the readability floor that makes 47 CFR § 79.1 satisfiable on legacy CEA-608 decoders, whose physical channel tops out near 30 bytes/s under CTA-608-E. The SHA-256 digest (FIPS 180-4) binds the source offset, line index, onset and text into one immutable identifier, so the audit manifest can prove exactly which bytes produced which verdict. The character-rate dimension is enforced as a standalone gate in enforcing character rate limits in QC; here it is one admission check among several.

process_chunk and run_gateway are the isolation and backpressure layer. Each batch of at most CHUNK_LINES records is submitted to a ProcessPoolExecutor, so parsing of an untrusted payload happens in a separate OS process that you can run with a read-only mount of the drop directory and a write-only quarantine path — a parser exploit cannot reach the rest of the host. The generator feeds batches lazily, so the producer never races ahead of the workers and the pending-futures list is the only growth term. Admitted records are written to an append-only (WORM) manifest keyed by hash; rejected cues are simply absent and recoverable from the same source bytes on re-ingest.

Memory-bounded ingestion gateway — size guard, zero-copy reader, sandboxed worker pool, WORM manifest An SCC archive of any size meets a size guard that rejects payloads over 256 MB or empty files before any read. Accepted bytes flow to an mmap zero-copy reader yielding one cue record at a time, then to a batcher that fills a 5000-line chunk (the backpressure unit). A dashed accent arrow loops from the worker pool back to the batcher: the generator is pulled lazily so the producer never races ahead. A ProcessPoolExecutor fans chunks to N filesystem-restricted workers running drop-frame tc_to_ms, a ≤20 cps density check and a SHA-256 audit hash. An as_completed collector gathers verdicts: compliant cues go to an append-only WORM manifest (highlighted), non-compliant cues are dropped and recoverable on re-ingest. Resident memory stays flat regardless of archive size because the file is never materialized. Memory-bounded ingestion gateway — guard, stream, sandbox, attest The file is never materialized: mmap + generator + a fixed chunk hold resident memory flat from 5 KB to 5 GB. backpressure — lazy pull, producer never races ahead SCC archive any size Size guard st_size before mmap mmap reader zero-copy · yields 1 cue/time Batcher 5000-line chunk ProcessPoolExecutor N FS-restricted workers tc_to_ms · drop-frame density ≤ 20 cps SHA-256 audit hash read-only mount · isolated process as_completed gathers verdicts reject compliant non-compliant Rejected > 256 MB or empty Dropped cues recoverable on re-ingest WORM manifest append-only · hash-keyed Resident memory stays flat regardless of archive size size guard bounds the input · mmap pages on demand · the chunk bounds each worker · only the pending-futures list grows

Edge cases and known gotchas

  • Drop-frame vs non-drop confusion. A : separator is non-drop (30 fps); a ; is drop-frame (29.97). Treating every file as 30 fps under-counts dropped frames and accumulates drift; tc_to_ms keys the compensation strictly on the separator, never on a config default.
  • Onset-only SCC has no native end time. SCC lines carry a cue start, not a duration, so the 1-second density window here is a conservative placeholder — derive the real end from the next cue onset (or explicit roll-up/pop-on logic) before treating peak_cps as authoritative.
  • BOM bytes counted as a character. A UTF-8 BOM that survives decoding inflates the visible-character count and can tip a borderline cue over 20 cps. Strip it during encoding normalization; the repair patterns live in fixing UTF-8 encoding errors in SCC files.
  • Orphaned control words. Padding (8080) and positioning words (942x) are not text; the >= 0x20 filter drops them per byte, but a parser that splits on them as cue boundaries will emit phantom cues. Decode with a proper state machine when fidelity matters — see parsing SCC with Python libraries.
  • Worker count above physical cores. This stage is CPU-light and I/O-bound on the mmap fault path; oversizing workers past the physical core count adds context-switching overhead without throughput. Size it to cores and scale horizontally for larger libraries.

Where this plugs into the pipeline

This gateway is the pre-parse, pre-validate boundary of the broader pipeline: it sits in front of the canonical cue model and decides whether each archive is allowed to become cues at all. It runs the same admission logic specified in secure caption pipeline design — signature, encoding, timecode and density gates — but wraps them in the memory-safe, sandboxed execution model that lets them survive multi-gigabyte SCC archives without an OOM kill. Its append-only manifest is the input contract for the acceptance gates in the FCC Part 79 compliance checklist, and its generator-driven batching is the same backpressure pattern detailed in async batch caption processing.

Part of: Broadcast Captioning Architecture & Compliance.