Async Batch Caption Processing

Captioning vendors and broadcast automation teams routinely hold archives of tens of thousands of SRT, SCC, and WebVTT assets, and the synchronous, one-file-at-a-time loop that works fine for a single delivery collapses at that scale. A blocking parser spends most of its wall-clock time waiting on disk and network I/O rather than on CPU, so a sequential run of 40,000 files can take hours while CPU sits near idle. Async batch caption processing fixes the throughput problem without sacrificing correctness: it streams files through a fixed pool of worker coroutines, applies the same frame-accurate compliance checks every file would get individually, and routes any asset that fails to a quarantine path instead of aborting the run. The engineering target for this stage is concrete — saturate available I/O while holding per-worker memory bounded, keep the event loop free of stalls longer than 50 ms, and enforce the same ±2-frame (FCC 47 CFR § 79.1) and reading-rate tolerances that a single-file validator would apply.

This page is part of SRT, SCC & WebVTT Parsing Workflows and assumes you already have working single-format decoders; here we focus on running them concurrently and safely at archive scale.

Problem framing

The failure mode this stage addresses is throughput collapse under synchronous I/O combined with all-or-nothing batch aborts. Three quantified constraints define a correct implementation:

• Concurrency must be bounded. Unbounded asyncio.create_task spawning over a 50,000-file tree exhausts file descriptors (the default soft limit is often 1024) and triggers OSError: Too many open files. The pool size must be capped, typically at os.cpu_count() * 2 for the I/O-bound mix of disk reads and object-store fetches.
• Memory must stay flat. Materializing an entire archive into resident memory is not an option. A bounded asyncio.Queue(maxsize=N) provides backpressure so the discovery phase pauses when downstream workers fall behind, holding peak memory to roughly N × average_file_size rather than the full corpus.
• Compliance thresholds are non-negotiable at scale. Each parsed asset must still satisfy the same limits enforced one file at a time — cue overlap under one frame, cumulative timing drift within tolerance, and reading rate within the per-territory cap. Batch mode changes the scheduling, never the rules.

A single corrupt or malformed file must never stall or abort the run. Head-of-line blocking — where one slow or broken payload holds up everything behind it — is designed out by decoupling discovery, parsing, and validation into separate queue stages.

Pipeline stage & prerequisites

Batch processing sits immediately after format detection and before automated sync drift detection and report generation in the end-to-end caption pipeline. It is the orchestration layer that fans the per-format decoders documented elsewhere out across an entire archive: SCC byte-level decoding from Parsing SCC with Python Libraries, SRT timestamp normalization, and WebVTT cue extraction & validation. The extension-based dispatch contract those pages establish is what makes a single worker able to handle a mixed-format tree.

Required tooling:

Install the parser stack with pip install aiofiles pysrt webvtt-py pycaption charset-normalizer. CPU-bound decode work — notably the SCC state machine — should be offloaded to a process pool via loop.run_in_executor, because pure-Python parsing under the GIL will otherwise serialize on the event loop thread.

Step-by-step implementation

Step 1 — Stream file discovery without blocking the loop

Discovery walks the archive lazily and yields control between filesystem entries so a deep tree never monopolizes the event loop before any work starts.

import asyncio
import os
from pathlib import Path
from typing import AsyncIterator

CAPTION_EXTS = {".srt", ".scc", ".vtt"}

async def discover_files(root_dir: Path) -> AsyncIterator[Path]:
    """Yield caption files lazily, releasing the loop between entries."""
    for path in root_dir.rglob("*"):
        if path.suffix.lower() in CAPTION_EXTS:
            yield path
            await asyncio.sleep(0)  # cooperative yield: keep the loop responsive

Step 2 — Detect encoding before decoding

Caption files arrive from heterogeneous workstations; a UTF-8 decode on a Latin-1 or UTF-16 export silently corrupts accented characters and breaks the 32-column CEA-608 grid. Sniff first, then decode.

from charset_normalizer import from_bytes

def decode_payload(raw: bytes) -> str:
    """Resolve encoding before parsing; strip a UTF-8 BOM if present.

    W3C WebVTT spec — a leading U+FEFF BOM must be stripped before the
    'WEBVTT' signature is matched, or cue parsing fails on byte 0.
    """
    best = from_bytes(raw).best()
    text = str(best) if best else raw.decode("utf-8", errors="replace")
    return text.lstrip("")

Step 3 — Dispatch to the right decoder by format

A single worker handles a mixed-format tree by routing on extension to the real per-format parsers. SCC decoding is CPU-bound and runs in an executor so it never blocks the loop.

import pysrt
import webvtt
from pycaption import SCCReader
from typing import Dict, List

