6e747e412a
Written from the extension's text: records read strictly and their JCS signatures, words and their alignment, the caption figures, description rules and DAPT documents (parsed with xml.etree), signing, reviews and claims; every mutated copy of the corpus must fail. The extension now links synthesis, merged on main, for the origin of a synthesized voice or signer and for its review vocabulary. Co-Authored-By: Claude Opus 5.5 <[email protected]>
990 lines
43 KiB
Python
990 lines
43 KiB
Python
#!/usr/bin/env python3
|
||
"""Independent verifier of the access-services corpus (extensions/access-services.md).
|
||
|
||
Written from the extension's text. It shares no code with wai-rs, which generates the
|
||
corpus, and never runs it. It verifies every event and the captions in it with
|
||
timed_text_verify.py (this directory), the timed-text corpus's second implementation, which
|
||
verifies events with live_events_verify.py; checks every record's signature with
|
||
../claim-conformance/_ed25519.py, WAI's strict Ed25519 rule written from RFC 8032; and writes
|
||
everything else from the extension's sections:
|
||
|
||
- the records (§2): a strict JSON reading, each record's shape from the tables, its JCS
|
||
form, its signature and its record id;
|
||
- words and their alignment (§3), with the walk back of §3.2;
|
||
- caption quality (§4): the tracks and cues, sources placed through a take's opening
|
||
commitment, reference transcripts, latency, stability and accuracy, and the check of a
|
||
stated figure;
|
||
- audio description (§5): the rules, the synchronisation bounds, the overlap with dialogue,
|
||
and the DAPT document, written byte for byte and parsed with xml.etree;
|
||
- sign language (§6): the rules, the binding to final revisions, offsets and the video
|
||
floor, and reviews;
|
||
- the completeness claim (§7) and verification (§8).
|
||
|
||
For every vector it requires the outcome, place, reason and figures the corpus states; for
|
||
every document, its bytes or its refusal; that every reason of §10 (read from the extension)
|
||
refuses a case; and that every mutated copy of the corpus fails.
|
||
|
||
python3 -I access_services_verify.py <live-conformance/access>
|
||
|
||
Exit status 0 only if every case verifies. Python 3.8 or later; standard library only.
|
||
"""
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import xml.etree.ElementTree as ET
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
import timed_text_verify as TT # noqa: E402
|
||
import live_events_verify as LEV # noqa: E402
|
||
import _ed25519 as ED # noqa: E402
|
||
|
||
EXTENSION = os.path.join(HERE, "..", "extensions", "access-services.md")
|
||
SIGN_DOMAIN = b"wai:access\x01"
|
||
MAX_SAFE = (1 << 53) - 1
|
||
MAX_MS = 359999999
|
||
NS_WAI = "tag:transaction.science,2026:wai/access"
|
||
NS_TT = "http://www.w3.org/ns/ttml"
|
||
NS_TTA = "http://www.w3.org/ns/ttml#audio"
|
||
NS_TTP = "http://www.w3.org/ns/ttml#parameter"
|
||
NS_DAPTM = "http://www.w3.org/ns/ttml/profile/dapt#metadata"
|
||
NS_XML = "http://www.w3.org/XML/1998/namespace"
|
||
DAPT_CONTENT = "http://www.w3.org/ns/ttml/profile/dapt1.0/content"
|
||
CAPTION, CONTROL, TAKE_COMMIT = 0x0A, 0x09, 0x0C
|
||
|
||
|
||
class Refused(Exception):
|
||
def __init__(self, reason):
|
||
super().__init__(reason)
|
||
self.reason = reason
|
||
|
||
|
||
def reasons_of_extension():
|
||
"""The codes of §10's table, in order."""
|
||
text = open(EXTENSION, encoding="utf-8").read()
|
||
at = text.index("\n## 10. Reasons")
|
||
end = text.find("\n## ", at + 1)
|
||
body = text[at:end if end >= 0 else len(text)]
|
||
return [l.split("`")[1] for l in body.split("\n") if l.startswith("| `")]
|
||
|
||
|
||
# ---------------------------------------------------------------- §2 records
|
||
|
||
class _Bad:
|
||
"""A JSON number this extension's records may not hold."""
|
||
|
||
|
||
BAD = _Bad()
|
||
|
||
|
||
def _no_dupes(pairs):
|
||
out = {}
|
||
for k, v in pairs:
|
||
if k in out:
|
||
raise Refused("record_malformed")
|
||
out[k] = v
|
||
return out
|
||
|
||
|
||
def _clean(v, depth):
|
||
if depth > 32:
|
||
raise Refused("record_malformed")
|
||
if isinstance(v, str):
|
||
if any(0xD800 <= ord(c) <= 0xDFFF for c in v):
|
||
raise Refused("record_malformed")
|
||
elif isinstance(v, list):
|
||
for x in v:
|
||
_clean(x, depth + 1)
|
||
elif isinstance(v, dict):
|
||
for k, x in v.items():
|
||
_clean(k, depth)
|
||
_clean(x, depth + 1)
|
||
|
||
|
||
def read_json(text):
|
||
"""jwp-receipts §3.1 step 1's reading: no BOM, no repeated member, no lone surrogate, at
|
||
most 32 levels. A number with a sign, a fraction or an exponent is kept as BAD."""
|
||
if text.startswith(""):
|
||
raise Refused("record_malformed")
|
||
try:
|
||
v = json.loads(text, object_pairs_hook=_no_dupes,
|
||
parse_int=lambda s: BAD if s.startswith("-") else int(s),
|
||
parse_float=lambda s: BAD,
|
||
parse_constant=lambda s: BAD)
|
||
except Refused:
|
||
raise
|
||
except ValueError:
|
||
raise Refused("record_malformed")
|
||
_clean(v, 1)
|
||
return v
|
||
|
||
|
||
HEX = re.compile(r"[0-9a-f]*")
|
||
SINT = re.compile(r"0|-?[1-9][0-9]*")
|
||
|
||
|
||
def is_hex(v, n):
|
||
return isinstance(v, str) and len(v) == 2 * n and HEX.fullmatch(v) is not None
|
||
|
||
|
||
def is_int(v):
|
||
return type(v) is int and 0 <= v <= MAX_SAFE
|
||
|
||
|
||
def is_sint(v):
|
||
return isinstance(v, str) and SINT.fullmatch(v) is not None and -(1 << 63) <= int(v) < (1 << 63)
|
||
|
||
|
||
# A type: "hex32", "hex64", "int", "sint", "str", ("one", [...]), ("obj", table),
|
||
# ("arr", type), ("null", type).
|
||
ORIGINS = ["human", "machine", "machine-reviewed"]
|
||
REPRESENTS = ["visual", "visual.nonText", "visual.text", "visual.text.title", "visual.text.credit",
|
||
"visual.text.location", "visual.dialogue"]
|
||
SIG = [("signer_pubkey", "hex32"), ("sig", "hex64")]
|
||
SOURCES = [("kind", ("one", ["wai-caption-sources"])), ("room", "hex32"), ("principal", "hex32"),
|
||
("sources", ("arr", ("obj", [("cue", "hex32"), ("basis", ("one", ["take", "room"])), ("take", ("null", "hex32")),
|
||
("start", "int"), ("end", "int")])))] + SIG
|
||
REFERENCE = [("kind", ("one", ["wai-caption-reference"])), ("room", "hex32"), ("principal", "hex32"),
|
||
("entries", ("arr", ("obj", [("cue", "hex32"), ("text", "str")])))] + SIG
|
||
COUNTS = ("obj", [("words", "int"), ("errors", "int"), ("substitutions", "int"), ("deletions", "int"), ("insertions", "int")])
|
||
QUALITY = [("kind", ("one", ["wai-caption-quality"])), ("room", "hex32"), ("principal", "hex32"), ("events", "hex32"),
|
||
("sources", ("null", "hex32")), ("reference", ("null", "hex32")), ("cues", "int"),
|
||
("latency", ("obj", [("class", ("one", ["none", "capture", "stated"])), ("n", "int")] +
|
||
[(k, ("null", "sint")) for k in ("min", "max", "mean", "p50", "p95")])),
|
||
("stability", ("obj", [("revisions", "int"), ("words_changed", "int"), ("cues_corrected", "int")])),
|
||
("accuracy", ("obj", [("class", ("one", ["transcript", "reviewed", "self"])), ("first", COUNTS), ("final", ("null", COUNTS))]))] + SIG
|
||
DESCRIPTION = [("kind", ("one", ["wai-description"])), ("lang", "str"),
|
||
("script", ("obj", [("origin", ("one", ORIGINS)), ("model", ("null", "hex32"))])),
|
||
("voicing", ("obj", [("form", ("one", ["track", "sink"])), ("origin", ("one", ORIGINS)),
|
||
("model", ("null", "hex32")), ("track", ("null", "hex32"))])),
|
||
("events", ("arr", ("obj", [("id", "str"), ("begin", "int"), ("end", "int"), ("represents", ("one", REPRESENTS)),
|
||
("text", "str"), ("duck", ("obj", [("gain", "int"), ("ramp", "int")])),
|
||
("voice", ("null", ("obj", [("begin", "int"), ("end", "int")])))])))] + SIG
|
||
SIGNING = [("kind", ("one", ["wai-signing"])), ("room", "hex32"), ("lang", "str"), ("source", "hex32"),
|
||
("form", ("one", ["pose", "video"])), ("track", "hex32"), ("origin", ("one", ORIGINS)), ("model", ("null", "hex32")),
|
||
("video", ("null", ("obj", [("width", "int"), ("height", "int"), ("rate_num", "int"), ("rate_den", "int")]))),
|
||
("segments", ("arr", ("obj", [("object", "hex32"), ("start", "int"), ("end", "int"), ("revisions", ("arr", "hex32"))])))] + SIG
|
||
REVIEW = [("kind", ("one", ["wai-signing-review"])), ("signing", "hex32"), ("scope", ("one", ["translation"])),
|
||
("verdict", ("one", ["approved", "approved-with-notes", "rejected"])), ("notes", ("null", "hex32"))] + SIG
|
||
TRACKS = {
|
||
"captions": [("service", ("one", ["captions"])), ("principal", "hex32"), ("lang", "str"),
|
||
("latency_mean_max", ("null", "int")), ("latency_max", ("null", "int"))],
|
||
"audio-description": [("service", ("one", ["audio-description"])), ("record", "hex32"), ("lang", "str")],
|
||
"sign-language": [("service", ("one", ["sign-language"])), ("record", "hex32"), ("lang", "str"),
|
||
("review", ("null", "hex32")), ("offset_max", ("null", "int"))],
|
||
}
|
||
CLAIM = [("kind", ("one", ["wai-access-completeness"])), ("room", "hex32"), ("from", "int"), ("to", "int"), ("pt0", "int"),
|
||
("tracks", "tracks")] + SIG
|
||
TABLES = {"wai-caption-sources": SOURCES, "wai-caption-reference": REFERENCE, "wai-caption-quality": QUALITY,
|
||
"wai-description": DESCRIPTION, "wai-signing": SIGNING, "wai-signing-review": REVIEW, "wai-access-completeness": CLAIM}
|
||
|
||
|
||
def has_type(v, t):
|
||
if v is BAD:
|
||
return False
|
||
if t == "hex32":
|
||
return is_hex(v, 32)
|
||
if t == "hex64":
|
||
return is_hex(v, 64)
|
||
if t == "int":
|
||
return is_int(v)
|
||
if t == "sint":
|
||
return is_sint(v)
|
||
if t == "str":
|
||
return isinstance(v, str)
|
||
if t == "tracks":
|
||
return isinstance(v, list) and all(isinstance(x, dict) and x.get("service") in TRACKS and fits(x, TRACKS[x["service"]]) for x in v)
|
||
tag, arg = t
|
||
if tag == "one":
|
||
return isinstance(v, str) and v in arg
|
||
if tag == "obj":
|
||
return fits(v, arg)
|
||
if tag == "arr":
|
||
return isinstance(v, list) and all(has_type(x, arg) for x in v)
|
||
if tag == "null":
|
||
return v is None or has_type(v, arg)
|
||
raise AssertionError(t)
|
||
|
||
|
||
def fits(v, table):
|
||
return isinstance(v, dict) and len(v) == len(table) and all(k in v and has_type(v[k], t) for k, t in table)
|
||
|
||
|
||
def jcs(v):
|
||
"""RFC 8785 over strings, unsigned integers, null, arrays and objects with ASCII member
|
||
names, which is what these records hold."""
|
||
return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
||
|
||
|
||
def read_record(text):
|
||
"""§2: (kind, value, record id)."""
|
||
v = read_json(text)
|
||
if not isinstance(v, dict) or v.get("kind") not in TABLES or not fits(v, TABLES[v["kind"]]):
|
||
raise Refused("record_malformed")
|
||
body = {k: x for k, x in v.items() if k != "sig"}
|
||
bad = ED.verify(bytes.fromhex(v["signer_pubkey"]), SIGN_DOMAIN + jcs(body), bytes.fromhex(v["sig"]))
|
||
if bad is not None:
|
||
raise Refused(bad)
|
||
return v["kind"], v, hashlib.sha256(jcs(v)).digest()
|
||
|
||
|
||
# ---------------------------------------------------------------- §3 words and alignment
|
||
|
||
EDGE = "!\"'(),-.:;?[]¡«»¿–—‘’“”…"
|
||
UPPER = {c: c + 0x20 for c in range(0x41, 0x5B)}
|
||
|
||
|
||
def words(text):
|
||
out = []
|
||
for piece in re.split("[ \n]", text):
|
||
if piece == "":
|
||
continue
|
||
w = piece.translate(UPPER).strip(EDGE)
|
||
if w:
|
||
out.append(w)
|
||
return out
|
||
|
||
|
||
def align(h, r, prefix):
|
||
"""§3.2: (words, errors, substitutions, deletions, insertions)."""
|
||
m, n = len(h), len(r)
|
||
d = [[0] * (n + 1) for _ in range(m + 1)]
|
||
for j in range(n + 1):
|
||
d[0][j] = j
|
||
for i in range(m + 1):
|
||
d[i][0] = i
|
||
for i in range(1, m + 1):
|
||
for j in range(1, n + 1):
|
||
c = 0 if h[i - 1] == r[j - 1] else 1
|
||
d[i][j] = min(d[i - 1][j - 1] + c, d[i - 1][j] + 1, d[i][j - 1] + 1)
|
||
if prefix:
|
||
least = min(d[m])
|
||
end = max(j for j in range(n + 1) if d[m][j] == least)
|
||
else:
|
||
end = n
|
||
s = de = ins = 0
|
||
i, j = m, end
|
||
while i > 0 or j > 0:
|
||
if i > 0 and j > 0:
|
||
c = 0 if h[i - 1] == r[j - 1] else 1
|
||
if d[i][j] == d[i - 1][j - 1] + c:
|
||
s += c
|
||
i, j = i - 1, j - 1
|
||
continue
|
||
if i > 0 and d[i][j] == d[i - 1][j] + 1:
|
||
ins += 1
|
||
i -= 1
|
||
else:
|
||
de += 1
|
||
j -= 1
|
||
assert s + de + ins == d[m][end]
|
||
return [end, d[m][end], s, de, ins]
|
||
|
||
|
||
def counts_json(c):
|
||
return {"words": c[0], "errors": c[1], "substitutions": c[2], "deletions": c[3], "insertions": c[4]}
|
||
|
||
|
||
def add(a, b):
|
||
return [x + y for x, y in zip(a, b)]
|
||
|
||
|
||
# ---------------------------------------------------------------- §4 the caption tracks
|
||
|
||
class Stream:
|
||
def __init__(self, room, cap):
|
||
self.room, self.revs, self.accepted, self.gaps = room, cap.revs, cap.ev.accepted, cap.ev.gaps
|
||
|
||
def principals(self):
|
||
out = []
|
||
for r in self.revs:
|
||
if r["event"]["principal"] not in out:
|
||
out.append(r["event"]["principal"])
|
||
return out
|
||
|
||
def cues(self, p):
|
||
"""§4.1: each cue as the positions of its revisions, first to last."""
|
||
out = []
|
||
for i, r in enumerate(self.revs):
|
||
if r["event"]["principal"] != p or r["sup"] is not None or r["sup_digest"] is not None:
|
||
continue
|
||
chain = [i]
|
||
while self.revs[chain[-1]]["by"] is not None:
|
||
chain.append(self.revs[chain[-1]]["by"])
|
||
out.append(chain)
|
||
return out
|
||
|
||
def head(self, c):
|
||
return self.revs[c[0]]["event"]["digest"]
|
||
|
||
def last(self, c):
|
||
return self.revs[c[-1]]
|
||
|
||
def whole(self, p):
|
||
cues = self.cues(p)
|
||
return (len(cues) > 0 and not any(g[0] == p for g in self.gaps)
|
||
and all(self.last(c)["final"] and not self.last(c)["unknown"] for c in cues))
|
||
|
||
def resolved(self, p):
|
||
return self.whole(p) and all(self.last(c)["duration"] > 0 for c in self.cues(p))
|
||
|
||
def events_digest(self, p):
|
||
mine = sorted((r for r in self.revs if r["event"]["principal"] == p), key=lambda r: r["event"]["seq"])
|
||
return hashlib.sha256(b"".join(r["event"]["digest"] for r in mine)).digest()
|
||
|
||
def take_offset(self, take):
|
||
"""§4.2 rule 6: the offset of the take's opening commitment, when the take is derived."""
|
||
opening = None
|
||
for _, ev, p in self.accepted:
|
||
if ev["kind"] != TAKE_COMMIT:
|
||
continue
|
||
c = read_commit(p["bytes"])
|
||
if c is not None and c["take"] == take and c["n"] == 0 and c["phase"] == 1:
|
||
opening = (ev["principal"], c["offset"])
|
||
break
|
||
if opening is None:
|
||
return None
|
||
author, offset = opening
|
||
for _, ev, p in self.accepted:
|
||
if ev["kind"] == CONTROL and p["command"] == 1 and p["target"] == author:
|
||
if hashlib.sha256(b"wai:capture-take\x01" + self.room + author + ev["digest"]).digest() == take:
|
||
return offset
|
||
return None
|
||
|
||
def intervals(self, p, pt0):
|
||
out = []
|
||
for c in self.cues(p):
|
||
h, l = self.revs[c[0]], self.last(c)
|
||
out.append((pt0 + h["event"]["t"] // 1000, pt0 + l["event"]["t"] // 1000 + l["duration"]))
|
||
return out
|
||
|
||
|
||
def read_commit(b):
|
||
"""capture-binding §5.1 and §5.2: a take_commit payload, or None when it does not read."""
|
||
if len(b) < 126:
|
||
return None
|
||
take, n, phase = b[0:32], int.from_bytes(b[32:36], "little"), b[36]
|
||
first, size = int.from_bytes(b[69:77], "little"), int.from_bytes(b[77:85], "little")
|
||
root, offset, k = b[85:117], int.from_bytes(b[117:125], "little", signed=True), b[125]
|
||
if phase not in (1, 2, 3) or size < first or k > 64 or len(b) != 126 + 32 * k or (size == 0 and root != bytes(32)):
|
||
return None
|
||
return {"take": take, "n": n, "phase": phase, "offset": offset}
|
||
|
||
|
||
def floor_div(a, n):
|
||
return a // n
|
||
|
||
|
||
class Run:
|
||
def __init__(self, stream, records):
|
||
self.s, self.records = stream, records # records: [(kind, value, id)]
|
||
self.sources = {} # principal -> (id, {cue: (start, on_take)})
|
||
self.references = {} # principal -> (id, {cue: text})
|
||
|
||
def room_ok(self, v):
|
||
if bytes.fromhex(v["room"]) != self.s.room:
|
||
raise Refused("wrong_room")
|
||
|
||
def latencies(self, p, cues):
|
||
src = self.sources.get(p)
|
||
if src is None:
|
||
return []
|
||
out = []
|
||
for c in cues:
|
||
placed = src[1].get(self.s.head(c))
|
||
if placed is not None:
|
||
out.append((self.s.revs[c[0]]["event"]["t"] - placed[0], placed[1]))
|
||
return out
|
||
|
||
def figure(self, p):
|
||
s = self.s
|
||
cues = s.cues(p)
|
||
lat = sorted(x[0] for x in self.latencies(p, cues))
|
||
n = len(lat)
|
||
if n == 0:
|
||
latency = {"class": "none", "n": 0, "min": None, "max": None, "mean": None, "p50": None, "p95": None}
|
||
else:
|
||
def rank(q):
|
||
return lat[-(-q * n // 100) - 1]
|
||
on_take = all(x[1] for x in self.latencies(p, cues))
|
||
latency = {"class": "capture" if on_take else "stated", "n": n, "min": str(lat[0]), "max": str(lat[-1]),
|
||
"mean": str(floor_div(sum(lat), n)), "p50": str(rank(50)), "p95": str(rank(95))}
|
||
revisions = changed_total = corrected = 0
|
||
for c in cues:
|
||
revisions += len(c) - 1
|
||
changed = 0
|
||
for a, b in zip(c, c[1:]):
|
||
al = align(words(s.revs[a]["text"]), words(s.revs[b]["text"]), False)
|
||
changed += al[2] + al[4]
|
||
changed_total += changed
|
||
corrected += 1 if changed > 0 else 0
|
||
ref = self.references.get(p)
|
||
if ref is not None:
|
||
cls = "transcript"
|
||
elif all(s.last(c)["origin"] in ("human", "machine-reviewed") for c in cues):
|
||
cls = "reviewed"
|
||
else:
|
||
cls = "self"
|
||
first, final = [0] * 5, [0] * 5
|
||
for c in cues:
|
||
rt = ref[1][s.head(c)] if ref is not None else s.last(c)["text"]
|
||
rw = words(rt)
|
||
first = add(first, align(words(s.revs[c[0]]["text"]), rw, True))
|
||
if ref is not None:
|
||
final = add(final, align(words(s.last(c)["text"]), rw, False))
|
||
src = self.sources.get(p)
|
||
return {"room": s.room.hex(), "principal": p.hex(), "events": s.events_digest(p).hex(),
|
||
"sources": None if src is None else src[0].hex(), "reference": None if ref is None else ref[0].hex(),
|
||
"cues": len(cues), "latency": latency,
|
||
"stability": {"revisions": revisions, "words_changed": changed_total, "cues_corrected": corrected},
|
||
"accuracy": {"class": cls, "first": counts_json(first), "final": counts_json(final) if ref is not None else None}}
|
||
|
||
def heads(self, p):
|
||
return [self.s.head(c) for c in self.s.cues(p)]
|
||
|
||
def earlier(self, at, kind, p):
|
||
return any(k == kind and bytes.fromhex(v["principal"]) == p for k, v, _ in self.records[:at])
|
||
|
||
# §4.2
|
||
def sources_rec(self, at):
|
||
_, v, rid = self.records[at]
|
||
self.room_ok(v)
|
||
p = bytes.fromhex(v["principal"])
|
||
if self.earlier(at, "wai-caption-sources", p):
|
||
raise Refused("record_conflict")
|
||
seen = set()
|
||
for e in v["sources"]:
|
||
if (e["basis"] == "take") != (e["take"] is not None) or e["start"] > e["end"] or e["cue"] in seen:
|
||
raise Refused("record_malformed")
|
||
seen.add(e["cue"])
|
||
if not self.s.whole(p):
|
||
raise Refused("stream_incomplete")
|
||
heads = self.heads(p)
|
||
if any(bytes.fromhex(e["cue"]) not in heads for e in v["sources"]):
|
||
raise Refused("source_unknown")
|
||
starts = {}
|
||
for e in v["sources"]:
|
||
if e["basis"] == "take":
|
||
o = self.s.take_offset(bytes.fromhex(e["take"]))
|
||
if o is None or e["start"] + o < 0:
|
||
raise Refused("source_take_unknown")
|
||
starts[bytes.fromhex(e["cue"])] = (e["start"] + o, True)
|
||
else:
|
||
starts[bytes.fromhex(e["cue"])] = (e["start"], False)
|
||
self.sources.setdefault(p, (rid, starts))
|
||
|
||
# §4.3
|
||
def reference_rec(self, at):
|
||
_, v, rid = self.records[at]
|
||
self.room_ok(v)
|
||
p = bytes.fromhex(v["principal"])
|
||
if self.earlier(at, "wai-caption-reference", p):
|
||
raise Refused("record_conflict")
|
||
texts = {}
|
||
for e in v["entries"]:
|
||
if e["cue"] in texts:
|
||
raise Refused("record_malformed")
|
||
texts[e["cue"]] = e["text"]
|
||
if not all(TT.text_ok(t) for t in texts.values()):
|
||
raise Refused("text_not_in_subset")
|
||
if not self.s.whole(p):
|
||
raise Refused("stream_incomplete")
|
||
heads = self.heads(p)
|
||
if any(bytes.fromhex(k) not in heads for k in texts):
|
||
raise Refused("reference_unknown")
|
||
if any(h.hex() not in texts for h in heads):
|
||
raise Refused("reference_incomplete")
|
||
self.references.setdefault(p, (rid, {bytes.fromhex(k): t for k, t in texts.items()}))
|
||
|
||
# §4.7
|
||
def quality_rec(self, at):
|
||
_, v, _ = self.records[at]
|
||
self.room_ok(v)
|
||
p = bytes.fromhex(v["principal"])
|
||
if not self.s.whole(p):
|
||
raise Refused("stream_incomplete")
|
||
f = self.figure(p)
|
||
if any(v[k] != f[k] for k in ("events", "sources", "reference")):
|
||
raise Refused("figure_inputs_mismatch")
|
||
if any(v[k] != x for k, x in f.items()):
|
||
raise Refused("figure_mismatch")
|
||
|
||
# §6.2
|
||
def signing_rec(self, at):
|
||
_, v, _ = self.records[at]
|
||
self.room_ok(v)
|
||
machine = v["origin"] != "human"
|
||
ok = TT.lang_ok(v["lang"]) and machine == (v["model"] is not None) and (v["form"] == "video") == (v["video"] is not None)
|
||
if v["video"] is not None:
|
||
ok = ok and all(v["video"][k] > 0 for k in ("width", "height", "rate_num", "rate_den"))
|
||
segs = v["segments"]
|
||
ok = ok and len(segs) > 0
|
||
named = []
|
||
for i, sg in enumerate(segs):
|
||
ok = ok and sg["start"] < sg["end"] and len(sg["revisions"]) > 0 and (i == 0 or sg["start"] >= segs[i - 1]["end"])
|
||
named += sg["revisions"]
|
||
ok = ok and len(set(named)) == len(named)
|
||
if not ok:
|
||
raise Refused("signing_malformed")
|
||
src = bytes.fromhex(v["source"])
|
||
if not self.s.whole(src):
|
||
raise Refused("stream_incomplete")
|
||
mine = {r["event"]["digest"]: r for r in self.s.revs if r["event"]["principal"] == src}
|
||
if any(bytes.fromhex(d) not in mine for d in named):
|
||
raise Refused("sign_revision_unknown")
|
||
for d in named:
|
||
r = mine[bytes.fromhex(d)]
|
||
if r["by"] is not None or not r["final"] or r["unknown"]:
|
||
raise Refused("sign_revision_not_final")
|
||
|
||
def segment_offset(self, src, sg):
|
||
cues = [c for c in self.s.cues(src) if self.s.last(c)["event"]["digest"].hex() in sg["revisions"]]
|
||
if not cues:
|
||
return None
|
||
earliest = min(cues, key=lambda c: (self.s.revs[c[0]]["event"]["t"], c[0]))
|
||
placed = self.sources.get(src, (None, {}))[1].get(self.s.head(earliest))
|
||
return None if placed is None else sg["start"] - placed[0]
|
||
|
||
# §6.4
|
||
def review_rec(self, at):
|
||
_, v, _ = self.records[at]
|
||
if (v["verdict"] == "approved") != (v["notes"] is None):
|
||
raise Refused("record_malformed")
|
||
if not any(k == "wai-signing" and rid.hex() == v["signing"] for k, _, rid in self.records):
|
||
raise Refused("review_subject_unknown")
|
||
|
||
def find(self, kind, rid):
|
||
return next((v for k, v, i in self.records if k == kind and i.hex() == rid), None)
|
||
|
||
# §7.2
|
||
def claim_rec(self, at):
|
||
_, v, _ = self.records[at]
|
||
self.room_ok(v)
|
||
frm, to, pt0, tracks = v["from"], v["to"], v["pt0"], v["tracks"]
|
||
ok = frm < to and len(tracks) > 0
|
||
principals, recs = [], []
|
||
for t in tracks:
|
||
ok = ok and TT.lang_ok(t["lang"])
|
||
if t["service"] == "captions":
|
||
ok = ok and t["principal"] not in principals
|
||
principals.append(t["principal"])
|
||
else:
|
||
ok = ok and t["record"] not in recs
|
||
recs.append(t["record"])
|
||
if not ok:
|
||
raise Refused("claim_malformed")
|
||
|
||
def session(p):
|
||
return [c for c in self.s.cues(p) if frm <= self.s.revs[c[0]]["event"]["t"] < to]
|
||
|
||
m_from, m_to = pt0 + frm // 1000, pt0 + to // 1000
|
||
for t in tracks:
|
||
if t["service"] == "captions":
|
||
p = bytes.fromhex(t["principal"])
|
||
cues = session(p)
|
||
if not cues:
|
||
raise Refused("track_missing")
|
||
if not self.s.resolved(p):
|
||
raise Refused("track_incomplete")
|
||
if any(self.s.revs[r]["lang"] != t["lang"] for c in cues for r in c):
|
||
raise Refused("track_mismatch")
|
||
if t["latency_mean_max"] is not None or t["latency_max"] is not None:
|
||
lat = [x[0] for x in self.latencies(p, cues)]
|
||
if len(lat) != len(cues):
|
||
raise Refused("sync_unmeasured")
|
||
if (t["latency_mean_max"] is not None and floor_div(sum(lat), len(lat)) > t["latency_mean_max"]) or \
|
||
(t["latency_max"] is not None and any(x > t["latency_max"] for x in lat)):
|
||
raise Refused("out_of_sync")
|
||
elif t["service"] == "audio-description":
|
||
d = self.find("wai-description", t["record"])
|
||
if d is None:
|
||
raise Refused("track_missing")
|
||
if d["lang"] != t["lang"] or any(e["begin"] < m_from or e["end"] > m_to for e in d["events"]):
|
||
raise Refused("track_mismatch")
|
||
if d["voicing"]["form"] == "track" and any(e["voice"] is None for e in d["events"]):
|
||
raise Refused("track_incomplete")
|
||
else:
|
||
g = self.find("wai-signing", t["record"])
|
||
if g is None:
|
||
raise Refused("track_missing")
|
||
if g["lang"] != t["lang"] or g["source"] not in principals:
|
||
raise Refused("track_mismatch")
|
||
src = bytes.fromhex(g["source"])
|
||
named = {d for sg in g["segments"] for d in sg["revisions"]}
|
||
if any(self.s.last(c)["event"]["digest"].hex() not in named for c in session(src)):
|
||
raise Refused("track_incomplete")
|
||
if t["review"] is not None:
|
||
r = self.find("wai-signing-review", t["review"])
|
||
if r is None:
|
||
raise Refused("review_missing")
|
||
if r["signing"] != t["record"]:
|
||
raise Refused("review_mismatch")
|
||
if r["verdict"] == "rejected":
|
||
raise Refused("review_rejected")
|
||
if t["offset_max"] is not None:
|
||
offs = [self.segment_offset(src, sg) for sg in g["segments"]]
|
||
if any(o is None for o in offs):
|
||
raise Refused("sync_unmeasured")
|
||
if any(abs(o) > t["offset_max"] for o in offs):
|
||
raise Refused("out_of_sync")
|
||
|
||
def overlap(self, v, rid):
|
||
dialogue = []
|
||
for t in v["tracks"]:
|
||
if t["service"] == "captions":
|
||
dialogue += self.s.intervals(bytes.fromhex(t["principal"]), v["pt0"])
|
||
union = []
|
||
for a, b in sorted(dialogue):
|
||
if union and a <= union[-1][1]:
|
||
union[-1][1] = max(union[-1][1], b)
|
||
else:
|
||
union.append([a, b])
|
||
out = []
|
||
for i, t in enumerate(v["tracks"]):
|
||
if t["service"] != "audio-description":
|
||
continue
|
||
d = self.find("wai-description", t["record"])
|
||
ms = n = 0
|
||
for e in d["events"]:
|
||
if e["voice"] is not None:
|
||
a, b = e["voice"]["begin"], e["voice"]["end"]
|
||
else:
|
||
a, b = e["begin"] + e["duck"]["ramp"], e["end"] - e["duck"]["ramp"]
|
||
o = sum(max(0, min(b, y) - max(a, x)) for x, y in union)
|
||
ms += o
|
||
n += 1 if o > 0 else 0
|
||
out.append({"track": i, "ms": ms, "events": n})
|
||
return {"record": rid.hex(), "overlap": out}
|
||
|
||
|
||
# ---------------------------------------------------------------- §5 audio description
|
||
|
||
ID = re.compile(r"[A-Za-z_][A-Za-z0-9_.\-]{0,63}")
|
||
|
||
|
||
def description_rules(v):
|
||
"""§5.2, in order."""
|
||
m = "description_malformed"
|
||
if not TT.lang_ok(v["lang"]):
|
||
raise Refused(m)
|
||
sc, vo = v["script"], v["voicing"]
|
||
sink = vo["form"] == "sink"
|
||
if (sc["origin"] != "human") != (sc["model"] is not None) or (vo["origin"] != "human") != (vo["model"] is not None) \
|
||
or sink == (vo["track"] is not None) or (sink and vo["origin"] == "human"):
|
||
raise Refused(m)
|
||
evs = v["events"]
|
||
if not evs:
|
||
raise Refused(m)
|
||
ids = set()
|
||
for i, e in enumerate(evs):
|
||
if ID.fullmatch(e["id"]) is None or e["id"] in ids or not e["begin"] < e["end"] <= MAX_MS or \
|
||
(i > 0 and e["begin"] < evs[i - 1]["end"]):
|
||
raise Refused(m)
|
||
ids.add(e["id"])
|
||
if not all(TT.text_ok(e["text"]) for e in evs):
|
||
raise Refused("text_not_in_subset")
|
||
for e in evs:
|
||
g, r = e["duck"]["gain"], e["duck"]["ramp"]
|
||
if not (g <= 1000 and (g == 1000) == (r == 0) and 2 * r <= e["end"] - e["begin"]):
|
||
raise Refused(m)
|
||
vc = e["voice"]
|
||
if vc is not None and (sink or not e["begin"] <= vc["begin"] < vc["end"] <= e["end"]):
|
||
raise Refused(m)
|
||
for e in evs:
|
||
vc = e["voice"]
|
||
if vc is None:
|
||
continue
|
||
a, z = e["begin"] + e["duck"]["ramp"], e["end"] - e["duck"]["ramp"]
|
||
if not (a - 35 <= vc["begin"] <= a + 45 and vc["end"] <= z + 45):
|
||
raise Refused("description_out_of_sync")
|
||
|
||
|
||
def description_figures(v, rid):
|
||
offs = [None if e["voice"] is None else str(e["voice"]["begin"] - (e["begin"] + e["duck"]["ramp"])) for e in v["events"]]
|
||
return {"record": rid.hex(), "events": len(v["events"]), "voiced": sum(1 for e in v["events"] if e["voice"] is not None),
|
||
"offsets": offs}
|
||
|
||
|
||
PARAM = re.compile(r"[!#-%'-;=?-~]{1,2048}")
|
||
|
||
|
||
def hhmmss(ms):
|
||
return "%02d:%02d:%02d.%03d" % (ms // 3600000, ms // 60000 % 60, ms // 1000 % 60, ms % 1000)
|
||
|
||
|
||
def gain(g):
|
||
if g == 0:
|
||
return "0"
|
||
if g == 1000:
|
||
return "1"
|
||
return "0." + ("%03d" % g).rstrip("0")
|
||
|
||
|
||
def esc(s):
|
||
return s.replace("&", "&").replace("<", "<").replace(">", ">").replace("\n", "<br/>")
|
||
|
||
|
||
def dapt(v, uri, typ):
|
||
"""§5.5's document."""
|
||
track = v["voicing"]["form"] == "track"
|
||
if track and not (PARAM.fullmatch(uri or "") and PARAM.fullmatch(typ or "")):
|
||
raise Refused("parameter_malformed")
|
||
reps = []
|
||
for e in v["events"]:
|
||
if e["represents"] not in reps:
|
||
reps.append(e["represents"])
|
||
|
||
def label(o, name):
|
||
s = ' wai:%s="%s"' % (name, o["origin"])
|
||
if o["model"] is not None:
|
||
s += ' wai:%s-model="%s"' % (name, o["model"])
|
||
return s
|
||
|
||
out = ['<?xml version="1.0" encoding="UTF-8"?>',
|
||
'<tt xmlns="%s" xmlns:ttp="%s" xmlns:tta="%s" xmlns:daptm="%s" xmlns:wai="%s" xml:lang="%s" ttp:contentProfiles="%s" '
|
||
'daptm:langSrc="zxx" daptm:scriptRepresents="%s" daptm:scriptType="%s"%s%s>'
|
||
% (NS_TT, NS_TTP, NS_TTA, NS_DAPTM, NS_WAI, v["lang"], DAPT_CONTENT, " ".join(reps),
|
||
"asRecorded" if track else "preRecording", label(v["script"], "script"), label(v["voicing"], "voice")),
|
||
"<body>"]
|
||
for e in v["events"]:
|
||
b, en, g, r = e["begin"], e["end"], e["duck"]["gain"], e["duck"]["ramp"]
|
||
out.append('<div xml:id="%s" begin="%s" end="%s" daptm:represents="%s">' % (e["id"], hhmmss(b), hhmmss(en), e["represents"]))
|
||
if g < 1000:
|
||
out.append('<animate begin="00:00:00.000" end="%s" tta:gain="1;%s" fill="freeze"/>' % (hhmmss(r), gain(g)))
|
||
out.append('<animate begin="%s" end="%s" tta:gain="%s;1"/>' % (hhmmss(en - b - r), hhmmss(en - b), gain(g)))
|
||
text = esc(e["text"])
|
||
if track and e["voice"] is not None:
|
||
vb, ve = e["voice"]["begin"], e["voice"]["end"]
|
||
content = '<span begin="%s" end="%s"><audio src="%s" type="%s" clipBegin="%s" clipEnd="%s"/>%s</span>' % (
|
||
hhmmss(vb - b), hhmmss(ve - b), uri, typ, hhmmss(vb), hhmmss(ve), text)
|
||
elif track:
|
||
content = text
|
||
else:
|
||
content = '<span tta:speak="normal">%s</span>' % text
|
||
out.append("<p>%s</p>" % content)
|
||
out.append("</div>")
|
||
out += ["</body>", "</tt>"]
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
def dapt_structure(doc, v):
|
||
"""Parse the document and check what DAPT §4 and §5 ask of it."""
|
||
root = ET.fromstring(doc.encode("utf-8"))
|
||
q = lambda ns, n: "{%s}%s" % (ns, n) # noqa: E731
|
||
fails = []
|
||
if root.tag != q(NS_TT, "tt"):
|
||
fails.append("the root is not tt")
|
||
if root.get(q(NS_TTP, "contentProfiles")) != DAPT_CONTENT or root.get(q(NS_TTP, "profile")) is not None:
|
||
fails.append("the profile is not signalled as DAPT §5.6 asks")
|
||
if root.get(q(NS_DAPTM, "scriptType")) not in ("preRecording", "asRecorded") or not root.get(q(NS_DAPTM, "scriptRepresents")):
|
||
fails.append("no script type or script represents")
|
||
if root.get(q(NS_XML, "lang")) != v["lang"]:
|
||
fails.append("xml:lang is not the record's")
|
||
reps = root.get(q(NS_DAPTM, "scriptRepresents"), "").split(" ")
|
||
divs = root.findall(q(NS_TT, "body") + "/" + q(NS_TT, "div"))
|
||
if len(divs) != len(v["events"]):
|
||
fails.append("not one Script Event per description")
|
||
for div, e in zip(divs, v["events"]):
|
||
if div.get(q(NS_XML, "id")) != e["id"] or div.get(q(NS_DAPTM, "represents")) not in reps:
|
||
fails.append("Script Event %s: its id or represents" % e["id"])
|
||
anim = div.findall(q(NS_TT, "animate"))
|
||
if len(anim) != (2 if e["duck"]["gain"] < 1000 else 0):
|
||
fails.append("Script Event %s: its mixing instructions" % e["id"])
|
||
ps = div.findall(q(NS_TT, "p"))
|
||
if len(ps) != 1:
|
||
fails.append("Script Event %s: not one Text" % e["id"])
|
||
continue
|
||
if "".join(ps[0].itertext()) != e["text"].replace("\n", ""):
|
||
fails.append("Script Event %s: its text" % e["id"])
|
||
audio = ps[0].findall(".//" + q(NS_TT, "audio"))
|
||
spoken = [s for s in ps[0].iter(q(NS_TT, "span")) if s.get(q(NS_TTA, "speak")) is not None]
|
||
want_audio = v["voicing"]["form"] == "track" and e["voice"] is not None
|
||
if (len(audio) == 1) != want_audio or (len(spoken) == 1) != (v["voicing"]["form"] == "sink"):
|
||
fails.append("Script Event %s: its audio" % e["id"])
|
||
return fails
|
||
|
||
|
||
# ---------------------------------------------------------------- §8 verification
|
||
|
||
KIND_ORDER = ["wai-caption-sources", "wai-caption-reference", "wai-description", "wai-signing", "wai-signing-review",
|
||
"wai-caption-quality", "wai-access-completeness"]
|
||
|
||
|
||
def verify(room, events, texts):
|
||
out, cap = TT.verify_events(room, events)
|
||
if out["outcome"] == "refuse":
|
||
return {"outcome": "refuse", "at": "events", "index": out["index"], "reason": out["reason"]}
|
||
stream = Stream(room, cap)
|
||
records = []
|
||
for i, t in enumerate(texts):
|
||
try:
|
||
records.append(read_record(t))
|
||
except Refused as e:
|
||
return {"outcome": "refuse", "at": "record", "index": i, "reason": e.reason}
|
||
run = Run(stream, records)
|
||
checks = {"wai-caption-sources": run.sources_rec, "wai-caption-reference": run.reference_rec,
|
||
"wai-description": lambda i: description_rules(records[i][1]), "wai-signing": run.signing_rec,
|
||
"wai-signing-review": run.review_rec, "wai-caption-quality": run.quality_rec,
|
||
"wai-access-completeness": run.claim_rec}
|
||
for kind in KIND_ORDER:
|
||
for i, (k, _, _) in enumerate(records):
|
||
if k != kind:
|
||
continue
|
||
try:
|
||
checks[kind](i)
|
||
except Refused as e:
|
||
return {"outcome": "refuse", "at": "record", "index": i, "reason": e.reason}
|
||
signings = []
|
||
for k, v, rid in records:
|
||
if k != "wai-signing":
|
||
continue
|
||
src = bytes.fromhex(v["source"])
|
||
offs = [run.segment_offset(src, sg) for sg in v["segments"]]
|
||
floor = None
|
||
if v["video"] is not None:
|
||
w, h = v["video"]["width"], v["video"]["height"]
|
||
floor = "meets" if min(w, h) >= 240 and max(w, h) >= 320 and v["video"]["rate_num"] >= 20 * v["video"]["rate_den"] else "below"
|
||
signings.append({"record": rid.hex(), "segments": len(v["segments"]), "revisions": sum(len(s["revisions"]) for s in v["segments"]),
|
||
"offsets": [None if o is None else str(o) for o in offs], "video_floor": floor})
|
||
figures = {
|
||
"captions": [run.figure(p) for p in stream.principals() if stream.whole(p)],
|
||
"descriptions": [description_figures(v, rid) for k, v, rid in records if k == "wai-description"],
|
||
"signings": signings,
|
||
"claims": [run.overlap(v, rid) for k, v, rid in records if k == "wai-access-completeness"],
|
||
}
|
||
return {"outcome": "accept", "figures": figures}
|
||
|
||
|
||
# ---------------------------------------------------------------- the corpus
|
||
|
||
def check(doc, files, quiet=False, only=None):
|
||
"""Every vector and document, or only those `only` names (a mutation's), with the corpus's
|
||
coverage of §10 checked when all are."""
|
||
fails, seen = [], set()
|
||
room = bytes.fromhex(doc["room"])
|
||
streams = {n: [bytes.fromhex(h) for h in evs] for n, evs in doc["streams"].items()}
|
||
texts_of = {}
|
||
for v in doc["vectors"]:
|
||
name = v["name"]
|
||
texts_of[name] = v["records"]
|
||
if only is not None and name not in only:
|
||
continue
|
||
before = len(fails)
|
||
try:
|
||
got = verify(room, streams[v["stream"]], v["records"])
|
||
except Exception as e: # a verifier that cannot read a case fails it
|
||
fails.append("%s: does not verify: %s: %s" % (name, type(e).__name__, e))
|
||
continue
|
||
for k in ("outcome", "at", "index", "reason", "figures"):
|
||
if got.get(k) != v.get(k):
|
||
fails.append("%s: %s is %s, the corpus states %s" % (name, k, json.dumps(got.get(k))[:300], json.dumps(v.get(k))[:300]))
|
||
if got.get("reason"):
|
||
seen.add(got["reason"])
|
||
if not quiet:
|
||
print("%s %s: %s" % ("PASS" if len(fails) == before else "FAIL", name, got["outcome"] + (" " + got["reason"] if got.get("reason") else "")))
|
||
for d in doc["documents"]:
|
||
if only is not None and d.get("file", d["vector"]) not in only:
|
||
continue
|
||
before = len(fails)
|
||
try:
|
||
_, v, _ = read_record(texts_of[d["vector"]][d["record"]])
|
||
doc_text, why = dapt(v, d["uri"], d["type"]), None
|
||
except Refused as e:
|
||
doc_text, why = None, e.reason
|
||
if why is not None:
|
||
seen.add(why)
|
||
if d.get("reason") != why:
|
||
fails.append("document of %s: refused %s, the corpus states %s" % (d["vector"], why, d.get("reason")))
|
||
elif doc_text is not None:
|
||
if files.get(d["file"]) != doc_text:
|
||
fails.append("document of %s: %s differs from the text's" % (d["vector"], d["file"]))
|
||
else:
|
||
fails += ["document %s: %s" % (d["file"], f) for f in dapt_structure(doc_text, v)]
|
||
if not quiet:
|
||
print("%s document of %s: %s" % ("PASS" if len(fails) == before else "FAIL", d["vector"], d.get("file") or why))
|
||
missing = [r for r in reasons_of_extension() if r not in seen]
|
||
if missing and only is None:
|
||
fails.append("nothing is refused %s" % ", ".join(missing))
|
||
return fails
|
||
|
||
|
||
def mutations(doc, files):
|
||
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
||
out = []
|
||
|
||
def mutate(label, f, only):
|
||
d, fs = copy.deepcopy(doc), dict(files)
|
||
f(d, fs)
|
||
out.append((label, d, fs, only))
|
||
|
||
def flip_sig(text):
|
||
v = json.loads(text)
|
||
v["sig"] = v["sig"][:-1] + ("1" if v["sig"][-1] == "0" else "0")
|
||
return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||
|
||
for i, v in enumerate(doc["vectors"]):
|
||
if v["outcome"] == "accept":
|
||
def stated(d, fs, i=i):
|
||
d["vectors"][i]["figures"]["captions"][0]["latency"]["n"] += 1
|
||
mutate("%s: a stated figure changed" % v["name"], stated, {v["name"]})
|
||
for k in range(len(v["records"])):
|
||
def sig(d, fs, i=i, k=k):
|
||
d["vectors"][i]["records"][k] = flip_sig(d["vectors"][i]["records"][k])
|
||
mutate("%s: record %d's signature flipped" % (v["name"], k), sig, {v["name"]})
|
||
for k in range(len(doc["streams"][v["stream"]])):
|
||
def ev(d, fs, i=i, k=k):
|
||
name = "mutated"
|
||
evs = list(d["streams"][d["vectors"][i]["stream"]])
|
||
b = bytearray(bytes.fromhex(evs[k]))
|
||
b[-1] ^= 0x01
|
||
evs[k] = b.hex()
|
||
d["streams"][name] = evs
|
||
d["vectors"][i]["stream"] = name
|
||
mutate("%s: event %d altered" % (v["name"], k), ev, {v["name"]})
|
||
else:
|
||
def place(d, fs, i=i):
|
||
d["vectors"][i]["index"] += 1
|
||
mutate("%s: the place of its refusal moved" % v["name"], place, {v["name"]})
|
||
|
||
def reason(d, fs, i=i):
|
||
d["vectors"][i]["reason"] = "record_malformed" if d["vectors"][i]["reason"] != "record_malformed" else "bad_key"
|
||
mutate("%s: its reason changed" % v["name"], reason, {v["name"]})
|
||
for d0 in doc["documents"]:
|
||
if "file" in d0:
|
||
def doc_byte(d, fs, f=d0["file"]):
|
||
fs[f] = fs[f].replace("<p>", "<p> ", 1)
|
||
mutate("%s: a byte added" % d0["file"], doc_byte, {d0["file"]})
|
||
return out
|
||
|
||
|
||
def main(corpus_dir):
|
||
files = {}
|
||
for n in sorted(os.listdir(corpus_dir)):
|
||
if n != "README.md":
|
||
with open(os.path.join(corpus_dir, n), encoding="utf-8") as f:
|
||
files[n] = f.read()
|
||
doc = json.loads(files["vectors.json"])
|
||
fails = check(doc, files)
|
||
for f in fails:
|
||
print("FAIL " + f)
|
||
caught = 0
|
||
muts = mutations(doc, files)
|
||
for label, d, fs, only in muts:
|
||
if check(d, fs, quiet=True, only=only):
|
||
caught += 1
|
||
else:
|
||
fails.append("a mutated copy verifies: " + label)
|
||
print("FAIL a mutated copy verifies: " + label)
|
||
print("%s every mutated copy of the corpus fails (%d of %d)" % ("PASS" if caught == len(muts) else "FAIL", caught, len(muts)))
|
||
print("%s %s" % ("PASS" if not fails else "FAIL", "the corpus verifies" if not fails else "%d failures" % len(fails)))
|
||
return 0 if not fails else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
sys.exit("usage: access_services_verify.py <live-conformance/access>")
|
||
sys.exit(main(sys.argv[1]))
|