1614 lines
67 KiB
Python
1614 lines
67 KiB
Python
#!/usr/bin/env python3
|
||
"""Independent verifier of the timed-text corpus (extensions/timed-text.md).
|
||
|
||
Written from the extension's text. It shares no code with wai-rs, which generates the
|
||
corpus, and never runs it. It reads and verifies every event record with
|
||
live_events_verify.py (this directory), the second implementation of live-events, and
|
||
writes everything else from the extension's sections:
|
||
|
||
- the cue model and its subset (§3);
|
||
- revisions and their annotation (§4.2, §4.3), what a sink shows (§4.4), the cues a
|
||
stream resolves to (§4.5), and the join set of the events track (§4.6);
|
||
- WebVTT documents, segments and the reader of the subset (§5);
|
||
- IMSC1 documents (§6.1), the check of the hypothetical render model (§6.2), and the
|
||
structure of every document checked with xml.etree (§6.3);
|
||
- the MoQ caption timelines and their catalog entries (§7);
|
||
- CTA-608: the character tables, the layout, pop-on and roll-up, the byte pairs, and a
|
||
decoder (§8);
|
||
- the disclosure label: HLS tags, C2PA statement, sidecar (§10).
|
||
|
||
For every vector it requires the outcome, reason, index, revisions, what is shown, the cues
|
||
and every byte of every output the corpus states; that every reason of §12 and every 608
|
||
code path of §8 appears; and that every mutated copy of the corpus fails.
|
||
|
||
python3 timed_text_verify.py <live-conformance/captions>
|
||
|
||
Exit status 0 only if every case verifies. Python 3.8 or later; standard library only.
|
||
"""
|
||
import copy
|
||
import json
|
||
import os
|
||
import sys
|
||
import xml.etree.ElementTree as ET
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
import live_events_verify as LEV # noqa: E402
|
||
|
||
# §12, in its order.
|
||
REASONS = [
|
||
"time_out_of_range", "cue_malformed", "model_missing", "text_not_in_subset",
|
||
"language_mismatch", "annotation_malformed", "annotation_misplaced", "supersede_unknown",
|
||
"supersede_final", "supersede_forked", "supersede_mismatch", "stability_broken",
|
||
"not_final", "duration_unknown", "vtt_malformed", "hrm_exceeded", "parameter_malformed",
|
||
"cues_overlap", "too_many_rows", "row_too_long", "char_not_in_608",
|
||
"style_not_expressible", "too_dense", "cc_malformed",
|
||
]
|
||
|
||
LAST_MS = 359999999
|
||
ORIGINS = {1: "human", 2: "machine", 3: "machine-reviewed"}
|
||
ANCHORS = ["bottom-center", "bottom-left", "bottom-right", "top-center", "top-left", "top-right"]
|
||
NS_WAI = "tag:transaction.science,2026:wai/timed-text"
|
||
NS_TT = "http://www.w3.org/ns/ttml"
|
||
NS_TTS = "http://www.w3.org/ns/ttml#styling"
|
||
NS_TTP = "http://www.w3.org/ns/ttml#parameter"
|
||
NS_XML = "http://www.w3.org/XML/1998/namespace"
|
||
NS_EBUTTM = "urn:ebu:tt:metadata"
|
||
PROFILES = ["http://www.w3.org/ns/ttml/profile/imsc1/text", "http://www.w3.org/ns/ttml/profile/imsc1.3/text"]
|
||
DST = "http://cv.iptc.org/newscodes/digitalsourcetype/"
|
||
MODEL_URI = "tag:transaction.science,2026:wai/model/"
|
||
CAPTION, META = 0x0A, 0x01
|
||
GROUP_US = 60000000 # live-events §8.1
|
||
|
||
# live-events §5's default policy: every kind of §4 but the lifecycle ones.
|
||
_kinds = bytearray(32)
|
||
for _k in range(0x01, 0x0D):
|
||
_kinds[_k // 8] |= 1 << (_k % 8)
|
||
DEFAULT_POLICY = {"kinds": bytes(_kinds), "max_payload": 1024, "max_text_cp": 500, "max_events": 50,
|
||
"window_us": 1000000, "max_reactions": 1, "reaction_window_us": 1000000}
|
||
|
||
|
||
class Refused(Exception):
|
||
def __init__(self, reason):
|
||
super().__init__(reason)
|
||
self.reason = reason
|
||
|
||
|
||
# ---------------------------------------------------------------- §3 the cue model
|
||
|
||
def align(anchor):
|
||
return anchor.split("-")[1]
|
||
|
||
|
||
def is_top(anchor):
|
||
return anchor.startswith("top")
|
||
|
||
|
||
def text_ok(s):
|
||
"""§3.2 rules 1 to 3."""
|
||
if s == "":
|
||
return False
|
||
for line in s.split("\n"):
|
||
if line == "" or line[0] == " " or line[-1] == " " or " " in line:
|
||
return False
|
||
for ch in line:
|
||
v = ord(ch)
|
||
inside = 0x20 <= v <= 0x7E or 0xA0 <= v <= 0xD7FF or 0xE000 <= v <= 0xFFFD or 0x10000 <= v <= 0x10FFFF
|
||
if not inside or v in (0x2028, 0x2029, 0xFEFF):
|
||
return False
|
||
return True
|
||
|
||
|
||
def ranges_ok(s, ranges):
|
||
"""§3.2 rule 4."""
|
||
end_before = None
|
||
for start, n in ranges:
|
||
if n < 1 or start + n > len(s):
|
||
return False
|
||
if end_before is not None:
|
||
if start <= end_before or set(s[end_before:start]) <= {" "}:
|
||
return False
|
||
piece = s[start:start + n]
|
||
if "\n" in piece or piece[0] == " " or piece[-1] == " ":
|
||
return False
|
||
end_before = start + n
|
||
return True
|
||
|
||
|
||
def lang_ok(tag):
|
||
return isinstance(tag, str) and 2 <= len(tag) <= 35 and LEV.LANG.fullmatch(tag) is not None
|
||
|
||
|
||
def check_cue(c):
|
||
"""§3.4, in its order."""
|
||
if not (c["start"] < c["end"] <= LAST_MS):
|
||
raise Refused("time_out_of_range")
|
||
if not lang_ok(c["lang"]) or c["origin"] not in ORIGINS.values():
|
||
raise Refused("cue_malformed")
|
||
if c["origin"] == "human" and c["model"] is not None:
|
||
raise Refused("cue_malformed")
|
||
if c["origin"] != "human" and c["model"] is None:
|
||
raise Refused("model_missing")
|
||
if not text_ok(c["text"]):
|
||
raise Refused("text_not_in_subset")
|
||
if not ranges_ok(c["text"], c["italic"]):
|
||
raise Refused("cue_malformed")
|
||
|
||
|
||
def check_doc(lang, cues):
|
||
if not lang_ok(lang):
|
||
raise Refused("parameter_malformed")
|
||
for c in cues:
|
||
check_cue(c)
|
||
if any(c["lang"] != lang for c in cues):
|
||
raise Refused("language_mismatch")
|
||
|
||
|
||
def in_order(cues):
|
||
return sorted(cues, key=lambda c: c["start"]) # stable: equal starts keep their order
|
||
|
||
|
||
def italic_flags(c):
|
||
flags = [False] * len(c["text"])
|
||
for start, n in c["italic"]:
|
||
for i in range(start, start + n):
|
||
flags[i] = True
|
||
return flags
|
||
|
||
|
||
def runs_of(c):
|
||
"""Each line of the cue, as [(italic, text)]."""
|
||
flags = italic_flags(c)
|
||
out, line = [], []
|
||
for i, ch in enumerate(c["text"]):
|
||
if ch == "\n":
|
||
out.append(line)
|
||
line = []
|
||
elif line and line[-1][0] == flags[i]:
|
||
line[-1] = (flags[i], line[-1][1] + ch)
|
||
else:
|
||
line.append((flags[i], ch))
|
||
out.append(line)
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- §5 WebVTT
|
||
|
||
def hhmmss(ms):
|
||
return "%02d:%02d:%02d.%03d" % (ms // 3600000, ms // 60000 % 60, ms // 1000 % 60, ms % 1000)
|
||
|
||
|
||
def settings(anchor):
|
||
line = "line:10%,start" if is_top(anchor) else "line:90%,end"
|
||
return "%s position:50%%,center size:80%% align:%s" % (line, align(anchor))
|
||
|
||
|
||
def esc(s):
|
||
return s.replace("&", "&").replace("<", "<").replace(">", ">")
|
||
|
||
|
||
def label_class(c):
|
||
return {"machine": "wai-machine", "machine-reviewed": "wai-machine-reviewed"}.get(c["origin"])
|
||
|
||
|
||
def vtt_payload(c):
|
||
lines = ["".join("<i>%s</i>" % esc(t) if it else esc(t) for it, t in line) for line in runs_of(c)]
|
||
body = "\n".join(lines)
|
||
cls = label_class(c)
|
||
if cls:
|
||
body = "<c.%s.wai-model-%s>%s</c>" % (cls, c["model"], body)
|
||
return body
|
||
|
||
|
||
def vtt_cue(c):
|
||
return "%s --> %s %s\n%s\n" % (hhmmss(c["start"]), hhmmss(c["end"]), settings(c["anchor"]), vtt_payload(c))
|
||
|
||
|
||
def webvtt(lang, cues):
|
||
check_doc(lang, cues)
|
||
return "WEBVTT\n\n" + "\n".join(vtt_cue(c) for c in in_order(cues))
|
||
|
||
|
||
def webvtt_segments(lang, cues, bounds, mpegts0):
|
||
check_doc(lang, cues)
|
||
if len(bounds) < 2 or any(a >= b for a, b in zip(bounds, bounds[1:])) or bounds[-1] > LAST_MS or mpegts0 >= 2 ** 33:
|
||
raise Refused("parameter_malformed")
|
||
out = []
|
||
for b, e in zip(bounds, bounds[1:]):
|
||
body = "\n".join(vtt_cue(c) for c in in_order(cues) if c["start"] < e and c["end"] > b)
|
||
out.append("WEBVTT\nX-TIMESTAMP-MAP=LOCAL:%s,MPEGTS:%d\n\n%s" % (hhmmss(b), (mpegts0 + 90 * b) % 2 ** 33, body))
|
||
return out
|
||
|
||
|
||
def read_hhmmss(s):
|
||
if len(s) != 12 or s[2] != ":" or s[5] != ":" or s[8] != "." or not (s[:2] + s[3:5] + s[6:8] + s[9:]).isdigit() or not s.isascii():
|
||
raise Refused("vtt_malformed")
|
||
h, m, sec, ms = int(s[:2]), int(s[3:5]), int(s[6:8]), int(s[9:])
|
||
if m > 59 or sec > 59:
|
||
raise Refused("vtt_malformed")
|
||
return ((h * 60 + m) * 60 + sec) * 1000 + ms
|
||
|
||
|
||
def unesc(s):
|
||
out, i = [], 0
|
||
while i < len(s):
|
||
if s[i] == "&":
|
||
for ref, ch in (("&", "&"), ("<", "<"), (">", ">")):
|
||
if s.startswith(ref, i):
|
||
out.append(ch)
|
||
i += len(ref)
|
||
break
|
||
else:
|
||
raise Refused("vtt_malformed")
|
||
elif s[i] in "<>":
|
||
raise Refused("vtt_malformed")
|
||
else:
|
||
out.append(s[i])
|
||
i += 1
|
||
return "".join(out)
|
||
|
||
|
||
def read_webvtt(doc, lang):
|
||
"""§5.3: (cues, (local, mpegts) or None)."""
|
||
bad = Refused("vtt_malformed")
|
||
if not doc.startswith("WEBVTT\n"):
|
||
raise bad
|
||
rest = doc[7:]
|
||
tsmap = None
|
||
if rest.startswith("X-TIMESTAMP-MAP=LOCAL:"):
|
||
line, nl, rest = rest[22:].partition("\n")
|
||
local, sep, mp = line.partition(",MPEGTS:")
|
||
if not nl or not sep or not mp.isdigit() or not mp.isascii() or (len(mp) > 1 and mp[0] == "0") or int(mp) >= 2 ** 33:
|
||
raise bad
|
||
tsmap = (read_hhmmss(local), int(mp))
|
||
if not rest.startswith("\n"):
|
||
raise bad
|
||
rest = rest[1:]
|
||
cues = []
|
||
blocks = rest.split("\n\n") if rest else []
|
||
for k, block in enumerate(blocks):
|
||
if not block.endswith("\n") and k == len(blocks) - 1:
|
||
raise bad
|
||
lines = block.split("\n")
|
||
if k == len(blocks) - 1:
|
||
if lines[-1] != "":
|
||
raise bad
|
||
lines = lines[:-1]
|
||
if len(lines) < 2 or any(x == "" for x in lines):
|
||
raise bad
|
||
head = lines[0].split(" ", 3)
|
||
if len(head) != 4 or head[1] != "-->":
|
||
raise bad
|
||
start, end = read_hhmmss(head[0]), read_hhmmss(head[2])
|
||
anchor = next((a for a in ANCHORS if settings(a) == head[3]), None)
|
||
if anchor is None:
|
||
raise bad
|
||
body = "\n".join(lines[1:])
|
||
origin, model = "human", None
|
||
if body.startswith("<c."):
|
||
tag, gt, inner = body[3:].partition(">")
|
||
if not gt or not inner.endswith("</c>"):
|
||
raise bad
|
||
classes = tag.split(".")
|
||
origin = {"wai-machine": "machine", "wai-machine-reviewed": "machine-reviewed"}.get(classes[0])
|
||
if origin is None:
|
||
raise bad
|
||
if len(classes) == 1:
|
||
raise Refused("model_missing")
|
||
mc = classes[1]
|
||
if len(classes) != 2 or not mc.startswith("wai-model-") or len(mc) != 74 or any(ch not in "0123456789abcdef" for ch in mc[10:]):
|
||
raise bad
|
||
model, body = mc[10:], inner[:-4]
|
||
text, italic, pos = [], [], 0
|
||
for n, line in enumerate(body.split("\n")):
|
||
if n:
|
||
text.append("\n")
|
||
pos += 1
|
||
runs, rest_line = [], line
|
||
while rest_line:
|
||
if rest_line.startswith("<i>"):
|
||
inner, sep, after = rest_line[3:].partition("</i>")
|
||
if not sep or inner == "" or "<" in inner:
|
||
raise bad
|
||
runs.append((True, unesc(inner)))
|
||
rest_line = after
|
||
else:
|
||
cut = rest_line.find("<")
|
||
cut = len(rest_line) if cut < 0 else cut
|
||
if cut == 0:
|
||
raise bad
|
||
runs.append((False, unesc(rest_line[:cut])))
|
||
rest_line = rest_line[cut:]
|
||
if any(a[0] == b[0] for a, b in zip(runs, runs[1:])):
|
||
raise bad
|
||
for it, t in runs:
|
||
if it:
|
||
if italic and italic[-1][0] + italic[-1][1] == pos:
|
||
italic[-1][1] += len(t)
|
||
else:
|
||
italic.append([pos, len(t)])
|
||
text.append(t)
|
||
pos += len(t)
|
||
cue = {"start": start, "end": end, "lang": lang, "origin": origin, "model": model,
|
||
"anchor": anchor, "text": "".join(text), "italic": italic}
|
||
if cues and cues[-1]["start"] > start:
|
||
raise bad
|
||
check_cue(cue)
|
||
cues.append(cue)
|
||
return cues, tsmap
|
||
|
||
|
||
# ---------------------------------------------------------------- §6 IMSC1
|
||
|
||
REGION_LINES = [
|
||
'<region xml:id="top" tts:origin="10% 10%" tts:extent="80% 35%" tts:displayAlign="before"/>',
|
||
'<region xml:id="bottom" tts:origin="10% 55%" tts:extent="80% 35%" tts:displayAlign="after"/>',
|
||
]
|
||
|
||
|
||
def region(anchor):
|
||
return "top" if is_top(anchor) else "bottom"
|
||
|
||
|
||
def imsc_p(c):
|
||
attrs = 'begin="%s" end="%s" region="%s" tts:textAlign="%s"' % (hhmmss(c["start"]), hhmmss(c["end"]), region(c["anchor"]), align(c["anchor"]))
|
||
if c["origin"] != "human":
|
||
attrs += ' wai:origin="%s" wai:model="%s"' % (c["origin"], c["model"])
|
||
lines = ["".join('<span tts:fontStyle="italic">%s</span>' % esc(t) if it else esc(t) for it, t in line) for line in runs_of(c)]
|
||
return "<p %s>%s</p>" % (attrs, "<br/>".join(lines))
|
||
|
||
|
||
def imsc(lang, cues):
|
||
check_doc(lang, cues)
|
||
cues = in_order(cues)
|
||
hrm(cues)
|
||
out = ['<?xml version="1.0" encoding="UTF-8"?>',
|
||
'<tt xmlns="%s" xmlns:ttp="%s" xmlns:tts="%s" xmlns:ebuttm="%s" xmlns:wai="%s" xml:lang="%s" ttp:profile="%s">'
|
||
% (NS_TT, NS_TTP, NS_TTS, NS_EBUTTM, NS_WAI, lang, PROFILES[0]),
|
||
"<head>", "<metadata>", "<ebuttm:documentMetadata>"]
|
||
out += ["<ebuttm:conformsToStandard>%s</ebuttm:conformsToStandard>" % p for p in PROFILES]
|
||
out += ["</ebuttm:documentMetadata>", "</metadata>", "<layout>"] + REGION_LINES + ["</layout>", "</head>", "<body>", "<div>"]
|
||
out += [imsc_p(c) for c in cues]
|
||
out += ["</div>", "</body>", "</tt>"]
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
def hrm(cues):
|
||
"""§6.2, in units of 1/2700 s."""
|
||
times = sorted(set([c["start"] for c in cues] + [c["end"] for c in cues]))
|
||
cache = {} # glyph -> flagged
|
||
last = None
|
||
for at, _ in zip(times, times[1:]):
|
||
live = [c for c in cues if c["start"] <= at < c["end"]]
|
||
if not live:
|
||
cache = {}
|
||
continue
|
||
budget_ms = at - last if last is not None and at - last < 1000 else 1000
|
||
cost = 225
|
||
for reg in ("top", "bottom"):
|
||
for c in live:
|
||
if region(c["anchor"]) != reg:
|
||
continue
|
||
flags = italic_flags(c)
|
||
for i, ch in enumerate(c["text"]):
|
||
if ch == "\n":
|
||
continue
|
||
g = (ch, flags[i])
|
||
cost += 4 if g in cache else 20
|
||
cache[g] = True
|
||
if sum(1 for f in cache.values() if f) > 225:
|
||
raise Refused("hrm_exceeded")
|
||
if 27 * budget_ms < 10 * cost:
|
||
raise Refused("hrm_exceeded")
|
||
cache = {g: False for g, f in cache.items() if f}
|
||
last = at
|
||
|
||
|
||
def percents(s):
|
||
"""Two lengths in percent, as numbers; None otherwise."""
|
||
parts = s.split(" ")
|
||
if len(parts) != 2 or not all(v.endswith("%") and v[:-1].replace(".", "", 1).isdigit() and v.isascii() for v in parts):
|
||
return None
|
||
return tuple(float(v[:-1]) for v in parts)
|
||
|
||
|
||
def clock(s):
|
||
return read_hhmmss(s)
|
||
|
||
|
||
def imsc_structure(doc, cues):
|
||
"""§6.3, with xml.etree: the failures."""
|
||
fails = []
|
||
if not doc.startswith('<?xml version="1.0" encoding="UTF-8"?>\n'):
|
||
fails.append("no UTF-8 XML declaration")
|
||
try:
|
||
root = ET.fromstring(doc.encode("utf-8"))
|
||
except ET.ParseError as e:
|
||
return ["not well-formed: %s" % e]
|
||
tt = lambda n: "{%s}%s" % (NS_TT, n) # noqa: E731
|
||
if root.tag != tt("tt") or not lang_ok(root.get("{%s}lang" % NS_XML)):
|
||
fails.append("the root is not tt with xml:lang")
|
||
named = [e.text for e in root.iter("{%s}conformsToStandard" % NS_EBUTTM)]
|
||
if named != PROFILES:
|
||
fails.append("the profiles named are %s" % named)
|
||
# IMSC 1.0.1 §6.9: ttp:profile on tt, the IMSC 1.0 Text Profile's designator; no
|
||
# ttp:profile element; never an Image Profile designator beside a Text Profile one.
|
||
if root.get("{%s}profile" % NS_TTP) != PROFILES[0]:
|
||
fails.append("tt's ttp:profile is %s" % root.get("{%s}profile" % NS_TTP))
|
||
if root.find(".//{%s}profile" % NS_TTP) is not None:
|
||
fails.append("a ttp:profile element is present")
|
||
if any("/image" in (d or "") for d in named + [root.get("{%s}profile" % NS_TTP)]):
|
||
fails.append("an Image Profile designator is signalled")
|
||
regions = {}
|
||
for r in root.iter(tt("region")):
|
||
rid = r.get("{%s}id" % NS_XML)
|
||
o, x = r.get("{%s}origin" % NS_TTS, ""), r.get("{%s}extent" % NS_TTS, "")
|
||
if r.get("{%s}position" % NS_TTS) is not None:
|
||
fails.append("region %s has tts:position" % rid)
|
||
po, px = percents(o), percents(x)
|
||
if po is None or px is None:
|
||
fails.append("region %s is not in percent" % rid)
|
||
continue
|
||
(ox, oy), (ex, ey) = po, px
|
||
if ox + ex > 100 or oy + ey > 100:
|
||
fails.append("region %s leaves the root container" % rid)
|
||
regions[rid] = (ox, oy, ox + ex, oy + ey)
|
||
if sorted(regions) != ["bottom", "top"]:
|
||
fails.append("the regions are %s" % sorted(regions))
|
||
rs = list(regions.values())
|
||
for i in range(len(rs)):
|
||
for j in range(i + 1, len(rs)):
|
||
a, b = rs[i], rs[j]
|
||
if a[0] < b[2] and b[0] < a[2] and a[1] < b[3] and b[1] < a[3]:
|
||
fails.append("two regions overlap")
|
||
allowed_tt = {"begin", "end", "region", "{%s}textAlign" % NS_TTS, "{%s}fontStyle" % NS_TTS}
|
||
body = root.find(tt("body"))
|
||
divs = [] if body is None else list(body)
|
||
if len(divs) != 1 or divs[0].tag != tt("div"):
|
||
fails.append("the body is not one div")
|
||
return fails
|
||
ps = list(divs[0])
|
||
if len(ps) != len(cues):
|
||
fails.append("%d p for %d cues" % (len(ps), len(cues)))
|
||
for p, c in zip(ps, in_order(cues)):
|
||
if p.tag != tt("p"):
|
||
fails.append("a child of div is not p")
|
||
continue
|
||
try:
|
||
b, e = clock(p.get("begin", "")), clock(p.get("end", ""))
|
||
if not b < e:
|
||
fails.append("a p does not begin before it ends")
|
||
except Refused:
|
||
fails.append("a p's times are not clock times")
|
||
if p.get("region") not in regions:
|
||
fails.append("a p names no region")
|
||
for k in p.attrib:
|
||
if k.startswith("{%s}" % NS_WAI):
|
||
continue
|
||
if k not in allowed_tt:
|
||
fails.append("p has attribute %s" % k)
|
||
has_label = (p.get("{%s}origin" % NS_WAI), p.get("{%s}model" % NS_WAI))
|
||
want = (c["origin"], c["model"]) if c["origin"] != "human" else (None, None)
|
||
if has_label != want:
|
||
fails.append("the label of a p is %s, its cue's %s" % (has_label, want))
|
||
for child in p.iter():
|
||
if child is p:
|
||
continue
|
||
if child.tag == tt("br"):
|
||
if child.attrib or (child.text or ""):
|
||
fails.append("a br has content")
|
||
elif child.tag == tt("span"):
|
||
if dict(child.attrib) != {"{%s}fontStyle" % NS_TTS: "italic"} or len(child):
|
||
fails.append("a span is not one italic run")
|
||
else:
|
||
fails.append("p holds %s" % child.tag)
|
||
if "".join(p.itertext()).replace("\n", "") != c["text"].replace("\n", ""):
|
||
fails.append("a p's text is not its cue's")
|
||
return fails
|
||
|
||
|
||
# ---------------------------------------------------------------- §7 timelines
|
||
|
||
def jstr(s):
|
||
out = ['"']
|
||
for ch in s:
|
||
if ch == '"':
|
||
out.append('\\"')
|
||
elif ch == "\\":
|
||
out.append("\\\\")
|
||
elif ch == "\n":
|
||
out.append("\\n")
|
||
elif ord(ch) < 0x20:
|
||
out.append("\\u%04x" % ord(ch))
|
||
else:
|
||
out.append(ch)
|
||
out.append('"')
|
||
return "".join(out)
|
||
|
||
|
||
def vtt_record(c):
|
||
line = "10%,start" if is_top(c["anchor"]) else "90%,end"
|
||
return ('{"m":%d,"data":{"start":%d,"end":%d,"text":%s,"settings":{"line":"%s","position":"50%%,center","size":"80%%","align":"%s"}}}'
|
||
% (c["start"], c["start"], c["end"], jstr(vtt_payload(c)), line, align(c["anchor"])))
|
||
|
||
|
||
IMSC_INIT = ('{"m":0,"data":{"init":{"profile":"http://www.w3.org/ns/ttml/profile/imsc1/text",'
|
||
'"styles":{"center":{"textAlign":"center"},"left":{"textAlign":"left"},"right":{"textAlign":"right"}},'
|
||
'"regions":{"bottom":{"origin":"10% 55%","extent":"80% 35%","displayAlign":"after"},'
|
||
'"top":{"origin":"10% 10%","extent":"80% 35%","displayAlign":"before"}}}}}')
|
||
|
||
|
||
def imsc_record(c):
|
||
content = []
|
||
for n, line in enumerate(runs_of(c)):
|
||
if n:
|
||
content.append('{"br":true}')
|
||
content += ['{"text":%s,"fontStyle":"italic"}' % jstr(t) if it else '{"text":%s}' % jstr(t) for it, t in line]
|
||
data = '"start":%d,"end":%d,"region":"%s","style":"%s","content":[%s]' % (c["start"], c["end"], region(c["anchor"]), align(c["anchor"]), ",".join(content))
|
||
if c["origin"] != "human":
|
||
data += ',"wai_origin":"%s","wai_model":"%s"' % (c["origin"], c["model"])
|
||
return '{"m":%d,"data":{%s}}' % (c["start"], data)
|
||
|
||
|
||
def timeline(kind, lang, published, group_ms, first, last):
|
||
check_doc(lang, [c for c, _ in published])
|
||
if group_ms < 1 or first > last or any(t < c["start"] for c, t in published):
|
||
raise Refused("parameter_malformed")
|
||
order = sorted(published, key=lambda ct: ct[1])
|
||
rec = vtt_record if kind == "webvtt" else imsc_record
|
||
groups = []
|
||
for g in range(first, last + 1):
|
||
lo, hi = g * group_ms, (g + 1) * group_ms
|
||
head = [IMSC_INIT] if kind == "imsc" else []
|
||
head += [rec(c) for c, t in order if t < lo and c["end"] > lo]
|
||
objects = ["[" + ",".join(head) + "]"] + ["[" + rec(c) + "]" for c, t in order if lo <= t < hi]
|
||
groups.append({"group": g, "objects": objects})
|
||
return groups
|
||
|
||
|
||
def catalog(kind, name, lang, depends, cues):
|
||
et, mode = ("urn:msf:captions:webvtt", "") if kind == "webvtt" else ("urn:msf:captions:imsc1", ',"imsc1Mode":"cue"')
|
||
mg = "true" if any(c["origin"] != "human" for c in cues) else "false"
|
||
return ('{"name":%s,"packaging":"eventtimeline","eventType":"%s","mimeType":"application/json","depends":[%s],"role":"caption","lang":%s%s,"wai_machine_generated":%s}'
|
||
% (jstr(name), et, ",".join(jstr(d) for d in depends), jstr(lang), mode, mg))
|
||
|
||
|
||
# ---------------------------------------------------------------- §10 the label
|
||
|
||
def attr_ok(s):
|
||
return s != "" and all(ch.isascii() and (ch.isalnum() or ch in " ._-/:") for ch in s)
|
||
|
||
|
||
def media_tag(kind, group, name, lang, chars, cues, uri=None):
|
||
check_doc(lang, cues)
|
||
if not attr_ok(group) or not attr_ok(name) or (uri is not None and not attr_ok(uri)) or any(not attr_ok(c) or "," in c or c == "public.machine-generated" for c in chars):
|
||
raise Refused("parameter_malformed")
|
||
cs = list(chars) + (["public.machine-generated"] if any(c["origin"] != "human" for c in cues) else [])
|
||
tag = '#EXT-X-MEDIA:TYPE=%s,GROUP-ID="%s",NAME="%s",LANGUAGE="%s"' % (kind, group, name, lang)
|
||
if kind == "CLOSED-CAPTIONS":
|
||
tag += ',INSTREAM-ID="CC1"'
|
||
if cs:
|
||
tag += ',CHARACTERISTICS="%s"' % ",".join(cs)
|
||
if uri is not None:
|
||
tag += ',URI="%s"' % uri
|
||
return tag
|
||
|
||
|
||
def source_type(cues):
|
||
kinds = sorted(set(c["origin"] for c in cues))
|
||
if kinds in ([], ["human"]):
|
||
return DST + "digitalCreation"
|
||
if kinds == ["machine"]:
|
||
return DST + "trainedAlgorithmicMedia"
|
||
if kinds == ["machine-reviewed"]:
|
||
return DST + "compositeWithTrainedAlgorithmicMedia"
|
||
return DST + "compositeSynthetic"
|
||
|
||
|
||
def c2pa(cues):
|
||
models = [] # [model, any machine]
|
||
for c in in_order(cues):
|
||
if c["model"] is None:
|
||
continue
|
||
hit = next((m for m in models if m[0] == c["model"]), None)
|
||
if hit is None:
|
||
models.append([c["model"], c["origin"] == "machine"])
|
||
else:
|
||
hit[1] = hit[1] or c["origin"] == "machine"
|
||
parts = []
|
||
for i, (m, auto) in enumerate(models):
|
||
label = "c2pa.ai-disclosure" if i == 0 else "c2pa.ai-disclosure__%d" % i
|
||
parts.append('{"label":"%s","data":{"modelType":"c2pa.types.model","modelIdentifier":"%s%s","contentProfile":{"humanOversightLevel":"%s"}}}'
|
||
% (label, MODEL_URI, m, "fully_autonomous" if auto else "human_validated"))
|
||
return '{"digitalSourceType":"%s","assertions":[%s]}' % (source_type(cues), ",".join(parts))
|
||
|
||
|
||
def sidecar(lang, cues):
|
||
check_doc(lang, cues)
|
||
rows = []
|
||
for c in in_order(cues):
|
||
m = ',"model":"%s"' % c["model"] if c["model"] is not None else ""
|
||
rows.append('{"start":%d,"end":%d,"origin":"%s"%s}' % (c["start"], c["end"], c["origin"], m))
|
||
return '{"wai_timed_text":1,"lang":%s,"cues":[%s]}' % (jstr(lang), ",".join(rows))
|
||
|
||
|
||
# ---------------------------------------------------------------- §8 CTA-608
|
||
|
||
SPECIAL = "®°½¿™¢£♪à èâêîôû"
|
||
EXT = {
|
||
0x12: "ÁÉÓÚÜü‘¡*\x00—©℠•“”ÀÂÇÈÊËëÎÏïÔÙùÛ«»",
|
||
0x13: "ÃãÍÌìÒòÕõ{}\\^_|~ÄäÖöߥ¤¦ÅåØø┌┐└┘",
|
||
}
|
||
SUBST = {"’": "'", "–": "-", "‐": "-", "‑": "-"}
|
||
BASIC_SWAP = {0x2A: "á", 0x5C: "é", 0x5E: "í", 0x5F: "ó", 0x60: "ú", 0x7B: "ç", 0x7C: "÷", 0x7D: "Ñ", 0x7E: "ñ"}
|
||
PAC_FIRST = {1: 0x11, 2: 0x11, 3: 0x12, 4: 0x12, 5: 0x15, 6: 0x15, 7: 0x16, 8: 0x16, 9: 0x17, 10: 0x17, 11: 0x10, 12: 0x13, 13: 0x13, 14: 0x14, 15: 0x14}
|
||
FALLBACK = {}
|
||
for _letters, _base in (("ÁÀÂÃÄÅ", "A"), ("ÉÈÊË", "E"), ("ÍÌÎÏ", "I"), ("ÓÒÔÕÖØ", "O"), ("ÚÙÛÜ", "U"), ("Ç", "C"),
|
||
("ãäå", "a"), ("ë", "e"), ("ìï", "i"), ("òõöø", "o"), ("ùü", "u"), ("ß", "s"), ("“”«»", '"'), ("‘", "'"), ("¡", "!")):
|
||
for _ch in _letters:
|
||
FALLBACK[_ch] = _base
|
||
|
||
|
||
def basic(b):
|
||
if b in BASIC_SWAP:
|
||
return BASIC_SWAP[b]
|
||
if 0x20 <= b <= 0x7A:
|
||
return chr(b)
|
||
return None
|
||
|
||
|
||
BASIC_OF = {basic(b): b for b in range(0x20, 0x7F) if basic(b) is not None}
|
||
|
||
|
||
def send(ch):
|
||
"""§8.4: (shown as, how sent) or None."""
|
||
ch = SUBST.get(ch, ch)
|
||
if ch in BASIC_OF:
|
||
return ch, ("b", BASIC_OF[ch])
|
||
if ch in SPECIAL:
|
||
return ch, ("s", 0x30 + SPECIAL.index(ch))
|
||
for first in (0x12, 0x13):
|
||
table = EXT[first]
|
||
if ch in table and ch != "\x00":
|
||
return ch, ("x", first, 0x20 + table.index(ch), ord(FALLBACK.get(ch, " ")))
|
||
return None
|
||
|
||
|
||
def row_base(row):
|
||
return 0x60 if row in (2, 4, 6, 8, 10, 13, 15) else 0x40
|
||
|
||
|
||
def odd(b):
|
||
return b | 0x80 if bin(b & 0x7F).count("1") % 2 == 0 else b & 0x7F
|
||
|
||
|
||
def layout(mode, c):
|
||
"""§8.5: [(row, col, text, italic ranges)], or Refused."""
|
||
lines = runs_of(c)
|
||
rows_max = 4 if mode == "pop-on" else int(mode[-1])
|
||
if len(lines) > rows_max:
|
||
raise Refused("too_many_rows")
|
||
shown = []
|
||
for line in lines:
|
||
text, flags = "", []
|
||
for it, t in line:
|
||
for ch in t:
|
||
s = send(ch)
|
||
if s is None:
|
||
raise Refused("char_not_in_608")
|
||
text += s[0]
|
||
flags.append(it)
|
||
shown.append((text, flags))
|
||
out = []
|
||
for k, (text, flags) in enumerate(shown):
|
||
width = len(text)
|
||
if width > 32:
|
||
raise Refused("row_too_long")
|
||
ranges = []
|
||
for i, f in enumerate(flags):
|
||
if f:
|
||
if ranges and ranges[-1][0] + ranges[-1][1] == i:
|
||
ranges[-1][1] += 1
|
||
else:
|
||
ranges.append([i, 1])
|
||
for s, n in ranges:
|
||
if (s > 0 and text[s - 1] != " ") or (s + n < width and text[s + n] != " "):
|
||
raise Refused("style_not_expressible")
|
||
n = len(shown)
|
||
if mode == "pop-on":
|
||
row = k + 1 if is_top(c["anchor"]) else 16 - n + k
|
||
else:
|
||
row = int(mode[-1]) if is_top(c["anchor"]) else 15
|
||
a = align(c["anchor"])
|
||
col = 0 if a == "left" else (32 - width if a == "right" else (32 - width) // 2)
|
||
out.append((row, col, text, ranges))
|
||
return out
|
||
|
||
|
||
def row_codes(row, col, text, ranges):
|
||
"""§8.5: [("b", byte) | ("2", b1, b2)]."""
|
||
flags = [False] * len(text)
|
||
for s, n in ranges:
|
||
for i in range(s, s + n):
|
||
flags[i] = True
|
||
|
||
def place(x):
|
||
out = [("2", PAC_FIRST[row], row_base(row) + 0x10 + 2 * (x // 4))]
|
||
if x % 4:
|
||
out.append(("2", 0x17, 0x20 + x % 4))
|
||
return out
|
||
if flags and flags[0] and col == 0:
|
||
codes = [("2", PAC_FIRST[row], row_base(row) + 0x0E)]
|
||
elif flags and flags[0]:
|
||
codes = place(col - 1) + [("2", 0x11, 0x2E)]
|
||
else:
|
||
codes = place(col)
|
||
for i, ch in enumerate(text):
|
||
if ch == " " and not flags[i] and i + 1 < len(text) and flags[i + 1]:
|
||
codes.append(("2", 0x11, 0x2E))
|
||
elif ch == " " and not flags[i] and i > 0 and flags[i - 1]:
|
||
codes.append(("2", 0x11, 0x20))
|
||
else:
|
||
how = send(ch)[1]
|
||
if how[0] == "b":
|
||
codes.append(("b", how[1]))
|
||
elif how[0] == "s":
|
||
codes.append(("2", 0x11, how[1]))
|
||
else:
|
||
codes.append(("b", how[3]))
|
||
codes.append(("2", how[1], how[2]))
|
||
return codes
|
||
|
||
|
||
def units_of(codes):
|
||
"""§8.3: [(pair, slots)] — one slot for a pair of standard bytes, two for a doubled code."""
|
||
out, held = [], None
|
||
for c in codes:
|
||
if c[0] == "b":
|
||
if held is None:
|
||
held = c[1]
|
||
else:
|
||
out.append(((held, c[1]), 1))
|
||
held = None
|
||
else:
|
||
if held is not None:
|
||
out.append(((held, 0), 1))
|
||
held = None
|
||
out.append(((c[1], c[2]), 2))
|
||
if held is not None:
|
||
out.append(((held, 0), 1))
|
||
return out
|
||
|
||
|
||
def slot_of(ms, rate):
|
||
num, den = rate
|
||
return -((-ms * num) // (1000 * den))
|
||
|
||
|
||
def encode608(mode, rate, lang, cues):
|
||
"""§8.6, §8.7: {slot: (b1, b2)} with parity."""
|
||
check_doc(lang, cues)
|
||
num, den = rate
|
||
if not (0 < num <= 1000000 and 0 < den <= 1000000) or mode not in ("pop-on", "roll-up-2", "roll-up-3", "roll-up-4"):
|
||
raise Refused("parameter_malformed")
|
||
cues = in_order(cues)
|
||
rows = [layout(mode, c) for c in cues]
|
||
for a, b in zip(cues, cues[1:]):
|
||
if b["start"] < a["end"]:
|
||
raise Refused("cues_overlap")
|
||
if mode != "pop-on" and len(set(is_top(c["anchor"]) for c in cues)) > 1:
|
||
raise Refused("style_not_expressible")
|
||
s = [slot_of(c["start"], rate) for c in cues]
|
||
e = [slot_of(c["end"], rate) for c in cues]
|
||
n = len(cues)
|
||
erase = [i == n - 1 or s[i + 1] >= e[i] + 2 for i in range(n)]
|
||
slots = {}
|
||
|
||
def put(at, pair, count):
|
||
for k in range(count):
|
||
slots[at + k] = (odd(pair[0]), odd(pair[1]))
|
||
if mode == "pop-on":
|
||
for i in range(n):
|
||
if e[i] < s[i] + 2:
|
||
raise Refused("too_dense")
|
||
put(s[i], (0x14, 0x2F), 2)
|
||
if erase[i]:
|
||
put(e[i], (0x14, 0x2C), 2)
|
||
for i in range(n):
|
||
codes = [("2", 0x14, 0x20), ("2", 0x14, 0x2E)]
|
||
for r in rows[i]:
|
||
codes += row_codes(*r)
|
||
floor = 0 if i == 0 else s[i - 1] + 2
|
||
ceiling = s[i]
|
||
for pair, count in reversed(units_of(codes)):
|
||
q = ceiling - count
|
||
while q >= floor and any(q + k in slots for k in range(count)):
|
||
q -= 1
|
||
if q < floor:
|
||
raise Refused("too_dense")
|
||
put(q, pair, count)
|
||
ceiling = q
|
||
else:
|
||
depth = int(mode[-1])
|
||
after, filled = 0, False
|
||
for i in range(n):
|
||
if s[i] < after:
|
||
raise Refused("too_dense")
|
||
codes = [("2", 0x14, 0x23 + depth)] if i == 0 else []
|
||
for r in rows[i]:
|
||
if filled:
|
||
codes.append(("2", 0x14, 0x2D))
|
||
codes += row_codes(*r)
|
||
filled = True
|
||
at = s[i]
|
||
for pair, count in units_of(codes):
|
||
put(at, pair, count)
|
||
at += count
|
||
if at > (e[i] if erase[i] else s[i + 1]):
|
||
raise Refused("too_dense")
|
||
if erase[i]:
|
||
put(e[i], (0x14, 0x2C), 2)
|
||
filled = False
|
||
after = e[i] + 2
|
||
else:
|
||
after = at
|
||
return slots
|
||
|
||
|
||
def to_hex(slots):
|
||
return "".join("%d %02x%02x\n" % (k, slots[k][0], slots[k][1]) for k in sorted(slots))
|
||
|
||
|
||
def from_hex(text):
|
||
bad = Refused("cc_malformed")
|
||
if not text.endswith("\n"):
|
||
raise bad
|
||
out = []
|
||
for line in text[:-1].split("\n"):
|
||
slot, sp, pair = line.partition(" ")
|
||
if not sp or not slot.isdigit() or not slot.isascii() or (len(slot) > 1 and slot[0] == "0"):
|
||
raise bad
|
||
if len(pair) != 4 or any(ch not in "0123456789abcdef" for ch in pair):
|
||
raise bad
|
||
if out and out[-1][0] >= int(slot):
|
||
raise bad
|
||
out.append((int(slot), (int(pair[:2], 16), int(pair[2:], 16))))
|
||
return out
|
||
|
||
|
||
def read_cells(row, cells):
|
||
real = [k for k in range(32) if cells[k] is not None and not cells[k][2]]
|
||
if not real:
|
||
return None
|
||
i, j = real[0], real[-1]
|
||
text, flags, space = "", [], []
|
||
for k in range(i, j + 1):
|
||
cell = cells[k]
|
||
ch = cell[0] if cell is not None and not cell[2] else " "
|
||
text += ch
|
||
flags.append(cell is not None and not cell[2] and cell[1] and ch != " ")
|
||
space.append(ch == " ")
|
||
ranges = []
|
||
for k in range(len(text)):
|
||
inside = flags[k] or (space[k] and 0 < k < len(text) - 1 and flags[k - 1] and flags[k + 1])
|
||
if inside:
|
||
if ranges and ranges[-1][0] + ranges[-1][1] == k:
|
||
ranges[-1][1] += 1
|
||
else:
|
||
ranges.append([k, 1])
|
||
return {"row": row, "col": i, "text": text, "italic": ranges}
|
||
|
||
|
||
def decode608(pairs):
|
||
"""§8.8: the events, as the corpus writes them."""
|
||
bad = Refused("cc_malformed")
|
||
blank = lambda: [[None] * 32 for _ in range(15)] # noqa: E731
|
||
shown, hidden = blank(), blank()
|
||
mode, row, col, it, base = None, 15, 0, False, 15
|
||
line_at, acted, prev, events = None, None, None, []
|
||
|
||
def rows_of(mem):
|
||
return [r for r in (read_cells(k + 1, mem[k]) for k in range(15)) if r is not None]
|
||
|
||
def commit():
|
||
nonlocal line_at
|
||
r = read_cells(base, shown[base - 1])
|
||
if r is not None and line_at is not None:
|
||
events.append({"line": line_at, "row": r})
|
||
line_at = None
|
||
|
||
def put(ch, mid_row=False):
|
||
nonlocal col
|
||
if mode is None:
|
||
raise bad
|
||
mem, r = (hidden, row) if mode == "pop" else (shown, base)
|
||
mem[r - 1][col] = (ch, it, mid_row)
|
||
col = min(col + 1, 31)
|
||
for slot, (x, y) in pairs:
|
||
if prev is not None and slot <= prev:
|
||
raise bad
|
||
prev = slot
|
||
if bin(x).count("1") % 2 == 0 or bin(y).count("1") % 2 == 0:
|
||
raise bad
|
||
a, b = x & 0x7F, y & 0x7F
|
||
if 0x10 <= a <= 0x1F:
|
||
if acted is not None and acted[0] == (a, b) and acted[1] + 1 == slot:
|
||
acted = None
|
||
continue
|
||
acted = ((a, b), slot)
|
||
if b >= 0x40:
|
||
r = next((k for k in range(1, 16) if PAC_FIRST[k] == a and row_base(k) == b & 0x60), None)
|
||
attr = b & 0x1F
|
||
if r is None:
|
||
raise bad
|
||
if attr == 0x00:
|
||
c0, i0 = 0, False
|
||
elif attr == 0x0E:
|
||
c0, i0 = 0, True
|
||
elif 0x10 <= attr <= 0x1E and attr % 2 == 0:
|
||
c0, i0 = (attr - 0x10) * 2, False
|
||
else:
|
||
raise bad
|
||
row, col, it = r, c0, i0
|
||
if mode not in (None, "pop"):
|
||
base = r
|
||
if line_at is None:
|
||
line_at = slot
|
||
elif a == 0x14 and b == 0x20:
|
||
mode = "pop"
|
||
elif a == 0x14 and b in (0x25, 0x26, 0x27):
|
||
if mode == "pop":
|
||
shown, hidden = blank(), blank()
|
||
mode, base, col = b - 0x23, 15, 0
|
||
elif a == 0x14 and b == 0x2C:
|
||
if mode not in (None, "pop"):
|
||
commit()
|
||
line_at = None
|
||
shown = blank()
|
||
events.append({"erase": slot})
|
||
elif a == 0x14 and b == 0x2D:
|
||
if mode in (None, "pop"):
|
||
raise bad
|
||
commit()
|
||
for k in range(base - mode, base - 1):
|
||
shown[k] = shown[k + 1]
|
||
shown[base - 1] = [None] * 32
|
||
col = 0
|
||
elif a == 0x14 and b == 0x2E:
|
||
hidden = blank()
|
||
elif a == 0x14 and b == 0x2F:
|
||
shown, hidden = hidden, shown
|
||
mode = "pop"
|
||
events.append({"show": slot, "rows": rows_of(shown)})
|
||
elif a == 0x17 and b in (0x21, 0x22, 0x23):
|
||
col = min(col + b - 0x20, 31)
|
||
elif a == 0x11 and b in (0x20, 0x2E):
|
||
it = b == 0x2E
|
||
put(" ", True)
|
||
elif a == 0x11 and 0x30 <= b <= 0x3F:
|
||
put(SPECIAL[b - 0x30])
|
||
elif a in (0x12, 0x13) and 0x20 <= b <= 0x3F:
|
||
ch = EXT[a][b - 0x20]
|
||
if ch == "\x00":
|
||
raise bad
|
||
col = max(col - 1, 0)
|
||
put(ch)
|
||
else:
|
||
raise bad
|
||
else:
|
||
acted = None
|
||
for v in (a, b):
|
||
if v == 0:
|
||
continue
|
||
ch = basic(v)
|
||
if ch is None:
|
||
raise bad
|
||
put(ch)
|
||
if mode not in (None, "pop"):
|
||
commit()
|
||
return events
|
||
|
||
|
||
def round_trip(mode, rate, cues, events):
|
||
"""§8.8's last paragraph: what a decoder reports is what the cues lay out."""
|
||
cues = in_order(cues)
|
||
n = len(cues)
|
||
s = [slot_of(c["start"], rate) for c in cues]
|
||
e = [slot_of(c["end"], rate) for c in cues]
|
||
erased = [i == n - 1 or s[i + 1] >= e[i] + 2 for i in range(n)]
|
||
rows = [[{"row": r, "col": c, "text": t, "italic": rg} for r, c, t, rg in layout(mode, cue)] for cue in cues]
|
||
if mode == "pop-on":
|
||
want = []
|
||
for i in range(n):
|
||
want.append({"show": s[i], "rows": rows[i]})
|
||
if erased[i]:
|
||
want.append({"erase": e[i]})
|
||
return events == want
|
||
k = 0
|
||
for i in range(n):
|
||
end = e[i] if erased[i] else s[i + 1]
|
||
for r in rows[i]:
|
||
if k >= len(events) or events[k].get("row") != r or not (s[i] <= events[k]["line"] < end):
|
||
return False
|
||
k += 1
|
||
if erased[i]:
|
||
if k >= len(events) or events[k] != {"erase": e[i]}:
|
||
return False
|
||
k += 1
|
||
return k == len(events)
|
||
|
||
|
||
def paths608(text):
|
||
seen = set()
|
||
for _, (x, y) in from_hex(text):
|
||
a, b = x & 0x7F, y & 0x7F
|
||
names = {(0x14, 0x20): "RCL", (0x14, 0x25): "RU2", (0x14, 0x26): "RU3", (0x14, 0x27): "RU4", (0x14, 0x2C): "EDM",
|
||
(0x14, 0x2D): "CR", (0x14, 0x2E): "ENM", (0x14, 0x2F): "EOC", (0x17, 0x21): "TO1", (0x17, 0x22): "TO2",
|
||
(0x17, 0x23): "TO3", (0x11, 0x20): "MRC white", (0x11, 0x2E): "MRC italics"}
|
||
if (a, b) in names:
|
||
seen.add(names[(a, b)])
|
||
elif a == 0x11 and 0x30 <= b <= 0x3F:
|
||
seen.add("special")
|
||
elif a in (0x12, 0x13) and 0x20 <= b <= 0x3F:
|
||
seen.add("extended 0x%x" % a)
|
||
elif 0x10 <= a <= 0x1F and b >= 0x40:
|
||
seen.add("PAC italics" if b & 0x1F == 0x0E else "PAC indent")
|
||
else:
|
||
seen.add("standard")
|
||
return seen
|
||
|
||
|
||
PATHS608 = {"RCL", "RU2", "RU3", "RU4", "EDM", "CR", "ENM", "EOC", "TO1", "TO2", "TO3", "MRC white", "MRC italics",
|
||
"special", "extended 0x12", "extended 0x13", "PAC italics", "PAC indent", "standard"}
|
||
|
||
|
||
# ---------------------------------------------------------------- §4 live captions
|
||
|
||
def read_annotation(b):
|
||
"""§4.2's layout."""
|
||
if len(b) < 42 or b[:4] != b"WTX1" or b[4] != 1 or b[37] not in (0, 1) or b[40] > 5 or len(b) != 42 + 4 * b[41]:
|
||
raise Refused("annotation_malformed")
|
||
italic = [[int.from_bytes(b[42 + 4 * i:44 + 4 * i], "little"), int.from_bytes(b[44 + 4 * i:46 + 4 * i], "little")] for i in range(b[41])]
|
||
return {"supersedes": b[5:37], "final": b[37] == 1, "stable": int.from_bytes(b[38:40], "little"),
|
||
"anchor": ANCHORS[b[40]], "italic": italic}
|
||
|
||
|
||
def annotation_of(p):
|
||
return p["kind"] == META and p["bytes"][:4] == b"WTX1"
|
||
|
||
|
||
class Captions:
|
||
"""§4.3 over a live-events verifier."""
|
||
|
||
def __init__(self, room):
|
||
self.ev = LEV.Verifier(room, DEFAULT_POLICY, {})
|
||
self.revs = [] # dicts, in stream order
|
||
self.held = {} # principal -> {seq: revision position or None}
|
||
self.whole = {} # principal -> no gap so far
|
||
self.waiting = {} # principal -> revision position
|
||
self.n = 0
|
||
|
||
def settle(self, k, unknown):
|
||
r = self.revs[k]
|
||
r.update(final=True, stable=len(r["text"]), unknown=unknown)
|
||
|
||
def offer(self, b):
|
||
index = self.n
|
||
self.n += 1
|
||
before = len(self.ev.accepted)
|
||
self.ev.offer(b) # raises LEV.Refuse
|
||
if len(self.ev.accepted) == before:
|
||
return # a duplicate
|
||
_, ev, p = self.ev.accepted[-1]
|
||
who, seq = ev["principal"], ev["seq"]
|
||
held = self.held.setdefault(who, {})
|
||
if who not in self.whole:
|
||
self.whole[who] = True
|
||
top = max(held) if held else None
|
||
if (top is None and seq > 0) or (top is not None and seq > top + 1):
|
||
self.whole[who] = False
|
||
mine = self.waiting.pop(who, None)
|
||
ann = annotation_of(p)
|
||
target = None
|
||
if mine is not None:
|
||
cseq = self.revs[mine]["event"]["seq"]
|
||
if seq == cseq + 1 and ann:
|
||
target = mine
|
||
else:
|
||
self.settle(mine, seq != cseq + 1)
|
||
if p["kind"] == CAPTION:
|
||
if not text_ok(p["text"]):
|
||
raise Refused("text_not_in_subset")
|
||
self.revs.append({"index": index, "event": ev, "text": p["text"], "lang": p["lang"], "duration": p["duration_ms"],
|
||
"origin": ORIGINS[p["origin"]], "model": None if p["model"] is None else p["model"].hex(),
|
||
"annotation": None, "unknown": False, "final": False, "stable": 0,
|
||
"anchor": "bottom-center", "italic": [], "sup": None, "sup_digest": None, "by": None})
|
||
held[seq] = len(self.revs) - 1
|
||
self.waiting[who] = len(self.revs) - 1
|
||
elif ann:
|
||
held[seq] = None
|
||
if target is not None:
|
||
self.attach(index, target, ev, p["bytes"])
|
||
elif seq > 0 and (seq - 1) in held:
|
||
raise Refused("annotation_misplaced")
|
||
else:
|
||
held[seq] = None
|
||
|
||
def attach(self, index, k, ev, b):
|
||
a = read_annotation(b)
|
||
r = self.revs[k]
|
||
if ev["t"] != r["event"]["t"]:
|
||
raise Refused("annotation_misplaced")
|
||
if a["stable"] > len(r["text"]) or (a["final"] and a["stable"] != len(r["text"])) or not ranges_ok(r["text"], a["italic"]):
|
||
raise Refused("annotation_malformed")
|
||
sup = None
|
||
if a["supersedes"] != bytes(32):
|
||
who = ev["principal"]
|
||
sup = next((j for j, x in enumerate(self.revs) if x["event"]["principal"] == who and x["event"]["digest"] == a["supersedes"]), None)
|
||
if sup is None:
|
||
if self.whole[who]:
|
||
raise Refused("supersede_unknown")
|
||
else:
|
||
old = self.revs[sup]
|
||
if old["final"]:
|
||
raise Refused("supersede_final")
|
||
if old["by"] is not None:
|
||
raise Refused("supersede_forked")
|
||
if old["lang"] != r["lang"] or old["anchor"] != a["anchor"]:
|
||
raise Refused("supersede_mismatch")
|
||
if not r["text"].startswith(old["text"][:old["stable"]]):
|
||
raise Refused("stability_broken")
|
||
old["by"] = k
|
||
r.update(annotation=index, final=a["final"], stable=a["stable"], anchor=a["anchor"], italic=a["italic"], sup=sup,
|
||
sup_digest=None if a["supersedes"] == bytes(32) else a["supersedes"])
|
||
|
||
def finish(self):
|
||
for k in list(self.waiting.values()):
|
||
self.settle(k, False)
|
||
self.waiting = {}
|
||
|
||
def shown(self, x, names):
|
||
out = []
|
||
for k, r in enumerate(self.revs):
|
||
t = r["event"]["t"]
|
||
if t > x or (r["by"] is not None and self.revs[r["by"]]["event"]["t"] <= x):
|
||
continue
|
||
if r["duration"]:
|
||
live = x < t + r["duration"] * 1000
|
||
else:
|
||
live = not any(n["event"]["principal"] == r["event"]["principal"] and n["event"]["seq"] > r["event"]["seq"]
|
||
and n["event"]["t"] <= x for n in self.revs[k + 1:])
|
||
if live:
|
||
out.append({"principal": names[r["event"]["principal"].hex()], "text": r["text"], "anchor": r["anchor"],
|
||
"origin": r["origin"], "italic": r["italic"], "final": r["final"]})
|
||
return {"at": x, "cues": out}
|
||
|
||
def revisions(self):
|
||
return [{"index": r["index"], "annotation": r["annotation"], "final": r["final"], "unknown": r["unknown"],
|
||
"stable_cp": r["stable"], "anchor": r["anchor"], "italic": r["italic"],
|
||
"supersedes": None if r["sup"] is None else self.revs[r["sup"]]["index"]} for r in self.revs]
|
||
|
||
def resolve(self, pt0):
|
||
"""§4.5: (cues with their publication times) or Refused with an index."""
|
||
out = []
|
||
for head in self.revs:
|
||
if head["sup"] is not None or head["sup_digest"] is not None:
|
||
continue
|
||
last = head
|
||
while last["by"] is not None:
|
||
last = self.revs[last["by"]]
|
||
if not last["final"] or last["unknown"]:
|
||
raise Refused(("not_final", last["index"]))
|
||
if last["duration"] == 0:
|
||
raise Refused(("duration_unknown", last["index"]))
|
||
pub = pt0 + last["event"]["t"] // 1000
|
||
cue = {"start": pt0 + head["event"]["t"] // 1000, "end": pub + last["duration"], "lang": last["lang"],
|
||
"origin": last["origin"], "model": last["model"], "anchor": last["anchor"], "text": last["text"],
|
||
"italic": last["italic"]}
|
||
if cue["end"] > LAST_MS:
|
||
raise Refused(("time_out_of_range", last["index"]))
|
||
out.append((cue["start"], head["index"], cue, pub))
|
||
out.sort(key=lambda x: (x[0], x[1]))
|
||
return [(c, p) for _, _, c, p in out]
|
||
|
||
|
||
def verify_events(room, records):
|
||
"""§4.3's outcome: (outcome dict, Captions)."""
|
||
cap = Captions(room)
|
||
for i, b in enumerate(records):
|
||
try:
|
||
cap.offer(b)
|
||
except (LEV.Refuse, Refused) as e:
|
||
return {"outcome": "refuse", "reason": e.reason, "index": i}, cap
|
||
cap.finish()
|
||
if cap.ev.gaps:
|
||
return {"outcome": "gap", "gaps": [[g[0], g[1], g[2]] for g in cap.ev.gaps]}, cap
|
||
bad = LEV.stated_tally_mismatch(cap.ev.accepted)
|
||
if bad is not None:
|
||
return {"outcome": "refuse", "reason": "tally_mismatch", "index": bad}, cap
|
||
return {"outcome": "accept"}, cap
|
||
|
||
|
||
def join_set(mine, group):
|
||
"""§4.6 on one principal's (event, payload) list: the seqs of its join set."""
|
||
start = group * GROUP_US
|
||
before = [(e, p) for e, p in mine if e["t"] < start]
|
||
by_seq = {e["seq"]: (e, p) for e, p in before}
|
||
keep = set()
|
||
# live-events §8.2.
|
||
closed = {p["poll"] for _, p in before if p["kind"] == 0x07}
|
||
open_polls = {e["digest"] for e, p in before if p["kind"] == 0x05 and e["digest"] not in closed}
|
||
last_vote, ins = {}, {}
|
||
for e, p in before:
|
||
if p["kind"] == 0x06 and p["poll"] in open_polls:
|
||
last_vote[p["poll"]] = e["seq"]
|
||
if p["kind"] == 0x02 and p["type"] == 2:
|
||
ins[p["cue_id"]] = e["seq"]
|
||
for e, p in before:
|
||
k = p["kind"]
|
||
if (k == 0x05 and e["digest"] in open_polls) or (k == 0x06 and last_vote.get(p["poll"]) == e["seq"]):
|
||
keep.add(e["seq"])
|
||
elif k == CAPTION and p["duration_ms"] > 0 and e["t"] + p["duration_ms"] * 1000 > start:
|
||
keep.add(e["seq"])
|
||
elif k == 0x02 and p["type"] in (1, 4) and not (p["cue_id"] in ins and ins[p["cue_id"]] > e["seq"]):
|
||
keep.add(e["seq"])
|
||
# §4.6.
|
||
superseded = set()
|
||
for e, p in before:
|
||
if annotation_of(p) and (e["seq"] - 1) in by_seq and by_seq[e["seq"] - 1][1]["kind"] == CAPTION:
|
||
try:
|
||
superseded.add(read_annotation(p["bytes"])["supersedes"])
|
||
except Refused:
|
||
pass
|
||
captions = [e["seq"] for e, p in before if p["kind"] == CAPTION]
|
||
for e, p in before:
|
||
if p["kind"] != CAPTION or p["duration_ms"] != 0 or e["seq"] != max(captions) or e["digest"] in superseded:
|
||
continue
|
||
nxt = by_seq.get(e["seq"] + 1)
|
||
if nxt and annotation_of(nxt[1]):
|
||
try:
|
||
if not read_annotation(nxt[1]["bytes"])["final"]:
|
||
keep.add(e["seq"])
|
||
except Refused:
|
||
pass
|
||
for s in sorted(keep):
|
||
if s in by_seq and by_seq[s][1]["kind"] == CAPTION and (s + 1) in by_seq and annotation_of(by_seq[s + 1][1]):
|
||
keep.add(s + 1)
|
||
return sorted(keep)
|
||
|
||
|
||
def join_stream(records, group):
|
||
parsed = []
|
||
for b in records:
|
||
e = LEV.read_record(b)
|
||
parsed.append((e, LEV.read_payload(e["kind"], e["payload"], 500)))
|
||
out, idx = [], []
|
||
for who in sorted(set(e["principal"] for e, _ in parsed)):
|
||
mine = [(e, p) for e, p in parsed if e["principal"] == who]
|
||
for s in join_set(mine, group):
|
||
i = next(k for k, (e, _) in enumerate(parsed) if e["principal"] == who and e["seq"] == s)
|
||
idx.append(i)
|
||
out.append(records[i])
|
||
out += [records[k] for k, (e, _) in enumerate(parsed) if e["t"] >= group * GROUP_US]
|
||
return out, idx
|
||
|
||
|
||
# ---------------------------------------------------------------- the corpus
|
||
|
||
def attempt(f, *a):
|
||
try:
|
||
return f(*a), None
|
||
except Refused as e:
|
||
return None, e.reason
|
||
|
||
|
||
def targets_of(name, lang, published, stated, written):
|
||
"""The targets object, from this text; `written` collects (file, contents)."""
|
||
cues = [c for c, _ in published]
|
||
out = {}
|
||
|
||
def file_or_reason(path, f, *a):
|
||
value, why = attempt(f, *a)
|
||
if why:
|
||
return {"reason": why}
|
||
written[path] = value
|
||
return {"file": path}
|
||
if "webvtt" in stated:
|
||
out["webvtt"] = file_or_reason(name + ".vtt", webvtt, lang, cues)
|
||
if "imsc" in stated:
|
||
out["imsc"] = file_or_reason(name + ".ttml", imsc, lang, cues)
|
||
if "segments" in stated:
|
||
s = stated["segments"]
|
||
o = {"bounds": s["bounds"], "mpegts0": s["mpegts0"]}
|
||
segs, why = attempt(webvtt_segments, lang, cues, s["bounds"], s["mpegts0"])
|
||
if why:
|
||
o["reason"] = why
|
||
else:
|
||
o["files"] = []
|
||
for k, seg in enumerate(segs):
|
||
p = "%s.seg%d.vtt" % (name, k)
|
||
written[p] = seg
|
||
o["files"].append(p)
|
||
out["segments"] = o
|
||
if "timelines" in stated:
|
||
s = stated["timelines"]
|
||
o = {"group_ms": s["group_ms"], "first": s["first"], "last": s["last"]}
|
||
for kind in ("webvtt", "imsc"):
|
||
groups, why = attempt(timeline, kind, lang, published, s["group_ms"], s["first"], s["last"])
|
||
if why:
|
||
o["reason"] = why
|
||
else:
|
||
o[kind] = groups
|
||
out["timelines"] = o
|
||
if "cc608" in stated:
|
||
lst = []
|
||
for e in stated["cc608"]:
|
||
mode, rate = e["mode"], tuple(e["rate"])
|
||
o = {"mode": mode, "rate": list(rate)}
|
||
slots, why = attempt(encode608, mode, rate, lang, cues)
|
||
if why:
|
||
o["reason"] = why
|
||
else:
|
||
suffix = "" if rate == (30000, 1001) else ".r%d-%d" % rate
|
||
p = "%s.%s%s.608.hex" % (name, mode, suffix)
|
||
written[p] = to_hex(slots)
|
||
o["file"] = p
|
||
o["decoded"] = decode608(sorted(slots.items()))
|
||
if not round_trip(mode, rate, cues, o["decoded"]):
|
||
o["decoded"] = "does not round-trip"
|
||
lst.append(o)
|
||
out["cc608"] = lst
|
||
if "disclosure" in stated:
|
||
chars = ["public.accessibility.transcribes-spoken-dialog", "public.accessibility.describes-music-and-sound"]
|
||
out["disclosure"] = {
|
||
"hls_subtitles": media_tag("SUBTITLES", "subs", "Captions", lang, chars, cues, name + "/captions.m3u8"),
|
||
"hls_closed_captions": media_tag("CLOSED-CAPTIONS", "cc", "Captions", lang, chars, cues),
|
||
"session_data": '#EXT-X-SESSION-DATA:DATA-ID="science.transaction.wai.timed-text",URI="%s.disclosure.json",LANGUAGE="%s"' % (name, lang),
|
||
"sidecar": sidecar(lang, cues),
|
||
"c2pa": c2pa(cues),
|
||
"catalog_webvtt": catalog("webvtt", "captions-" + lang, lang, ["video"], cues),
|
||
"catalog_imsc": catalog("imsc", "captions-%s-imsc" % lang, lang, ["video"], cues),
|
||
}
|
||
return out
|
||
|
||
|
||
def check_targets(name, lang, published, stated, files, fails, seen):
|
||
written = {}
|
||
try:
|
||
got = targets_of(name, lang, published, stated, written)
|
||
except Exception as e: # a mutated copy that this text cannot map is a failure, not a crash
|
||
fails.append("%s: the targets do not map: %s: %s" % (name, type(e).__name__, e))
|
||
return
|
||
if got != stated:
|
||
keys = sorted(k for k in set(got) | set(stated) if got.get(k) != stated.get(k))
|
||
fails.append("%s: targets %s differ from this text's" % (name, keys))
|
||
for p, body in written.items():
|
||
if files.get(p) != body:
|
||
fails.append("%s: %s differs from this text's rendering" % (name, p))
|
||
cues = in_order([c for c, _ in published])
|
||
w = stated.get("webvtt", {})
|
||
if "file" in w:
|
||
back, why = attempt(read_webvtt, files.get(w["file"], ""), lang)
|
||
if why or back != (cues, None):
|
||
fails.append("%s: its WebVTT does not read back (%s)" % (name, why))
|
||
s = stated.get("segments", {})
|
||
for k, f in enumerate(s.get("files", [])):
|
||
lo, hi = s["bounds"][k], s["bounds"][k + 1]
|
||
back, why = attempt(read_webvtt, files.get(f, ""), lang)
|
||
want = [c for c in cues if c["start"] < hi and c["end"] > lo]
|
||
if why or back[0] != want or back[1][0] != lo:
|
||
fails.append("%s: segment %d does not read back (%s)" % (name, k, why))
|
||
m = stated.get("imsc", {})
|
||
if "file" in m:
|
||
for f in imsc_structure(files.get(m["file"], ""), cues):
|
||
fails.append("%s: IMSC1 structure: %s" % (name, f))
|
||
for e in stated.get("cc608", []):
|
||
if "reason" in e:
|
||
seen.add(e["reason"])
|
||
if "file" in e:
|
||
try:
|
||
dec = decode608(from_hex(files.get(e["file"], "")))
|
||
except Refused as r:
|
||
dec = r.reason
|
||
if dec != e.get("decoded"):
|
||
fails.append("%s: %s decodes otherwise than stated" % (name, e["file"]))
|
||
for k in ("webvtt", "imsc", "segments", "timelines"):
|
||
if "reason" in stated.get(k, {}):
|
||
seen.add(stated[k]["reason"])
|
||
|
||
|
||
def check(doc, files, quiet=False):
|
||
fails, seen = [], set()
|
||
pt0 = doc["pt0_ms"]
|
||
room = bytes.fromhex(doc["room"])
|
||
names = {v: k for k, v in doc["principals"].items()}
|
||
for v in doc["event_vectors"]:
|
||
name = v["name"]
|
||
try:
|
||
records = [bytes.fromhex(h) for h in v["events"]]
|
||
out, cap = verify_events(room, records)
|
||
except Exception as e:
|
||
fails.append("%s: does not verify: %s: %s" % (name, type(e).__name__, e))
|
||
continue
|
||
if "gaps" in out:
|
||
out["gaps"] = [[names[g[0].hex()], g[1], g[2]] for g in out["gaps"]]
|
||
for k in ("outcome", "reason", "index", "gaps"):
|
||
if out.get(k) != v.get(k):
|
||
fails.append("%s: %s is %s, the corpus states %s" % (name, k, out.get(k), v.get(k)))
|
||
if v.get("reason"):
|
||
seen.add(v["reason"])
|
||
if not quiet:
|
||
print("%s event vector %s: %s" % ("PASS" if not fails or not fails[-1].startswith(name + ":") else "FAIL", name, out["outcome"] + (" " + out.get("reason", "") if out.get("reason") else "")))
|
||
if out["outcome"] == "refuse":
|
||
continue
|
||
if cap.revisions() != v.get("revisions"):
|
||
fails.append("%s: the revisions differ" % name)
|
||
for s in v.get("shown", []):
|
||
if cap.shown(s["at"], names) != s:
|
||
fails.append("%s: what is shown at %d differs" % (name, s["at"]))
|
||
try:
|
||
published = cap.resolve(pt0)
|
||
except Refused as r:
|
||
why, at = r.reason
|
||
seen.add(why)
|
||
if v.get("resolve") != {"reason": why, "index": at}:
|
||
fails.append("%s: resolves to %s at %d, the corpus states %s" % (name, why, at, v.get("resolve")))
|
||
published = None
|
||
if published is not None:
|
||
want = {"cues": [c for c, _ in published], "published": [p for _, p in published]}
|
||
if v.get("resolve") != want:
|
||
fails.append("%s: the resolved cues differ" % name)
|
||
check_targets(name, "en", published, v.get("targets", {}), files, fails, seen)
|
||
j = v.get("join")
|
||
if j is not None:
|
||
stream, idx = join_stream(records, j["group"])
|
||
if [b.hex() for b in stream] != j["stream"] or idx != j["join_set"]:
|
||
fails.append("%s: the joiner's records differ" % name)
|
||
jo, jcap = verify_events(room, stream)
|
||
if jo.get("outcome") != j.get("outcome") or [[names[g[0].hex()], g[1], g[2]] for g in jo.get("gaps", [])] != j.get("gaps"):
|
||
fails.append("%s: the joiner's outcome differs" % name)
|
||
for s in j.get("shown", []):
|
||
if jcap.shown(s["at"], names) != s or cap.shown(s["at"], names) != s:
|
||
fails.append("%s: the joiner does not show what the whole stream shows at %d" % (name, s["at"]))
|
||
# On a timeline, a joiner at any group holds every cue not yet ended (§7.3).
|
||
tl = v.get("targets", {}).get("timelines", {})
|
||
for g in tl.get("webvtt", []):
|
||
lo = g["group"] * tl["group_ms"]
|
||
held = set()
|
||
for later in tl["webvtt"]:
|
||
if later["group"] >= g["group"]:
|
||
for obj in later["objects"]:
|
||
held |= {r["data"]["start"] for r in json.loads(obj)}
|
||
need = {c["start"] for c, _ in (published or []) if c["end"] > lo}
|
||
if not need <= held:
|
||
fails.append("%s: a joiner at group %d lacks a cue" % (name, g["group"]))
|
||
for v in doc["cue_vectors"]:
|
||
published = [(c, c["start"]) for c in v["cues"]]
|
||
before = len(fails)
|
||
check_targets(v["name"], v["lang"], published, v["targets"], files, fails, seen)
|
||
if not quiet:
|
||
print("%s cue vector %s" % ("PASS" if len(fails) == before else "FAIL", v["name"]))
|
||
for c in doc["read_cases"]:
|
||
_, why = attempt(read_webvtt, c["text"], c["lang"])
|
||
seen.add(why)
|
||
if why != c["reason"]:
|
||
fails.append("read case %s: %s, the corpus states %s" % (c["name"], why, c["reason"]))
|
||
for c in doc["decode_cases"]:
|
||
try:
|
||
decode608(from_hex(c["hex"]))
|
||
why = None
|
||
except Refused as r:
|
||
why = r.reason
|
||
seen.add(why)
|
||
if why != c["reason"]:
|
||
fails.append("decode case %s: %s, the corpus states %s" % (c["name"], why, c["reason"]))
|
||
missing = [r for r in REASONS if r not in seen]
|
||
if missing:
|
||
fails.append("nothing is refused %s" % ", ".join(missing))
|
||
paths = set()
|
||
for p, body in files.items():
|
||
if p.endswith(".608.hex"):
|
||
try:
|
||
paths |= paths608(body)
|
||
except Refused:
|
||
fails.append("%s does not read" % p)
|
||
if PATHS608 - paths:
|
||
fails.append("no 608 file sends %s" % ", ".join(sorted(PATHS608 - paths)))
|
||
return fails
|
||
|
||
|
||
def mutations(doc, files):
|
||
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
||
out = []
|
||
|
||
def mutate(label, f):
|
||
d, fs = copy.deepcopy(doc), dict(files)
|
||
f(d, fs)
|
||
out.append((label, d, fs))
|
||
|
||
def vec(d, kind, name):
|
||
return next(v for v in d[kind] if v["name"] == name)
|
||
|
||
def flip_hex(h, at):
|
||
b = bytearray(bytes.fromhex(h))
|
||
b[at] ^= 0x01
|
||
return b.hex()
|
||
|
||
def swap_lines(text, i):
|
||
lines = text.split("\n")
|
||
lines[i], lines[i + 1] = lines[i + 1], lines[i]
|
||
return "\n".join(lines)
|
||
mutate("a letter of the machine lesson's WebVTT", lambda d, fs: fs.update({"lesson_machine.vtt": fs["lesson_machine.vtt"].replace("twice", "twise")}))
|
||
mutate("the machine label taken out of disclosure_mix.ttml",
|
||
lambda d, fs: fs.update({"disclosure_mix.ttml": fs["disclosure_mix.ttml"].replace(' wai:origin="machine-reviewed"', "", 1)}))
|
||
mutate("the machine label taken out of disclosure_mix.vtt",
|
||
lambda d, fs: fs.update({"disclosure_mix.vtt": fs["disclosure_mix.vtt"].replace("wai-machine-reviewed", "wai-machine", 1)}))
|
||
mutate("a byte pair of the human lesson's 608", lambda d, fs: fs.update(
|
||
{"lesson_human.pop-on.608.hex": fs["lesson_human.pop-on.608.hex"].replace(" 9420\n", " 9429\n", 1)}))
|
||
mutate("two lines of the roll-up 608 swapped", lambda d, fs: fs.update({"cc_rollup.roll-up-3.608.hex": swap_lines(fs["cc_rollup.roll-up-3.608.hex"], 3)}))
|
||
mutate("a byte of the machine lesson's third record", lambda d, fs: vec(d, "event_vectors", "lesson_machine")["events"].__setitem__(
|
||
2, flip_hex(vec(d, "event_vectors", "lesson_machine")["events"][2], 150)))
|
||
mutate("the human lesson's stated end of its first cue", lambda d, fs: vec(d, "event_vectors", "lesson_human")["resolve"]["cues"][0].update(end=13600))
|
||
mutate("an anchor of styles_and_anchors", lambda d, fs: vec(d, "cue_vectors", "styles_and_anchors")["cues"][1].update(anchor="bottom-center"))
|
||
mutate("an italic range of styles_and_anchors", lambda d, fs: vec(d, "cue_vectors", "styles_and_anchors")["cues"][7].update(italic=[[5, 7]]))
|
||
mutate("the stated reason of supersede_forked", lambda d, fs: vec(d, "event_vectors", "supersede_forked").update(reason="supersede_final"))
|
||
mutate("object 0 of the join vector's group 35", lambda d, fs: vec(d, "event_vectors", "mid_stream_join")["targets"]["timelines"]["webvtt"][2]["objects"].__setitem__(0, "[]"))
|
||
mutate("what the joiner shows at 60 s", lambda d, fs: vec(d, "event_vectors", "mid_stream_join")["join"]["shown"][0]["cues"].pop())
|
||
mutate("the C2PA statement of disclosure_mix", lambda d, fs: vec(d, "cue_vectors", "disclosure_mix")["targets"]["disclosure"].update(
|
||
c2pa=vec(d, "cue_vectors", "disclosure_mix")["targets"]["disclosure"]["c2pa"].replace("human_validated", "fully_autonomous", 1)))
|
||
mutate("a sidecar origin of disclosure_mix", lambda d, fs: vec(d, "cue_vectors", "disclosure_mix")["targets"]["disclosure"].update(
|
||
sidecar=vec(d, "cue_vectors", "disclosure_mix")["targets"]["disclosure"]["sidecar"].replace('"origin":"machine"', '"origin":"human"', 1)))
|
||
mutate("the machine characteristic of the machine lesson's HLS tag", lambda d, fs: vec(d, "event_vectors", "lesson_machine")["targets"]["disclosure"].update(
|
||
hls_closed_captions=vec(d, "event_vectors", "lesson_machine")["targets"]["disclosure"]["hls_closed_captions"].replace(",public.machine-generated", "")))
|
||
mutate("a stated show slot of the human lesson's 608", lambda d, fs: vec(d, "event_vectors", "lesson_human")["targets"]["cc608"][0]["decoded"][0].update(
|
||
show=vec(d, "event_vectors", "lesson_human")["targets"]["cc608"][0]["decoded"][0]["show"] + 1))
|
||
mutate("the stated reason of a read case", lambda d, fs: d["read_cases"][0].update(reason="cue_malformed"))
|
||
mutate("a segment's X-TIMESTAMP-MAP", lambda d, fs: fs.update({"lesson_human.seg1.vtt": fs["lesson_human.seg1.vtt"].replace("MPEGTS:2160000", "MPEGTS:2160090")}))
|
||
mutate("ttp:profile taken off lesson_human.ttml's tt", lambda d, fs: fs.update(
|
||
{"lesson_human.ttml": fs["lesson_human.ttml"].replace(' ttp:profile="http://www.w3.org/ns/ttml/profile/imsc1/text"', "")}))
|
||
mutate("a span of styles_and_anchors.ttml no longer italic", lambda d, fs: fs.update(
|
||
{"styles_and_anchors.ttml": fs["styles_and_anchors.ttml"].replace('tts:fontStyle="italic"', 'tts:fontStyle="normal"', 1)}))
|
||
return out
|
||
|
||
|
||
def main(corpus_dir):
|
||
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
files = {}
|
||
for n in sorted(os.listdir(corpus_dir)):
|
||
if n not in ("vectors.json", "README.md"):
|
||
with open(os.path.join(corpus_dir, n), encoding="utf-8", newline="") as f:
|
||
files[n] = f.read()
|
||
fails = check(doc, files)
|
||
# Every file is one this text writes: none left over.
|
||
written = set()
|
||
for v in doc["event_vectors"] + doc["cue_vectors"]:
|
||
t = v.get("targets", {})
|
||
for k in ("webvtt", "imsc"):
|
||
if "file" in t.get(k, {}):
|
||
written.add(t[k]["file"])
|
||
written |= set(t.get("segments", {}).get("files", []))
|
||
written |= {e["file"] for e in t.get("cc608", []) if "file" in e}
|
||
if set(files) != written:
|
||
fails.append("files not stated by a vector: %s; stated and absent: %s" % (sorted(set(files) - written), sorted(written - set(files))))
|
||
print("%s every vector, case and file agrees with this text (%d event vectors, %d cue vectors, %d files)"
|
||
% ("PASS" if not fails else "FAIL", len(doc["event_vectors"]), len(doc["cue_vectors"]), len(files)))
|
||
for label, d, fs in mutations(doc, files):
|
||
try:
|
||
caught = check(d, fs, quiet=True)
|
||
except Exception as e: # a crash on a mutated copy is a failure to read it
|
||
caught = ["%s: %s" % (type(e).__name__, e)]
|
||
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:110] if caught else ""))
|
||
if not caught:
|
||
fails.append("a mutated copy verifies: %s" % label)
|
||
for f in fails:
|
||
print("FAIL " + f)
|
||
print(("PASS" if not fails else "FAIL") + " timed-text corpus")
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: timed_text_verify.py <live-conformance/captions>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|