2b54ac2054
A second falsification: L2's squarings rounded instead of truncated still regenerated the corpus (the variant differs on about 3 in 100000 mantissas). vectors.json now states the SHA-256 of L2 at 2^18 spread values, of E2's mantissa for every fraction and of tan for every angle; both implementations check them. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1365 lines
51 KiB
Python
1365 lines
51 KiB
Python
#!/usr/bin/env python3
|
||
"""Second implementation of extensions/exact-tonemap.md (`wai.media.tonemap`, `wai.meta.gainmap`,
|
||
`wai.meta.st2094_50`), over the corpus `live-conformance/tonemap/`.
|
||
|
||
Written from the extension's text. It shares no code with wai-rs: it computes both derivations in
|
||
integer arithmetic of its own (the fixed-point log2 and exp2 of §3, the sRGB table, the gain map
|
||
of §4, the Annex C reader, slopes, gain curves and weights of §5), reads PNG with the standard
|
||
library's zlib and its own unfiltering, and reads records as §2 has them read. It verifies every
|
||
signature with WAI's strict Ed25519 rule (claim-conformance/_ed25519.py) and every link hash and
|
||
content hash with BLAKE3 written from its specification (claim-conformance/_blake3.py), and reads
|
||
envelopes and SPEC §7.1's component rules with the staged-delivery corpus's second implementation
|
||
(staged-conformance/staged_verify.py). It reads §3.3's table of K, §5.4's constant P and §9's codes
|
||
from the extension.
|
||
|
||
python3 -I tonemap_verify.py <live-conformance/tonemap>
|
||
|
||
It checks that frames.rgb8 holds the frames whose digests the int_video sidecar publishes, the
|
||
arithmetic at the corpus's stated points (L2, E2, the sRGB table, tan, the video's slopes), every
|
||
envelope's content hash and SPEC §7.1 role, and gives every case the outcome, the reason, the place
|
||
and the statements the corpus states; it checks that every reason of §9 refuses a case, and
|
||
recomputes §6's measurement in binary64 (within one code value of the largest difference and 0.01
|
||
of the mean). Then it mutates copies of the corpus and requires every copy to fail, and drops each
|
||
of its own checks in turn and requires the corpus to catch the weakened verifier. Exit status 0
|
||
only if all of that holds. Python 3.8 or later; standard library only.
|
||
"""
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import os
|
||
import re
|
||
import struct
|
||
import sys
|
||
import zlib
|
||
from fractions import Fraction
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
sys.path.insert(0, os.path.join(HERE, "..", "staged-conformance"))
|
||
import _ed25519 as ED # noqa: E402
|
||
from _blake3 import blake3 # noqa: E402
|
||
import staged_verify as SV # noqa: E402
|
||
|
||
WAI = os.path.join(HERE, "..")
|
||
EXTENSION = os.path.join(WAI, "extensions", "exact-tonemap.md")
|
||
|
||
SIGN_DOMAIN = b"wai:tonemap\x01" # §2
|
||
LINK_DOMAIN = b"wai:tonemap-link\x01" # §2
|
||
KIND = "wai-tonemap" # §7.1
|
||
GAINMAP, ST2094 = "wai-tonemap:gainmap", "wai-tonemap:st2094-50"
|
||
CAP_GAINMAP, CAP_ST2094 = "wai.meta.gainmap", "wai.meta.st2094_50"
|
||
CAP_PNG = "wai.image.png"
|
||
CLASSES = ["captured", "reconstructed", "assistive", "synthesized"]
|
||
MAX_INT = (1 << 53) - 1
|
||
MAX_HEADROOM, MAX_SIDE, MAX_PICTURES, MAX_INPUTS = 60000, 4096, 256, 16
|
||
MAX_METADATA, MAX_SETS = 4096, 256
|
||
|
||
# A check named here is left out: the verifier weakened on purpose, to show the corpus catches it.
|
||
SKIP = set()
|
||
|
||
|
||
class Refuse(Exception):
|
||
def __init__(self, reason, at=None):
|
||
super().__init__(reason)
|
||
self.reason = reason
|
||
self.at = at
|
||
|
||
|
||
def on(check):
|
||
return check not in SKIP
|
||
|
||
|
||
# ------------------------------------------------------------------ the extension's text
|
||
|
||
def extension_text():
|
||
with open(EXTENSION, encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
|
||
def section(text, heading):
|
||
at = text.index(heading)
|
||
end = text.find("\n## ", at + 1)
|
||
return text[at:end if end >= 0 else len(text)]
|
||
|
||
|
||
def read_k_table(text):
|
||
"""§3.3's table: K_1..K_16."""
|
||
s = section(text, "### 3.3 ")
|
||
k = {}
|
||
for line in s.splitlines():
|
||
cells = [c.strip().strip("*") for c in line.strip().strip("|").split("|")]
|
||
if len(cells) == 8 and cells[0].isdigit():
|
||
for i in range(0, 8, 2):
|
||
k[int(cells[i])] = int(cells[i + 1])
|
||
return [k[i] for i in range(1, 17)]
|
||
|
||
|
||
def read_pi(text):
|
||
"""§5.4's constant P = ⌊π · 2^62⌋."""
|
||
m = re.search(r"2\^62⌋ =\s*([0-9]+)", section(text, "### 5.4 "))
|
||
return int(m.group(1))
|
||
|
||
|
||
def read_reasons(text):
|
||
return [l.split("`")[1] for l in section(text, "\n## 9. Reasons").splitlines() if l.startswith("| `")]
|
||
|
||
|
||
TEXT = extension_text()
|
||
K = read_k_table(TEXT)
|
||
PI62 = read_pi(TEXT)
|
||
REASONS = read_reasons(TEXT)
|
||
|
||
|
||
# ------------------------------------------------------------------ §3 integer arithmetic
|
||
|
||
def rnd(a, b):
|
||
"""rnd(a, b) = ⌊(a + ⌊b/2⌋) / b⌋, b > 0 (§3.1)."""
|
||
return (a + b // 2) // b
|
||
|
||
|
||
def L2(v):
|
||
"""§3.2: log2(v) in Q16, from below."""
|
||
n = v.bit_length() - 1
|
||
m = v << (30 - n) if n <= 30 else v >> (n - 30)
|
||
f = 0
|
||
for i in range(1, 17):
|
||
m = (m * m) >> 30
|
||
if m >= 1 << 31:
|
||
m >>= 1
|
||
f += 1 << (16 - i)
|
||
return (n << 16) + f
|
||
|
||
|
||
def k_by_rule():
|
||
"""§3.3: R_0 = 2^61, R_i = ⌊√(R_{i−1} · 2^60)⌋, K_i = rnd(R_i, 2^30)."""
|
||
r, out = 1 << 61, []
|
||
for _ in range(16):
|
||
r = math.isqrt(r << 60)
|
||
out.append(rnd(r, 1 << 30))
|
||
return out
|
||
|
||
|
||
_E2 = {}
|
||
|
||
|
||
def E2(y):
|
||
"""§3.3: (p, e) with 2^(y / 2^16) ≈ p · 2^(e − 30)."""
|
||
e = y >> 16
|
||
f = y - (e << 16)
|
||
p = _E2.get(f)
|
||
if p is None:
|
||
p = 1 << 30
|
||
for i in range(1, 17):
|
||
if (f >> (16 - i)) & 1:
|
||
p = rnd(p * K[i - 1], 1 << 30)
|
||
_E2[f] = p
|
||
return p, e
|
||
|
||
|
||
def scale(x, y, f):
|
||
"""§3.3: X · 2^(y / 2^16) / 2^F as an integer."""
|
||
p, e = E2(y)
|
||
t = e - 30 - f
|
||
return (x * p) << t if t >= 0 else rnd(x * p, 1 << -t)
|
||
|
||
|
||
def srgb_table():
|
||
"""§3.4."""
|
||
base = L2(269025)
|
||
s = []
|
||
for k in range(256):
|
||
if k <= 10:
|
||
s.append(rnd(k * 100 * (1 << 30), 255 * 1292))
|
||
else:
|
||
lam = L2(1000 * k + 14025) - base
|
||
s.append(scale(1 << 30, rnd(12 * lam, 5), 0))
|
||
return s
|
||
|
||
|
||
S = srgb_table()
|
||
|
||
|
||
def hq16(h):
|
||
"""§3.5: a headroom in Q16."""
|
||
return rnd(h * (1 << 16), 10000)
|
||
|
||
|
||
def code(v):
|
||
return min(max(v, 0), 65535)
|
||
|
||
|
||
def rgb16(samples):
|
||
return struct.pack("<%dH" % len(samples), *samples)
|
||
|
||
|
||
def sha(b):
|
||
return hashlib.sha256(b).digest()
|
||
|
||
|
||
# ------------------------------------------------------------------ PNG (8-bit grey or RGB)
|
||
|
||
PNG_SIG = b"\x89PNG\r\n\x1a\n"
|
||
|
||
|
||
def png_decode(b, grey_ok):
|
||
"""A PNG of 8-bit samples, colour type 2 (and 0 where grey_ok), no interlace: (w, h, channels, samples)."""
|
||
if b[:8] != PNG_SIG:
|
||
raise ValueError("signature")
|
||
at, ihdr, z, ended, palette, idat = 8, None, b"", False, False, 0
|
||
while at < len(b):
|
||
if ended:
|
||
raise ValueError("after IEND")
|
||
if at + 8 > len(b):
|
||
raise ValueError("truncated")
|
||
ln = struct.unpack_from(">I", b, at)[0]
|
||
kind = b[at + 4:at + 8]
|
||
if at + 12 + ln > len(b):
|
||
raise ValueError("truncated")
|
||
data = b[at + 8:at + 8 + ln]
|
||
if not all(65 <= c <= 90 or 97 <= c <= 122 for c in kind):
|
||
raise ValueError("type")
|
||
if struct.unpack_from(">I", b, at + 8 + ln)[0] != zlib.crc32(kind + data) & 0xFFFFFFFF:
|
||
raise ValueError("crc")
|
||
if idat == 1 and kind != b"IDAT":
|
||
idat = 2
|
||
if kind == b"IHDR":
|
||
if ihdr is not None or at != 8 or ln != 13:
|
||
raise ValueError("IHDR")
|
||
w, h = struct.unpack_from(">II", data, 0)
|
||
tail = data[8:]
|
||
if tail == bytes([8, 2, 0, 0, 0]):
|
||
ch = 3
|
||
elif grey_ok and tail == bytes([8, 0, 0, 0, 0]):
|
||
ch = 1
|
||
else:
|
||
raise ValueError("format")
|
||
if w == 0 or h == 0 or w * h * ch > 1 << 26:
|
||
raise ValueError("size")
|
||
ihdr = (w, h, ch)
|
||
elif ihdr is None:
|
||
raise ValueError("before IHDR")
|
||
elif kind == b"PLTE":
|
||
if ihdr[2] == 1 or palette or idat != 0 or ln == 0 or ln % 3 or ln > 768:
|
||
raise ValueError("PLTE")
|
||
palette = True
|
||
elif kind == b"IDAT":
|
||
if idat == 2:
|
||
raise ValueError("IDAT")
|
||
idat = 1
|
||
z += data
|
||
elif kind == b"IEND":
|
||
if ln:
|
||
raise ValueError("IEND")
|
||
ended = True
|
||
elif not kind[0] & 0x20:
|
||
raise ValueError("critical")
|
||
at += 12 + ln
|
||
if ihdr is None or not ended:
|
||
raise ValueError("incomplete")
|
||
w, h, ch = ihdr
|
||
d = zlib.decompressobj()
|
||
raw = d.decompress(z)
|
||
if not d.eof or d.unused_data:
|
||
raise ValueError("zlib")
|
||
stride = ch * w
|
||
if len(raw) != (stride + 1) * h:
|
||
raise ValueError("rows")
|
||
out = bytearray(stride * h)
|
||
for y in range(h):
|
||
f = raw[y * (stride + 1)]
|
||
row = raw[y * (stride + 1) + 1:(y + 1) * (stride + 1)]
|
||
for i in range(stride):
|
||
a = out[y * stride + i - ch] if i >= ch else 0
|
||
up = out[(y - 1) * stride + i] if y else 0
|
||
c = out[(y - 1) * stride + i - ch] if y and i >= ch else 0
|
||
if f == 0:
|
||
p = 0
|
||
elif f == 1:
|
||
p = a
|
||
elif f == 2:
|
||
p = up
|
||
elif f == 3:
|
||
p = (a + up) // 2
|
||
elif f == 4:
|
||
q = a + up - c
|
||
pa, pb, pc = abs(q - a), abs(q - up), abs(q - c)
|
||
p = a if pa <= pb and pa <= pc else (up if pb <= pc else c)
|
||
else:
|
||
raise ValueError("filter")
|
||
out[y * stride + i] = (row[i] + p) & 0xFF
|
||
return w, h, ch, bytes(out)
|
||
|
||
|
||
# ------------------------------------------------------------------ §4 gain maps
|
||
|
||
def read_iso(b):
|
||
"""§4.2 steps 1 and 2: the fields, big-endian."""
|
||
if len(b) < 2:
|
||
raise Refuse("metadata_malformed")
|
||
minimum = struct.unpack_from(">H", b, 0)[0]
|
||
if minimum != 0 and on("iso_version"):
|
||
raise Refuse("metadata_unsupported")
|
||
if len(b) < 5:
|
||
raise Refuse("metadata_malformed")
|
||
writer = struct.unpack_from(">H", b, 2)[0]
|
||
if writer < minimum:
|
||
raise Refuse("metadata_malformed")
|
||
flags = b[4]
|
||
multi, base_space = bool(flags & 0x80), bool(flags & 0x40)
|
||
at = [5]
|
||
|
||
def pair(signed):
|
||
if at[0] + 8 > len(b):
|
||
raise Refuse("metadata_malformed")
|
||
n = struct.unpack_from(">i" if signed else ">I", b, at[0])[0]
|
||
d = struct.unpack_from(">I", b, at[0] + 4)[0]
|
||
at[0] += 8
|
||
return n, d
|
||
|
||
bh, ah = pair(False), pair(False)
|
||
chans = [[pair(True), pair(True), pair(False), pair(True), pair(True)] for _ in range(3 if multi else 1)]
|
||
if writer == 0 and at[0] != len(b) and on("iso_length"):
|
||
raise Refuse("metadata_malformed")
|
||
return {"multi": multi, "base_space": base_space, "bh": bh, "ah": ah, "chans": chans}
|
||
|
||
|
||
def iso_params(m):
|
||
"""§4.2 steps 3 and 4, and the channels' Q16 / Q30 values."""
|
||
def le(nd, limit):
|
||
return abs(nd[0]) <= limit * nd[1]
|
||
ok = m["bh"][1] != 0 and m["ah"][1] != 0
|
||
ok = ok and le(m["bh"], 6) and le(m["ah"], 6)
|
||
ok = ok and m["ah"][0] * m["bh"][1] != m["bh"][0] * m["ah"][1]
|
||
for mn, mx, g, bo, ao in m["chans"]:
|
||
ok = ok and all(x[1] != 0 for x in (mn, mx, g, bo, ao))
|
||
if not ok:
|
||
break
|
||
ok = ok and le(mn, 16) and le(mx, 16) and Fraction(*mn) <= Fraction(*mx)
|
||
ok = ok and g[0] >= 1 and Fraction(1, 64) <= Fraction(*g) <= 64
|
||
ok = ok and le(bo, 1) and le(ao, 1)
|
||
if not ok and on("iso_range"):
|
||
raise Refuse("metadata_out_of_range")
|
||
if not m["base_space"] and on("gain_space"):
|
||
raise Refuse("gain_space_unsupported")
|
||
return [{"min": rnd(mn[0] << 16, mn[1]), "max": rnd(mx[0] << 16, mx[1]), "g": g,
|
||
"bo": rnd(bo[0] << 30, bo[1]), "ao": rnd(ao[0] << 30, ao[1])} for mn, mx, g, bo, ao in m["chans"]]
|
||
|
||
|
||
def read_gainmap(p):
|
||
"""§4.1 to §4.3."""
|
||
if (len(p) < 8 or p[:4] != b"WGM1" or p[4] != 1 or p[5] != 0) and on("wgm1_frame"):
|
||
raise Refuse("companion_malformed")
|
||
md_len = struct.unpack_from("<H", p, 6)[0]
|
||
if (md_len == 0 or md_len > MAX_METADATA or len(p) < 8 + md_len) and on("wgm1_frame"):
|
||
raise Refuse("companion_malformed")
|
||
try:
|
||
w, h, ch, samples = png_decode(p[8 + md_len:], True)
|
||
except (ValueError, zlib.error, struct.error, IndexError):
|
||
raise Refuse("companion_malformed")
|
||
m = read_iso(p[8:8 + md_len])
|
||
chans = iso_params(m)
|
||
if ch != (3 if m["multi"] else 1) and on("channels"):
|
||
raise Refuse("map_channels_mismatch")
|
||
return {"multi": m["multi"], "bh": m["bh"], "ah": m["ah"], "chans": chans, "w": w, "h": h, "ch": ch, "samples": samples}
|
||
|
||
|
||
def gain_weight(g, hd):
|
||
"""§4.4 step 1."""
|
||
(bn, bd), (an, ad) = g["bh"], g["ah"]
|
||
a = hd * bd * ad - 10000 * bn * ad
|
||
d = 10000 * (an * bd - bn * ad)
|
||
if d > 0:
|
||
return min(max(rnd(a << 16, d), 0), 1 << 16)
|
||
return -min(max(rnd((-a) << 16, -d), 0), 1 << 16)
|
||
|
||
|
||
def gain_factor(g, w, h):
|
||
for s in (1, 2, 4, 8):
|
||
if g["w"] * s == w and g["h"] * s == h:
|
||
return s
|
||
if on("map_dims"):
|
||
raise Refuse("map_dimensions_mismatch")
|
||
return 1
|
||
|
||
|
||
def render_gainmap(g, base, w, h, hd):
|
||
"""§4.4."""
|
||
s = gain_factor(g, w, h)
|
||
wt, ht = gain_weight(g, hd), hq16(hd)
|
||
gw, gh, ch, smp = g["w"], g["h"], g["ch"], g["samples"]
|
||
q = 255 * 4 * s * s
|
||
lq = L2(q)
|
||
out = []
|
||
for y in range(h):
|
||
yy = max(0, 2 * y + 1 - s)
|
||
y0, ay = divmod(yy, 2 * s)
|
||
y1 = min(y0 + 1, gh - 1)
|
||
for x in range(w):
|
||
xx = max(0, 2 * x + 1 - s)
|
||
x0, ax = divmod(xx, 2 * s)
|
||
x1 = min(x0 + 1, gw - 1)
|
||
for c in range(3):
|
||
j = c if g["multi"] else 0
|
||
p = g["chans"][j]
|
||
|
||
def at(xi, yi):
|
||
return smp[(yi * gw + xi) * ch + j]
|
||
n = (at(x0, y0) * (2 * s - ax) * (2 * s - ay) + at(x1, y0) * ax * (2 * s - ay)
|
||
+ at(x0, y1) * (2 * s - ax) * ay + at(x1, y1) * ax * ay)
|
||
gn, gd = p["g"]
|
||
if gn == gd:
|
||
v = rnd(n << 30, q)
|
||
elif n == 0:
|
||
v = 0
|
||
else:
|
||
v = scale(1 << 30, rnd((L2(n) - lq) * gd, gn), 0)
|
||
lg = rnd(p["min"] * ((1 << 30) - v) + p["max"] * v, 1 << 30)
|
||
gg = rnd(lg * wt, 1 << 16)
|
||
t = scale((S[base[(y * w + x) * 3 + c]] + p["bo"]) * 65535, gg - ht, 0) - scale(p["ao"] * 65535, -ht, 0)
|
||
out.append(code(rnd(t, 1 << 30)))
|
||
return out
|
||
|
||
|
||
def float_gainmap(g, base, w, h, hd):
|
||
"""§6's binary64 evaluation of §4.4's formulas."""
|
||
s = gain_factor(g, w, h)
|
||
|
||
def srgb(k):
|
||
e = k / 255.0
|
||
return e / 12.92 if e <= 0.04045 else ((e + 0.055) / 1.055) ** 2.4
|
||
hb = g["bh"][0] / g["bh"][1]
|
||
ha = g["ah"][0] / g["ah"][1]
|
||
ht = hd / 10000.0
|
||
wt = min(max((ht - hb) / (ha - hb), 0.0), 1.0)
|
||
if ha < hb:
|
||
wt = -wt
|
||
gw, gh, ch, smp = g["w"], g["h"], g["ch"], g["samples"]
|
||
out = []
|
||
for y in range(h):
|
||
fy = max(0.0, (y + 0.5) / s - 0.5)
|
||
y0 = math.floor(fy)
|
||
ay = fy - y0
|
||
y1 = min(y0 + 1, gh - 1)
|
||
for x in range(w):
|
||
fx = max(0.0, (x + 0.5) / s - 0.5)
|
||
x0 = math.floor(fx)
|
||
ax = fx - x0
|
||
x1 = min(x0 + 1, gw - 1)
|
||
for c in range(3):
|
||
j = c if g["multi"] else 0
|
||
p = g["chans"][j]
|
||
|
||
def at(xi, yi):
|
||
return smp[(yi * gw + xi) * ch + j] / 255.0
|
||
v = (at(x0, y0) * (1 - ax) * (1 - ay) + at(x1, y0) * ax * (1 - ay)
|
||
+ at(x0, y1) * (1 - ax) * ay + at(x1, y1) * ax * ay)
|
||
mn, mx = p["min"] / 65536.0, p["max"] / 65536.0
|
||
gam = p["g"][0] / p["g"][1]
|
||
l = mn + (mx - mn) * v ** (1.0 / gam)
|
||
t = (srgb(base[(y * w + x) * 3 + c]) + p["bo"] / 2.0 ** 30) * 2.0 ** (l * wt) - p["ao"] / 2.0 ** 30
|
||
out.append(int(min(max(round_half_away(t / 2.0 ** ht * 65535.0), 0), 65535)))
|
||
return out
|
||
|
||
|
||
def round_half_away(x):
|
||
"""Rounding to nearest, halves away from zero (binary64's `round`)."""
|
||
return math.floor(x + 0.5) if x >= 0 else -math.floor(-x + 0.5)
|
||
|
||
|
||
# ------------------------------------------------------------------ §5 ST 2094-50
|
||
|
||
class Bits:
|
||
def __init__(self, b):
|
||
self.b, self.at = b, 0
|
||
|
||
def u(self, n):
|
||
v = 0
|
||
for _ in range(n):
|
||
if self.at // 8 >= len(self.b):
|
||
raise Refuse("metadata_malformed")
|
||
v = (v << 1) | ((self.b[self.at // 8] >> (7 - self.at % 8)) & 1)
|
||
self.at += 1
|
||
return v
|
||
|
||
def zero(self, n):
|
||
if self.u(n) != 0 and on("annexc_reserved"):
|
||
raise Refuse("metadata_malformed")
|
||
|
||
|
||
def tan_q30(theta):
|
||
"""§5.4: tan of the angle, through Q62 series."""
|
||
d = min(max(theta, 1), 35999) - 18000
|
||
a = rnd(d * PI62, 36000)
|
||
q = rnd(a * a, 1 << 62)
|
||
t = sig = a
|
||
for k in range(1, 16):
|
||
t = rnd(-t * q, (1 << 62) * (2 * k) * (2 * k + 1))
|
||
sig += t
|
||
t = kap = 1 << 62
|
||
for k in range(1, 16):
|
||
t = rnd(-t * q, (1 << 62) * (2 * k - 1) * (2 * k))
|
||
kap += t
|
||
return rnd(sig << 30, kap)
|
||
|
||
|
||
def sign(fr):
|
||
return (fr > 0) - (fr < 0)
|
||
|
||
|
||
def pchip(x, y):
|
||
"""C.3.9 on exact rationals (§5.4); slopes in stops per unit of x."""
|
||
n = len(x)
|
||
h = [x[i + 1] - x[i] for i in range(n - 1)]
|
||
s = [Fraction(0) if h[i] == 0 else Fraction(y[i + 1] - y[i], 10 * h[i]) for i in range(n - 1)]
|
||
out = []
|
||
for i in range(n):
|
||
prev = i >= 1 and x[i - 1] < x[i]
|
||
nxt = i + 1 < n and x[i] < x[i + 1]
|
||
if prev and nxt:
|
||
s0, s1 = s[i - 1], s[i]
|
||
if sign(s0) != sign(s1) or (s0 == 0 and s1 == 0):
|
||
m = Fraction(0)
|
||
else:
|
||
h0, h1 = h[i - 1], h[i]
|
||
m = 3 * (h0 + h1) * s0 * s1 / ((2 * h0 + h1) * s0 + (h0 + 2 * h1) * s1)
|
||
elif nxt:
|
||
if i + 2 < n and x[i + 1] < x[i + 2]:
|
||
m = ((2 * h[i] + h[i + 1]) * s[i] - h[i] * s[i + 1]) / (h[i] + h[i + 1])
|
||
if sign(m) != sign(s[i]):
|
||
m = Fraction(0)
|
||
if sign(m) != sign(s[i + 1]) and abs(m) > 3 * abs(s[i]):
|
||
m = 3 * s[i]
|
||
else:
|
||
m = s[i]
|
||
elif prev:
|
||
if i >= 2 and x[i - 2] < x[i - 1]:
|
||
m = ((2 * h[i - 1] + h[i - 2]) * s[i - 1] - h[i - 1] * s[i - 2]) / (h[i - 1] + h[i - 2])
|
||
if sign(m) != sign(s[i - 1]):
|
||
m = Fraction(0)
|
||
if sign(m) != sign(s[i - 2]) and abs(m) > 3 * abs(s[i - 1]):
|
||
m = 3 * s[i - 1]
|
||
else:
|
||
m = s[i - 1]
|
||
else:
|
||
m = Fraction(0)
|
||
out.append(m)
|
||
return out
|
||
|
||
|
||
def triangle_ok(c):
|
||
p = [(c[2 * i], c[2 * i + 1]) for i in range(4)]
|
||
|
||
def cross(a, b, q):
|
||
return (b[0] - a[0]) * (q[1] - a[1]) - (b[1] - a[1]) * (q[0] - a[0])
|
||
r, g, b, w = p
|
||
area = cross(r, g, b)
|
||
if area == 0:
|
||
return False
|
||
sgn = sign(area)
|
||
return sign(cross(r, g, w)) == sgn and sign(cross(g, b, w)) == sgn and sign(cross(b, r, w)) == sgn
|
||
|
||
|
||
MIX_TYPES = {0: [0, 0, 0, 150000, 0, 0], 1: [0, 0, 0, 0, 0, 150000], 2: [25000, 25000, 25000, 75000, 0, 0]}
|
||
|
||
|
||
def read_set(b, first, count):
|
||
"""§5.2 over one Annex C metadata set."""
|
||
if not b:
|
||
raise Refuse("metadata_malformed")
|
||
if (b[0] >> 2) & 7 and on("annexc_version"):
|
||
raise Refuse("metadata_unsupported")
|
||
r = Bits(b)
|
||
r.u(3)
|
||
r.u(3)
|
||
r.zero(2)
|
||
custom, atm = r.u(1), r.u(1)
|
||
r.zero(6)
|
||
if custom:
|
||
r.u(16)
|
||
if not atm:
|
||
return {"first": first, "count": count, "tm": None}
|
||
baseline_raw = r.u(16)
|
||
if r.u(1):
|
||
r.zero(7)
|
||
if on("annexc_gain_space"):
|
||
raise Refuse("gain_space_unsupported")
|
||
return {"first": first, "count": count, "tm": (min(baseline_raw, 60000), [])}
|
||
num, mode, common_mix, common_curve = r.u(3), r.u(2), r.u(1), r.u(1)
|
||
chroma = [min(r.u(16), 50000) for _ in range(8)] if mode == 3 else None
|
||
raw = []
|
||
for a in range(min(num, 4)):
|
||
hr = r.u(16)
|
||
if a == 0 or not common_mix:
|
||
t = r.u(2)
|
||
if t != 3:
|
||
r.zero(6)
|
||
k = list(MIX_TYPES[t])
|
||
else:
|
||
flags = [r.u(1) for _ in range(6)]
|
||
k = [3 * min(r.u(16), 50000) if f else 0 for f in flags]
|
||
else:
|
||
k = list(raw[0][1])
|
||
if a == 0 or not common_curve:
|
||
ncp = r.u(5) + 1
|
||
pch = r.u(1)
|
||
r.zero(2)
|
||
xs = [r.u(16) for _ in range(ncp)]
|
||
else:
|
||
pch, xs = raw[0][2], list(raw[0][3])
|
||
ys = [r.u(16) for _ in range(len(xs))]
|
||
th = [r.u(16) for _ in range(len(xs))] if not pch else None
|
||
raw.append((hr, k, pch, xs, ys, th))
|
||
baseline = min(baseline_raw, 60000)
|
||
alts, bad = [], False
|
||
for hr, k, pch, xs, ys, th in raw:
|
||
sg = 1 if baseline_raw < hr else -1
|
||
x = [min(v, 64000) for v in xs]
|
||
y = [sg * min(v, 60000) for v in ys]
|
||
for i in range(len(x) - 1):
|
||
if x[i] > x[i + 1] or (x[i] == x[i + 1] and y[i] != y[i + 1]):
|
||
bad = True
|
||
if x[-1] == 0 or sum(k) == 0:
|
||
bad = True
|
||
alts.append({"h": min(hr, 60000), "k": k, "x": x, "y": y, "pchip": pch, "theta": th})
|
||
for i, a in enumerate(alts):
|
||
if a["h"] == baseline or (i + 1 < len(alts) and a["h"] >= alts[i + 1]["h"]):
|
||
bad = True
|
||
if mode == 3 and not triangle_ok(chroma):
|
||
bad = True
|
||
if bad and on("annexc_range"):
|
||
raise Refuse("metadata_out_of_range")
|
||
if mode != 0 and on("annexc_gain_space"):
|
||
raise Refuse("gain_space_unsupported")
|
||
for a in alts:
|
||
if a["pchip"]:
|
||
fr = pchip(a["x"], a["y"])
|
||
a["m"] = [rnd(f.numerator << 30, f.denominator) for f in fr]
|
||
a["mf"] = [float(f) for f in fr]
|
||
else:
|
||
a["m"] = [tan_q30(t) for t in a["theta"]]
|
||
a["mf"] = [math.tan((min(max(t, 1), 35999) - 18000.0) * math.pi / 36000.0) for t in a["theta"]]
|
||
a["cache"] = {}
|
||
return {"first": first, "count": count, "tm": (baseline, alts)}
|
||
|
||
|
||
def read_wd50(p):
|
||
"""§5.1, then §5.2 for each set."""
|
||
if len(p) < 8 or p[:4] != b"WD50" or p[4] != 1 or p[5] != 0:
|
||
raise Refuse("companion_malformed")
|
||
n = struct.unpack_from("<H", p, 6)[0]
|
||
if not 1 <= n <= MAX_SETS:
|
||
raise Refuse("companion_malformed")
|
||
at, end, raw = 8, 0, []
|
||
for _ in range(n):
|
||
if at + 10 > len(p):
|
||
raise Refuse("companion_malformed")
|
||
first, count, ln = struct.unpack_from("<IIH", p, at)
|
||
if (count == 0 or first < end or first + count > 0xFFFFFFFF or not 1 <= ln <= MAX_METADATA) and on("wd50_frame"):
|
||
raise Refuse("companion_malformed")
|
||
if at + 10 + ln > len(p):
|
||
raise Refuse("companion_malformed")
|
||
raw.append((first, count, p[at + 10:at + 10 + ln]))
|
||
end = first + count
|
||
at += 10 + ln
|
||
if at != len(p):
|
||
raise Refuse("companion_malformed")
|
||
return [read_set(b, f, c) for f, c, b in raw]
|
||
|
||
|
||
def curve(a, m):
|
||
"""§5.5: the gain curve at M (Q30), in Q16."""
|
||
x, y, ms = a["x"], a["y"], a["m"]
|
||
n = len(x)
|
||
xm = 1000 * m
|
||
if xm < x[0] << 30:
|
||
return rnd(y[0] << 16, 10000)
|
||
for i in range(n):
|
||
if xm == x[i] << 30:
|
||
return rnd(y[i] << 16, 10000)
|
||
for i in range(n - 1):
|
||
if x[i] << 30 < xm < x[i + 1] << 30:
|
||
h = x[i + 1] - x[i]
|
||
t = rnd(xm - (x[i] << 30), h)
|
||
y0, y1 = rnd(y[i] << 30, 10000), rnd(y[i + 1] << 30, 10000)
|
||
m0, m1 = rnd(h * ms[i], 1000), rnd(h * ms[i + 1], 1000)
|
||
c3 = 2 * y0 + m0 - 2 * y1 + m1
|
||
c2 = -3 * y0 + 3 * y1 - 2 * m0 - m1
|
||
u = rnd(c3 * t, 1 << 30) + c2
|
||
u = rnd(u * t, 1 << 30) + m0
|
||
u = rnd(u * t, 1 << 30) + y0
|
||
return rnd(u, 1 << 14)
|
||
return rnd(y[n - 1] << 16, 10000) + L2(x[n - 1] << 30) - L2(xm)
|
||
|
||
|
||
def gain(a, c):
|
||
"""Formulas (7), (9), (10)."""
|
||
hit = a["cache"].get(c)
|
||
if hit is None:
|
||
k = a["k"]
|
||
ksum = sum(k)
|
||
common = k[0] * c[0] + k[1] * c[1] + k[2] * c[2] + k[3] * max(c) + k[4] * min(c)
|
||
hit = tuple(curve(a, rnd(cj * k[5] + common, ksum)) for cj in c)
|
||
a["cache"][c] = hit
|
||
return hit
|
||
|
||
|
||
def plan(st, hd):
|
||
baseline, alts = st["tm"]
|
||
hs = sorted([a["h"] for a in alts] + [baseline])
|
||
entries = []
|
||
for hv in hs:
|
||
entries.append(None if hv == baseline else next(a for a in alts if a["h"] == hv))
|
||
hc = min(max(hd, hs[0]), hs[-1])
|
||
return hs, entries, hc
|
||
|
||
|
||
def render_st(st, frame, hd):
|
||
"""§5.6."""
|
||
ht = hq16(hd)
|
||
out = []
|
||
if st["tm"] is not None:
|
||
hs, entries, hc = plan(st, hd)
|
||
eq = hs.index(hc) if hc in hs else None
|
||
i = eq if eq is not None else next(i for i in range(len(hs) - 1) if hs[i] < hc < hs[i + 1])
|
||
|
||
def g_at(idx, c):
|
||
return (0, 0, 0) if entries[idx] is None else gain(entries[idx], c)
|
||
for px in range(0, len(frame), 3):
|
||
c = (S[frame[px]], S[frame[px + 1]], S[frame[px + 2]])
|
||
if st["tm"] is None:
|
||
g = (0, 0, 0)
|
||
elif eq is not None:
|
||
g = g_at(i, c)
|
||
else:
|
||
g0, g1 = g_at(i, c), g_at(i + 1, c)
|
||
h0, h1 = hs[i], hs[i + 1]
|
||
g = tuple(rnd(g0[k] * (h1 - hc) + g1[k] * (hc - h0), h1 - h0) for k in range(3))
|
||
for k in range(3):
|
||
out.append(code(scale(c[k] * 65535, g[k] - ht, 30)))
|
||
return out
|
||
|
||
|
||
def float_st(st, frame, hd):
|
||
"""§6's binary64 evaluation of §5.6's formulas."""
|
||
def srgb(k):
|
||
e = k / 255.0
|
||
return e / 12.92 if e <= 0.04045 else ((e + 0.055) / 1.055) ** 2.4
|
||
|
||
def fcurve(a, xv):
|
||
xs = [v / 1000.0 for v in a["x"]]
|
||
ys = [v / 10000.0 for v in a["y"]]
|
||
n = len(xs)
|
||
if xv < xs[0]:
|
||
return ys[0]
|
||
for i in range(n):
|
||
if xv == xs[i]:
|
||
return ys[i]
|
||
for i in range(n - 1):
|
||
if xs[i] < xv < xs[i + 1]:
|
||
h = xs[i + 1] - xs[i]
|
||
t = (xv - xs[i]) / h
|
||
m0, m1 = h * a["mf"][i], h * a["mf"][i + 1]
|
||
c3 = 2 * ys[i] + m0 - 2 * ys[i + 1] + m1
|
||
c2 = -3 * ys[i] + 3 * ys[i + 1] - 2 * m0 - m1
|
||
return ((c3 * t + c2) * t + m0) * t + ys[i]
|
||
return ys[n - 1] + math.log2(xs[n - 1] / xv)
|
||
|
||
def fgain(a, c):
|
||
k = [v / 150000.0 for v in a["k"]]
|
||
ksum = sum(k)
|
||
common = k[0] * c[0] + k[1] * c[1] + k[2] * c[2] + k[3] * max(c) + k[4] * min(c)
|
||
return [fcurve(a, (cj * k[5] + common) / ksum) for cj in c]
|
||
ht = hd / 10000.0
|
||
out = []
|
||
for px in range(0, len(frame), 3):
|
||
c = [srgb(frame[px]), srgb(frame[px + 1]), srgb(frame[px + 2])]
|
||
if st["tm"] is None:
|
||
g = [0.0] * 3
|
||
else:
|
||
baseline, alts = st["tm"]
|
||
lst = sorted([(a["h"] / 10000.0, a) for a in alts] + [(baseline / 10000.0, None)], key=lambda e: e[0])
|
||
hc = min(max(ht, lst[0][0]), lst[-1][0])
|
||
|
||
def fg(e):
|
||
return [0.0] * 3 if e[1] is None else fgain(e[1], c)
|
||
hit = [e for e in lst if e[0] == hc]
|
||
if hit:
|
||
g = fg(hit[0])
|
||
else:
|
||
i = next(i for i in range(len(lst) - 1) if lst[i][0] < hc < lst[i + 1][0])
|
||
w0 = (hc - lst[i + 1][0]) / (lst[i][0] - lst[i + 1][0])
|
||
g0, g1 = fg(lst[i]), fg(lst[i + 1])
|
||
g = [w0 * g0[k] + (1 - w0) * g1[k] for k in range(3)]
|
||
for k in range(3):
|
||
t = c[k] * 2.0 ** g[k]
|
||
out.append(int(min(max(round_half_away(t / 2.0 ** ht * 65535.0), 0), 65535)))
|
||
return out
|
||
|
||
|
||
# ------------------------------------------------------------------ §2 and §7 records
|
||
|
||
class Malformed(ValueError):
|
||
pass
|
||
|
||
|
||
def strict_json(text):
|
||
def pairs(kv):
|
||
d = {}
|
||
for k, v in kv:
|
||
if k in d:
|
||
raise Malformed("a member named twice")
|
||
d[k] = v
|
||
return d
|
||
|
||
def bad(_):
|
||
raise Malformed("a number outside the subset")
|
||
|
||
def an_int(s):
|
||
if s.startswith("-"):
|
||
raise Malformed("a signed integer")
|
||
return int(s)
|
||
v = json.loads(text, object_pairs_hook=pairs, parse_float=bad, parse_constant=bad, parse_int=an_int)
|
||
|
||
def depth(x, d):
|
||
if d > 32:
|
||
raise Malformed("nesting")
|
||
if isinstance(x, dict):
|
||
for y in x.values():
|
||
depth(y, d + 1)
|
||
elif isinstance(x, list):
|
||
for y in x:
|
||
depth(y, d + 1)
|
||
depth(v, 1)
|
||
return v
|
||
|
||
|
||
def subset_ok(v):
|
||
if v is None or isinstance(v, str):
|
||
return True
|
||
if isinstance(v, bool):
|
||
return False
|
||
if isinstance(v, int):
|
||
return 0 <= v <= MAX_INT
|
||
if isinstance(v, list):
|
||
return all(subset_ok(x) for x in v)
|
||
if isinstance(v, dict):
|
||
return all(subset_ok(x) for x in v.values())
|
||
return False
|
||
|
||
|
||
def jcs(v):
|
||
return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||
|
||
|
||
def is_hex(s, n):
|
||
return isinstance(s, str) and len(s) == 2 * n and all(c in "0123456789abcdef" for c in s)
|
||
|
||
|
||
def is_int(v, lo=0, hi=MAX_INT):
|
||
return isinstance(v, int) and not isinstance(v, bool) and lo <= v <= hi
|
||
|
||
|
||
def exact(v, members):
|
||
return isinstance(v, dict) and set(v) == set(members)
|
||
|
||
|
||
WS = set("\t\n\x0b\x0c\r \x85\xa0 "
|
||
"
")
|
||
|
||
|
||
def term_ok(s):
|
||
return (isinstance(s, str) and 0 < len(s.encode("utf-8")) <= 2048
|
||
and not any(c in WS or ord(c) < 32 or 127 <= ord(c) < 160 for c in s))
|
||
|
||
|
||
DATE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})[Tt](\d{2}):(\d{2}):(\d{2})(\.\d+)?([Zz]|[+-](\d{2}):(\d{2}))$")
|
||
|
||
|
||
def date_ok(s):
|
||
m = DATE.match(s) if isinstance(s, str) else None
|
||
if not m:
|
||
return False
|
||
y, mo, d, hh, mi, ss = (int(m.group(i)) for i in range(1, 7))
|
||
if not 1 <= mo <= 12 or hh > 23 or mi > 59 or ss > 60:
|
||
return False
|
||
leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
|
||
days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][mo - 1]
|
||
if not 1 <= d <= days:
|
||
return False
|
||
if m.group(9) is not None and (int(m.group(9)) > 23 or int(m.group(10)) > 59):
|
||
return False
|
||
return True
|
||
|
||
|
||
SIG_MEMO = {}
|
||
|
||
|
||
def read_record(text):
|
||
"""§2, §7.1 and §8.2 step 1."""
|
||
try:
|
||
v = strict_json(text)
|
||
except (ValueError, RecursionError):
|
||
raise Refuse("record_malformed")
|
||
members = ["kind", "derivation", "base", "companion", "pictures", "class", "inputs", "issued_at", "signer_pubkey", "sig"]
|
||
ok = exact(v, members) and on("members") or (isinstance(v, dict) and not on("members"))
|
||
if not ok or v.get("kind") != KIND or not subset_ok(v):
|
||
raise Refuse("record_malformed")
|
||
d = v.get("derivation")
|
||
if not term_ok(d) or d not in (GAINMAP, ST2094):
|
||
raise Refuse("record_malformed")
|
||
b = v.get("base")
|
||
if not exact(b, ["object", "rendition", "width", "height"]) or not is_hex(b["object"], 32) or not is_int(b["rendition"]):
|
||
raise Refuse("record_malformed")
|
||
if not is_int(b["width"], 1, MAX_SIDE) or not is_int(b["height"], 1, MAX_SIDE):
|
||
raise Refuse("record_malformed")
|
||
c = v.get("companion")
|
||
if not exact(c, ["index", "sha256"]) or not is_int(c["index"]) or not is_hex(c["sha256"], 32):
|
||
raise Refuse("record_malformed")
|
||
pics = v.get("pictures")
|
||
if not isinstance(pics, list) or not 1 <= len(pics) <= MAX_PICTURES:
|
||
raise Refuse("record_malformed")
|
||
prev, high = None, False
|
||
for p in pics:
|
||
if not exact(p, ["frame", "headroom", "input", "output"]) or not is_int(p["frame"]) or not is_int(p["headroom"]):
|
||
raise Refuse("record_malformed")
|
||
if not is_hex(p["input"], 32) or not is_hex(p["output"], 32):
|
||
raise Refuse("record_malformed")
|
||
if d == GAINMAP and p["frame"] != 0:
|
||
raise Refuse("record_malformed")
|
||
key = (p["frame"], p["headroom"])
|
||
if prev is not None and key <= prev:
|
||
raise Refuse("record_malformed")
|
||
prev = key
|
||
high = high or p["headroom"] > MAX_HEADROOM
|
||
if v.get("class") not in CLASSES:
|
||
raise Refuse("record_malformed")
|
||
ins = v.get("inputs")
|
||
if not isinstance(ins, list) or not 1 <= len(ins) <= MAX_INPUTS:
|
||
raise Refuse("record_malformed")
|
||
seen = set()
|
||
for i in ins:
|
||
if not exact(i, ["content_hash", "origin", "class"]) or not is_hex(i["content_hash"], 32):
|
||
raise Refuse("record_malformed")
|
||
if i["origin"] is not None and not is_hex(i["origin"], 32):
|
||
raise Refuse("record_malformed")
|
||
if i["content_hash"] in seen or i["class"] not in CLASSES:
|
||
raise Refuse("record_malformed")
|
||
seen.add(i["content_hash"])
|
||
if b["object"] not in seen:
|
||
raise Refuse("record_malformed")
|
||
if not date_ok(v.get("issued_at")) or not is_hex(v.get("signer_pubkey"), 32) or not is_hex(v.get("sig"), 64):
|
||
raise Refuse("record_malformed")
|
||
body = {k: x for k, x in v.items() if k != "sig"}
|
||
key = (bytes.fromhex(v["signer_pubkey"]), SIGN_DOMAIN + jcs(body).encode("utf-8"), bytes.fromhex(v["sig"]))
|
||
if key not in SIG_MEMO:
|
||
SIG_MEMO[key] = ED.verify(*key)
|
||
why = SIG_MEMO[key]
|
||
if why is not None and on("signature"):
|
||
raise Refuse(why)
|
||
if high and on("headroom"):
|
||
raise Refuse("headroom_out_of_range")
|
||
v["_link"] = blake3(LINK_DOMAIN + jcs({k: x for k, x in v.items() if k != "_link"}).encode("utf-8")).hex()
|
||
return v
|
||
|
||
|
||
# ------------------------------------------------------------------ §8 verification
|
||
|
||
RENDER_MEMO = {}
|
||
|
||
|
||
def wai1_role(b):
|
||
"""SPEC §5 / §7.1 for a WAI1: a companion-only capability is neither its capability nor its fallback."""
|
||
m, _, _ = SV.read_wai1(b)
|
||
mr = m["model_requirement"]
|
||
if SV._companion_only(mr["capability"]):
|
||
return "companion-only-as-capability"
|
||
if mr.get("fallback") is not None and SV._companion_only(mr["fallback"]):
|
||
return "companion-only-as-fallback"
|
||
return None
|
||
|
||
|
||
def envelope_role(b):
|
||
if b[:4] == b"WAI2":
|
||
_, entries, _ = SV.read_wai2(b)
|
||
return SV.component_roles(entries)
|
||
return wai1_role(b)
|
||
|
||
|
||
def selectable(e):
|
||
return e.get("role") not in ("companion", "refinement") and SV._reg_selectable(e)
|
||
|
||
|
||
def bound(entries, base_i, comp_i):
|
||
"""SPEC §7.1: the companion at comp_i binds to the entry at base_i."""
|
||
e, base = entries[comp_i], entries[base_i]
|
||
if e.get("companion_of_id") is not None:
|
||
return base.get("id") == e["companion_of_id"]
|
||
return e.get("companion_of") == base["capability"]
|
||
|
||
|
||
def envelope_step(r, held):
|
||
"""§8.2 step 3. None where the envelope is not held, else (companion, still frame or None)."""
|
||
env = held["envelopes"].get(r["base"]["object"])
|
||
if env is None:
|
||
return None
|
||
want = CAP_GAINMAP if r["derivation"] == GAINMAP else CAP_ST2094
|
||
if env[:4] == b"WAI2":
|
||
try:
|
||
_, entries, payloads = SV.read_wai2(env)
|
||
except Exception:
|
||
raise Refuse("envelope_nonconforming")
|
||
if SV.component_roles(entries) is not None and on("roles"):
|
||
raise Refuse("envelope_nonconforming")
|
||
bi, ci = r["base"]["rendition"], r["companion"]["index"]
|
||
if on("binding"):
|
||
if bi >= len(entries) or not selectable(entries[bi]):
|
||
raise Refuse("companion_mismatch")
|
||
if ci >= len(entries) or entries[ci].get("role") != "companion" or entries[ci]["capability"] != want:
|
||
raise Refuse("companion_mismatch")
|
||
if not bound(entries, bi, ci):
|
||
raise Refuse("companion_mismatch")
|
||
if ci >= len(payloads):
|
||
raise Refuse("companion_mismatch")
|
||
payload = payloads[ci]
|
||
if sha(payload).hex() != r["companion"]["sha256"] and on("companion_digest"):
|
||
raise Refuse("companion_mismatch")
|
||
comp = read_gainmap(payload) if want == CAP_GAINMAP else read_wd50(payload)
|
||
still = None
|
||
if bi < len(entries) and entries[bi]["capability"] == CAP_PNG:
|
||
try:
|
||
still = png_decode(payloads[bi], False)[3]
|
||
except (ValueError, zlib.error, struct.error, IndexError):
|
||
still = None
|
||
return comp, still, sha(payload)
|
||
try:
|
||
role = wai1_role(env)
|
||
except Exception:
|
||
raise Refuse("envelope_nonconforming")
|
||
if role is not None:
|
||
raise Refuse("envelope_nonconforming")
|
||
raise Refuse("companion_mismatch")
|
||
|
||
|
||
def verify_record(i, text, held, policy):
|
||
r = read_record(text)
|
||
want = max([1] + [CLASSES.index(x["class"]) for x in r["inputs"]])
|
||
if CLASSES.index(r["class"]) != want and on("class"):
|
||
raise Refuse("class_mismatch", ["record", i])
|
||
step = envelope_step(r, held)
|
||
statuses = []
|
||
w, h = r["base"]["width"], r["base"]["height"]
|
||
for j, p in enumerate(r["pictures"]):
|
||
at = ["record", i, "picture", j]
|
||
if step is None:
|
||
statuses.append("not-recomputed")
|
||
continue
|
||
comp, still, comp_digest = step
|
||
if r["derivation"] == ST2094 and on("coverage"):
|
||
if not any(s["first"] <= p["frame"] < s["first"] + s["count"] for s in comp):
|
||
raise Refuse("frame_uncovered", at)
|
||
frame = still if (still is not None and p["frame"] == 0) else held["frames"].get((r["base"]["object"], p["frame"]))
|
||
if frame is None:
|
||
statuses.append("not-recomputed")
|
||
continue
|
||
if (len(frame) != w * h * 3 or sha(frame).hex() != p["input"]) and on("input_digest"):
|
||
raise Refuse("input_samples_mismatch", at)
|
||
key = (comp_digest, sha(frame), w, h, p["frame"], p["headroom"])
|
||
out = RENDER_MEMO.get(key)
|
||
if out is None:
|
||
try:
|
||
if r["derivation"] == GAINMAP:
|
||
out = sha(rgb16(render_gainmap(comp, frame, w, h, p["headroom"])))
|
||
else:
|
||
st = next(s for s in comp if s["first"] <= p["frame"] < s["first"] + s["count"])
|
||
out = sha(rgb16(render_st(st, frame, p["headroom"])))
|
||
except Refuse as e:
|
||
raise Refuse(e.reason, at)
|
||
if not SKIP:
|
||
RENDER_MEMO[key] = out
|
||
if out.hex() != p["output"] and on("recompute"):
|
||
raise Refuse("rendition_mismatch", at)
|
||
statuses.append("proved")
|
||
if policy == "require-proof" and on("require_proof"):
|
||
for j, s in enumerate(statuses):
|
||
if s != "proved":
|
||
raise Refuse("rendition_not_proved", ["record", i, "picture", j])
|
||
return {"record": r["_link"], "class": r["class"], "pictures": statuses}
|
||
|
||
|
||
def run_case(case, held):
|
||
out = []
|
||
for i, t in enumerate(case["records"]):
|
||
try:
|
||
out.append(verify_record(i, t, held, case["policy"]))
|
||
except Refuse as e:
|
||
return {"outcome": "refuse", "reason": e.reason, "at": e.at or ["record", i]}
|
||
except (IndexError, KeyError, ValueError, ZeroDivisionError, StopIteration) as e:
|
||
# Only a weakened verifier gets here (a check dropped in the self-test).
|
||
return {"outcome": "error", "error": "%s: %s" % (type(e).__name__, e)}
|
||
return {"outcome": "accept", "statements": out}
|
||
|
||
|
||
# ------------------------------------------------------------------ the corpus
|
||
|
||
def load(d):
|
||
with open(os.path.join(d, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
files = {}
|
||
for e in doc["envelopes"]:
|
||
with open(os.path.join(d, e["file"]), "rb") as f:
|
||
files[e["name"]] = f.read()
|
||
with open(os.path.join(d, doc["frames"]["file"]), "rb") as f:
|
||
frames = f.read()
|
||
return doc, files, frames
|
||
|
||
|
||
def held_for(case, doc, files, frames):
|
||
held = {"envelopes": {}, "frames": {}}
|
||
for n in case["held"]["envelopes"]:
|
||
b = files[n]
|
||
held["envelopes"][blake3(b"moq-jwp:object\x01" + b).hex()] = b
|
||
if case["held"]["frames"]:
|
||
fr = doc["frames"]
|
||
size = fr["width"] * fr["height"] * 3
|
||
for k in range(fr["count"]):
|
||
held["frames"][(fr["object"], k)] = frames[k * size:(k + 1) * size]
|
||
return held
|
||
|
||
|
||
def check_corpus(doc, files, frames, measure=False):
|
||
"""Every check of the corpus; a list of failures."""
|
||
fails = []
|
||
# The frames are the int_video decode whose digests the sidecar publishes.
|
||
fr = doc["frames"]
|
||
with open(os.path.join(WAI, fr["sidecar"]), encoding="utf-8") as f:
|
||
side = json.load(f)["int_video"]["frames_sha256"]
|
||
size = fr["width"] * fr["height"] * 3
|
||
got = [sha(frames[k * size:(k + 1) * size]).hex() for k in range(fr["count"])]
|
||
if len(frames) != size * fr["count"] or got != side:
|
||
fails.append("frames.rgb8: not the int_video frames the sidecar names")
|
||
with open(os.path.join(WAI, fr["source"]), "rb") as f:
|
||
wiv1 = f.read()
|
||
video = files.get("video", b"")
|
||
try:
|
||
_, entries, payloads = SV.read_wai2(video)
|
||
if payloads[0] != wiv1 or entries[0]["capability"] != "wai.video.int_motion":
|
||
fails.append("the video envelope does not carry the int_video clip")
|
||
except Exception:
|
||
fails.append("the video envelope does not read")
|
||
if blake3(b"moq-jwp:object\x01" + video).hex() != fr["object"]:
|
||
fails.append("frames: object is not the video envelope's content hash")
|
||
for e in doc["envelopes"]:
|
||
if blake3(b"moq-jwp:object\x01" + files[e["name"]]).hex() != e["content_hash"]:
|
||
fails.append("envelope %s: content hash" % e["name"])
|
||
for ec in doc["envelope_cases"]:
|
||
try:
|
||
role = envelope_role(files[ec["envelope"]])
|
||
except Exception as x:
|
||
role = "unreadable: %s" % x
|
||
if role != ec["role"]:
|
||
fails.append("%s: role %r where the corpus states %r" % (ec["name"], role, ec["role"]))
|
||
refused = set()
|
||
for case in doc["cases"]:
|
||
got = run_case(case, held_for(case, doc, files, frames))
|
||
if got != case["expect"]:
|
||
fails.append("case %r: %s where the corpus states %s" % (case["name"], json.dumps(got), json.dumps(case["expect"])))
|
||
if got["outcome"] == "refuse":
|
||
refused.add(got["reason"])
|
||
missing = [r for r in REASONS if r not in refused]
|
||
if missing:
|
||
fails.append("no case refused with %s" % missing)
|
||
unlisted = sorted(refused - set(REASONS))
|
||
if unlisted:
|
||
fails.append("refused with %s, which §9 does not list" % unlisted)
|
||
fails += check_arithmetic(doc, files)
|
||
if measure:
|
||
fails += check_measurement(doc, files, frames)
|
||
return fails
|
||
|
||
|
||
RANGES = {}
|
||
|
||
|
||
def check_arithmetic(doc, files):
|
||
"""§3 and §5.4 at the corpus's stated points: L2, E2, S, tan, and the video's slopes."""
|
||
a, fails = doc["arithmetic"], []
|
||
if not RANGES:
|
||
RANGES["l2"] = sha(b"".join(struct.pack("<q", L2(1 + (k * 2654435761) % (1 << 40))) for k in range(1 << 18))).hex()
|
||
RANGES["e2"] = sha(b"".join(struct.pack("<q", E2(f)[0]) for f in range(1 << 16))).hex()
|
||
RANGES["tan"] = sha(b"".join(struct.pack("<q", tan_q30(t)) for t in range(36001))).hex()
|
||
for k in ("l2", "e2", "tan"):
|
||
if RANGES[k] != a[k + "_sha256"]:
|
||
fails.append("%s over its stated range does not give the corpus's digest" % k)
|
||
for v, want in a["l2"]:
|
||
if L2(int(v)) != want:
|
||
fails.append("L2(%s) is %d, the corpus states %d" % (v, L2(int(v)), want))
|
||
for y, p, e in a["e2"]:
|
||
if E2(int(y)) != (int(p), int(e)):
|
||
fails.append("E2(%s) is %s, the corpus states %s" % (y, E2(int(y)), (p, e)))
|
||
if [str(x) for x in S] != a["srgb"]:
|
||
fails.append("the sRGB table is not the corpus's")
|
||
for t, want in a["tan_q30"]:
|
||
if tan_q30(t) != int(want):
|
||
fails.append("tan at %d is %d, the corpus states %s" % (t, tan_q30(t), want))
|
||
try:
|
||
_, _, payloads = SV.read_wai2(files["video"])
|
||
alts = [x for st in read_wd50(payloads[1]) for x in (st["tm"][1] if st["tm"] else [])]
|
||
except Exception as x:
|
||
return fails + ["the video's companion does not read: %s" % x]
|
||
got = [{"x": x["x"], "y": x["y"], "m": [str(m) for m in x["m"]]} for x in alts]
|
||
if got != a["slopes"]:
|
||
fails.append("the video's slopes are not the corpus's")
|
||
return fails
|
||
|
||
|
||
def check_measurement(doc, files, frames):
|
||
"""§6: the binary64 evaluation against the integer renditions, every distinct rendition of
|
||
an accepted record."""
|
||
seen, stats = set(), {}
|
||
for case in doc["cases"]:
|
||
if case["expect"]["outcome"] != "accept":
|
||
continue
|
||
held = held_for(case, doc, files, frames)
|
||
for t in case["records"]:
|
||
r = read_record(t)
|
||
step = envelope_step(r, held)
|
||
if step is None:
|
||
continue
|
||
comp, still, _ = step
|
||
w, h = r["base"]["width"], r["base"]["height"]
|
||
for p in r["pictures"]:
|
||
k = (r["base"]["object"], p["frame"], p["headroom"])
|
||
frame = still if (still is not None and p["frame"] == 0) else held["frames"].get((r["base"]["object"], p["frame"]))
|
||
if k in seen or frame is None:
|
||
continue
|
||
seen.add(k)
|
||
if r["derivation"] == GAINMAP:
|
||
name, a, b = "gainmap", render_gainmap(comp, frame, w, h, p["headroom"]), float_gainmap(comp, frame, w, h, p["headroom"])
|
||
else:
|
||
st = next(s for s in comp if s["first"] <= p["frame"] < s["first"] + s["count"])
|
||
name, a, b = "st2094-50", render_st(st, frame, p["headroom"]), float_st(st, frame, p["headroom"])
|
||
e = stats.setdefault(name, [0, 0, 0, 0])
|
||
e[0] += 1
|
||
for x, y in zip(a, b):
|
||
dd = abs(x - y)
|
||
e[1] += 1
|
||
e[2] = max(e[2], dd)
|
||
e[3] += dd
|
||
fails = []
|
||
for name, (n, samples, largest, total) in stats.items():
|
||
st = doc["measurement"].get(name)
|
||
if st is None:
|
||
fails.append("measurement: no %s" % name)
|
||
continue
|
||
if (st["renditions"], st["samples"]) != (n, samples):
|
||
fails.append("measurement %s: %d renditions, %d samples where the corpus states %d, %d" % (name, n, samples, st["renditions"], st["samples"]))
|
||
if abs(largest - st["largest"]) > 1 or abs(total / samples - float(st["mean"])) > 0.01:
|
||
fails.append("measurement %s: largest %d, mean %.4f where the corpus states %d, %s" % (name, largest, total / samples, st["largest"], st["mean"]))
|
||
# §6's table states the same figures.
|
||
six = section(TEXT, "\n## 6. ")
|
||
for name, st in doc["measurement"].items():
|
||
row = "| `wai-tonemap:%s` | %d | %d | %d | %s |" % (name, st["renditions"], st["samples"], st["largest"], st["mean"])
|
||
if row not in six:
|
||
fails.append("§6 does not state the corpus's figures for %s" % name)
|
||
return fails
|
||
|
||
|
||
# ------------------------------------------------------------------ self-tests
|
||
|
||
CHECKS = ["members", "signature", "headroom", "class", "roles", "binding", "companion_digest", "wgm1_frame",
|
||
"iso_version", "iso_length", "iso_range", "gain_space", "channels", "wd50_frame", "annexc_version",
|
||
"annexc_reserved", "annexc_range", "annexc_gain_space", "coverage", "input_digest", "map_dims",
|
||
"recompute", "require_proof"]
|
||
|
||
|
||
def mutations(doc, files, frames):
|
||
"""Copies of the corpus, each changed in one place; every one must fail."""
|
||
out = []
|
||
|
||
def case_named(d, name):
|
||
return next(c for c in d["cases"] if c["name"] == name)
|
||
# A rendition digest changed in an accepted record (and re-signed is impossible here: the
|
||
# signature then fails; the stated statements no longer match either way).
|
||
d = copy.deepcopy(doc)
|
||
c = case_named(d, "video: two metadata sets, proved at five headrooms a frame")
|
||
c["expect"]["statements"][0]["pictures"][3] = "not-recomputed"
|
||
out.append(("a stated status changed", d, files, frames))
|
||
# One byte of a frame.
|
||
fr = bytearray(frames)
|
||
fr[1000] ^= 1
|
||
out.append(("a frame's byte flipped", doc, files, bytes(fr)))
|
||
# One byte of the gain map picture inside an envelope: its content hash changes, so the
|
||
# accepted still is no longer held and require-proof refuses it.
|
||
f2 = dict(files)
|
||
b = bytearray(f2["still-luma-quarter"])
|
||
b[-30] ^= 1
|
||
f2["still-luma-quarter"] = bytes(b)
|
||
out.append(("an envelope's byte flipped", doc, f2, frames))
|
||
# A stated reason changed.
|
||
d = copy.deepcopy(doc)
|
||
c = case_named(d, "a map whose size does not divide the base's")
|
||
c["expect"]["reason"] = "rendition_mismatch"
|
||
out.append(("a stated reason changed", d, files, frames))
|
||
# A stated place changed.
|
||
d = copy.deepcopy(doc)
|
||
c = case_named(d, "a rendition one sample of which was changed")
|
||
c["expect"]["at"] = ["record", 0, "picture", 2]
|
||
out.append(("a stated place changed", d, files, frames))
|
||
# A stated role changed.
|
||
d = copy.deepcopy(doc)
|
||
next(e for e in d["envelope_cases"] if e["envelope"] == "gainmap-on-video")["role"] = None
|
||
out.append(("a stated role changed", d, files, frames))
|
||
# The policy relaxed.
|
||
d = copy.deepcopy(doc)
|
||
case_named(d, "require-proof without the frames")["policy"] = "accept-unrecomputed"
|
||
out.append(("a policy relaxed", d, files, frames))
|
||
# A stated tan value changed by one.
|
||
d = copy.deepcopy(doc)
|
||
d["arithmetic"]["tan_q30"][-1][1] = str(int(d["arithmetic"]["tan_q30"][-1][1]) + 1)
|
||
out.append(("a stated tan changed", d, files, frames))
|
||
# A measurement figure changed.
|
||
d = copy.deepcopy(doc)
|
||
d["measurement"]["gainmap"]["largest"] += 3
|
||
out.append(("a measurement figure changed", d, files, frames))
|
||
return out
|
||
|
||
|
||
def self_test(doc, files, frames):
|
||
fails = []
|
||
for name, d, f, fr in mutations(doc, files, frames):
|
||
if not check_corpus(d, f, fr, measure=name.startswith("a measurement")):
|
||
fails.append("mutation %r: the corpus still passes" % name)
|
||
for check in CHECKS:
|
||
SKIP.clear()
|
||
SKIP.add(check)
|
||
if not check_corpus(doc, files, frames):
|
||
fails.append("check %r dropped: the corpus does not catch it" % check)
|
||
SKIP.clear()
|
||
# The table of K is its rule, and a K one unit off changes a rendition.
|
||
if K != k_by_rule():
|
||
fails.append("§3.3's table of K is not its rule")
|
||
return fails
|
||
|
||
|
||
def main(d):
|
||
fails = []
|
||
if K != k_by_rule():
|
||
fails.append("§3.3's table of K is not its rule")
|
||
if L2(3) != 103872 or E2(1 << 15)[0] != K[0] or S[255] != 1 << 30 or tan_q30(27000) - (1 << 30) not in (-1, 0, 1):
|
||
fails.append("the arithmetic does not reproduce its known points")
|
||
doc, files, frames = load(d)
|
||
fails += check_corpus(doc, files, frames, measure=True)
|
||
if not fails:
|
||
fails += self_test(doc, files, frames)
|
||
for f in fails:
|
||
print("FAIL", f)
|
||
n = len(doc["cases"])
|
||
print("%d cases, %d envelope cases, %d mutations, %d checks dropped: %d failure(s)"
|
||
% (n, len(doc["envelope_cases"]), len(mutations(doc, files, frames)), len(CHECKS), len(fails)))
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "tonemap")))
|