def _parse_srt(text: str) -> List[Dict]:
    subs = pysrt.from_string(text)
    return [
        {"start": s.start.ordinal / 1000.0,  # ms -> seconds
         "end": s.end.ordinal / 1000.0,
         "text": s.text}
        for s in subs
    ]

def _parse_vtt(text: str) -> List[Dict]:
    return [
        {"start": c.start_in_seconds, "end": c.end_in_seconds, "text": c.text}
        for c in webvtt.from_string(text)
    ]

def _parse_scc(text: str) -> List[Dict]:
    # pycaption returns microsecond timestamps for CEA-608 payloads
    caps = SCCReader().read(text)
    lang = caps.get_languages()[0]
    return [
        {"start": c.start / 1_000_000.0, "end": c.end / 1_000_000.0,
         "text": c.get_text()}
        for c in caps.get_captions(lang)
    ]

_DISPATCH = {".srt": _parse_srt, ".vtt": _parse_vtt, ".scc": _parse_scc}

async def parse_file(path: Path, text: str, loop) -> List[Dict]:
    """Run the format-specific decoder; offload CPU-bound SCC to an executor."""
    fn = _DISPATCH[path.suffix.lower()]
    if path.suffix.lower() == ".scc":
        return await loop.run_in_executor(None, fn, text)  # GIL-bound work off-loop
    return fn(text)

Step 4 — Validate against compliance thresholds

Validation runs as its own coroutine stage so a dense or overlapping file is flagged without halting the workers feeding it.

OVERLAP_TOLERANCE_S = 0.040   # FCC 47 CFR § 79.1 — < 1 frame @ 25 fps overlap allowed
MAX_CPS             = 17.0    # reading-rate ceiling (Netflix/EBU adult guideline)

def validate_cues(cues: List[Dict]) -> List[str]:
    """Return a list of violation strings; empty list means the file passes."""
    violations: List[str] = []
    for i in range(len(cues) - 1):
        # Overlap beyond one-frame tolerance is a synchronicity defect
        if cues[i + 1]["start"] < cues[i]["end"] - OVERLAP_TOLERANCE_S:
            violations.append(f"overlap@cue{i}")
    for i, c in enumerate(cues):
        dur = c["end"] - c["start"]
        if dur > 0:
            cps = len(c["text"].replace("\n", "")) / dur
            if cps > MAX_CPS:                       # reading rate exceeds cap
                violations.append(f"cps>{MAX_CPS:.0f}@cue{i}")
    return violations

The reading-rate logic here mirrors the dedicated cluster on enforcing character-rate limits in QC; keep the two in sync so batch and single-file runs report identical defects.

Step 5 — Wire workers, backpressure, and quarantine together

The orchestrator caps concurrency with a semaphore, bounds memory with a sized queue, and uses a TaskGroup so a crash in any worker cancels the rest cleanly.

import logging

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s | %(levelname)s | %(message)s")
log = logging.getLogger("batch")

MAX_CONCURRENCY = os.cpu_count() * 2   # I/O-bound: oversubscribe CPU count
QUEUE_CAPACITY  = 500                   # bounds peak resident memory

async def worker(name, queue, semaphore, results, quarantine, loop):
    import aiofiles
    while True:
        path: Path = await queue.get()
        try:
            async with semaphore:                       # cap concurrent open FDs
                async with aiofiles.open(path, "rb") as f:
                    raw = await f.read()
            text = decode_payload(raw)
            cues = await parse_file(path, text, loop)
            violations = validate_cues(cues)
            if violations:
                quarantine.append((path, violations))
            else:
                results.append({"path": str(path), "cues": len(cues)})
        except Exception as exc:                          # never abort the batch
            log.error("worker %s failed on %s: %s", name, path.name, exc)
            quarantine.append((path, [f"exception:{type(exc).__name__}"]))
        finally:
            queue.task_done()

async def orchestrate(root_dir: Path):
    loop       = asyncio.get_running_loop()
    queue      = asyncio.Queue(maxsize=QUEUE_CAPACITY)   # backpressure to discovery
    semaphore  = asyncio.Semaphore(MAX_CONCURRENCY)
    results, quarantine = [], []

    async with asyncio.TaskGroup() as tg:
        workers = [
            tg.create_task(worker(i, queue, semaphore, results, quarantine, loop))
            for i in range(MAX_CONCURRENCY)
        ]
        async for path in discover_files(root_dir):
            await queue.put(path)        # blocks when queue is full -> backpressure
        await queue.join()               # drain all enqueued work
        for w in workers:
            w.cancel()                   # workers loop forever; cancel after drain

    log.info("done: %d passed, %d quarantined", len(results), len(quarantine))
    return results, quarantine

