Build a video QC gate that catches black frames before your viewers do

TL;DR

We're building a QC gate that runs after your transcode: tier 1 is ffprobe structural checks (no decode), tier 2 chains blackdetect + freezedetect + silencedetect into a single decode pass, and tier 3 is sampled SSIM. The part that makes it usable is the rules layer that turns raw detections into pass/warn/fail based on where in the timeline they land.

An encoder exiting 0 means a file was produced. It doesn't mean the file has picture in it. Black intros, frozen middles and silent audio all ship with a green build, and get found by a viewer weeks later.

Let's build the step that catches them. Requirements: ffmpeg 7.0+ and python 3.11+.

ffmpeg -version | head -1
ffmpeg -hide_banner -filters | grep -E "blackdetect|freezedetect|silencedetect"

 ... blackdetect       V->V       Detect video intervals that are (almost) black.
 ... freezedetect      V->V       Detects frozen video input.
 ... silencedetect     A->A       Detect silence.

1. Tier 1: structural checks, no decoding 🔍

Start here. It costs nothing and catches more than you'd expect.

# qc/structural.py
import json, subprocess
from dataclasses import dataclass

@dataclass
class Structural:
    duration: float
    video_streams: int
    audio_streams: int
    codecs: list[str]

def probe(path: str) -> Structural:
    cmd = [
        "ffprobe", "-v", "error",
        "-show_entries", "format=duration",
        "-show_entries", "stream=codec_type,codec_name",
        "-of", "json", path,
    ]
    data = json.loads(subprocess.run(cmd, capture_output=True, text=True,
                                     check=True).stdout)
    streams = data.get("streams", [])
    return Structural(
        duration=float(data["format"].get("duration") or 0),
        video_streams=sum(1 for s in streams if s["codec_type"] == "video"),
        audio_streams=sum(1 for s in streams if s["codec_type"] == "audio"),
        codecs=[s["codec_name"] for s in streams],
    )

def check_structural(output: str, source_duration: float, expect_audio: bool):
    s = probe(output)
    findings = []

    if s.video_streams == 0:
        findings.append(("fail", "no video stream in output"))
    if expect_audio and s.audio_streams == 0:
        findings.append(("fail", "expected audio, output has none"))

    drift = abs(s.duration - source_duration)
    if drift > max(2.0, source_duration * 0.02):
        findings.append(("fail", f"duration drift {drift:.1f}s "
                                 f"(out {s.duration:.1f}s vs src {source_duration:.1f}s)"))
    return findings

💡 Tip: missing audio is the single most common real-world hit here. It happens when the source puts audio on a non-first stream and your mapping assumed -map 0:a:0. Nothing else in your pipeline notices.

$ python3 -m qc.structural out/lesson-14.mp4 --source-duration 1842
fail: expected audio, output has none

One ffprobe call, sub-second, and you just caught a defect that would have shipped.

2. Tier 2: three detectors, one decode pass

The trick is chaining them so you decode once. -f null - means no output file is written at all.

ffmpeg -hide_banner -nostats -i out/lesson-14.mp4 \
  -vf "blackdetect=d=0.5:pic_th=0.98:pix_th=0.10,freezedetect=n=-60dB:d=2" \
  -af "silencedetect=n=-50dB:d=2" \
  -f null - 2> qc.log

The filters report to stderr, not stdout, and not as JSON. Parsing log lines is the integration path, annoying but stable:

[blackdetect @ 0x5f2a] black_start:0 black_end:22.4067 black_duration:22.4067
[freezedetect @ 0x5f2b] lavfi.freezedetect.freeze_start: 604.12
[freezedetect @ 0x5f2b] lavfi.freezedetect.freeze_duration: 11.3
[freezedetect @ 0x5f2b] lavfi.freezedetect.freeze_end: 615.42
[silencedetect @ 0x5f2c] silence_start: 0
[silencedetect @ 0x5f2c] silence_end: 1842.31 | silence_duration: 1842.31

Parse it:

# qc/detect.py
import re, subprocess
from dataclasses import dataclass

BLACK = re.compile(r"black_start:(?P<start>[\d.]+) black_end:(?P<end>[\d.]+) "
                   r"black_duration:(?P<dur>[\d.]+)")
FREEZE_START = re.compile(r"freeze_start:\s*(?P<start>[\d.]+)")
FREEZE_DUR = re.compile(r"freeze_duration:\s*(?P<dur>[\d.]+)")
SILENCE_START = re.compile(r"silence_start:\s*(?P<start>-?[\d.]+)")
SILENCE_DUR = re.compile(r"silence_duration:\s*(?P<dur>[\d.]+)")

@dataclass
class Event:
    kind: str      # black | freeze | silence
    start: float
    duration: float

def run_detectors(path: str, black_d=0.5, freeze_d=2.0, silence_d=2.0) -> list[Event]:
    cmd = [
        "ffmpeg", "-hide_banner", "-nostats", "-i", path,
        "-vf", f"blackdetect=d={black_d}:pic_th=0.98:pix_th=0.10,"
               f"freezedetect=n=-60dB:d={freeze_d}",
        "-af", f"silencedetect=n=-50dB:d={silence_d}",
        "-f", "null", "-",
    ]
    log = subprocess.run(cmd, capture_output=True, text=True).stderr
    events, pending = [], {}

    for line in log.splitlines():
        if m := BLACK.search(line):
            events.append(Event("black", float(m["start"]), float(m["dur"])))
        elif m := FREEZE_START.search(line):
            pending["freeze"] = float(m["start"])
        elif m := FREEZE_DUR.search(line):
            events.append(Event("freeze", pending.pop("freeze", 0.0), float(m["dur"])))
        elif m := SILENCE_START.search(line):
            pending["silence"] = max(0.0, float(m["start"]))
        elif m := SILENCE_DUR.search(line):
            events.append(Event("silence", pending.pop("silence", 0.0), float(m["dur"])))

    return events

⚠️ Note: silence_start can be reported as a small negative number at the head of a file. Clamp it to 0 or your position rules will misfire on exactly the case you care most about.

3. The rules layer (this is the actual product) ⚖️

Raw detections are not failures. A 1.5s black frame at 00:00 is a fade-in. A 22s black frame at 00:00 is a broken render. Same detector, opposite verdicts.

If you alert on every detection, the channel gets muted inside a month and you've built nothing.

# qc/rules.py
from dataclasses import dataclass, field
from .detect import Event

RULES_VERSION = "2026.09.1"
HEAD = 3.0   # seconds considered "the start"
TAIL = 3.0   # seconds considered "the end"

@dataclass
class Verdict:
    status: str = "pass"          # pass | warn | fail
    notes: list = field(default_factory=list)
    rules_version: str = RULES_VERSION

    def raise_to(self, level: str, note: str):
        order = {"pass": 0, "warn": 1, "fail": 2}
        if order[level] > order[self.status]:
            self.status = level
        self.notes.append(f"{level}: {note}")

def judge(events: list[Event], total_duration: float) -> Verdict:
    v = Verdict()

    for e in events:
        at_head = e.start <= HEAD
        at_tail = (e.start + e.duration) >= (total_duration - TAIL)
        spans_all = e.duration >= total_duration * 0.95

        if e.kind == "black":
            if at_head and e.duration <= 3:
                continue                                    # fade-in, fine
            if at_head:
                v.raise_to("fail", f"black intro {e.duration:.1f}s")
            elif at_tail and e.duration <= 5:
                continue                                    # fade-out, fine
            elif e.duration >= 5:
                v.raise_to("warn", f"black at {e.start:.0f}s for {e.duration:.1f}s")

        elif e.kind == "freeze":
            if at_tail:
                continue                                    # held last frame
            v.raise_to("fail", f"freeze at {e.start:.0f}s for {e.duration:.1f}s")

        elif e.kind == "silence":
            if spans_all:
                v.raise_to("fail", "audio silent for entire duration")
            elif e.duration >= 30:
                v.raise_to("warn", f"silence at {e.start:.0f}s for {e.duration:.1f}s")

    return v

