#!/usr/bin/env python3 """Independent verifier of the audio renditions corpus (extensions/packaging.md §8.10). Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and never runs it. It takes from the second implementation of the text this section builds on what that text defines: a description record's reading, signature and rules from the access-services corpus's (`../live-conformance/access_services_verify.py`, access-services §2 and §5.2). The entries' rules (§8.10.1, §8.10.2), the HLS tags (§8.10.3) and the DASH adaptation sets (§8.10.4) are written here. For every case it requires the outcome the corpus states, line for line; that every reason of §8.10.6's table appears; then it mutates copies of the corpus, one defect each, and requires every copy to fail; and it drops each of its own checks in turn and requires the corpus to catch the weakened verifier. python3 renditions_verify.py Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier is caught. Python 3.8 or later and the standard library. """ import copy import hashlib import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "..", "live-conformance")) sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance")) import access_services_verify as AV # noqa: E402 EXTENSION = os.path.join(HERE, "..", "extensions", "packaging.md") MEMBERS = {"bandwidth", "capability", "channels", "codecs", "default", "description", "group", "id", "language", "name", "role", "uri"} ROLES = {"main": "main", "alternate": "alternate", "dub": "dub", "commentary": "commentary", "description": "alternate"} CODECS = { "wai.audio.aac": ["mp4a.40.2", "mp4a.40.5", "mp4a.40.29"], "wai.audio.opus.packet": ["opus"], "wai.audio.flac": ["fLaC"], "wai.audio.ac3": ["ac-3"], "wai.audio.eac3": ["ec-3"], } IDENT = re.compile(r"[A-Za-z0-9._-]{1,64}") NAME = re.compile(r"[A-Za-z0-9 ._()-]{1,64}") URI = re.compile(r"[A-Za-z0-9._/:~%-]{1,1024}") HEX32 = re.compile(r"[0-9a-f]{64}") LANG = re.compile(r"[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*") DESCRIBES = "public.accessibility.describes-video" MACHINE = "public.machine-generated" # A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything. DROPPED = None def drops(name): return DROPPED == name def reasons_table(): """§8.10.6's codes, in order.""" text = open(EXTENSION, encoding="utf-8").read() at = text.index("\n**8.10.6**") ends = [e for e in (text.find("\n**8.", at + 1), text.find("\n## ", at + 1)) if e >= 0] rows = text[at:min(ends) if ends else len(text)].splitlines() return [r.split("`")[1] for r in rows if r.startswith("| `")] class Refused(Exception): def __init__(self, reason): super().__init__(reason) self.reason = reason def is_int(v): return isinstance(v, int) and not isinstance(v, bool) def entry(v): """§8.10.1: an entry's shape.""" bad = Refused("rendition-malformed") if not isinstance(v, dict) or set(v) != MEMBERS: raise bad strs = ("id", "group", "name", "language", "role", "capability", "codecs", "uri", "default") if not all(isinstance(v[k], str) for k in strs): raise bad if not (IDENT.fullmatch(v["id"]) and IDENT.fullmatch(v["group"]) and URI.fullmatch(v["uri"])): raise bad if not NAME.fullmatch(v["name"]) or v["name"][0] == " " or v["name"][-1] == " ": raise bad if not (2 <= len(v["language"]) <= 35 and LANG.fullmatch(v["language"])) and not drops("language"): raise bad if v["role"] not in ROLES or v["default"] not in ("yes", "no"): raise bad if not (is_int(v["channels"]) and 1 <= v["channels"] <= 32) and not drops("channels"): raise bad if not (is_int(v["bandwidth"]) and 1 <= v["bandwidth"] <= 0xFFFFFFFF): raise bad if v["description"] is not None and not (isinstance(v["description"], str) and HEX32.fullmatch(v["description"])): raise bad return dict(v, default=v["default"] == "yes") def autoselect(e): return e["role"] in ("main", "dub", "description") def characteristics(e, machine): if e["role"] != "description": return [] return [DESCRIBES, MACHINE] if machine and not drops("machine") else [DESCRIBES] def write(entries, records): """§8.10.2 to §8.10.4: (hls, dash), or Refused.""" es = [entry(v) for v in entries] if not es: raise Refused("rendition-malformed") for e in es: if e["capability"] not in CODECS: raise Refused("rendition-capability") if e["codecs"] not in CODECS[e["capability"]] and not drops("codecs"): raise Refused("rendition-codecs") if not drops("duplicate"): for key in (lambda e: e["id"], lambda e: e["uri"], lambda e: (e["group"], e["name"])): keys = [key(e) for e in es] if len(set(keys)) != len(keys): raise Refused("rendition-duplicate") seen = {} for e in es: if e["default"]: seen[e["group"]] = seen.get(e["group"], 0) + 1 if (e["role"] != "main" or seen[e["group"]] > 1) and not drops("default"): raise Refused("rendition-default") texts = {} for t in records: try: texts[hashlib.sha256(AV.jcs(json.loads(t))).hexdigest()] = t except ValueError: pass described = {} for e in es: if (e["role"] == "description") != (e["description"] is not None) and not drops("bound"): raise Refused("rendition-description-unbound") d = e["description"] if d is None: continue if d not in described: if d not in texts: raise Refused("rendition-description-unbound") try: kind, v, _ = AV.read_record(texts[d]) if kind != "wai-description": raise AV.Refused("record_malformed") if not drops("record_rules"): AV.description_rules(v) except AV.Refused: if not drops("record"): raise Refused("rendition-description-invalid") v = json.loads(texts[d]) described[d] = (v["lang"], v["voicing"]["origin"] != "human") if described[d][0] != e["language"] and not drops("record_language"): raise Refused("rendition-description-language") def machine(e): return e["description"] is not None and described[e["description"]][1] if not drops("ambiguous"): keys = [(e["group"], e["language"], tuple(characteristics(e, machine(e)))) for e in es if autoselect(e)] if len(set(keys)) != len(keys): raise Refused("rendition-ambiguous") groups = [] for e in es: if e["group"] not in groups: groups.append(e["group"]) def members(g): return {e["name"]: (e["language"], e["role"], e["default"], e["description"]) for e in es if e["group"] == g} if not drops("groups") and any(members(g) != members(groups[0]) for g in groups[1:]): raise Refused("rendition-groups-differ") hls, dash = [], [] for k, e in enumerate(es): ch = characteristics(e, machine(e)) t = ('#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="{}",NAME="{}",LANGUAGE="{}",DEFAULT={},AUTOSELECT={},CHANNELS="{}"' .format(e["group"], e["name"], e["language"], "YES" if e["default"] else "NO", "YES" if autoselect(e) or drops("autoselect") else "NO", e["channels"])) if ch: t += ',CHARACTERISTICS="{}"'.format(",".join(ch)) if e["description"] is not None: t += ',X-WAI-DESCRIPTION="{}"'.format(e["description"]) hls.append(t + ',URI="{}"'.format(e["uri"])) role = e["role"] if drops("role_map") else ROLES[e["role"]] dash.append(''.format(k + 2, e["language"], e["codecs"])) dash.append(' '.format(role)) if e["description"] is not None and not drops("accessibility"): dash.append(' ') dash.append(' '.format(e["description"])) dash.append(' '.format(e["channels"])) dash.append(' '.format(e["id"], e["bandwidth"])) dash.append("") return hls, dash def run(c): try: hls, dash = write(c["entries"], c["records"]) except Refused as r: return {"refuse": r.reason} return {"hls": hls, "dash": dash} def check(doc, quiet=False): fails = [] given = set() for c in doc["cases"]: try: got = run(c) except (KeyError, TypeError, ValueError, IndexError) as e: got = {"error": repr(e)} if got != c["outcome"]: fails.append(c["name"]) if not quiet: print(f"FAIL {c['name']}\n stated {json.dumps(c['outcome'])}\n got {json.dumps(got)}") if "refuse" in c["outcome"]: given.add(c["outcome"]["refuse"]) for code in reasons_table(): if code not in given: fails.append("no case gives " + code) return fails def mutations(doc): def case(d, name): return next(c for c in d["cases"] if c["name"] == name) def m_autoselect(d): c = case(d, "languages") c["outcome"]["hls"][2] = c["outcome"]["hls"][2].replace("AUTOSELECT=NO", "AUTOSELECT=YES") def m_role(d): c = case(d, "two_codecs_described") c["outcome"]["dash"] = [x.replace('value="alternate"', 'value="description"') for x in c["outcome"]["dash"]] def m_channels(d): case(d, "two_codecs_described")["entries"][2]["channels"] = 8 def m_record_lang(d): c = case(d, "two_codecs_described") rec = json.loads(c["records"][0]) rec["lang"] = "en-GB" c["records"][0] = json.dumps(rec, sort_keys=True, separators=(",", ":")) def m_machine(d): c = case(d, "machine_voice") c["outcome"]["hls"][1] = c["outcome"]["hls"][1].replace(",public.machine-generated", "") def m_codecs(d): case(d, "alternate")["entries"][0]["codecs"] = "ec-3" def m_order(d): c = case(d, "languages") c["entries"].reverse() def m_reason_gone(d): d["cases"].remove(case(d, "groups_language_differs")) d["cases"].remove(case(d, "groups_member_missing")) def m_reason_swapped(d): case(d, "record_language")["outcome"]["refuse"] = "rendition-description-invalid" return [m_autoselect, m_role, m_channels, m_record_lang, m_machine, m_codecs, m_order, m_reason_gone, m_reason_swapped] WEAKENINGS = ["language", "channels", "codecs", "duplicate", "default", "bound", "record", "record_rules", "record_language", "ambiguous", "groups", "machine", "autoselect", "role_map", "accessibility"] def main(argv): global DROPPED if len(argv) != 2: print(__doc__) return 2 doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8")) extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"}) fails = check(doc) if extra: fails.append(f"files the corpus does not hold: {extra}") print(f"{len(doc['cases'])} cases: {len(fails)} failure(s)") caught = 0 for m in mutations(doc): d = copy.deepcopy(doc) m(d) if check(d, quiet=True): caught += 1 else: fails.append(f"mutation {m.__name__} was not caught") print(f"{caught} of {len(mutations(doc))} mutated copies fail") weak = 0 for w in WEAKENINGS: DROPPED = w if check(doc, quiet=True): weak += 1 else: fails.append(f"a verifier without its {w} check passes the corpus") DROPPED = None print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught") for f in fails: print("FAIL", f) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main(sys.argv))