Run it with asyncio.run(orchestrate(Path("/mnt/archive"))). Because the queue is bounded, await queue.put(path) blocks the discovery loop whenever workers fall behind — that single line is the entire backpressure mechanism.

Threshold reference table

Every numeric limit the batch validator enforces, with its governing source:

Verification & test pattern

Confirm correctness on a known fixture before pointing the runner at a live archive. The assertion checks both that a valid file is accepted and that an overlap-injected file lands in quarantine.

import asyncio
import pytest
from pathlib import Path

@pytest.fixture
def fixture_tree(tmp_path: Path) -> Path:
    good = "1\n00:00:01,000 --> 00:00:03,000\nHello world\n"
    # second cue starts 0.5 s before the first ends -> overlap > 1 frame
    bad  = ("1\n00:00:01,000 --> 00:00:03,000\nLine A\n\n"
            "2\n00:00:02,500 --> 00:00:04,000\nLine B\n")
    (tmp_path / "good.srt").write_text(good, encoding="utf-8")
    (tmp_path / "bad.srt").write_text(bad, encoding="utf-8")
    return tmp_path

def test_batch_routes_overlap_to_quarantine(fixture_tree):
    results, quarantine = asyncio.run(orchestrate(fixture_tree))
    passed = {Path(r["path"]).name for r in results}
    quar   = {p.name for p, _ in quarantine}
    assert passed == {"good.srt"}              # clean file accepted
    assert quar == {"bad.srt"}                 # overlap defect quarantined
    assert any("overlap" in v for _, vs in quarantine for v in vs)

For a regression gate, fold this fixture into the CI/CD gating for caption builds workflow so a parser change that alters quarantine behaviour fails the build.

Troubleshooting / failure modes

OSError: [Errno 24] Too many open files : Root cause: concurrency was not bounded, or the semaphore guards parsing but not the aiofiles.open call. Fix: ensure every file open happens inside async with semaphore, and raise the soft FD limit with ulimit -n 4096 if the cap legitimately needs to be higher.

Event loop stalls; queue depth climbs then validation starves : Root cause: CPU-bound SCC decoding is running on the loop thread, blocking it past the 50 ms budget. Fix: route _parse_scc through loop.run_in_executor with a ProcessPoolExecutor, and tune gc.set_threshold() to reduce pause frequency on long runs.

Memory grows unbounded across a large tree : Root cause: an unbounded asyncio.Queue (no maxsize) lets discovery enqueue the entire archive before workers drain it. Fix: set maxsize=QUEUE_CAPACITY so queue.put applies backpressure to discovery.

One malformed file aborts the whole batch : Root cause: the parser exception propagates out of the worker and cancels the TaskGroup. Fix: wrap the parse/validate body in try/except Exception and append to quarantine in the handler, as shown in Step 5 — only truly unexpected errors should escape.

Accented characters mangled in output : Root cause: a UTF-16 or Latin-1 export was force-decoded as UTF-8. Fix: run charset-normalizer detection (Step 2) before decode; this is the same class of defect covered in depth under fixing UTF-8 encoding errors in SCC files.

Re-run reprocesses thousands of already-validated files : Root cause: no checkpointing, so an interrupted run restarts from zero. Fix: persist processed paths to a SQLite table in WAL mode and skip any path already recorded — see operational notes below.

Operational notes

For archives that exceed available RAM, three patterns keep a batch run stable:

• Idempotent checkpointing. Write each completed path and its result to a SQLite database opened with PRAGMA journal_mode=WAL. On restart, discover_files skips any path already present, making the runner safe to kill and resume after spot-instance preemption.
• Worker-pool sizing. Start at os.cpu_count() * 2. If profiling shows workers blocked on object-store latency rather than CPU, raise the multiplier; if SCC decode dominates, move that work to a ProcessPoolExecutor sized to the physical core count and keep the async pool lean.
• I/O patterns at scale. Read multi-hour broadcast files in chunks rather than a single await f.read(); for object stores, batch range requests and prefer streaming decode so peak resident memory stays near QUEUE_CAPACITY × avg_file_size. Emit queue depth, worker utilization, and quarantine rate as metrics so SLA regressions surface before they breach.

Telemetry from this stage feeds directly into scheduled QC report generation, and the whole runner should execute inside a hardened boundary as described in secure caption pipeline design.

Related

• Per-format decoders this stage fans out — Parsing SCC with Python Libraries
• Timestamp quantization before batch validation — SRT Timestamp Normalization
• WebVTT cue model the workers consume — WebVTT Cue Extraction & Validation
• Drift checks run downstream of the batch — Automated Sync Drift Detection
• Gating batch behaviour in CI — CI/CD Gating for Caption Builds

Part of: SRT, SCC & WebVTT Parsing Workflows