Now the same detections produce useful output:

$ python3 -m qc out/lesson-14.mp4
{
  "status": "fail",
  "notes": [
    "fail: black intro 22.4s",
    "fail: freeze at 604s for 11.3s",
    "fail: audio silent for entire duration"
  ],
  "rules_version": "2026.09.1"
}

$ python3 -m qc out/promo-clip.mp4
{
  "status": "pass",
  "notes": [],
  "rules_version": "2026.09.1"
}

The second file also had a black detection (a 1.2s fade-in) and a tail freeze. It passed, correctly, because the rules know where those landed.

4. Tier 3: sampled reference comparison

SSIM against the source catches "the encode ran but produced garbage," and it costs two decodes. Don't put it in the synchronous path.

ffmpeg -hide_banner \
  -i out/lesson-14.mp4 -i src/lesson-14.mov \
  -lavfi "[0:v]scale=1280:720[a];[1:v]scale=1280:720[b];[a][b]ssim=stats_file=ssim.log" \
  -f null -

SSIM Y:0.981234 U:0.992011 V:0.991455 All:0.984772 (18.156204)

# qc/reference.py: run nightly or on sampled assets only
SSIM_FLOOR = 0.93

def check_ssim(all_score: float):
    if all_score < SSIM_FLOOR:
        return [("fail", f"SSIM {all_score:.3f} below floor {SSIM_FLOOR}")]
    return []

⚠️ Scale both inputs to a common resolution before comparing, or ssim errors out on mismatched dimensions. And pick the floor empirically from your own known-good output; a number copied from someone else's pipeline will either never fire or fire constantly.

5. Wire it into the pipeline

# qc/__main__.py
import sys, json
from .structural import check_structural, probe
from .detect import run_detectors
from .rules import judge

def gate(output: str, source_duration: float, expect_audio: bool = True) -> dict:
    # tier 1: fail fast, no decode
    structural = check_structural(output, source_duration, expect_audio)
    if any(level == "fail" for level, _ in structural):
        return {"status": "fail",
                "notes": [f"{l}: {m}" for l, m in structural],
                "tier": 1}

    # tier 2: one decode, three detectors
    info = probe(output)
    verdict = judge(run_detectors(output), info.duration)
    return {"status": verdict.status, "notes": verdict.notes,
            "rules_version": verdict.rules_version, "tier": 2}

if __name__ == "__main__":
    result = gate(sys.argv[1], float(sys.argv[2]))
    print(json.dumps(result, indent=2))
    sys.exit(1 if result["status"] == "fail" else 0)

Exiting nonzero on fail lets you drop this straight into a worker or a CI step.

6. Store the verdict with its rules version

ALTER TABLE assets
  ADD COLUMN qc_status text,
  ADD COLUMN qc_notes jsonb,
  ADD COLUMN qc_rules_version text;

This matters the first time you loosen a threshold. When a customer's library legitimately opens on slow fades and you relax the black-intro rule, you need to find everything that failed under the old version and re-judge it. Without the stamp, you re-run QC on your entire library.

The scarier direction is the same query in reverse: when you tighten a rule because something got through, you want the list of everything that passed under the looser version.

What's next

Run tier 1 against last week's output before you build anything else. One ffprobe call per asset, three assertions, a count at the end:

$ for f in out/*.mp4; do python3 -m qc.structural "$f" || echo "FAIL $f"; done | grep -c FAIL
7

Seven broken assets you're already serving is a better argument for this project than any design doc.

From there, the things worth adding are ebur128 for loudness (catches "audio exists but is inaudible," which silencedetect misses entirely), a check that every rendition in your ABR ladder has matching duration and keyframe positions, and per-tenant rule overrides so the customer with the slow fades gets their own thresholds instead of forcing you to loosen everyone's.