VO delivery audit

The reconciliation pair to the loc-table manifest driver (issue #42): audit_delivery compares a delivered audio folder against the loc-table the way a localization vendor's pre-delivery QA pass does — missing lines, orphan files, naming-convention violations, empty takes, and script↔audio duration mismatches. It is deterministic set/arithmetic over file stats + the manifest (no ML), and read-only over the delivered folder: it only walks it and reads WAV headers/samples, never writing there. It shares the #40 read_manifest parser and reuses pipeline.wav_duration for stats.

python -m openfacefx audit --manifest loc.csv --delivered vo/ \
  --duration-tolerance 0.4 --cps 14 --json

The report (a superset of the batch_summary.json shape — format/version self-describing, every list sorted, keyed by loc-ID) itemizes:

kind	flagged when
missing	a manifest row's declared audio is absent from the delivery
orphan	a delivered .wav no manifest row references
duration	actual length outside ±--duration-tolerance of the len(text)/--cps estimate (a take inside tolerance is never flagged)
empty	a zero-duration or near-silent (~0 RMS) take
naming	a delivered file whose stem doesn't match the loc-ID convention
unreadable	a file wav_duration cannot parse

plus a language-coverage matrix ({loc-ID: {locale: present}}) surfacing per-locale holes. The audit command exits nonzero when any issue is found (a CI QA gate) and prints a human worst-first table, or the full JSON report with --json. Audio paths resolve relative to --delivered (the delivery root). Deterministic; stdlib + numpy (RMS only).

openfacefx.vo_audit

VO delivery QA at scale (issue #42): reconcile a delivered folder against the loc-table manifest.

Localization vendors run a pre-delivery QA pass that reconciles the delivered VO against the script — missing lines, orphan files, naming-convention violations, empty takes, and script<->audio duration mismatches. This is that pass as deterministic set/arithmetic over file stats + the #40 manifest — no ML, and read-only over the delivered folder (it only walks it and reads WAV headers; it never writes there). It is the reconciliation pair to :mod:openfacefx.batch_manifest: it shares that module's :func:~openfacefx.batch_manifest.read_manifest parser and reuses :func:openfacefx.pipeline.wav_duration for stats.

:func:audit_delivery returns a deterministic, itemized report — a superset of the batch_summary.json shape (format/version self-describing, stable key order, every list sorted) — keyed by loc-ID:

• missing — a manifest row whose declared audio is absent from the delivery;
• orphan — a delivered .wav that no manifest row references;
• duration — actual length outside a configurable tolerance of the length estimated from the transcript (len(text) / cps seconds);
• empty — a zero-duration or near-silent (~0 RMS) take;
• naming — a delivered file whose stem does not match the loc-ID convention;
• unreadable — a file wav_duration cannot parse.

plus a language-coverage matrix ({loc-ID: {locale: present}}) that surfaces per-locale holes. Audio paths in the manifest are resolved relative to the delivered folder (the delivery root). Deterministic and stdlib + numpy (RMS only).

AUDIT_FORMAT = 'openfacefx.vo_audit' module-attribute

AUDIT_VERSION = 1 module-attribute

EMPTY_DURATION = 0.02 module-attribute

SILENT_RMS = 0.0001 module-attribute

DEFAULT_CPS = 14.0 module-attribute

DEFAULT_TOLERANCE = 0.5 module-attribute

audit_delivery(manifest_path: str, delivered_dir: str, *, duration_tolerance: float = DEFAULT_TOLERANCE, cps: float = DEFAULT_CPS) -> Dict

Reconcile delivered_dir against the loc-table manifest_path.

Returns the deterministic audit report dict. duration_tolerance is the fraction a take may differ from its len(text)/cps estimate before it is flagged — a take inside [expected*(1-tol), expected*(1+tol)] is never a duration issue. Nothing under delivered_dir is written.

Source code in src/openfacefx/vo_audit.py

def audit_delivery(manifest_path: str, delivered_dir: str, *,
                   duration_tolerance: float = DEFAULT_TOLERANCE,
                   cps: float = DEFAULT_CPS) -> Dict:
    """Reconcile ``delivered_dir`` against the loc-table ``manifest_path``.

    Returns the deterministic audit report dict. ``duration_tolerance`` is the
    fraction a take may differ from its ``len(text)/cps`` estimate before it is
    flagged — a take inside ``[expected*(1-tol), expected*(1+tol)]`` is never a
    duration issue. Nothing under ``delivered_dir`` is written."""
    rows = read_manifest(manifest_path)
    delivered = set(_scan_wavs(delivered_dir))
    languages = sorted({(r.get("language") or "") for r in rows})
    referenced: set = set()
    coverage: Dict[str, Dict[str, bool]] = {}
    issues: List[Dict] = []

    for i, row in enumerate(rows):
        loc = row.get("id") or (
            os.path.splitext(os.path.basename(row["audio"]))[0]
            if row.get("audio") else "row%d" % (i + 1))
        lang = row.get("language") or ""
        audio = row.get("audio")
        present = False
        if not audio:
            issues.append(dict(id=loc, language=lang, kind="missing", audio=None,
                               detail="no audio path declared in the manifest"))
        else:
            rel = _norm_rel(audio)
            referenced.add(rel)
            full = os.path.join(delivered_dir, audio)
            if not os.path.isfile(full):
                issues.append(dict(id=loc, language=lang, kind="missing",
                                   audio=rel, detail="declared audio not delivered"))
            else:
                present = _audit_present(loc, lang, rel, full, row, cps,
                                         duration_tolerance, issues)
        cov = coverage.setdefault(loc, {})
        cov[lang] = cov.get(lang, False) or present

    for f in sorted(delivered - referenced):
        issues.append(dict(id=None, language=None, kind="orphan", audio=f,
                           detail="delivered audio has no manifest key"))

    issues.sort(key=lambda it: (_SEVERITY.get(it["kind"], 9), it.get("id") or "",
                                it.get("language") or "", it.get("audio") or ""))
    matrix = {loc: {lang: coverage[loc].get(lang, False) for lang in languages}
              for loc in sorted(coverage)}
    counts = dict(rows=len(rows), delivered=len(delivered),
                  referenced=len(referenced), issues=len(issues))
    for kind in _SEVERITY:
        counts[kind] = sum(1 for it in issues if it["kind"] == kind)
    return dict(format=AUDIT_FORMAT, version=AUDIT_VERSION,
                manifest=manifest_path, delivered=delivered_dir,
                cps=cps, duration_tolerance=duration_tolerance,
                counts=counts, languages=languages, coverage=matrix,
                issues=issues)

audit_report_text(report: Dict) -> str

A human, worst-first summary of an :func:audit_delivery report: the itemized issues then the language-coverage matrix (present * / hole .).

Source code in src/openfacefx/vo_audit.py

def audit_report_text(report: Dict) -> str:
    """A human, worst-first summary of an :func:`audit_delivery` report: the
    itemized issues then the language-coverage matrix (present ``*`` / hole ``.``)."""
    c = report["counts"]
    lines = ["VO delivery audit: %d row(s), %d delivered file(s), %d issue(s)"
             % (c["rows"], c["delivered"], c["issues"])]
    if c["issues"]:
        lines.append("  " + "  ".join("%s=%d" % (k, c[k]) for k in _SEVERITY
                                      if c[k]))
    for it in report["issues"]:
        who = it.get("id") or "-"
        lang = (" [%s]" % it["language"]) if it.get("language") else ""
        lines.append("  %-9s %s%s  %s" % (it["kind"], who, lang, it["detail"]))
    langs = report["languages"]
    if langs and report["coverage"]:
        lines.append("\ncoverage  " + "  ".join(langs))
        for loc, cov in report["coverage"].items():
            cells = "  ".join(("*" if cov[la] else ".").center(max(len(la), 1))
                              for la in langs)
            lines.append("  %-16s %s" % (loc, cells))
    return "\n".join(lines)