93a8d5929b
The checker's list of rendering groups is read from the README's layout. D7-5 adds the text group to the corpus and to that sentence of the README, so the list gains it. No computation changes. Co-Authored-By: Claude Opus 5.5 <[email protected]>
2450 lines
86 KiB
Python
2450 lines
86 KiB
Python
#!/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.
|
|
|
|
# The groups the corpus README names as rendering: their expected files list frames.
|
|
RENDER_GROUPS = ("clock", "composite", "fills", "motion", "runs", "slicing", "text")
|
|
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:]))
|