Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 15 Packages Wiki Activity Actions 36
open-standards/wai/tools/wai_vector_check.py
Transaction Science fd3de87f4e wai-vector checker: two fixes from the first comparison; the per-fill density bound; the sidecar exercise 
The first comparison run found two faults in the checker, both mine; the text is
clear on each:
  - text_hash hashed from the TEXTS offset up to STRINGS, taking in ITEMS, GLYPHS
    and CLUSTERS. Section 2.11: the TEXTS table followed by the STRINGS area. Fixed.
  - the Wos1 record lacked the slot the load stores a set's bytes in (a crash).

Also:
  - section 3.7's per-fill bound D_r <= 256*p is asserted for every fill, with p its
    stored point count; fills/comb-16382 attains it (gap 0);
  - the corpus carries no section 2.10 sidecar, so for every accepted payload that
    renders, the driver builds the authoring tool's sidecar from the frames, parses
    and rewrites it, checks every probe and clip, and requires sidecar_malformed,
    sidecar_mismatch and layer_mismatch from three corruptions;
  - the probe set's level 1 takes the first and last frame only when the document
    has a frame at that rate, per the text fix that follows.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-11 08:45:34 -04:00

2449 lines
86 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""Independent conformance checker for `wai.graphics.vector`.
Written from the extension text alone: wai/extensions/graphics-vector.md, sections
1-3 (the normative revision), with the notation of integer-operators section 2 and
the digests of SPEC section 7, which that text cites. It shares no code with the
reference crate, and was written without reading it.
Python 3.11 or later, standard library only. Run it with `python3 -I`.
What it computes, each from the section that defines it:
- WOS1 (section 2.1): the load checks in their order, the parse, and the canonical
rewrite from the parsed fields (counts derived, zero fields written as zero);
- WVG1 (sections 2.2-2.5, 2.7): the refusal order, steps 1-9, with the exact code
and step of the first failure; the parse; and the canonical rewrite;
- the load-time interval check, section 3.4(a), and the per-size cost, 3.4(b);
- the frame procedure R(P, S, z, r, g), section 3.10: the clock (3.1), tracks
(3.2), placement (3.3), flattening by the closed form (3.5), the row and cell
splits by the closed-form crossings, the floors of exact rationals in big
integers (3.6), accumulation and the scan (3.7), coverage (3.8) and compositing
(3.9), into RGBA8P (2.9); and RGBA8S;
- the digest sidecar (2.10): its canonical form and checks, the probe digests,
the clip digests, and the authoring tool's probe set;
- the text digests (2.11);
- section 3.12's bounds, asserted on every value the procedure forms.
python3 -I wai_vector_check.py load PAYLOAD [--sets DIR]
python3 -I wai_vector_check.py render PAYLOAD WxH NUM/DEN G [--sets DIR] [--out F]
python3 -I wai_vector_check.py sidecar PAYLOAD SIDECAR [--sets DIR] [--verify]
python3 -I wai_vector_check.py selftest
python3 -I wai_vector_check.py corpus vector-conformance [--groups a,b] [--quiet]
`corpus` recomputes every expected file under vector-conformance/ from its vector and
sets/ alone and requires the committed bytes; rewrites every payload and set to its
own bytes; checks each refusal's step where the vector's name or description pins
one; and recomputes every digest of sha256.json.
Exit status 1 on any failure.
"""
import bisect
import hashlib
import json
import os
import struct
import sys
from collections import namedtuple
# ------------------------------------------------------------------ refusals
class Refusal(Exception):
"""A refusal of section 2.7: its code, the step that found it, and where."""
def __init__(self, code, step, where=""):
super().__init__(f"{code} (step {step}) {where}")
self.code = code
self.step = step
self.where = where
class RequestError(Exception):
"""A render request error of section 2.7: size_unlisted, rate_unlisted, frame_range."""
def __init__(self, code):
super().__init__(code)
self.code = code
class SidecarCode(Exception):
"""sidecar_malformed or sidecar_mismatch (section 2.10)."""
def __init__(self, code, why=""):
super().__init__(f"{code}: {why}")
self.code = code
self.why = why
class BoundError(Exception):
"""A value past a bound section 3.12 (or the section it cites) states."""
# ------------------------------------------------------------------ section 3.12's bounds
#
# Each entry is (kind, limit): "lt" means |v| < limit, "le" means |v| <= limit.
# The checker asserts these on every value it forms; a tightened entry is how the
# falsification shows that a vector reaches a bound.
BOUNDS = {
# 3.2 track evaluation
"track.linear": ("lt", 1 << 56), # 2*delta*r + Delta
"track.table_r": ("lt", 1 << 47), # r*2^16
"track.table_step": ("lt", 1 << 28), # (E[k+1]-E[k])*phi
"track.value_step": ("lt", 1 << 40), # delta*e
# 3.3 placement, and 3.4(a)'s end products
"place.L_num": ("lt", 1 << 58),
"place.L": ("le", 1 << 30),
"place.point_sum": ("lt", 1 << 47), # L*p with its rounding term
"place.origin_num": ("lt", 1 << 58),
# 3.4(a): every device coordinate
"device.lo": ("ge", -(1 << 22)),
"device.hi": ("le_signed", (1 << 22) - 1),
# 3.4(a): every segment after the check
"segment.d": ("lt", 1 << 23),
# 3.5 flattening
"flatten.quad_n": ("le", 9),
"flatten.cubic_n": ("le", 10),
"flatten.quad_num": ("lt", 1 << 43),
"flatten.cubic_num": ("lt", 1 << 57),
# 3.6 row split: closed form, and the 32-bit form's first product
"rows.cross": ("lt", 1 << 46),
"rows.cross32": ("lt", 1 << 31),
# 3.6 cell split
"cells.product": ("lt", 1 << 31),
"cells.cover": ("le", 1 << 8),
"cells.area": ("le", 1 << 17),
# 3.7
"density": ("le", (1 << 22) - 256),
"accum": ("le", (1 << 31) - (1 << 17)),
# 3.8
"union.S": ("le", 1 << 29),
"alpha.product": ("lt", 1 << 25),
# 3.9
"composite.product": ("le", 65025),
"div255.intermediate": ("lt", 1 << 24),
# 2.9
"unpremul.num": ("le", 255 * 255 + 127),
# 3.4(b)
"cost.fill": ("lt", 1 << 28),
"cost.run": ("lt", 1 << 40),
"cost.sum": ("lt", 1 << 58),
}
# Section 3.7's per-fill density bound D_r <= 256*p is asserted with this slack.
DENSITY_SLACK = 0
# The largest |value| each bound saw, so a run can say which bounds a corpus reaches.
BOUND_PEAK = {}
def bound(name, v):
kind, lim = BOUNDS[name]
if kind == "lt":
a = abs(v)
ok = a < lim
elif kind == "le":
a = abs(v)
ok = a <= lim
elif kind == "ge":
a = v
ok = v >= lim
else: # le_signed
a = v
ok = v <= lim
if not ok:
raise BoundError(f"bound {name}: {v} against {kind} {lim}")
p = BOUND_PEAK.get(name)
if kind == "ge":
if p is None or a < p:
BOUND_PEAK[name] = a
elif p is None or a > p:
BOUND_PEAK[name] = a
return v
# ------------------------------------------------------------------ section 3.0 notation
def R_rat(n, d):
"""R_rat(n, d) = floor((2n + d) / (2d)), d > 0 (section 3.0)."""
assert d > 0
return (2 * n + d) // (2 * d)
def R_half_up(p, s):
"""integer-operators section 2: floor((p + 2^(s-1)) / 2^s), and p for s = 0."""
if s == 0:
return p
return (p + (1 << (s - 1))) >> s
def cdiv(a, b):
"""ceil(a / b) = -floor(-a / b), b > 0."""
assert b > 0
return -((-a) // b)
def div255(x):
"""Section 3.9: floor(257*(x + 128) / 2^16), x in [0, 65025]."""
if not 0 <= x <= 65025:
raise BoundError(f"div255 argument {x}")
t = 257 * (x + 128)
bound("div255.intermediate", t)
return t >> 16
def unpremul(px):
"""RGBA8S of one RGBA8P pixel (section 2.9)."""
r, g, b, a = px
if a == 0:
return (0, 0, 0, 0)
out = []
for c in (r, g, b):
bound("unpremul.num", 255 * c + a // 2)
out.append(R_rat(255 * c, a))
return (out[0], out[1], out[2], a)
def rgba8s(buf):
out = bytearray(len(buf))
for i in range(0, len(buf), 4):
out[i:i + 4] = bytes(unpremul(buf[i:i + 4]))
return bytes(out)
def sha256(b):
return hashlib.sha256(b).digest()
# ------------------------------------------------------------------ section 2.1: WOS1
WOS1_HDR = struct.Struct("<4sIHHIIIII")
WOS1_GLYPH = struct.Struct("<HHHHhhhh")
class Wos1Glyph:
__slots__ = ("gid", "n_contours", "n_points", "box", "ends", "points", "prims")
class Wos1:
__slots__ = ("upem2", "flags", "glyphs", "notes", "by_gid", "digest", "length", "bytes")
def _om(where):
return Refusal("outline_malformed", "6", where)
def walk_contour(tags):
"""Section 2.1's contour grammar over one contour's tags.
Returns the emitted primitives as tuples of point indices within the contour,
index m written as 0, or None when the contour is malformed."""
m = len(tags)
if m == 0 or tags[0] != 0:
return None
if m == 1:
return [] # "A contour of one point emits nothing."
def tag(j):
return tags[j if j < m else 0]
prims = []
i = 0
while i < m:
t1 = tag(i + 1)
if t1 == 0:
prims.append(("L", i, (i + 1) % m))
i += 1
elif t1 == 1:
if i + 2 > m or tag(i + 2) != 0:
return None
prims.append(("Q", i, i + 1, (i + 2) % m))
i += 2
elif t1 == 2:
if i + 3 > m or tag(i + 2) != 2 or tag(i + 3) != 0:
return None
prims.append(("C", i, i + 1, i + 2, (i + 3) % m))
i += 3
else:
return None
return prims
def exact_box(points):
if not points:
return (0, 0, 0, 0)
xs = [p[0] for p in points]
ys = [p[1] for p in points]
return (min(xs), min(ys), max(xs), max(ys))
def parse_wos1(b):
"""Section 2.1's load checks, in order; returns the parsed set."""
# 1. magic, format, zero, reserved bits of flags
if len(b) < 32:
raise _om("wos1 step 1: shorter than the header")
magic, fmt, upem2, flags, n_glyphs, points_total, contours_total, notes_len, zero = \
WOS1_HDR.unpack_from(b, 0)
if magic != b"WOS1":
raise _om("wos1 step 1: magic")
if fmt != 1:
raise _om("wos1 step 1: format")
if flags & 0xFFFE:
raise _om("wos1 step 1: reserved flags")
if zero != 0:
raise _om("wos1 step 1: zero")
# 2. the counts, in field order
if n_glyphs > 65536:
raise Refusal("too_large", "6", "wos1 step 2: n_glyphs")
if n_glyphs == 0:
raise _om("wos1 step 2: n_glyphs = 0")
if points_total > 1 << 22:
raise Refusal("too_large", "6", "wos1 step 2: points_total")
if contours_total > 1 << 22:
raise Refusal("too_large", "6", "wos1 step 2: contours_total")
if notes_len > 1 << 20:
raise Refusal("too_large", "6", "wos1 step 2: notes_len")
if notes_len == 0:
raise _om("wos1 step 2: notes_len = 0")
# 3. exact length
exact = 32 + 16 * n_glyphs + 2 * contours_total + 5 * points_total + notes_len
if exact != len(b):
raise _om(f"wos1 step 3: length {len(b)} against {exact}")
# 4. each glyph record
glyphs = []
off = 32
for k in range(n_glyphs):
gid, nc, npt, z, x0, y0, x1, y1 = WOS1_GLYPH.unpack_from(b, off)
off += 16
if npt > 16383:
raise Refusal("points_limit", "6", f"wos1 step 4: glyph {k} n_points")
if nc > npt:
raise _om(f"wos1 step 4: glyph {k} n_contours > n_points")
if npt >= 1 and nc == 0:
raise _om(f"wos1 step 4: glyph {k} n_contours = 0")
if z != 0:
raise _om(f"wos1 step 4: glyph {k} zero")
g = Wos1Glyph()
g.gid, g.n_contours, g.n_points, g.box = gid, nc, npt, (x0, y0, x1, y1)
glyphs.append(g)
# 5. gids strictly increase
for k in range(1, n_glyphs):
if glyphs[k].gid <= glyphs[k - 1].gid:
raise _om(f"wos1 step 5: gid of glyph {k}")
# 6. sums
if sum(g.n_points for g in glyphs) != points_total:
raise _om("wos1 step 6: points_total")
if sum(g.n_contours for g in glyphs) != contours_total:
raise _om("wos1 step 6: contours_total")
# 7. contour ends
ends_off = 32 + 16 * n_glyphs
for k, g in enumerate(glyphs):
g.ends = list(struct.unpack_from(f"<{g.n_contours}H", b, ends_off))
ends_off += 2 * g.n_contours
for j in range(1, len(g.ends)):
if g.ends[j] <= g.ends[j - 1]:
raise _om(f"wos1 step 7: glyph {k} contour ends")
if g.ends and g.ends[-1] != g.n_points - 1:
raise _om(f"wos1 step 7: glyph {k} last end")
# 8. tags
pts_off = ends_off
for g in glyphs:
pts = []
for _ in range(g.n_points):
pts.append(struct.unpack_from("<hhB", b, pts_off))
pts_off += 5
g.points = pts
for k, g in enumerate(glyphs):
for p in g.points:
if p[2] > 2:
raise _om(f"wos1 step 8: glyph {k} tag {p[2]}")
# 9. contour grammar
for k, g in enumerate(glyphs):
g.prims = contours_of(g.ends, g.points)
if g.prims is None:
raise _om(f"wos1 step 9: glyph {k} contour grammar")
# 10. exact box
for k, g in enumerate(glyphs):
if exact_box(g.points) != g.box:
raise _om(f"wos1 step 10: glyph {k} box")
# 11. notes
notes = b[pts_off:]
assert len(notes) == notes_len
try:
notes.decode("utf-8", errors="strict")
except UnicodeDecodeError:
raise _om("wos1 step 11: notes are not UTF-8")
s = Wos1()
s.upem2, s.flags, s.glyphs, s.notes = upem2, flags, glyphs, bytes(notes)
s.by_gid = {g.gid: g for g in glyphs}
s.digest = sha256(b)
s.length = len(b)
return s
def contours_of(ends, points):
"""Per contour: (start index, its primitives with indices into `points`), or None."""
out = []
start = 0
for e in ends:
tags = [points[j][2] for j in range(start, e + 1)]
prims = walk_contour(tags)
if prims is None:
return None
out.append((start, [(p[0],) + tuple(start + j for j in p[1:]) for p in prims]))
start = e + 1
return out
def write_wos1(s):
"""Section 2.1's layout from the parsed fields; counts derived, zero written as 0."""
out = bytearray()
pts_total = sum(len(g.points) for g in s.glyphs)
con_total = sum(len(g.ends) for g in s.glyphs)
out += WOS1_HDR.pack(b"WOS1", 1, s.upem2, s.flags, len(s.glyphs), pts_total, con_total,
len(s.notes), 0)
for g in s.glyphs:
out += WOS1_GLYPH.pack(g.gid, len(g.ends), len(g.points), 0, *exact_box(g.points))
for g in s.glyphs:
out += struct.pack(f"<{len(g.ends)}H", *g.ends)
for g in s.glyphs:
for p in g.points:
out += struct.pack("<hhB", *p)
out += s.notes
return bytes(out)
# ------------------------------------------------------------------ sections 2.2-2.4: WVG1
HDR = struct.Struct("<4sIIIIHHBBBBHHHHIIIIIIIIIQ")
HDR_FIELDS = ("magic", "format", "tb", "t0", "dur", "design_w", "design_h", "n_sizes",
"n_rates", "n_sets", "n_tables", "n_paints", "n_texts", "n_tracks", "n_paths",
"n_items", "n_keys", "n_glyphs", "n_clusters", "path_contours", "path_points",
"strings_len", "inline_len", "held_len", "zero")
assert HDR.size == 80
# Step 2's caps, in field order.
CAPS = (("n_sizes", 8), ("n_rates", 4), ("n_sets", 64), ("n_tables", 32), ("n_paints", 4096),
("n_paths", 4096), ("n_items", 65536), ("n_keys", 1 << 22), ("n_glyphs", 1 << 20),
("n_clusters", 1 << 20), ("path_contours", 1 << 20), ("path_points", 1 << 22),
("strings_len", 1 << 22), ("inline_len", 1 << 25), ("held_len", 1 << 26))
S_SIZE = struct.Struct("<HHI")
S_RATE = struct.Struct("<II")
S_SET = struct.Struct("<B3sI32s")
S_TABLE = struct.Struct("<17I")
S_PAINT = struct.Struct("<4B")
S_TRACK = struct.Struct("<BBHI")
S_KEY = struct.Struct("<IiB3s")
S_PATH = struct.Struct("<HHIhhhh")
S_PEND = struct.Struct("<H")
S_PPT = struct.Struct("<hhB")
S_TEXT = struct.Struct("<BBBBHHIII")
S_ITEM = struct.Struct("<BBHHHHHHHIII7H6s")
S_GLYPH = struct.Struct("<HHii")
S_CLUSTER = struct.Struct("<HHII")
for _s, _n in ((S_SIZE, 8), (S_RATE, 8), (S_SET, 40), (S_TABLE, 68), (S_PAINT, 4), (S_TRACK, 8),
(S_KEY, 12), (S_PATH, 16), (S_PEND, 2), (S_PPT, 5), (S_TEXT, 20), (S_ITEM, 48),
(S_GLYPH, 12), (S_CLUSTER, 12)):
assert _s.size == _n
Size = namedtuple("Size", "w h cost")
Rate = namedtuple("Rate", "num den")
SetRec = namedtuple("SetRec", "carriage zero length sha256")
Paint = namedtuple("Paint", "r g b a")
Key = namedtuple("Key", "t value interp zero")
Text = namedtuple("Text", "purpose dir lang_len mode group ruby ruby_start ruby_len str_len")
Item = namedtuple("Item", "kind rule zero0 ref paint text zero1 n_glyphs n_clusters t_on t_off "
"unit_scale slots zero2")
Glyph = namedtuple("Glyph", "gid zero x y")
Cluster = namedtuple("Cluster", "glyph_count zero byte_first byte_len")
SLOT_NAMES = ("m00", "m01", "m10", "m11", "tx", "ty", "op")
SLOT_DEFAULT = (65536, 0, 0, 65536, 0, 0, 255)
SLOT_KIND = (3, 3, 3, 3, 2, 2, 1)
ABSENT = 0xFFFF
T31 = 1 << 31
class TrackRec:
__slots__ = ("kind", "zero0", "zero1", "n_keys", "keys", "ts", "vs", "lo", "hi")
class PathRec:
__slots__ = ("n_contours", "n_points", "zero", "box", "ends", "points", "prims")
class Payload:
"""A loaded WVG1 and the sets it names."""
def exact_length(H):
return (80 + 8 * H["n_sizes"] + 8 * H["n_rates"] + 40 * H["n_sets"] + 68 * H["n_tables"]
+ 4 * H["n_paints"] + 8 * H["n_tracks"] + 12 * H["n_keys"] + 16 * H["n_paths"]
+ 2 * H["path_contours"] + 5 * H["path_points"] + 20 * H["n_texts"]
+ 48 * H["n_items"] + 12 * H["n_glyphs"] + 12 * H["n_clusters"]
+ H["strings_len"] + H["inline_len"])
def utf8_ok(b):
try:
b.decode("utf-8", errors="strict")
return True
except UnicodeDecodeError:
return False
def string_ok(b):
"""Section 2.5 rule 6: valid UTF-8, no C0 but U+000A, no U+007F, no C1."""
try:
s = b.decode("utf-8", errors="strict")
except UnicodeDecodeError:
return False
for ch in s:
o = ord(ch)
if (o < 0x20 and o != 0x0A) or o == 0x7F or 0x80 <= o <= 0x9F:
return False
return True
def tag_ok(b):
"""Section 2.5 rule 6: ASCII letters, digits and hyphens; no leading, trailing
or doubled hyphen."""
if not b:
return False
for c in b:
if not (0x30 <= c <= 0x39 or 0x41 <= c <= 0x5A or 0x61 <= c <= 0x7A or c == 0x2D):
return False
if b[0] == 0x2D or b[-1] == 0x2D or b"--" in b:
return False
return True
def cp_boundary(b, o):
"""An offset within b is a code-point boundary of the valid UTF-8 string b."""
if o == 0 or o == len(b):
return True
if o < 0 or o > len(b):
return False
return (b[o] & 0xC0) != 0x80
def gcd(a, b):
while b:
a, b = b, a % b
return a
def load(payload, store=None):
"""Section 2.7's load, steps 1-9. `store` maps a SHA-256 digest to the bytes of a
held set the sink holds (its resolution policy). Returns a Payload, or raises
Refusal with the first failure."""
store = store or {}
b = payload
# ---- step 1
if len(b) < 80:
raise Refusal("payload_malformed", "1", "shorter than the header")
hv = HDR.unpack_from(b, 0)
H = dict(zip(HDR_FIELDS, hv))
if H["magic"] != b"WVG1":
raise Refusal("payload_malformed", "1", "magic")
if H["format"] != 1:
raise Refusal("payload_malformed", "1", "format")
if H["zero"] != 0:
raise Refusal("payload_malformed", "1", "header zero")
# ---- step 2
for name, cap in CAPS:
if H[name] > cap:
raise Refusal("too_large", "2", name)
if H["inline_len"] + H["held_len"] > 1 << 26:
raise Refusal("too_large", "2", "inline_len + held_len")
exact = exact_length(H)
if exact > 1 << 26:
raise Refusal("too_large", "2", "exact length")
# ---- step 3
if len(b) != exact:
raise Refusal("payload_malformed", "3", f"length {len(b)} against {exact}")
# ---- step 4: the record pass
P = Payload()
P.bytes = bytes(b)
P.H = H
def fr(where):
return Refusal("field_range", "4", where)
def pm(where):
return Refusal("payload_malformed", "4", where)
tb, t0, dur, dw, dh = H["tb"], H["t0"], H["dur"], H["design_w"], H["design_h"]
if not 1 <= tb <= 1 << 20:
raise fr("tb")
if not 0 <= t0 <= T31 - 1:
raise fr("t0")
if not 1 <= dur <= T31 - t0:
raise fr("dur")
if not 16 <= dw <= 4096:
raise fr("design_w")
if not 16 <= dh <= 4096:
raise fr("design_h")
if H["n_sizes"] < 1:
raise fr("n_sizes")
if H["n_rates"] < 1:
raise fr("n_rates")
if H["n_items"] < 1:
raise fr("n_items")
off = 80
sizes = []
for k in range(H["n_sizes"]):
w, h, cost = S_SIZE.unpack_from(b, off)
off += 8
if not 16 <= w <= 4096 or not 8 * w >= dw or not w <= 4 * dw:
raise fr(f"SIZES[{k}].w")
if not 16 <= h <= 4096 or w * dh != h * dw:
raise fr(f"SIZES[{k}].h")
sizes.append(Size(w, h, cost))
rates = []
for k in range(H["n_rates"]):
num, den = S_RATE.unpack_from(b, off)
off += 8
if not 1 <= num <= 1 << 20:
raise fr(f"RATES[{k}].num")
if not 1 <= den <= 1 << 20:
raise fr(f"RATES[{k}].den")
if gcd(num, den) != 1 or num > 240 * den or (den * tb) % num != 0:
raise fr(f"RATES[{k}].den")
rates.append(Rate(num, den))
sets = []
for k in range(H["n_sets"]):
carriage, z, length, digest = S_SET.unpack_from(b, off)
off += 40
if carriage > 1:
raise fr(f"SETS[{k}].carriage")
if z != b"\0\0\0":
raise pm(f"SETS[{k}].zero")
if not 32 <= length <= 1 << 25:
raise fr(f"SETS[{k}].length")
sets.append(SetRec(carriage, z, length, digest))
tables = []
for k in range(H["n_tables"]):
E = S_TABLE.unpack_from(b, off)
off += 68
for i, e in enumerate(E):
if e > 65536 or (i == 0 and e != 0) or (i == 16 and e != 65536):
raise fr(f"TABLES[{k}].E[{i}]")
tables.append(tuple(E))
paints = []
for k in range(H["n_paints"]):
r, g, bb, a = S_PAINT.unpack_from(b, off)
off += 4
if r > a or g > a or bb > a:
raise fr(f"PAINTS[{k}].a")
paints.append(Paint(r, g, bb, a))
tracks = []
for k in range(H["n_tracks"]):
kind, z0, z1, nk = S_TRACK.unpack_from(b, off)
off += 8
if kind not in (1, 2, 3):
raise fr(f"TRACKS[{k}].kind")
if z0 != 0:
raise pm(f"TRACKS[{k}].zero(u8)")
if z1 != 0:
raise pm(f"TRACKS[{k}].zero(u16)")
if not 1 <= nk <= 4096:
raise fr(f"TRACKS[{k}].n_keys")
t = TrackRec()
t.kind, t.zero0, t.zero1, t.n_keys = kind, z0, z1, nk
tracks.append(t)
keys = []
for k in range(H["n_keys"]):
t, v, interp, z = S_KEY.unpack_from(b, off)
off += 12
if t >= T31:
raise fr(f"KEYS[{k}].t")
if interp > 1 + H["n_tables"]:
raise fr(f"KEYS[{k}].interp")
if z != b"\0\0\0":
raise pm(f"KEYS[{k}].zero")
keys.append(Key(t, v, interp, z))
paths = []
for k in range(H["n_paths"]):
nc, npt, z, x0, y0, x1, y1 = S_PATH.unpack_from(b, off)
off += 16
if nc < 1:
raise fr(f"PATHS[{k}].n_contours")
if npt > 16383:
raise Refusal("points_limit", "4", f"PATHS[{k}].n_points")
if npt < 1 or nc > npt:
raise fr(f"PATHS[{k}].n_points")
if z != 0:
raise pm(f"PATHS[{k}].zero")
p = PathRec()
p.n_contours, p.n_points, p.zero, p.box = nc, npt, z, (x0, y0, x1, y1)
paths.append(p)
pends = []
for k in range(H["path_contours"]):
pends.append(S_PEND.unpack_from(b, off)[0])
off += 2
ppts = []
for k in range(H["path_points"]):
x, y, tag = S_PPT.unpack_from(b, off)
off += 5
if tag > 2:
raise fr(f"PATH_POINTS[{k}].tag")
ppts.append((x, y, tag))
texts = []
for k in range(H["n_texts"]):
t = Text(*S_TEXT.unpack_from(b, off))
off += 20
if t.purpose == 0:
raise fr(f"TEXTS[{k}].purpose")
if t.dir > 1:
raise fr(f"TEXTS[{k}].dir")
if not 1 <= t.lang_len <= 63:
raise fr(f"TEXTS[{k}].lang_len")
if t.mode > 2:
raise fr(f"TEXTS[{k}].mode")
if t.ruby > H["n_texts"]:
raise fr(f"TEXTS[{k}].ruby")
if t.ruby == 0 and t.ruby_start != 0:
raise fr(f"TEXTS[{k}].ruby_start")
if (t.ruby == 0 and t.ruby_len != 0) or (t.ruby != 0 and t.ruby_len < 1):
raise fr(f"TEXTS[{k}].ruby_len")
if not 1 <= t.str_len <= 4096:
raise fr(f"TEXTS[{k}].str_len")
texts.append(t)
items = []
for k in range(H["n_items"]):
v = S_ITEM.unpack_from(b, off)
off += 48
it = Item(v[0], v[1], v[2], v[3], v[4], v[5], v[6], v[7], v[8], v[9], v[10], v[11],
tuple(v[12:19]), v[19])
check_item_fields(it, k, H)
items.append(it)
glyphs = []
for k in range(H["n_glyphs"]):
gid, z, x, y = S_GLYPH.unpack_from(b, off)
off += 12
if z != 0:
raise pm(f"GLYPHS[{k}].zero")
if not -(1 << 23) <= x < 1 << 23:
raise fr(f"GLYPHS[{k}].x")
if not -(1 << 23) <= y < 1 << 23:
raise fr(f"GLYPHS[{k}].y")
glyphs.append(Glyph(gid, z, x, y))
clusters = []
for k in range(H["n_clusters"]):
gc, z, bf, bl = S_CLUSTER.unpack_from(b, off)
off += 12
if z != 0:
raise pm(f"CLUSTERS[{k}].zero")
clusters.append(Cluster(gc, z, bf, bl))
strings_off = off
strings = b[off:off + H["strings_len"]]
off += H["strings_len"]
inline = b[off:off + H["inline_len"]]
off += H["inline_len"]
assert off == len(b)
texts_off = strings_off - 12 * H["n_clusters"] - 12 * H["n_glyphs"] - 48 * H["n_items"] \
- 20 * H["n_texts"]
P.text_hash = sha256(b[texts_off:texts_off + 20 * H["n_texts"]] + strings)
# ---- step 5: cross-record structure
def pm5(sub, where):
return Refusal("payload_malformed", "5" + sub, where)
# (a) the sum rules
if sum(s.length for s in sets if s.carriage == 1) != H["inline_len"]:
raise pm5("a", "inline set lengths")
if sum(s.length for s in sets if s.carriage == 0) != H["held_len"]:
raise pm5("a", "held set lengths")
if sum(t.n_keys for t in tracks) != H["n_keys"]:
raise pm5("a", "track n_keys")
if sum(p.n_contours for p in paths) != H["path_contours"]:
raise pm5("a", "path n_contours")
if sum(p.n_points for p in paths) != H["path_points"]:
raise pm5("a", "path n_points")
if sum(t.lang_len + t.str_len for t in texts) != H["strings_len"]:
raise pm5("a", "text lengths")
runs = [it for it in items if it.kind == 3]
if sum(it.n_glyphs for it in runs) != H["n_glyphs"]:
raise pm5("a", "run n_glyphs")
if sum(it.n_clusters for it in runs) != H["n_clusters"]:
raise pm5("a", "run n_clusters")
# ownership, now that the sums hold
o = 0
for t in tracks:
t.keys = keys[o:o + t.n_keys]
o += t.n_keys
t.ts = [k.t for k in t.keys]
t.vs = [k.value for k in t.keys]
t.lo, t.hi = min(t.vs), max(t.vs)
oe = op_ = 0
for p in paths:
p.ends = pends[oe:oe + p.n_contours]
oe += p.n_contours
p.points = ppts[op_:op_ + p.n_points]
op_ += p.n_points
so = 0
text_lang, text_str = [], []
for t in texts:
text_lang.append(bytes(strings[so:so + t.lang_len]))
so += t.lang_len
text_str.append(bytes(strings[so:so + t.str_len]))
so += t.str_len
run_glyphs, run_clusters = {}, {}
og = oc = 0
for i, it in enumerate(items):
if it.kind == 3:
run_glyphs[i] = glyphs[og:og + it.n_glyphs]
og += it.n_glyphs
run_clusters[i] = clusters[oc:oc + it.n_clusters]
oc += it.n_clusters
# (b) ordering
for k in range(1, len(sizes)):
if sizes[k].w <= sizes[k - 1].w:
raise pm5("b", f"SIZES[{k}] order")
for k in range(1, len(rates)):
a, c = rates[k - 1], rates[k]
if not a.num * c.den < c.num * a.den:
raise pm5("b", f"RATES[{k}] order")
for ti, t in enumerate(tracks):
for j in range(1, len(t.keys)):
if t.keys[j].t <= t.keys[j - 1].t:
raise pm5("b", f"TRACKS[{ti}] key {j} order")
if t.keys[-1].interp != 0:
raise pm5("b", f"TRACKS[{ti}] last interp")
# (c) typing
ki = 0
for t in tracks:
for k in t.keys:
lo, hi = {1: (0, 255), 2: (-(1 << 22), 1 << 22), 3: (-(1 << 20), 1 << 20)}[t.kind]
if not lo <= k.value <= hi:
raise Refusal("field_range", "5c", f"KEYS[{ki}].value")
ki += 1
for i, it in enumerate(items):
if it.kind in (0, 2, 3):
for s, ti in enumerate(it.slots):
if ti != ABSENT and tracks[ti].kind != SLOT_KIND[s]:
raise Refusal("field_range", "5c", f"ITEMS[{i}].{SLOT_NAMES[s]}")
# (d) group balance and depth
depth = 0
open_ = []
match = {}
parents = [] # for each item, the group begins enclosing it, outermost first
for i, it in enumerate(items):
if it.kind == 0:
parents.append(tuple(open_))
depth += 1
if depth > 4:
raise pm5("d", f"ITEMS[{i}] depth 5")
open_.append(i)
elif it.kind == 1:
if depth == 0:
raise pm5("d", f"ITEMS[{i}] group end with no group open")
depth -= 1
match[open_.pop()] = i
parents.append(tuple(open_))
else:
parents.append(tuple(open_))
if depth != 0:
raise pm5("d", "a group is open after the last item")
# (e) paths: contour ends, grammar, box
for pi, p in enumerate(paths):
for j in range(1, len(p.ends)):
if p.ends[j] <= p.ends[j - 1]:
raise pm5("e", f"PATHS[{pi}] contour ends")
if p.ends[-1] != p.n_points - 1:
raise pm5("e", f"PATHS[{pi}] last end")
p.prims = contours_of(p.ends, p.points)
if p.prims is None:
raise pm5("e", f"PATHS[{pi}] contour grammar")
if exact_box(p.points) != p.box:
raise pm5("e", f"PATHS[{pi}] box")
# (f) STRINGS
for k in range(len(texts)):
if not tag_ok(text_lang[k]):
raise pm5("f", f"TEXTS[{k}] language tag")
if not string_ok(text_str[k]):
raise pm5("f", f"TEXTS[{k}] string")
# (g) the canonical rules of section 2.5
canonical_rules(items, sets, paints, tracks, paths, tables, texts, text_str, run_clusters)
# (h) a path's text naming a decorative block
for i, it in enumerate(items):
if it.kind == 2 and it.text != ABSENT and texts[it.text].purpose == 13:
raise pm5("h", f"ITEMS[{i}].text names a decorative block")
# ---- step 6: sets, one at a time in SET order
parsed_sets = []
io = 0
for k, s in enumerate(sets):
if s.carriage == 1:
data = bytes(inline[io:io + s.length])
io += s.length
else:
data = store.get(bytes(s.sha256))
if data is None:
raise Refusal("outline_not_held", "6", f"SETS[{k}]")
data = data[:s.length + 1]
if len(data) != s.length or sha256(data) != bytes(s.sha256):
raise Refusal("outline_mismatch", "6", f"SETS[{k}]")
try:
ws = parse_wos1(data)
except Refusal as r:
raise Refusal(r.code, "6", f"SETS[{k}] {r.where}")
ws.bytes = data
parsed_sets.append(ws)
# ---- step 7: glyph_missing
for i, it in enumerate(items):
if it.kind == 3:
ws = parsed_sets[it.ref]
for j, g in enumerate(run_glyphs[i]):
if g.gid not in ws.by_gid:
raise Refusal("glyph_missing", "7", f"ITEMS[{i}] glyph {j}")
P.sizes, P.rates, P.sets, P.tables, P.paints = sizes, rates, sets, tables, paints
P.tracks, P.keys, P.paths, P.texts, P.items = tracks, keys, paths, texts, items
P.glyphs, P.clusters, P.strings, P.inline = glyphs, clusters, bytes(strings), bytes(inline)
P.text_lang, P.text_str = text_lang, text_str
P.run_glyphs, P.run_clusters = run_glyphs, run_clusters
P.match, P.parents = match, parents
P.wos = parsed_sets
# ---- step 8: the interval check, at every listed size
P.device = {}
for z in sizes:
P.device[(z.w, z.h)] = interval_check(P, z.w, z.h)
# ---- step 9: the cost, in SIZE order
P.costs = []
for z in sizes:
c = size_cost(P, z.w, z.h)
P.costs.append(c)
if c > (1 << 32) - 1:
raise Refusal("too_large", "9", f"cost {c} at {z.w}x{z.h}")
if c != z.cost:
raise Refusal("payload_malformed", "9", f"cost {c} at {z.w}x{z.h}, stated {z.cost}")
return P
def check_item_fields(it, k, H):
"""Section 2.4's per-kind rules, in field offset order (section 2.7 step 4)."""
def fr(f):
return Refusal("field_range", "4", f"ITEMS[{k}].{f}")
def pm(f):
return Refusal("payload_malformed", "4", f"ITEMS[{k}].{f}")
kind = it.kind
if kind > 3:
raise fr("kind")
# rule (1)
if kind in (1, 3):
if it.rule != 0:
raise fr("rule")
elif it.rule > 1:
raise fr("rule")
# zero (2)
if it.zero0 != 0:
raise pm("zero@2")
# ref (4)
if kind == 0:
if it.ref != ABSENT and it.ref >= H["n_paths"]:
raise fr("ref")
if it.ref == ABSENT and it.rule != 0:
raise fr("ref (rule with no clip)")
elif kind == 1:
if it.ref != 0:
raise fr("ref")
elif kind == 2:
if it.ref >= H["n_paths"]:
raise fr("ref")
else:
if it.ref >= H["n_sets"]:
raise fr("ref")
# paint (6)
if kind == 0:
if it.paint != ABSENT:
raise fr("paint")
elif kind == 1:
if it.paint != 0:
raise fr("paint")
elif it.paint >= H["n_paints"]:
raise fr("paint")
# text (8)
if kind == 0:
if it.text != ABSENT:
raise fr("text")
elif kind == 1:
if it.text != 0:
raise fr("text")
elif kind == 2:
if it.text != ABSENT and it.text >= H["n_texts"]:
raise fr("text")
elif it.text >= H["n_texts"]:
raise fr("text")
# zero (10)
if it.zero1 != 0:
raise pm("zero@10")
# n_glyphs (12), n_clusters (14)
if kind == 3:
if not 1 <= it.n_glyphs <= 4096:
raise fr("n_glyphs")
if not 1 <= it.n_clusters <= 8192:
raise fr("n_clusters")
else:
if it.n_glyphs != 0:
raise fr("n_glyphs")
if it.n_clusters != 0:
raise fr("n_clusters")
# t_on (16), t_off (20)
if kind == 1:
if it.t_on != 0:
raise fr("t_on")
if it.t_off != 0:
raise fr("t_off")
else:
if it.t_on >= T31:
raise fr("t_on")
if it.t_off > T31 or not it.t_on < it.t_off:
raise fr("t_off")
# unit_scale (24)
us = it.unit_scale
if kind == 1:
if us != 0:
raise fr("unit_scale")
elif kind == 0:
if us > 1 << 24 or (it.ref == ABSENT and us != 0) or (it.ref != ABSENT and us < 1):
raise fr("unit_scale")
elif not 1 <= us <= 1 << 24:
raise fr("unit_scale")
# track slots (28..41)
for s, ti in enumerate(it.slots):
name = SLOT_NAMES[s]
if kind == 1:
if ti != 0:
raise fr(name)
elif kind == 0 and s < 4:
if ti != ABSENT:
raise fr(name)
elif ti != ABSENT and ti >= H["n_tracks"]:
raise fr(name)
# zero (42)
if it.zero2 != b"\0" * 6:
raise pm("zero@42")
def canonical_rules(items, sets, paints, tracks, paths, tables, texts, text_str, run_clusters):
"""Section 2.5 rules 2-5 (section 2.7 step 5(g)), in that order."""
def pm(where):
return Refusal("payload_malformed", "5g", where)
# rule 2: first-reference order
nxt = {"path": 0, "set": 0, "paint": 0, "track": 0}
size = {"path": len(paths), "set": len(sets), "paint": len(paints), "track": len(tracks)}
def ref(pool, x, where):
if x == nxt[pool]:
nxt[pool] += 1
elif x > nxt[pool]:
raise pm(f"rule 2: {pool} {x} first referenced before {pool} {nxt[pool]} ({where})")
for i, it in enumerate(items):
if it.kind == 1:
continue
if it.ref != ABSENT:
ref("set" if it.kind == 3 else "path", it.ref, f"ITEMS[{i}].ref")
if it.paint != ABSENT:
ref("paint", it.paint, f"ITEMS[{i}].paint")
for s, ti in enumerate(it.slots):
if ti != ABSENT:
ref("track", ti, f"ITEMS[{i}].{SLOT_NAMES[s]}")
for pool in ("path", "set", "paint", "track"):
if nxt[pool] != size[pool]:
raise pm(f"rule 2: {pool} {nxt[pool]} unreferenced")
tn = 0
for ti, t in enumerate(tracks):
for k in t.keys:
if k.interp >= 2:
j = k.interp - 2
if j == tn:
tn += 1
elif j > tn:
raise pm(f"rule 2: table {j} first referenced before table {tn}")
if tn != len(tables):
raise pm(f"rule 2: table {tn} unreferenced")
# rule 3: no duplicates, by hashing each pool
def nodup(pool, keys):
seen = set()
for k, v in enumerate(keys):
if v in seen:
raise pm(f"rule 3: duplicate {pool} {k}")
seen.add(v)
nodup("set", [bytes(s.sha256) for s in sets])
nodup("paint", [tuple(p) for p in paints])
nodup("track", [(t.kind, tuple((k.t, k.value, k.interp) for k in t.keys)) for t in tracks])
nodup("path", [(p.n_contours, p.n_points, p.box, tuple(p.ends), tuple(p.points))
for p in paths])
nodup("table", list(tables))
# rule 4: text blocks
named = set()
for it in items:
if it.kind == 3 or (it.kind == 2 and it.text != ABSENT):
named.add(it.text)
for k in range(len(texts)):
if k not in named:
raise pm(f"rule 4: block {k} is drawn by nothing")
mx = 0
for k, t in enumerate(texts):
if t.group != 0 and t.group > mx + 1:
raise pm(f"rule 4: block {k} group number skips")
mx = max(mx, t.group)
seen_groups = set()
for k, t in enumerate(texts):
if t.group != 0:
if t.group in seen_groups and texts[k - 1].group != t.group:
raise pm(f"rule 4: group {t.group} not contiguous at block {k}")
seen_groups.add(t.group)
for k, t in enumerate(texts):
if t.ruby != 0:
base = t.ruby - 1
if not base < k:
raise pm(f"rule 4: block {k} ruby names block {base}, not an earlier one")
s = text_str[base]
a, e = t.ruby_start, t.ruby_start + t.ruby_len
if e > len(s) or not cp_boundary(s, a) or not cp_boundary(s, e):
raise pm(f"rule 4: block {k} ruby range")
# rule 5: clusters of each run
for i, it in enumerate(items):
if it.kind != 3:
continue
cl = run_clusters[i]
if sum(c.glyph_count for c in cl) != it.n_glyphs:
raise pm(f"rule 5: ITEMS[{i}] cluster glyph counts")
s = text_str[it.text]
for j, c in enumerate(cl):
a, e = c.byte_first, c.byte_first + c.byte_len
if e > len(s) or not cp_boundary(s, a) or not cp_boundary(s, e):
raise pm(f"rule 5: ITEMS[{i}] cluster {j} range")
if c.glyph_count == 0 and c.byte_len < 1:
raise pm(f"rule 5: ITEMS[{i}] cluster {j} empty")
if c.byte_len == 0 and c.glyph_count < 1:
raise pm(f"rule 5: ITEMS[{i}] cluster {j} empty")
def write_wvg1(P):
"""Sections 2.2-2.4's layout from the parsed fields: every count derived from the
tables, every zero field written as 0."""
out = bytearray()
inline = b"".join(ws.bytes for s, ws in zip(P.sets, P.wos) if s.carriage == 1) \
if hasattr(P, "wos") else P.inline
held = sum(s.length for s in P.sets if s.carriage == 0)
strings = b"".join(l + s for l, s in zip(P.text_lang, P.text_str))
keys = [k for t in P.tracks for k in t.keys]
ends = [e for p in P.paths for e in p.ends]
pts = [q for p in P.paths for q in p.points]
runs = [i for i, it in enumerate(P.items) if it.kind == 3]
glyphs = [g for i in runs for g in P.run_glyphs[i]]
clusters = [c for i in runs for c in P.run_clusters[i]]
H = P.H
out += HDR.pack(b"WVG1", 1, H["tb"], H["t0"], H["dur"], H["design_w"], H["design_h"],
len(P.sizes), len(P.rates), len(P.sets), len(P.tables), len(P.paints),
len(P.texts), len(P.tracks), len(P.paths), len(P.items), len(keys),
len(glyphs), len(clusters), len(ends), len(pts), len(strings), len(inline),
held, 0)
for z in P.sizes:
out += S_SIZE.pack(z.w, z.h, z.cost)
for r in P.rates:
out += S_RATE.pack(r.num, r.den)
for s in P.sets:
out += S_SET.pack(s.carriage, b"\0\0\0", s.length, s.sha256)
for t in P.tables:
out += S_TABLE.pack(*t)
for p in P.paints:
out += S_PAINT.pack(*p)
for t in P.tracks:
out += S_TRACK.pack(t.kind, 0, 0, len(t.keys))
for k in keys:
out += S_KEY.pack(k.t, k.value, k.interp, b"\0\0\0")
for p in P.paths:
out += S_PATH.pack(len(p.ends), len(p.points), 0, *exact_box(p.points))
for e in ends:
out += S_PEND.pack(e)
for q in pts:
out += S_PPT.pack(*q)
for t, l, s in zip(P.texts, P.text_lang, P.text_str):
out += S_TEXT.pack(t.purpose, t.dir, len(l), t.mode, t.group, t.ruby, t.ruby_start,
t.ruby_len, len(s))
for i, it in enumerate(P.items):
ng = len(P.run_glyphs[i]) if it.kind == 3 else 0
nc = len(P.run_clusters[i]) if it.kind == 3 else 0
out += S_ITEM.pack(it.kind, it.rule, 0, it.ref, it.paint, it.text, 0, ng, nc, it.t_on,
it.t_off, it.unit_scale, *it.slots, b"\0" * 6)
for g in glyphs:
out += S_GLYPH.pack(g.gid, 0, g.x, g.y)
for c in clusters:
out += S_CLUSTER.pack(c.glyph_count, 0, c.byte_first, c.byte_len)
out += strings
out += inline
return bytes(out)
def block_identity(P, k):
"""Section 2.11: a block's identity."""
t = P.texts[k]
return sha256(b"wai:text-block" + b"\x01" + bytes([t.purpose, t.dir, t.mode, t.lang_len])
+ P.text_lang[k] + struct.pack("<I", t.str_len) + P.text_str[k])
# ------------------------------------------------------------------ section 3.4
def iv_scale(iv, k):
a, b = iv[0] * k, iv[1] * k
return (min(a, b), max(a, b))
def iv_add(a, b):
return (a[0] + b[0], a[1] + b[1])
def hull(I, J):
ps = (I[0] * J[0], I[0] * J[1], I[1] * J[0], I[1] * J[1])
return (min(ps), max(ps))
def slot_interval(P, it, s):
ti = it.slots[s]
if ti == ABSENT:
return (SLOT_DEFAULT[s], SLOT_DEFAULT[s])
t = P.tracks[ti]
return (t.lo, t.hi)
def instance_intervals(P, w, Mx, My, O, s, records):
"""Section 3.4(a) steps 3-4 for the records (g_x, g_y, box) of one item; returns
each record's device interval ((x_lo, x_hi), (y_lo, y_hi))."""
dw = P.H["design_w"]
den = dw << 16
def Liv(m):
lo_n, hi_n = m[0] * s * w, m[1] * s * w
bound("place.L_num", lo_n)
bound("place.L_num", hi_n)
L = (R_rat(lo_n, den), R_rat(hi_n, den))
bound("place.L", L[0])
bound("place.L", L[1])
return L
L00, L01 = Liv(Mx[0]), Liv(Mx[1])
L10, L11 = Liv(My[0]), Liv(My[1])
out = []
for gx, gy, box in records:
xmin, ymin, xmax, ymax = box
res = []
for Oa, m1, m2, L1, L2 in ((O[0], Mx[0], Mx[1], L00, L01), (O[1], My[0], My[1], L10, L11)):
numr = iv_add(iv_add(iv_scale(Oa, 1 << 16), iv_scale(m1, gx)), iv_scale(m2, gy))
for e in numr:
bound("place.origin_num", e * w)
org = (R_rat(numr[0] * w, den), R_rat(numr[1] * w, den))
pt = iv_add(hull(L1, (xmin, xmax)), hull(L2, (ymin, ymax)))
for e in pt:
bound("place.point_sum", e + (1 << 15))
pt = (R_half_up(pt[0], 16), R_half_up(pt[1], 16))
res.append(iv_add(org, pt))
out.append((res[0], res[1]))
return out
def interval_check(P, w, h):
"""Section 3.4(a) at one size. Returns {item index: [device intervals]}; raises
geometry_range."""
lim_lo, lim_hi = -(1 << 22), (1 << 22) - 1
dev = {}
ident = ((65536, 65536), (0, 0))
for i, it in enumerate(P.items):
if it.kind == 1 or (it.kind == 0 and it.ref == ABSENT):
continue
groups = P.parents[i]
if it.kind == 0:
chain = groups + (i,)
Mx, My = ident, ((0, 0), (65536, 65536))
else:
chain = groups
Mx = (slot_interval(P, it, 0), slot_interval(P, it, 1))
My = (slot_interval(P, it, 2), slot_interval(P, it, 3))
Ox, Oy = (0, 0), (0, 0)
if it.kind != 0:
Ox, Oy = slot_interval(P, it, 4), slot_interval(P, it, 5)
for gi in chain:
g = P.items[gi]
Ox = iv_add(Ox, slot_interval(P, g, 4))
Oy = iv_add(Oy, slot_interval(P, g, 5))
if it.kind == 3:
ws = P.wos[it.ref]
recs = [(g.x, g.y, ws.by_gid[g.gid].box) for g in P.run_glyphs[i]]
else:
recs = [(0, 0, P.paths[it.ref].box)]
ivs = instance_intervals(P, w, Mx, My, (Ox, Oy), it.unit_scale, recs)
for k, (xi, yi) in enumerate(ivs):
for e in xi + yi:
if not lim_lo <= e <= lim_hi:
raise Refusal("geometry_range", "8",
f"{w}x{h}: ITEMS[{i}] record {k} interval {xi}x{yi}")
dev[i] = ivs
return dev
def fill_box(xi, yi, w, h):
c_lo = max(0, xi[0] // 256)
c_hi = min(w - 1, xi[1] // 256)
r_lo = max(0, yi[0] // 256)
r_hi = min(h - 1, yi[1] // 256)
bw = max(0, c_hi - c_lo + 1)
bh = max(0, r_hi - r_lo + 1)
return c_lo, c_hi, r_lo, r_hi, bw, bh
def fill_cost(p, bw, bh):
return bound("cost.fill", p * (1024 + bw + bh) + bw * bh)
def item_costs(P, w, h):
dev = P.device[(w, h)]
costs = {}
for i, it in enumerate(P.items):
if it.kind == 1:
continue
if it.kind == 0:
c = w * h
if it.ref != ABSENT:
xi, yi = dev[i][0]
_, _, _, _, bw, bh = fill_box(xi, yi, w, h)
c += fill_cost(P.paths[it.ref].n_points, bw, bh)
elif it.kind == 2:
xi, yi = dev[i][0]
_, _, _, _, bw, bh = fill_box(xi, yi, w, h)
c = fill_cost(P.paths[it.ref].n_points, bw, bh)
else:
ws = P.wos[it.ref]
c = 0
cl = ch = rl = rh = None
for g, (xi, yi) in zip(P.run_glyphs[i], dev[i]):
c_lo, c_hi, r_lo, r_hi, bw, bh = fill_box(xi, yi, w, h)
c += fill_cost(ws.by_gid[g.gid].n_points, bw, bh)
if bw > 0 and bh > 0:
cl = c_lo if cl is None else min(cl, c_lo)
ch = c_hi if ch is None else max(ch, c_hi)
rl = r_lo if rl is None else min(rl, r_lo)
rh = r_hi if rh is None else max(rh, r_hi)
if cl is not None:
c += (ch - cl + 1) * (rh - rl + 1)
bound("cost.run", c)
costs[i] = c
return costs
def effective_window(P, i):
it = P.items[i]
a, b = it.t_on, it.t_off
for gi in P.parents[i]:
g = P.items[gi]
a, b = max(a, g.t_on), min(b, g.t_off)
t0, dur = P.H["t0"], P.H["dur"]
a, b = max(a, t0), min(b, t0 + dur)
return a, b
def size_cost(P, w, h):
"""Section 3.4(b) step 5: w*h + the largest sum over any tick."""
costs = item_costs(P, w, h)
ev = []
for i, c in costs.items():
a, b = effective_window(P, i)
if a < b:
ev.append((a, 1, c))
ev.append((b, 0, -c)) # ends (0) before starts (1) at equal ticks
ev.sort(key=lambda e: (e[0], e[1]))
cur = best = 0
for _, _, d in ev:
cur += d
bound("cost.sum", cur)
best = max(best, cur)
total = w * h + best
bound("cost.sum", total)
return total
# ------------------------------------------------------------------ section 3.1, 3.2
def ticks_per_frame(P, num, den):
return den * P.H["tb"] // num
def frame_range(P, num, den):
Pf = ticks_per_frame(P, num, den)
t0, dur = P.H["t0"], P.H["dur"]
return cdiv(t0, Pf), (t0 + dur - 1) // Pf
def eval_track(P, ti, tau):
t = P.tracks[ti]
ts, vs = t.ts, t.vs
n = len(ts)
if tau <= ts[0]:
return vs[0]
if tau >= ts[-1]:
return vs[-1]
i = bisect.bisect_right(ts, tau) - 1
D = ts[i + 1] - ts[i]
r = tau - ts[i]
d = vs[i + 1] - vs[i]
interp = t.keys[i].interp
if interp == 0:
return vs[i]
if interp == 1:
bound("track.linear", 2 * d * r + D)
return vs[i] + R_rat(d * r, D)
E = P.tables[interp - 2]
bound("track.table_r", r << 16)
u = (r << 16) // D
assert 0 <= u <= 65535
k = u // 4096
phi = u % 4096
step = (E[k + 1] - E[k]) * phi
bound("track.table_step", step)
e = E[k] + step // 4096
assert 0 <= e <= 65536
bound("track.value_step", d * e)
return vs[i] + R_half_up(d * e, 16)
def slot_value(P, it, s, tau):
ti = it.slots[s]
if ti == ABSENT:
return SLOT_DEFAULT[s]
return eval_track(P, ti, tau)
# ------------------------------------------------------------------ section 3.3, 3.5
def place(P, w, M, O, s, g, points):
"""Section 3.3: the device points of one glyph record (or path, or clip)."""
dw = P.H["design_w"]
den = dw << 16
L = []
for m in M:
n = m * s * w
bound("place.L_num", n)
L.append(bound("place.L", R_rat(n, den)))
L00, L01, L10, L11 = L
m00, m01, m10, m11 = M
nx = ((O[0] << 16) + m00 * g[0] + m01 * g[1]) * w
ny = ((O[1] << 16) + m10 * g[0] + m11 * g[1]) * w
bound("place.origin_num", nx)
bound("place.origin_num", ny)
tx, ty = R_rat(nx, den), R_rat(ny, den)
out = []
for px, py, _tag in points:
sx = L00 * px + L01 * py
sy = L10 * px + L11 * py
bound("place.point_sum", sx + (1 << 15))
bound("place.point_sum", sy + (1 << 15))
dx = tx + R_half_up(sx, 16)
dy = ty + R_half_up(sy, 16)
bound("device.lo", dx)
bound("device.hi", dx)
bound("device.lo", dy)
bound("device.hi", dy)
out.append((dx, dy))
return out
def flatten_quad(P0, P1, P2):
Ax, Ay = P0[0] - 2 * P1[0] + P2[0], P0[1] - 2 * P1[1] + P2[1]
M = max(abs(Ax), abs(Ay))
n = 0
while M > 64 * (4 ** n):
n += 1
bound("flatten.quad_n", n)
Bx, By = P1[0] - P0[0], P1[1] - P0[1]
N = 1 << n
sh = 2 * n
out = []
for k in range(1, N + 1):
nx = (1 << (n + 1)) * k * Bx + k * k * Ax
ny = (1 << (n + 1)) * k * By + k * k * Ay
bound("flatten.quad_num", nx)
bound("flatten.quad_num", ny)
out.append((P0[0] + (nx >> sh), P0[1] + (ny >> sh)))
assert out[-1] == tuple(P2)
return out
def flatten_cubic(P0, P1, P2, P3):
M = 0
for a, b_, c in ((P0, P1, P2), (P1, P2, P3)):
for ax in (0, 1):
M = max(M, abs(a[ax] - 2 * b_[ax] + c[ax]))
n = 0
while 3 * M > 64 * (4 ** n):
n += 1
bound("flatten.cubic_n", n)
B = [P1[a] - P0[a] for a in (0, 1)]
C = [P0[a] - 2 * P1[a] + P2[a] for a in (0, 1)]
D = [P3[a] - 3 * P2[a] + 3 * P1[a] - P0[a] for a in (0, 1)]
N = 1 << n
sh = 3 * n
out = []
for k in range(1, N + 1):
v = []
for a in (0, 1):
num = 3 * (4 ** n) * k * B[a] + 3 * (2 ** n) * k * k * C[a] + k * k * k * D[a]
bound("flatten.cubic_num", num)
v.append(P0[a] + (num >> sh))
out.append((v[0], v[1]))
assert out[-1] == tuple(P3)
return out
def polylines(prims, dev):
"""Section 3.5: each contour's polyline from its primitives and device points."""
out = []
for start, plist in prims:
pl = [dev[start]]
for p in plist:
if p[0] == "L":
pl.append(dev[p[2]])
elif p[0] == "Q":
pl.extend(flatten_quad(dev[p[1]], dev[p[2]], dev[p[3]]))
else:
pl.extend(flatten_cubic(dev[p[1]], dev[p[2]], dev[p[3]], dev[p[4]]))
out.append(pl)
return out
# ------------------------------------------------------------------ sections 3.6, 3.7
def fill_values(pls, w, h, p=None):
"""Sections 3.6-3.7 for one fill: {row: [(c_start, c_end, v)]}, the runs of
constant nonzero v on the canvas, with the left fold. With `p`, the stored point
count of the fill's outline, it also asserts section 3.7's D_r <= 256*p.
Crossings are the closed forms, floors of exact rationals in big integers. Column
boundaries left of x = 0 and right of x = 256*w are not split at: every sub-piece
left of the canvas folds 512*cover into column 0 and every sub-piece right of it
adds nothing, so their sum is the same (section 3.7; section 3.11's freedom)."""
d = {}
D = {}
for pl in pls:
for j in range(len(pl) - 1):
x0, y0 = pl[j]
x1, y1 = pl[j + 1]
bound("segment.d", x1 - x0)
bound("segment.d", y1 - y0)
if y0 == y1:
continue
if y0 < y1:
xt, yt, xb, yb = x0, y0, x1, y1
else:
xt, yt, xb, yb = x1, y1, x0, y0
dx, dy = xb - xt, yb - yt
first = 256 * (yt // 256 + 1)
bound("rows.cross32", (first - yt) * abs(dx))
r0 = max(yt // 256, 0)
r1 = min((yb - 1) // 256, h - 1)
for r in range(r0, r1 + 1):
ytop = max(yt, 256 * r)
ybot = min(yb, 256 * r + 256)
D[r] = D.get(r, 0) + (ybot - ytop)
bound("rows.cross", (ytop - yt) * dx)
bound("rows.cross", (ybot - yt) * dx)
Xt = xt + ((ytop - yt) * dx) // dy
Xb = xt + ((ybot - yt) * dx) // dy
if y0 < y1:
cell_split(Xt, ytop - 256 * r, Xb, ybot - 256 * r, d.setdefault(r, {}), w)
else:
cell_split(Xb, ybot - 256 * r, Xt, ytop - 256 * r, d.setdefault(r, {}), w)
out = {}
for r, drow in d.items():
bound("density", D[r])
if p is not None:
# section 3.7, "the point cap makes the condition hold": D_r <= 256*p.
# DENSITY_SLACK is 0; the falsification raises it to tighten the bound.
if D[r] > 256 * p - DENSITY_SLACK:
raise BoundError(f"D_{r} = {D[r]} past 256*p = {256 * p} - {DENSITY_SLACK}")
gap = 256 * p - D[r]
if gap < BOUND_PEAK.get("density.per_point.gap", 1 << 62):
BOUND_PEAK["density.per_point.gap"] = gap
lim = 512 * D[r]
cols = sorted(drow)
v = 0
spans = []
for k, c in enumerate(cols):
if abs(drow[c]) > lim:
raise BoundError(f"d[{r}][{c}] = {drow[c]} past 512*D_r = {lim}")
v += drow[c]
if abs(v) > lim:
raise BoundError(f"v[{r}][{c}] = {v} past 512*D_r = {lim}")
bound("accum", v)
nxt = cols[k + 1] if k + 1 < len(cols) else w
if v != 0:
spans.append((c, nxt, v))
if spans:
out[r] = spans
return out
def cell_split(xa, ya, xb, yb, drow, w):
"""Section 3.6's cell split of one row piece and section 3.7's accumulation."""
xs, ys = [xa], [ya]
if xa != xb:
if xa < xb:
cs = range(max(xa // 256 + 1, 0), min((xb - 1) // 256, w) + 1)
else:
cs = range(min((xa - 1) // 256, w), max(xb // 256 + 1, 0) - 1, -1)
num_y = yb - ya
den = xb - xa
for c in cs:
xc = 256 * c
bound("cells.product", abs(xc - xa) * abs(num_y))
n = (xc - xa) * num_y
if den > 0:
yc = ya + n // den
else:
yc = ya + (-n) // (-den)
xs.append(xc)
ys.append(yc)
xs.append(xb)
ys.append(yb)
for j in range(len(xs) - 1):
cover = ys[j + 1] - ys[j]
bound("cells.cover", cover)
if cover == 0:
continue
c = min(xs[j], xs[j + 1]) // 256
if c < 0:
drow[0] = drow.get(0, 0) + 512 * cover
bound("accum", drow[0])
elif c < w:
fa = xs[j] - 256 * c
fb = xs[j + 1] - 256 * c
assert 0 <= fa <= 256 and 0 <= fb <= 256
area = cover * (fa + fb)
bound("cells.area", area)
drow[c] = drow.get(c, 0) + 512 * cover - area
bound("accum", drow[c])
if c + 1 < w:
drow[c + 1] = drow.get(c + 1, 0) + area
bound("accum", drow[c + 1])
# ------------------------------------------------------------------ section 3.8
def alpha_nonzero(v):
m = min(abs(v), 1 << 17)
bound("alpha.product", 255 * m)
return R_half_up(255 * m, 17)
def alpha_evenodd(v):
m = v % (1 << 18)
m2 = min(m, (1 << 18) - m)
bound("alpha.product", 255 * m2)
return R_half_up(255 * m2, 17)
def path_alpha(vals, rule, w):
"""{pixel index: alpha} for a path or clip fill."""
q = alpha_evenodd if rule == 1 else alpha_nonzero
out = {}
for r, spans in vals.items():
base = r * w
for c0, c1, v in spans:
a = q(v)
if a:
for c in range(c0, c1):
out[base + c] = a
return out
def run_alpha(S):
out = {}
for px, s in S.items():
bound("union.S", s)
m = min(s, 1 << 17)
bound("alpha.product", 255 * m)
a = R_half_up(255 * m, 17)
if a:
out[px] = a
return out
# ------------------------------------------------------------------ section 3.9, 3.10
def composite_fill(T, alpha, paint, o):
pp = []
for ch in paint:
bound("composite.product", ch * o)
pp.append(div255(ch * o))
for px, a in alpha.items():
s = []
for ch in pp:
bound("composite.product", ch * a)
s.append(div255(ch * a))
sA = s[3]
b = 4 * px
for ch in range(4):
bound("composite.product", T[b + ch] * (255 - sA))
T[b + ch] = s[ch] + div255(T[b + ch] * (255 - sA))
assert T[b + ch] <= 255
def composite_group(parent, G, mask, og):
n = len(G) // 4
for px in range(n):
b = 4 * px
if G[b + 3] == 0:
continue # R, G, B <= A = 0, so s = 0 and the parent is unchanged
m = 255 if mask is None else mask.get(px, 0)
bound("composite.product", m * og)
k = div255(m * og)
s = []
for ch in range(4):
bound("composite.product", G[b + ch] * k)
s.append(div255(G[b + ch] * k))
sA = s[3]
for ch in range(4):
parent[b + ch] = s[ch] + div255(parent[b + ch] * (255 - sA))
def check_request(P, w, h, num, den, g):
if not any(z.w == w and z.h == h for z in P.sizes):
raise RequestError("size_unlisted")
if not any(r.num == num and r.den == den for r in P.rates):
raise RequestError("rate_unlisted")
lo, hi = frame_range(P, num, den)
if not (isinstance(g, int) and g >= 0 and lo <= g <= hi):
raise RequestError("frame_range")
def render(P, w, h, num, den, g):
"""R(P, S, z, r, g), section 3.10: the frame's RGBA8P."""
check_request(P, w, h, num, den, g)
tau = g * ticks_per_frame(P, num, den)
stack = [bytearray(4 * w * h)]
gstack = [] # (item index, tx, ty, op) of each open, active group
items = P.items
i = 0
while i < len(items):
it = items[i]
if it.kind == 0:
if it.t_on <= tau < it.t_off:
gstack.append((i, slot_value(P, it, 4, tau), slot_value(P, it, 5, tau),
slot_value(P, it, 6, tau)))
stack.append(bytearray(4 * w * h))
else:
i = P.match[i] + 1
continue
elif it.kind == 1:
gi, _, _, og = gstack[-1]
grp = items[gi]
mask = None
if grp.ref != ABSENT:
O = (sum(e[1] for e in gstack), sum(e[2] for e in gstack))
path = P.paths[grp.ref]
dev = place(P, w, (65536, 0, 0, 65536), O, grp.unit_scale, (0, 0), path.points)
mask = path_alpha(fill_values(polylines(path.prims, dev), w, h, path.n_points),
grp.rule, w)
gstack.pop()
G = stack.pop()
composite_group(stack[-1], G, mask, og)
elif it.t_on <= tau < it.t_off:
M = tuple(slot_value(P, it, s, tau) for s in range(4))
O = (slot_value(P, it, 4, tau) + sum(e[1] for e in gstack),
slot_value(P, it, 5, tau) + sum(e[2] for e in gstack))
o = slot_value(P, it, 6, tau)
if it.kind == 2:
path = P.paths[it.ref]
dev = place(P, w, M, O, it.unit_scale, (0, 0), path.points)
alpha = path_alpha(fill_values(polylines(path.prims, dev), w, h, path.n_points),
it.rule, w)
else:
ws = P.wos[it.ref]
S = {}
for gr in P.run_glyphs[i]:
gl = ws.by_gid[gr.gid]
if not gl.points:
continue
dev = place(P, w, M, O, it.unit_scale, (gr.x, gr.y), gl.points)
vals = fill_values(polylines(gl.prims, dev), w, h, gl.n_points)
for r, spans in vals.items():
base = r * w
for c0, c1, v in spans:
ck = min(abs(v), 1 << 17)
for c in range(c0, c1):
S[base + c] = S.get(base + c, 0) + ck
alpha = run_alpha(S)
composite_fill(stack[-1], alpha, P.paints[it.paint], o)
i += 1
assert len(stack) == 1
root = stack[0]
for j in range(0, len(root), 4):
assert root[j] <= root[j + 3] and root[j + 1] <= root[j + 3] and root[j + 2] <= root[j + 3]
return bytes(root)
# ------------------------------------------------------------------ section 2.10
SIDECAR_SCHEMA = "wai.graphics.digests/1"
I31 = (1 << 31) - 1
def write_sidecar(sc):
def ints(xs):
return ",".join(str(x) for x in xs)
clips = ",".join(f'[{ints(c[:4])},"{c[4]}"]' for c in sc["clips"])
probes = ",".join(f'[{ints(p[:5])},"{p[5]}"]' for p in sc["probes"])
return (f'{{"clips":[{clips}],"instr":"{sc["instr"]}","probes":[{probes}],'
f'"schema":"{SIDECAR_SCHEMA}"}}').encode("ascii")
def _hex64(s):
return isinstance(s, str) and len(s) == 64 and all(c in "0123456789abcdef" for c in s)
def _int31(v):
return type(v) is int and 0 <= v <= I31
def parse_sidecar(b, P):
"""Section 2.10's sidecar checks 1-2 against a loaded payload; returns the parsed
sidecar, or raises SidecarCode."""
def bad(why):
return SidecarCode("sidecar_malformed", why)
if len(b) > 1 << 17:
raise bad("over 2^17 bytes")
def no_const(x):
raise bad(f"constant {x}")
def no_float(x):
raise bad(f"fraction or exponent {x}")
def pairs(kv):
d = {}
for k, v in kv:
if k in d:
raise bad(f"repeated member {k}")
d[k] = v
return d
try:
text = b.decode("utf-8", errors="strict")
sc = json.loads(text, parse_float=no_float, parse_constant=no_const,
object_pairs_hook=pairs)
except SidecarCode:
raise
except (UnicodeDecodeError, ValueError) as e:
raise bad(f"not JSON: {e}")
if not isinstance(sc, dict) or set(sc) != {"clips", "instr", "probes", "schema"}:
raise bad("members")
if sc["schema"] != SIDECAR_SCHEMA:
raise bad("schema")
if not _hex64(sc["instr"]):
raise bad("instr")
clips, probes = sc["clips"], sc["probes"]
if not isinstance(clips, list) or not 0 <= len(clips) <= 32:
raise bad("clip count")
if not isinstance(probes, list) or not 1 <= len(probes) <= 1024:
raise bad("probe count")
sizes = {(z.w, z.h) for z in P.sizes}
rates = {(r.num, r.den) for r in P.rates}
prev = None
for c in clips:
if not (isinstance(c, list) and len(c) == 5 and all(_int31(x) for x in c[:4])
and _hex64(c[4])):
raise bad(f"clip entry {c}")
if (c[0], c[1]) not in sizes or (c[2], c[3]) not in rates:
raise bad(f"clip size or rate {c[:4]}")
k = tuple(c[:4])
if prev is not None and not prev < k:
raise bad("clips out of order or repeated")
prev = k
prev = None
for p in probes:
if not (isinstance(p, list) and len(p) == 6 and all(_int31(x) for x in p[:5])
and _hex64(p[5])):
raise bad(f"probe entry {p}")
if (p[0], p[1]) not in sizes or (p[2], p[3]) not in rates:
raise bad(f"probe size or rate {p[:4]}")
lo, hi = frame_range(P, p[2], p[3])
if not lo <= p[4] <= hi:
raise bad(f"probe g {p[4]} out of range")
k = tuple(p[:5])
if prev is not None and not prev < k:
raise bad("probes out of order or repeated")
prev = k
if write_sidecar(sc) != bytes(b):
raise bad("not the canonical form")
if sc["instr"] != hashlib.sha256(P.bytes).hexdigest():
raise SidecarCode("sidecar_mismatch", "instr")
return sc
def frame_digest(P, w, h, num, den, g, cache=None):
key = (w, h, num, den, g)
if cache is not None and key in cache:
return cache[key]
d = sha256(render(P, w, h, num, den, g))
if cache is not None:
cache[key] = d
return d
def clip_digest(P, w, h, num, den, cache=None):
lo, hi = frame_range(P, num, den)
return sha256(b"".join(frame_digest(P, w, h, num, den, g, cache) for g in range(lo, hi + 1)))
def check_probes(P, sc, cache=None):
"""The sink rule: every probe frame rendered as published, compared by SHA-256.
Returns None, or the first mismatching probe (layer_mismatch)."""
for p in sc["probes"]:
if frame_digest(P, *p[:5], cache).hex() != p[5]:
return p
return None
def probe_set(P, pairs):
"""Section 2.10's authoring-tool probe set over the given (w, h, num, den) pairs."""
levels = [[], [], []]
t0, dur = P.H["t0"], P.H["dur"]
for (w, h, num, den) in pairs:
Pf = ticks_per_frame(P, num, den)
lo, hi = frame_range(P, num, den)
if lo <= hi:
# Only when the document has a frame at this rate: with none, the first
# exceeds the last and neither is in range (the text fix of this PR).
levels[0] += [(w, h, num, den, lo), (w, h, num, den, hi)]
for i, it in enumerate(P.items):
a, b = effective_window(P, i)
if a >= b:
continue
if cdiv(a, Pf) <= cdiv(b, Pf) - 1:
levels[1] += [(w, h, num, den, cdiv(a, Pf)), (w, h, num, den, cdiv(b, Pf) - 1)]
for t in P.tracks:
for k in t.keys:
g = cdiv(k.t, Pf)
if lo <= g <= hi:
levels[2].append((w, h, num, den, g))
held = set()
for lv in levels:
for c in sorted(lv):
if len(held) >= 1024:
break
held.add(c)
return sorted(held)
# ------------------------------------------------------------------ set store
def load_store(dirpath):
store = {}
if dirpath and os.path.isdir(dirpath):
for name in sorted(os.listdir(dirpath)):
if name.endswith(".wos1"):
with open(os.path.join(dirpath, name), "rb") as f:
data = f.read()
store[sha256(data)] = data
return store
# ------------------------------------------------------------------ self-test from the text
def selftest():
"""Worked examples and claims the text states, recomputed."""
fails = []
def eq(what, got, want):
if got != want:
fails.append(f"{what}: {got} != {want}")
# 3.8: the triangle (0,0)-(512,3)-(0,3) gives |v| = 1280 and 512 in its two pixels.
vals = fill_values([[(0, 0), (512, 3), (0, 3), (0, 0)]], 4, 4)
got = sorted(abs(v) for spans in vals.values() for c0, c1, v in spans for _ in range(c0, c1))
eq("3.8 triangle", got, [512, 1280])
# 3.8: a full pixel gives 255, half a pixel 128.
eq("3.8 full pixel", alpha_nonzero(1 << 17), 255)
eq("3.8 half pixel", alpha_nonzero(1 << 16), 128)
# 3.9: div255 is the half-up rounding of x/255 over [0, 65025].
for x in range(0, 65026):
if div255(x) != (2 * x + 255) // 510:
fails.append(f"3.9 div255({x})")
break
# 3.9: div255(255*x) = x.
for x in range(256):
eq(f"3.9 div255(255*{x})", div255(255 * x), x)
# 3.0: R_rat's quotient-remainder form.
for n in range(-600, 601, 7):
for d in (1, 2, 3, 255, 256):
q, r = n // d, n - (n // d) * d
eq(f"3.0 R_rat({n},{d})", R_rat(n, d), q + (1 if 2 * r >= d else 0))
# 2.1: the WOS1 length at the caps.
eq("2.1 WOS1 caps", 32 + 16 * 65536 + 2 * (1 << 22) + 5 * (1 << 22) + (1 << 20), 31457312)
# 2.9: a 1920x1080 frame.
eq("2.9 frame bytes", 4 * 1920 * 1080, 8294400)
# 3.5: V_(2^n) is the end point (asserted inside), and a straight quadratic is n = 0.
eq("3.5 straight quad", flatten_quad((0, 0), (256, 0), (512, 0)), [(512, 0)])
# 3.6: X(y) is the same from either end (the row split of a reversed segment).
a = fill_values([[(10, 7), (900, 700), (10, 700), (10, 7)]], 8, 4)
b = fill_values([[(10, 7), (10, 700), (900, 700), (10, 7)]], 8, 4)
eq("3.11 P1 reversal", {r: [(c0, c1, -v) for c0, c1, v in s] for r, s in a.items()}, b)
if fails:
for f in fails:
print("FAIL", f)
return 1
print("selftest: every worked example and stated claim recomputes")
return 0
# ------------------------------------------------------------------ the corpus driver
#
# Written after the checker above was committed, from the corpus README's layout: it
# maps each vector file to the computations above and writes the expected file's JSON.
# It takes no expected value as an input, with one exception it names: the `v` values
# at which the `alpha` vector probes the quantisers are listed only in its expected
# file, so they are read from there and the values at them recomputed.
RENDER_GROUPS = ("clock", "composite", "fills", "motion", "runs", "slicing")
RGBA8S_GROUPS = ("composite",)
def canon(obj):
return (json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+ "\n").encode("utf-8")
def frames_of(P, group, only=None):
out = []
for z in P.sizes:
for r in P.rates:
if only is not None and (z.w, z.h, r.num, r.den) not in only:
continue
lo, hi = frame_range(P, r.num, r.den)
fr, s8 = [], []
digests = []
for g in range(lo, hi + 1):
buf = render(P, z.w, z.h, r.num, r.den, g)
d = sha256(buf)
digests.append(d)
fr.append([g, d.hex()])
if group in RGBA8S_GROUPS:
s8.append([g, sha256(rgba8s(buf)).hex()])
e = {"clip": sha256(b"".join(digests)).hex(), "den": r.den, "frames": fr, "h": z.h,
"num": r.num, "w": z.w}
if group in RGBA8S_GROUPS:
e["rgba8s"] = s8
out.append(e)
return out
def payload_result(data, store, group):
"""The expected-file object for one WVG1, and whether it rewrites to its own bytes
(None when refused), and the refusal (or None)."""
try:
P = load(data, store)
except Refusal as r:
return {"code": r.code}, None, r, None
d = {"blocks": [block_identity(P, k).hex() for k in range(len(P.texts))],
"code": "accepted", "instr": sha256(data).hex(),
"rates": [[r.num, r.den, ticks_per_frame(P, r.num, r.den), *frame_range(P, r.num, r.den)]
for r in P.rates],
"sizes": [[z.w, z.h, c] for z, c in zip(P.sizes, P.costs)],
"text_hash": P.text_hash.hex()}
if group in RENDER_GROUPS:
d["frames"] = frames_of(P, group)
return d, write_wvg1(P) == data, None, P
def wos1_result(data):
try:
s = parse_wos1(data)
except Refusal as r:
return {"code": r.code}, None, r
return ({"code": "accepted", "glyphs": len(s.glyphs),
"points": sum(len(g.points) for g in s.glyphs), "sha256": sha256(data).hex()},
write_wos1(s) == data, None)
def arith_result(v, expected_path):
op = v["op"]
if op == "div255":
table = bytes(div255(x) for x in range(65026))
first = []
for x in range(65026):
if table[x] == len(first):
first.append(x)
return {"count": 65026, "first_x_of_each_value": first, "op": op,
"sha256": sha256(table).hex()}
if op == "alpha":
lo, hi = -(1 << 18), 1 << 18
nz = bytes(alpha_nonzero(x) for x in range(lo, hi + 1))
eo = bytes(alpha_evenodd(x) for x in range(lo, hi + 1))
with open(expected_path, "rb") as f:
vs = [p[0] for p in json.load(f)["probes"]] # the probe inputs (see above)
return {"domain": [lo, hi], "evenodd_sha256": sha256(eo).hex(),
"nonzero_sha256": sha256(nz).hex(), "op": op,
"probes": [[x, alpha_nonzero(x), alpha_evenodd(x)] for x in vs]}
if op == "unpremul":
table = bytearray()
for a in range(256):
for c in range(a + 1):
table.append(unpremul((c, c, c, a))[0])
return {"count": len(table), "op": op, "sha256": sha256(bytes(table)).hex()}
if op == "half_up":
return {"op": op, "values": [R_half_up(p, s) for p, s in v["cases"]]}
if op == "r_rat":
return {"op": op, "values": [R_rat(n, d) for n, d in v["cases"]]}
if op == "interp":
P = Payload()
P.tables = [tuple(t) for t in v["tables"]]
P.tracks = []
for t in v["tracks"]:
tr = TrackRec()
tr.kind = t["kind"]
tr.keys = [Key(k[0], k[1], k[2], b"\0\0\0") for k in t["keys"]]
tr.ts = [k.t for k in tr.keys]
tr.vs = [k.value for k in tr.keys]
P.tracks.append(tr)
return {"op": op, "values": [[eval_track(P, ti, tau) for tau in v["ticks"]]
for ti in range(len(P.tracks))]}
raise ValueError(f"unknown op {op}")
def flatten_result(v):
out = []
for c in v["curves"]:
pts = [tuple(p) for p in c[1:]]
verts = flatten_quad(*pts) if c[0] == "q" else flatten_cubic(*pts)
out.append({"n": len(verts).bit_length() - 1, "vertices": [list(p) for p in verts]})
return {"curves": out, "op": v.get("op", "flatten")}
def pinned_step(name, desc):
"""The step a refusal vector's own name or description pins, or None."""
import re
m = re.match(r"s(\d)([a-h]?)-", name)
if m:
return m.group(1) + m.group(2)
m = re.search(r"\((?:step|check) (\d+[a-h]?) before", desc)
if m:
return m.group(1)
m = re.match(r"(?:step|check) (\d+[a-h]?):", desc)
if m:
return m.group(1)
return None
def json_diff(a, b, path=""):
"""Field paths where two JSON values differ."""
if isinstance(a, dict) and isinstance(b, dict):
out = []
for k in sorted(set(a) | set(b)):
if k not in a or k not in b:
out.append(f"{path}.{k} present in one only")
else:
out += json_diff(a[k], b[k], f"{path}.{k}")
return out
if isinstance(a, list) and isinstance(b, list):
if len(a) != len(b):
return [f"{path} length {len(a)} != {len(b)}"]
out = []
for i, (x, y) in enumerate(zip(a, b)):
out += json_diff(x, y, f"{path}[{i}]")
return out
return [] if a == b else [f"{path}: checker {a!r}, expected {b!r}"]
def sidecar_exercise(P, frames):
"""The digest sidecar (section 2.10) the authoring tool writes for an accepted
payload, built from its rendered frames: the probe set over every listed pair and
a clip per pair. It must parse and rewrite to its own bytes, every probe must
match, and three corruptions must give sidecar_malformed, sidecar_mismatch and
layer_mismatch. Returns a list of failures."""
cache = {}
for e in frames:
for g, d in e["frames"]:
cache[(e["w"], e["h"], e["num"], e["den"], g)] = bytes.fromhex(d)
pairs = [(z.w, z.h, r.num, r.den) for z in P.sizes for r in P.rates]
probes = probe_set(P, pairs)
if not probes:
return [] # no frame at any pair: no sidecar can list a probe
sc = {"clips": sorted([e["w"], e["h"], e["num"], e["den"], e["clip"]] for e in frames),
"instr": hashlib.sha256(P.bytes).hexdigest(),
"probes": [list(p) + [cache[p].hex()] for p in probes]}
b = write_sidecar(sc)
fails = []
try:
got = parse_sidecar(b, P)
if write_sidecar(got) != b:
fails.append("sidecar does not rewrite to its own bytes")
if check_probes(P, got, cache) is not None:
fails.append("a probe of the sidecar mismatches")
for c in got["clips"]:
if clip_digest(P, *c[:4], cache).hex() != c[4]:
fails.append(f"clip {c[:4]} mismatches in verify mode")
except SidecarCode as e:
fails.append(f"the tool's own sidecar refused: {e}")
for bad, want in ((b.replace(b'":', b'": ', 1), "sidecar_malformed"),
(b.replace(sc["instr"].encode(), b"0" * 64), "sidecar_mismatch"),
(b[:-1] + b" ", "sidecar_malformed")):
try:
parse_sidecar(bad, P)
fails.append(f"a corrupted sidecar was accepted, {want} expected")
except SidecarCode as e:
if e.code != want:
fails.append(f"corrupted sidecar gave {e.code}, {want} expected")
flipped = dict(sc, probes=[list(sc["probes"][0][:5]) + ["0" * 64]] + sc["probes"][1:])
if "0" * 64 != sc["probes"][0][5]:
if check_probes(P, parse_sidecar(write_sidecar(flipped), P), cache) is None:
fails.append("a wrong probe digest was not a layer_mismatch")
return fails
def corpus(root, only_groups=None, verbose=True):
import time
store = load_store(os.path.join(root, "sets"))
agree = total = 0
rewrite_ok = rewrite_total = 0
step_ok = step_total = 0
disagreements = []
step_disagreements = []
rendered = {}
sidecars = 0
vroot = os.path.join(root, "vectors")
for group in sorted(os.listdir(vroot)):
if only_groups and group not in only_groups:
continue
gdir = os.path.join(vroot, group)
with open(os.path.join(gdir, "index.json"), encoding="utf-8") as f:
index = json.load(f)
for name in sorted(index):
t = time.time()
exp_path = os.path.join(root, "expected", group, name + ".json")
with open(exp_path, "rb") as f:
exp_bytes = f.read()
refusal = None
rw = None
P = None
try:
if os.path.exists(os.path.join(gdir, name + ".wvg1")):
with open(os.path.join(gdir, name + ".wvg1"), "rb") as f:
data = f.read()
got, rw, refusal, P = payload_result(data, store, group)
if P is not None and group in RENDER_GROUPS:
rendered[(group, name)] = got.get("frames")
sidecars += 1
for fl in sidecar_exercise(P, got["frames"]):
disagreements.append((f"{group}/{name} sidecar", [fl]))
elif os.path.exists(os.path.join(gdir, name + ".wos1")):
with open(os.path.join(gdir, name + ".wos1"), "rb") as f:
data = f.read()
got, rw, refusal = wos1_result(data)
else:
with open(os.path.join(gdir, name + ".json"), encoding="utf-8") as f:
v = json.load(f)
got = flatten_result(v) if group == "flatten" else arith_result(v, exp_path)
mine = canon(got)
except BoundError as e:
mine = None
got = {"bound": str(e)}
total += 1
ok = mine == exp_bytes
if ok:
agree += 1
else:
try:
exp = json.loads(exp_bytes)
diff = json_diff(got, exp)
except ValueError:
diff = ["expected file is not JSON"]
disagreements.append((f"{group}/{name}", diff[:6] or ["byte form differs"]))
if rw is not None:
rewrite_total += 1
rewrite_ok += bool(rw)
if not rw:
disagreements.append((f"{group}/{name}", ["does not rewrite to its own bytes"]))
if refusal is not None:
want = pinned_step(name, index[name])
mystep = refusal.step
if group == "refusals" and name.startswith("wos1"):
mystep = refusal.where.split(":")[0].replace("wos1 step ", "")
if want is not None:
step_total += 1
if want == mystep:
step_ok += 1
else:
step_disagreements.append((f"{group}/{name}", mystep, want,
refusal.where))
if verbose:
tag = "ok " if ok else "DIFF"
extra = f" [{refusal.code} step {refusal.step}: {refusal.where}]" if refusal else ""
print(f"{tag} {group}/{name} ({time.time() - t:.1f}s){extra}", flush=True)
# the sets the payloads hold, each rewritten
for name in sorted(os.listdir(os.path.join(root, "sets"))):
with open(os.path.join(root, "sets", name), "rb") as f:
data = f.read()
got, rw, refusal = wos1_result(data)
if rw is not None:
rewrite_total += 1
rewrite_ok += bool(rw)
if verbose:
print(f"set {name}: {got['code']}" + (f", rewrites: {rw}" if rw is not None else
f" ({refusal.where})"))
# sha256.json: every digest recomputed
sha_ok = sha_total = 0
sha_path = os.path.join(root, "sha256.json")
if os.path.exists(sha_path) and not only_groups:
with open(sha_path, encoding="utf-8") as f:
sj = json.load(f)
for key, ent in sj.items():
sha_total += 1
if key == "arith/div255":
good = ent["sha256"] == sha256(bytes(div255(x) for x in range(65026))).hex()
else:
vec, wh, rate = key.split("@")
group, name = vec.split("/")
w, h = (int(x) for x in wh.split("x"))
num, den = (int(x) for x in rate.split("/"))
fr = rendered.get((group, name)) or []
e = [x for x in fr if (x["w"], x["h"], x["num"], x["den"]) == (w, h, num, den)]
good = bool(e) and ent["frames_sha256"] == [d for _, d in e[0]["frames"]] \
and ent["clip_sha256"] == e[0]["clip"]
sha_ok += good
if not good:
disagreements.append((f"sha256.json {key}", ["digest differs"]))
print()
print(f"expected files recomputed byte for byte: {agree} of {total}")
print(f"payloads and sets rewritten to their own bytes: {rewrite_ok} of {rewrite_total}")
print(f"refusal steps the vectors pin, reported at that step: {step_ok} of {step_total}")
if sha_total:
print(f"sha256.json entries recomputed: {sha_ok} of {sha_total}")
print(f"digest sidecars built, parsed, rewritten and corrupted: {sidecars}")
for v, diff in disagreements:
print(f"DISAGREE {v}")
for d in diff:
print(f" {d}")
for v, mine, want, where in step_disagreements:
print(f"STEP {v}: checker step {mine} ({where}), vector pins step {want}")
peaks = ", ".join(f"{k}={v}" for k, v in sorted(BOUND_PEAK.items()))
print(f"bound peaks: {peaks}")
return 0 if not disagreements and not step_disagreements and agree == total else 1
# ------------------------------------------------------------------ command line
def _opt(args, name, default=None):
if name in args:
i = args.index(name)
v = args[i + 1]
del args[i:i + 2]
return v
return default
def main(argv):
args = list(argv)
if not args:
print(__doc__)
return 2
cmd = args.pop(0)
if cmd == "selftest":
return selftest()
if cmd == "corpus":
quiet = "--quiet" in args
if quiet:
args.remove("--quiet")
groups = _opt(args, "--groups")
rc = selftest()
return rc or corpus(args[0], groups.split(",") if groups else None, not quiet)
sets_dir = _opt(args, "--sets")
out = _opt(args, "--out")
verify = "--verify" in args
if verify:
args.remove("--verify")
store = load_store(sets_dir)
if cmd == "load":
data = open(args[0], "rb").read()
try:
P = load(data, store)
except Refusal as r:
print(json.dumps({"refused": r.code, "step": r.step, "where": r.where}))
return 0
same = write_wvg1(P) == data
print(json.dumps({"accepted": True, "costs": P.costs, "rewrite_identical": same,
"text_hash": P.text_hash.hex()}))
return 0 if same else 1
if cmd == "render":
data = open(args[0], "rb").read()
w, h = (int(x) for x in args[1].split("x"))
num, den = (int(x) for x in args[2].split("/"))
g = int(args[3])
P = load(data, store)
buf = render(P, w, h, num, den, g)
if out:
open(out, "wb").write(buf)
print(sha256(buf).hex())
return 0
if cmd == "sidecar":
data = open(args[0], "rb").read()
sc_bytes = open(args[1], "rb").read()
P = load(data, store)
try:
sc = parse_sidecar(sc_bytes, P)
except SidecarCode as e:
print(json.dumps({"sidecar": e.code, "why": e.why}))
return 0
bad = check_probes(P, sc)
if bad is not None:
print(json.dumps({"render": "layer_mismatch", "probe": bad[:5]}))
return 1
if verify:
for c in sc["clips"]:
if clip_digest(P, *c[:4]).hex() != c[4]:
print(json.dumps({"verify": "clip mismatch", "clip": c[:4]}))
return 1
print(json.dumps({"sidecar": "ok", "probes": len(sc["probes"])}))
return 0
print(f"unknown command {cmd}")
return 2
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
Reference in a new issue View git blame Copy permalink