a1ecded5a2
extensions/scene-carriage.md: the bundle, the numbers that cross the mapping, the glTF document (MPEG_media, MPEG_buffer_circular, MPEG_accessor_timed, MPEG_audio_spatial, MPEG_texture_video) and the USD layer each direction writes and reads, the refusals and the corpus. scene_verify.py implements it from the text and agrees with wai-rs on every vector; it also flips bytes of every exported document and requires each copy to be caught. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1626 lines
58 KiB
Python
1626 lines
58 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the scene-carriage corpus (extensions/scene-carriage.md §9).
|
|
|
|
Written from the extension's text: §2 (the bundle), §3 (numbers), §4 and §5 (glTF), §6 and
|
|
§7 (USD). It shares no code with wai-rs, which generates the corpus; it mints node ids with
|
|
BLAKE3 written from its specification (claim-conformance/_blake3.py) and reads §8's codes
|
|
from the extension.
|
|
|
|
For every vector it requires the outcome the corpus states: for an export, the document's
|
|
bytes or the code; for an import, the bundle's bytes, its media list and its report of
|
|
presentation not carried, or the code. Then it flips one byte of each document the corpus
|
|
exports, in several places, and requires every copy to be refused or to read to another
|
|
bundle.
|
|
|
|
python3 -I scene_verify.py <live-conformance/scene>
|
|
|
|
Exit status 0 only if every vector verifies and every mutated copy is caught.
|
|
"""
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
from fractions import Fraction
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
WAI = os.path.normpath(os.path.join(HERE, ".."))
|
|
sys.path.insert(0, os.path.join(WAI, "claim-conformance"))
|
|
from _blake3 import blake3 # noqa: E402
|
|
|
|
EXTENSION = os.path.join(WAI, "extensions", "scene-carriage.md")
|
|
PROFILE = "wai-scene/1"
|
|
URN = "urn:wai:asset:"
|
|
KIN = "wai.world.sim.kinematic0"
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, code, why=""):
|
|
super().__init__(f"{code}: {why}")
|
|
self.code = code
|
|
|
|
|
|
def codes():
|
|
text = open(EXTENSION, encoding="utf-8").read()
|
|
sec = text.split("\n## 8.")[1].split("\n## ")[0]
|
|
return [m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", sec, re.M)]
|
|
|
|
|
|
# ------------------------------------------------------------------ §3 numbers
|
|
|
|
def round_binary(q, mant_bits, emin):
|
|
"""The binary floating-point value nearest the Fraction q, ties to even, as a Fraction:
|
|
mant_bits significand bits, emin the least normal exponent (unbounded above)."""
|
|
if q == 0:
|
|
return Fraction(0)
|
|
sign = -1 if q < 0 else 1
|
|
a = abs(q)
|
|
e = a.numerator.bit_length() - a.denominator.bit_length()
|
|
if Fraction(2) ** e > a:
|
|
e -= 1
|
|
while Fraction(2) ** (e + 1) <= a:
|
|
e += 1
|
|
e = max(e, emin)
|
|
scale = Fraction(2) ** (e - (mant_bits - 1))
|
|
m = a / scale
|
|
fl = m.numerator // m.denominator
|
|
rem = m - fl
|
|
if rem > Fraction(1, 2) or (rem == Fraction(1, 2) and fl % 2 == 1):
|
|
fl += 1
|
|
return sign * fl * scale
|
|
|
|
|
|
F32_MAX = (2 - Fraction(1, 2 ** 23)) * Fraction(2) ** 127
|
|
|
|
|
|
def to_f32(q):
|
|
"""§3.1, §3.2: the binary32 nearest q, ties to even; None past the range."""
|
|
v = round_binary(Fraction(q), 24, -126)
|
|
return None if abs(v) > F32_MAX else v
|
|
|
|
|
|
def fx_frac(r):
|
|
return Fraction(r, 1 << 32)
|
|
|
|
|
|
def to_fx(v):
|
|
"""§3.1: a value as Q32.32 when exact, else None."""
|
|
if isinstance(v, float) and (v != v or v in (float("inf"), float("-inf"))):
|
|
return None
|
|
q = Fraction(v) * (1 << 32)
|
|
if q.denominator != 1 or not -(1 << 63) <= q.numerator < (1 << 63):
|
|
return None
|
|
return q.numerator
|
|
|
|
|
|
def shortest(v):
|
|
"""(digits, e) with v = d1.d2... x 10^e, the shortest digits reading back to v."""
|
|
s = repr(abs(float(v)))
|
|
if "e" in s:
|
|
m, ex = s.split("e")
|
|
ex = int(ex)
|
|
else:
|
|
m, ex = s, 0
|
|
if "." in m:
|
|
ip, fp = m.split(".")
|
|
else:
|
|
ip, fp = m, ""
|
|
digits = (ip + fp).lstrip("0")
|
|
lead = len(ip.lstrip("0")) if ip.strip("0") else -(len(fp) - len(fp.lstrip("0")))
|
|
e = ex + (lead - 1 if ip.strip("0") else lead - 1)
|
|
digits = digits.rstrip("0") or "0"
|
|
return digits, e
|
|
|
|
|
|
def json_number(v):
|
|
"""§3.2: RFC 8785's spelling of a binary64."""
|
|
v = float(v)
|
|
if v == 0:
|
|
return "0"
|
|
d, e = shortest(v)
|
|
k, n = len(d), e + 1
|
|
if k <= n <= 21:
|
|
body = d + "0" * (n - k)
|
|
elif 0 < n <= 21:
|
|
body = d[:n] + "." + d[n:]
|
|
elif -6 < n <= 0:
|
|
body = "0." + "0" * (-n) + d
|
|
else:
|
|
body = d[0] + ("." + d[1:] if k > 1 else "") + "e" + ("+" if n - 1 >= 0 else "-") + str(abs(n - 1))
|
|
return ("-" if v < 0 else "") + body
|
|
|
|
|
|
def usd_number(v):
|
|
"""§3.3: the USD Core Specification's spelling of a binary64."""
|
|
v = float(v)
|
|
if v == 0:
|
|
return "0"
|
|
d, e = shortest(v)
|
|
k = len(d)
|
|
if e < -6 or e >= 15:
|
|
body = d[0] + ("." + d[1:] if k > 1 else "") + "e" + str(e)
|
|
elif e >= k - 1:
|
|
body = d + "0" * (e - (k - 1))
|
|
elif e >= 0:
|
|
body = d[:e + 1] + "." + d[e + 1:]
|
|
else:
|
|
body = "0." + "0" * (-e - 1) + d
|
|
return ("-" if v < 0 else "") + body
|
|
|
|
|
|
# ------------------------------------------------------------------ JSON
|
|
|
|
def jquote(s):
|
|
return json.dumps(s, ensure_ascii=False)
|
|
|
|
|
|
def strict_json(b):
|
|
"""§5.1: RFC 8259 over UTF-8, no member twice, at most 64 levels; numbers as spelt."""
|
|
try:
|
|
text = b.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise Refuse("document_malformed", "not UTF-8")
|
|
|
|
def pairs(ps):
|
|
seen = set()
|
|
for k, _ in ps:
|
|
if k in seen:
|
|
raise Refuse("document_malformed", "member twice")
|
|
seen.add(k)
|
|
return Obj(ps)
|
|
|
|
def bad(c):
|
|
raise Refuse("document_malformed", c)
|
|
try:
|
|
v = json.loads(text, object_pairs_hook=pairs, parse_int=Num, parse_float=Num, parse_constant=bad)
|
|
except Refuse:
|
|
raise
|
|
except (ValueError, RecursionError) as e:
|
|
raise Refuse("document_malformed", str(e))
|
|
|
|
def walk(x, d):
|
|
if d > 64:
|
|
raise Refuse("document_malformed", "too deep")
|
|
if isinstance(x, str):
|
|
try:
|
|
x.encode("utf-8")
|
|
except UnicodeEncodeError:
|
|
raise Refuse("document_malformed", "surrogate")
|
|
elif isinstance(x, list):
|
|
for y in x:
|
|
walk(y, d + 1)
|
|
elif isinstance(x, Obj):
|
|
for k, y in x.items:
|
|
walk(k, d)
|
|
walk(y, d + 1)
|
|
walk(v, 1)
|
|
return v
|
|
|
|
|
|
class Num(str):
|
|
pass
|
|
|
|
|
|
class Obj:
|
|
def __init__(self, items):
|
|
self.items = items
|
|
|
|
def get(self, k, default=None):
|
|
for n, v in self.items:
|
|
if n == k:
|
|
return v
|
|
return default
|
|
|
|
def has(self, k):
|
|
return any(n == k for n, _ in self.items)
|
|
|
|
|
|
def uint(x):
|
|
if isinstance(x, Num) and re.fullmatch(r"0|[1-9][0-9]*", x):
|
|
return int(x)
|
|
return None
|
|
|
|
|
|
# ------------------------------------------------------------------ §2 the bundle
|
|
|
|
def sections(b, magic, code):
|
|
if len(b) < 6 or b[:4] != magic:
|
|
raise Refuse(code, "magic")
|
|
n = struct.unpack_from("<H", b, 4)[0]
|
|
base = 6 + 9 * n
|
|
if base > len(b):
|
|
raise Refuse(code, "table")
|
|
out = {}
|
|
for i in range(n):
|
|
k, off, ln = struct.unpack_from("<BII", b, 6 + 9 * i)
|
|
if base + off + ln > len(b):
|
|
raise Refuse(code, "section past the end")
|
|
if k in out:
|
|
raise Refuse(code, "kind twice")
|
|
out[k] = b[base + off:base + off + ln]
|
|
return out
|
|
|
|
|
|
def assets_of(a, code):
|
|
if len(a) < 4:
|
|
raise Refuse(code, "assets")
|
|
n = struct.unpack_from("<I", a, 0)[0]
|
|
at, out = 4, []
|
|
for _ in range(n):
|
|
if at + 37 > len(a):
|
|
raise Refuse(code, "assets")
|
|
h, role, nl = a[at:at + 32], a[at + 32], struct.unpack_from("<I", a, at + 33)[0]
|
|
at += 37
|
|
if at + nl > len(a):
|
|
raise Refuse(code, "assets")
|
|
try:
|
|
name = a[at:at + nl].decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise Refuse(code, "asset name")
|
|
at += nl
|
|
out.append((h, role, name))
|
|
if at != len(a):
|
|
raise Refuse(code, "assets")
|
|
return out
|
|
|
|
|
|
def position(sim, st):
|
|
"""§3.4."""
|
|
def take(b):
|
|
if len(b) < 24:
|
|
raise Refuse("world_malformed", "short state")
|
|
return list(struct.unpack_from("<3q", b, 0))
|
|
if sim in ("wai.world.sim.kinematic0", "wai.world.sim.ballistic0"):
|
|
return take(st)
|
|
if sim == "wai.world.sim.truss0":
|
|
if not st:
|
|
raise Refuse("world_malformed", "empty truss node")
|
|
return take(st[1:]) if st[0] == 0 else None
|
|
return None
|
|
|
|
|
|
def read_world(b):
|
|
s = sections(b, b"WWLD", "world_malformed")
|
|
if 1 not in s or 2 not in s:
|
|
raise Refuse("world_malformed", "sections")
|
|
try:
|
|
c = strict_json(s[1])
|
|
except Refuse:
|
|
raise Refuse("world_malformed", "contract")
|
|
sim = c.get("sim") if isinstance(c, Obj) else None
|
|
if not isinstance(sim, str):
|
|
raise Refuse("world_malformed", "sim")
|
|
g = s[2]
|
|
if len(g) < 4:
|
|
raise Refuse("world_malformed", "graph")
|
|
n = struct.unpack_from("<I", g, 0)[0]
|
|
at, nodes = 4, []
|
|
for _ in range(n):
|
|
if at + 20 > len(g):
|
|
raise Refuse("world_malformed", "graph")
|
|
nid, ln = g[at:at + 16], struct.unpack_from("<I", g, at + 16)[0]
|
|
at += 20
|
|
if at + ln > len(g):
|
|
raise Refuse("world_malformed", "graph")
|
|
if nodes and nodes[-1][0] >= nid:
|
|
raise Refuse("world_malformed", "order")
|
|
nodes.append((nid, g[at:at + ln]))
|
|
at += ln
|
|
if at != len(g):
|
|
raise Refuse("world_malformed", "graph")
|
|
assets = assets_of(s[3], "world_malformed") if 3 in s else []
|
|
for _, st in nodes:
|
|
position(sim, st)
|
|
return {"sim": sim, "nodes": nodes, "assets": assets}
|
|
|
|
|
|
def read_audio(b):
|
|
s = sections(b, b"WSCN", "audio_malformed")
|
|
if 1 not in s or 2 not in s:
|
|
raise Refuse("audio_malformed", "sections")
|
|
try:
|
|
c = strict_json(s[1])
|
|
except Refuse:
|
|
raise Refuse("audio_malformed", "contract")
|
|
rate = uint(c.get("sample_rate")) if isinstance(c, Obj) else None
|
|
dur = uint(c.get("duration")) if isinstance(c, Obj) else None
|
|
if rate is None or not 1 <= rate <= 0xFFFFFFFF or dur is None:
|
|
raise Refuse("audio_malformed", "contract")
|
|
t = s[2]
|
|
|
|
def rd(at, n):
|
|
if at + n > len(t):
|
|
raise Refuse("audio_malformed", "objects")
|
|
return t[at:at + n]
|
|
n = struct.unpack("<I", rd(0, 4))[0]
|
|
at, objs = 4, []
|
|
for _ in range(n):
|
|
src = rd(at, 32)
|
|
nk = struct.unpack("<H", rd(at + 40, 2))[0]
|
|
at += 42
|
|
keys = []
|
|
for _ in range(nk):
|
|
k = rd(at, 32)
|
|
keys.append((struct.unpack_from("<Q", k, 0)[0], list(struct.unpack_from("<3q", k, 8))))
|
|
at += 32
|
|
objs.append((bytes(src), keys))
|
|
if at != len(t):
|
|
raise Refuse("audio_malformed", "objects")
|
|
if 3 in s:
|
|
assets_of(s[3], "audio_malformed")
|
|
if 4 in s and len(s[4]) != 104:
|
|
raise Refuse("audio_malformed", "seal")
|
|
return {"rate": rate, "duration": dur, "objects": objs}
|
|
|
|
|
|
def cap_ok(c):
|
|
if not c.startswith("wai."):
|
|
return False
|
|
rest = c[4:]
|
|
if not rest.startswith(("video.", "audio.", "image.")):
|
|
return False
|
|
segs = rest.split(".")
|
|
return bool(re.fullmatch(r"[a-z0-9]+", segs[0])) and len(segs) > 1 and all(re.fullmatch(r"[a-z0-9_]+", x) for x in segs[1:])
|
|
|
|
|
|
def picture(c):
|
|
return c.startswith(("wai.video.", "wai.image."))
|
|
|
|
|
|
def check_media(m, code="media_malformed"):
|
|
if not cap_ok(m["capability"]):
|
|
raise Refuse(code, "capability")
|
|
if picture(m["capability"]):
|
|
if m.get("width") is None or not (1 <= m["width"] <= 65535 and 1 <= m["height"] <= 65535):
|
|
raise Refuse(code, "size")
|
|
elif m.get("width") is not None:
|
|
raise Refuse(code, "size")
|
|
|
|
|
|
def media_text(media):
|
|
out = []
|
|
for m in media:
|
|
s = f'{{"name":{jquote(m["name"])},"asset":"{m["asset"].hex()}","capability":{jquote(m["capability"])}'
|
|
if m.get("width") is not None:
|
|
s += f',"width":{m["width"]},"height":{m["height"]}'
|
|
out.append(s + "}")
|
|
return "[" + ",".join(out) + "]"
|
|
|
|
|
|
def media_from(v):
|
|
if not isinstance(v, list):
|
|
return None
|
|
out = []
|
|
for it in v:
|
|
if not isinstance(it, Obj):
|
|
return None
|
|
keys = [k for k, _ in it.items]
|
|
if keys not in (["name", "asset", "capability"], ["name", "asset", "capability", "width", "height"]):
|
|
return None
|
|
name, asset, cap = it.get("name"), it.get("asset"), it.get("capability")
|
|
if not (isinstance(name, str) and isinstance(asset, str) and isinstance(cap, str) and re.fullmatch(r"[0-9a-f]{64}", asset)):
|
|
return None
|
|
m = {"name": name, "asset": bytes.fromhex(asset), "capability": cap, "width": None, "height": None}
|
|
if len(keys) == 5:
|
|
w, h = uint(it.get("width")), uint(it.get("height"))
|
|
if w is None or h is None or w >= 1 << 32 or h >= 1 << 32:
|
|
return None
|
|
m["width"], m["height"] = w, h
|
|
out.append(m)
|
|
return out
|
|
|
|
|
|
def check_bundle(b):
|
|
w = read_world(b["world"]) if b["world"] is not None else None
|
|
a = read_audio(b["audio"]) if b["audio"] is not None else None
|
|
for m in b["media"]:
|
|
check_media(m)
|
|
return w, a
|
|
|
|
|
|
def mime(cap):
|
|
return "application/wai" if cap is None else f'application/wai; codecs="wai1.{cap[4:]}"'
|
|
|
|
|
|
def mime_cap(m):
|
|
mt = re.fullmatch(r'application/wai; codecs="wai1\.([^"]*)"', m or "")
|
|
if not mt:
|
|
return None
|
|
c = "wai." + mt.group(1)
|
|
return c if cap_ok(c) else None
|
|
|
|
|
|
# ------------------------------------------------------------------ §4 writing glTF
|
|
|
|
def jw(v):
|
|
"""Write the JSON of §4.1: dicts in insertion order, F(...) numbers per §3.2."""
|
|
if isinstance(v, dict):
|
|
return "{" + ",".join(f"{jquote(k)}:{jw(x)}" for k, x in v.items()) + "}"
|
|
if isinstance(v, list):
|
|
return "[" + ",".join(jw(x) for x in v) + "]"
|
|
if isinstance(v, bool):
|
|
return "true" if v else "false"
|
|
if isinstance(v, int):
|
|
return str(v)
|
|
if isinstance(v, F):
|
|
return json_number(v.v)
|
|
return jquote(v)
|
|
|
|
|
|
class F:
|
|
def __init__(self, v):
|
|
self.v = v
|
|
|
|
|
|
def f32(q):
|
|
return float(to_f32(q))
|
|
|
|
|
|
def vec3(p):
|
|
return [F(f32(fx_frac(x))) for x in p]
|
|
|
|
|
|
def export_gltf(b):
|
|
w, a = check_bundle(b)
|
|
nodes, used = [], []
|
|
if w:
|
|
for nid, st in w["nodes"]:
|
|
n = {"name": "node-" + nid.hex()}
|
|
p = position(w["sim"], st)
|
|
if p is not None:
|
|
n["translation"] = vec3(p)
|
|
nodes.append(n)
|
|
items = []
|
|
first = {}
|
|
for m in b["media"]:
|
|
first.setdefault(m["asset"], m)
|
|
if a:
|
|
for src, _ in a["objects"]:
|
|
if all(i[1] != src for i in items):
|
|
m = first.get(src)
|
|
items.append((m["name"] if m else "source-" + src.hex()[:16], src, m["capability"] if m else None, None))
|
|
nsrc = len(items)
|
|
for m in b["media"]:
|
|
if all(i[1] != m["asset"] for i in items):
|
|
items.append((m["name"], m["asset"], m["capability"], (m["width"], m["height"]) if m["width"] is not None else None))
|
|
anim = bytearray()
|
|
views, accs, samplers, channels, cams, plan, objnodes = [], [], [], [], [], [], []
|
|
if a:
|
|
cams.append({"type": "perspective", "perspective": {"yfov": F(1.0), "znear": F(0.25)}})
|
|
nodes.append({"name": "listener", "camera": 0, "extensions": {"MPEG_audio_spatial": {"listener": {"id": 0}}}})
|
|
used.append("MPEG_audio_spatial")
|
|
for i, (src, keys) in enumerate(a["objects"]):
|
|
if any(keys[j + 1][0] <= keys[j][0] for j in range(len(keys) - 1)):
|
|
raise Refuse("not_presentable", "order")
|
|
times = [to_f32(Fraction(s, a["rate"])) for s, _ in keys]
|
|
if any(times[j + 1] <= times[j] for j in range(len(times) - 1)):
|
|
raise Refuse("not_presentable", "times")
|
|
objnodes.append(len(nodes))
|
|
if len(keys) >= 2:
|
|
plan.append((len(nodes), times, [p for _, p in keys]))
|
|
nodes.append(None)
|
|
buffers = [None] if plan else []
|
|
for node, times, poss in plan:
|
|
ti = len(anim)
|
|
for t in times:
|
|
anim += struct.pack("<f", float(t))
|
|
to = len(anim)
|
|
for p in poss:
|
|
for x in p:
|
|
anim += struct.pack("<f", f32(fx_frac(x)))
|
|
vi = len(views)
|
|
views.append({"buffer": 0, "byteOffset": ti, "byteLength": 4 * len(times)})
|
|
views.append({"buffer": 0, "byteOffset": to, "byteLength": 12 * len(times)})
|
|
ai = len(accs)
|
|
accs.append({"bufferView": vi, "componentType": 5126, "count": len(times), "type": "SCALAR", "min": [F(float(times[0]))], "max": [F(float(times[-1]))]})
|
|
accs.append({"bufferView": vi + 1, "componentType": 5126, "count": len(times), "type": "VEC3"})
|
|
si = len(samplers)
|
|
samplers.append({"input": ai, "interpolation": "LINEAR", "output": ai + 1})
|
|
channels.append({"sampler": si, "target": {"node": node, "path": "translation"}})
|
|
if plan:
|
|
buffers[0] = {"byteLength": len(anim), "uri": "data:application/octet-stream;base64," + base64.b64encode(bytes(anim)).decode()}
|
|
item_acc = [None] * len(items)
|
|
textures = []
|
|
for k, (_, _, _, size) in enumerate(items):
|
|
if k >= nsrc and size is None:
|
|
continue
|
|
if size:
|
|
cnt = size[0] * size[1]
|
|
nb, acc = 3 * cnt, {"componentType": 5121, "normalized": True, "count": cnt, "type": "VEC3"}
|
|
else:
|
|
nb, acc = 4096, {"componentType": 5126, "count": 1024, "type": "SCALAR"}
|
|
bi = len(buffers)
|
|
buffers.append({"byteLength": nb, "extensions": {"MPEG_buffer_circular": {"media": k}}})
|
|
vi = len(views)
|
|
views.append({"buffer": bi, "byteLength": nb})
|
|
full = {"bufferView": vi}
|
|
full.update(acc)
|
|
full["extensions"] = {"MPEG_accessor_timed": {"immutable": True}}
|
|
item_acc[k] = len(accs)
|
|
accs.append(full)
|
|
if size:
|
|
textures.append({"extensions": {"MPEG_texture_video": {"accessor": item_acc[k], "width": size[0], "height": size[1], "format": "RGB"}}})
|
|
if a:
|
|
for i, (src, keys) in enumerate(a["objects"]):
|
|
k = next(j for j, it in enumerate(items) if it[1] == src)
|
|
n = {"name": f"object-{i}"}
|
|
if keys:
|
|
n["translation"] = vec3(keys[0][1])
|
|
n["extensions"] = {"MPEG_audio_spatial": {"sources": [{"id": i, "type": "Object", "attenuation": "custom", "accessors": [item_acc[k]]}]}}
|
|
nodes[objnodes[i]] = n
|
|
if items:
|
|
used.append("MPEG_media")
|
|
if any("extensions" in x for x in buffers):
|
|
used += ["MPEG_buffer_circular", "MPEG_accessor_timed"]
|
|
if textures:
|
|
used.append("MPEG_texture_video")
|
|
used.sort()
|
|
media = [{"name": nm, "autoplay": True, "alternatives": [{"uri": URN + h.hex(), "mimeType": mime(c)}]} for nm, h, c, _ in items]
|
|
doc = {"asset": {"version": "2.0", "generator": PROFILE}}
|
|
if used:
|
|
doc["extensionsUsed"] = used
|
|
if media:
|
|
doc["extensions"] = {"MPEG_media": {"media": media}}
|
|
doc["scene"] = 0
|
|
doc["scenes"] = [{"nodes": list(range(len(nodes)))} if nodes else {}]
|
|
for k, v in (("nodes", nodes), ("cameras", cams), ("buffers", buffers), ("bufferViews", views), ("accessors", accs)):
|
|
if v:
|
|
doc[k] = v
|
|
if channels:
|
|
doc["animations"] = [{"name": "audio", "channels": channels, "samplers": samplers}]
|
|
if textures:
|
|
doc["textures"] = textures
|
|
wai = {"profile": PROFILE}
|
|
if b["world"] is not None:
|
|
wai["world"] = b["world"].hex()
|
|
if b["audio"] is not None:
|
|
wai["audio"] = b["audio"].hex()
|
|
if b["media"]:
|
|
wai["media"] = media_text(b["media"])
|
|
doc["extras"] = {"wai": wai}
|
|
return (jw(doc) + "\n").encode()
|
|
|
|
|
|
# ------------------------------------------------------------------ §5 reading glTF
|
|
|
|
def DM(why):
|
|
return Refuse("document_malformed", why)
|
|
|
|
|
|
def arr(o, k):
|
|
v = o.get(k) if isinstance(o, Obj) else None
|
|
return v if isinstance(v, list) else []
|
|
|
|
|
|
def index(v, n, what):
|
|
i = uint(v)
|
|
if i is None or i >= n:
|
|
raise DM(what)
|
|
return i
|
|
|
|
|
|
def f32_of(v):
|
|
if not isinstance(v, Num):
|
|
return None
|
|
return to_f32(Fraction(v))
|
|
|
|
|
|
def mint(actor, i):
|
|
return blake3(b"wai:world-node\x01" + actor + struct.pack("<Q", i))[:16]
|
|
|
|
|
|
def kin_state(p):
|
|
return struct.pack("<3q4q", *p, 0, 0, 0, 0)
|
|
|
|
|
|
def container(magic, secs):
|
|
out = bytearray(magic + struct.pack("<H", len(secs)))
|
|
off = 0
|
|
for k, body in secs:
|
|
out += struct.pack("<BII", k, off, len(body))
|
|
off += -(-len(body) // 8) * 8
|
|
for _, body in secs:
|
|
out += body + b"\0" * ((-len(body)) % 8)
|
|
return bytes(out)
|
|
|
|
|
|
def asset_table(assets):
|
|
o = bytearray(struct.pack("<I", len(assets)))
|
|
for h, r, n in assets:
|
|
nb = n.encode()
|
|
o += h + bytes([r]) + struct.pack("<I", len(nb)) + nb
|
|
return bytes(o)
|
|
|
|
|
|
def world_bytes(contract, nodes, assets):
|
|
nodes = sorted(nodes)
|
|
g = bytearray(struct.pack("<I", len(nodes)))
|
|
for nid, st in nodes:
|
|
g += nid + struct.pack("<I", len(st)) + st
|
|
secs = [(1, contract), (2, bytes(g))]
|
|
if assets:
|
|
secs.append((3, asset_table(assets)))
|
|
secs.append((4, b""))
|
|
return container(b"WWLD", secs)
|
|
|
|
|
|
def audio_bytes(contract, objs, assets):
|
|
t = bytearray(struct.pack("<I", len(objs)))
|
|
for src, keys in objs:
|
|
t += src + struct.pack("<qH", 1 << 32, len(keys))
|
|
for s, p in keys:
|
|
t += struct.pack("<Q3q", s, *p)
|
|
secs = [(1, contract), (2, bytes(t))]
|
|
if assets:
|
|
secs.append((3, asset_table(assets)))
|
|
return container(b"WSCN", secs)
|
|
|
|
|
|
def contract_json(c, code="contract_absent"):
|
|
if c is None:
|
|
raise Refuse(code, "no contract")
|
|
try:
|
|
return strict_json(c)
|
|
except Refuse:
|
|
raise Refuse(code, "not JSON")
|
|
|
|
|
|
def audio_rate(params):
|
|
j = contract_json(params.get("audio_contract"))
|
|
r = uint(j.get("sample_rate")) if isinstance(j, Obj) else None
|
|
if r is None or not 1 <= r <= 0xFFFFFFFF:
|
|
raise Refuse("contract_absent", "sample_rate")
|
|
return r
|
|
|
|
|
|
def unhex(s):
|
|
if not isinstance(s, str) or len(s) % 2 or not re.fullmatch(r"[0-9a-f]*", s):
|
|
return None
|
|
return bytes.fromhex(s)
|
|
|
|
|
|
def import_gltf(b, params):
|
|
doc = strict_json(b)
|
|
asset = doc.get("asset") if isinstance(doc, Obj) else None
|
|
if not isinstance(doc, Obj) or not isinstance(asset, Obj) or asset.get("version") != "2.0":
|
|
raise DM("not glTF 2.0")
|
|
extras = doc.get("extras")
|
|
if isinstance(extras, Obj) and extras.has("wai"):
|
|
bundle = carried_gltf(extras.get("wai"))
|
|
try:
|
|
again = export_gltf(bundle)
|
|
except Refuse as r:
|
|
raise Refuse("carried_malformed", r.code)
|
|
if again != b:
|
|
raise Refuse("presentation_mismatch")
|
|
return bundle, []
|
|
return foreign_gltf(doc, params)
|
|
|
|
|
|
def carried_gltf(w):
|
|
if not isinstance(w, Obj):
|
|
raise Refuse("carried_malformed", "not an object")
|
|
b = {"world": None, "audio": None, "media": []}
|
|
for k, v in w.items:
|
|
if k == "profile" and v == PROFILE:
|
|
continue
|
|
if k in ("world", "audio"):
|
|
x = unhex(v)
|
|
if x is None:
|
|
raise Refuse("carried_malformed", k)
|
|
b[k] = x
|
|
elif k == "media":
|
|
if not isinstance(v, str):
|
|
raise Refuse("carried_malformed", "media")
|
|
try:
|
|
j = strict_json(v.encode())
|
|
except Refuse:
|
|
raise Refuse("carried_malformed", "media")
|
|
m = media_from(j)
|
|
if m is None:
|
|
raise Refuse("carried_malformed", "media")
|
|
b["media"] = m
|
|
else:
|
|
raise Refuse("carried_malformed", k)
|
|
if w.get("profile") != PROFILE:
|
|
raise Refuse("carried_malformed", "profile")
|
|
return b
|
|
|
|
|
|
def foreign_gltf(doc, params):
|
|
nodes, scenes = arr(doc, "nodes"), arr(doc, "scenes")
|
|
# Step 1
|
|
roots = []
|
|
if scenes:
|
|
s = index(doc.get("scene"), len(scenes), "scene") if doc.has("scene") else 0
|
|
roots = [index(n, len(nodes), "node") for n in arr(scenes[s], "nodes")]
|
|
order, seen, parent = [], set(), {}
|
|
stack = [(r, None) for r in reversed(roots)]
|
|
while stack:
|
|
n, par = stack.pop()
|
|
if n in seen:
|
|
raise DM("twice")
|
|
seen.add(n)
|
|
parent[n] = par
|
|
order.append(n)
|
|
ch = [index(c, len(nodes), "child") for c in arr(nodes[n], "children")]
|
|
for c in reversed(ch):
|
|
stack.append((c, n))
|
|
# Step 2
|
|
pos = {}
|
|
for n in order:
|
|
nd = nodes[n]
|
|
if not isinstance(nd, Obj):
|
|
raise DM("node")
|
|
if nd.has("matrix"):
|
|
raise Refuse("transform_not_carried", "matrix")
|
|
|
|
def floats(k, ln):
|
|
if not nd.has(k):
|
|
return None
|
|
v = nd.get(k)
|
|
if not isinstance(v, list) or len(v) != ln:
|
|
raise DM(k)
|
|
out = [f32_of(x) for x in v]
|
|
if any(x is None for x in out):
|
|
raise DM(k)
|
|
return out
|
|
r = floats("rotation", 4)
|
|
if r is not None and r != [0, 0, 0, 1]:
|
|
raise Refuse("transform_not_carried", "rotation")
|
|
sc = floats("scale", 3)
|
|
if sc is not None and sc != [1, 1, 1]:
|
|
raise Refuse("transform_not_carried", "scale")
|
|
t = floats("translation", 3) or [0, 0, 0]
|
|
tf = [to_fx(x) for x in t]
|
|
if any(x is None for x in tf):
|
|
raise Refuse("position_not_exact", "translation")
|
|
base = pos[parent[n]] if parent[n] is not None else [0, 0, 0]
|
|
p = [base[i] + tf[i] for i in range(3)]
|
|
if any(not -(1 << 63) <= x < (1 << 63) for x in p):
|
|
raise Refuse("position_not_exact", "range")
|
|
pos[n] = p
|
|
# Step 3
|
|
def spatial(n):
|
|
e = nodes[n].get("extensions")
|
|
return e.get("MPEG_audio_spatial") if isinstance(e, Obj) else None
|
|
audio_nodes, world_nodes, dropped = [], [], []
|
|
for n in order:
|
|
sp = spatial(n)
|
|
if isinstance(sp, Obj) and sp.has("sources"):
|
|
audio_nodes.append(n)
|
|
elif nodes[n].has("camera") or sp is not None:
|
|
continue
|
|
else:
|
|
if nodes[n].has("mesh"):
|
|
dropped.append(f"node {n}: mesh")
|
|
world_nodes.append(pos[n])
|
|
# Step 4
|
|
anims = arr(doc, "animations")
|
|
anim_of = {}
|
|
for ai, an in enumerate(anims):
|
|
smp = arr(an, "samplers")
|
|
for c in arr(an, "channels"):
|
|
si = index(c.get("sampler") if isinstance(c, Obj) else None, len(smp), "sampler")
|
|
tg = c.get("target")
|
|
if not isinstance(tg, Obj):
|
|
raise DM("target")
|
|
if not tg.has("node"):
|
|
continue
|
|
node = index(tg.get("node"), len(nodes), "node")
|
|
path = tg.get("path") if isinstance(tg.get("path"), str) else ""
|
|
it = smp[si].get("interpolation") if isinstance(smp[si], Obj) else None
|
|
interp = it if isinstance(it, str) else "LINEAR"
|
|
if node not in audio_nodes or path != "translation" or interp != "LINEAR" or node in anim_of:
|
|
raise Refuse("animation_not_carried", f"{node} {path} {interp}")
|
|
anim_of[node] = (ai, si)
|
|
# Step 5
|
|
ext = doc.get("extensions")
|
|
mm = ext.get("MPEG_media") if isinstance(ext, Obj) else None
|
|
items = mm.get("media") if isinstance(mm, Obj) else None
|
|
items = items if isinstance(items, list) else []
|
|
accs, views, bufs = arr(doc, "accessors"), arr(doc, "bufferViews"), arr(doc, "buffers")
|
|
|
|
def media_of(a):
|
|
v = index(accs[a].get("bufferView"), len(views), "view")
|
|
bb = index(views[v].get("buffer"), len(bufs), "buffer")
|
|
e = bufs[bb].get("extensions")
|
|
cb = e.get("MPEG_buffer_circular") if isinstance(e, Obj) else None
|
|
return index(cb.get("media") if isinstance(cb, Obj) else None, len(items), "media")
|
|
|
|
def wai_alt(k):
|
|
for alt in arr(items[k], "alternatives"):
|
|
if not isinstance(alt, Obj):
|
|
continue
|
|
u = alt.get("uri")
|
|
if isinstance(u, str) and u.startswith(URN) and re.fullmatch(r"[0-9a-f]{64}", u[len(URN):]):
|
|
mt = alt.get("mimeType")
|
|
return bytes.fromhex(u[len(URN):]), mime_cap(mt) if isinstance(mt, str) else None
|
|
return None
|
|
|
|
def floats_of(a, ty):
|
|
acc = accs[a]
|
|
if uint(acc.get("componentType")) != 5126 or acc.get("type") != ty or acc.has("sparse"):
|
|
raise DM("accessor type")
|
|
cnt = uint(acc.get("count"))
|
|
if cnt is None:
|
|
raise DM("count")
|
|
comps = 3 if ty == "VEC3" else 1
|
|
v = index(acc.get("bufferView"), len(views), "view")
|
|
view = views[v]
|
|
if view.has("byteStride"):
|
|
raise DM("stride")
|
|
bb = index(view.get("buffer"), len(bufs), "buffer")
|
|
uri = bufs[bb].get("uri")
|
|
pre = "data:application/octet-stream;base64,"
|
|
if not isinstance(uri, str) or not uri.startswith(pre):
|
|
raise DM("buffer")
|
|
try:
|
|
data = base64.b64decode(uri[len(pre):], validate=True)
|
|
except (binascii.Error, ValueError):
|
|
raise DM("base64")
|
|
if base64.b64encode(data).decode() != uri[len(pre):]:
|
|
raise DM("base64")
|
|
|
|
def opt(o, k):
|
|
if not o.has(k):
|
|
return 0
|
|
x = uint(o.get(k))
|
|
if x is None:
|
|
raise DM(k)
|
|
return x
|
|
off = opt(view, "byteOffset") + opt(acc, "byteOffset")
|
|
ln = uint(view.get("byteLength"))
|
|
if ln is None:
|
|
raise DM("byteLength")
|
|
need = cnt * 4 * comps
|
|
vstart = uint(view.get("byteOffset")) or 0
|
|
if off + need > len(data) or off + need > vstart + ln:
|
|
raise DM("past the view")
|
|
return [struct.unpack_from("<f", data, off + 4 * i)[0] for i in range(cnt * comps)]
|
|
objects, src_items, src_names = [], [], []
|
|
for n in audio_nodes:
|
|
for s in arr(spatial(n), "sources"):
|
|
if not isinstance(s, Obj) or s.get("type") != "Object":
|
|
raise Refuse("audio_not_object")
|
|
if s.has("pregain") and f32_of(s.get("pregain")) != 0:
|
|
raise Refuse("gain_not_exact")
|
|
a0s = arr(s, "accessors")
|
|
a0 = index(a0s[0] if a0s else None, len(accs), "accessor")
|
|
k = media_of(a0)
|
|
alt = wai_alt(k)
|
|
if alt is None:
|
|
raise Refuse("media_not_wai")
|
|
h = alt[0]
|
|
if k not in src_items:
|
|
src_items.append(k)
|
|
if all(x[0] != h for x in src_names):
|
|
nm = items[k].get("name")
|
|
src_names.append((h, nm if isinstance(nm, str) else ""))
|
|
if n not in anim_of:
|
|
keys = [(0, pos[n])]
|
|
else:
|
|
rate = audio_rate(params)
|
|
base = pos[parent[n]] if parent[n] is not None else [0, 0, 0]
|
|
ai, si = anim_of[n]
|
|
smp = arr(anims[ai], "samplers")[si]
|
|
ti = index(smp.get("input"), len(accs), "input")
|
|
to = index(smp.get("output"), len(accs), "output")
|
|
times, vals = floats_of(ti, "SCALAR"), floats_of(to, "VEC3")
|
|
if len(vals) != 3 * len(times):
|
|
raise DM("counts")
|
|
keys = []
|
|
for j, t in enumerate(times):
|
|
if t != t or t in (float("inf"), float("-inf")) or t < 0:
|
|
raise Refuse("time_not_exact")
|
|
q = Fraction(t) * rate
|
|
if q.denominator != 1 or q >= 1 << 64:
|
|
raise Refuse("time_not_exact")
|
|
p = []
|
|
for c in range(3):
|
|
x = to_fx(vals[3 * j + c])
|
|
if x is None:
|
|
raise Refuse("position_not_exact")
|
|
p.append(base[c] + x)
|
|
if any(not -(1 << 63) <= x < (1 << 63) for x in p):
|
|
raise Refuse("position_not_exact")
|
|
keys.append((q.numerator, p))
|
|
objects.append((h, keys))
|
|
# Step 6
|
|
texs = arr(doc, "textures")
|
|
media = []
|
|
for k in range(len(items)):
|
|
if k in src_items:
|
|
continue
|
|
alt = wai_alt(k)
|
|
if alt is None or alt[1] is None:
|
|
raise Refuse("media_not_wai")
|
|
nm = items[k].get("name") if isinstance(items[k], Obj) else None
|
|
m = {"name": nm if isinstance(nm, str) else "", "asset": alt[0], "capability": alt[1], "width": None, "height": None}
|
|
if picture(alt[1]):
|
|
for t in texs:
|
|
e = t.get("extensions") if isinstance(t, Obj) else None
|
|
tv = e.get("MPEG_texture_video") if isinstance(e, Obj) else None
|
|
if not isinstance(tv, Obj):
|
|
continue
|
|
a = index(tv.get("accessor"), len(accs), "texture accessor")
|
|
if media_of(a) == k:
|
|
w, h = uint(tv.get("width")), uint(tv.get("height"))
|
|
if w is None or h is None:
|
|
raise DM("texture size")
|
|
m["width"], m["height"] = (w if w < 1 << 32 else 0), (h if h < 1 << 32 else 0)
|
|
break
|
|
check_media(m, "media_not_wai")
|
|
media.append(m)
|
|
# Step 7
|
|
bundle = {"world": None, "audio": None, "media": media}
|
|
if world_nodes:
|
|
j = contract_json(params.get("world_contract"))
|
|
if not isinstance(j, Obj) or j.get("sim") != KIN:
|
|
raise Refuse("sim_not_importable")
|
|
ns = [(mint(params["actor"], i + 1), kin_state(p)) for i, p in enumerate(world_nodes)]
|
|
bundle["world"] = world_bytes(params["world_contract"], ns, [])
|
|
if objects:
|
|
audio_rate(params)
|
|
bundle["audio"] = audio_bytes(params["audio_contract"], objects, [(h, 0, n) for h, n in src_names])
|
|
return bundle, dropped
|
|
|
|
|
|
# ------------------------------------------------------------------ §6 writing USD
|
|
|
|
def usd_char_ok(c):
|
|
u = ord(c)
|
|
return (u <= 0x7F or 0xA0 <= u <= 0xD7FF or 0xF900 <= u <= 0xFDCF or 0xFDF0 <= u <= 0xFFEF
|
|
or (0x10000 <= u <= 0xEFFFD and (u & 0xFFFE) != 0xFFFE))
|
|
|
|
|
|
def usd_string(s):
|
|
o = '"'
|
|
for c in s:
|
|
if not usd_char_ok(c):
|
|
raise Refuse("not_presentable", "character")
|
|
o += {'"': '\\"', "\\": "\\\\", "\n": "\\n", "\r": "\\r", "\t": "\\t"}.get(c) or (f"\\x{ord(c):02X}" if ord(c) < 0x20 or ord(c) == 0x7F else c)
|
|
return o + '"'
|
|
|
|
|
|
def tuple3(p):
|
|
return "(" + ", ".join(usd_number(float(fx_frac(x))) for x in p) + ")"
|
|
|
|
|
|
def export_usda(b):
|
|
w, a = check_bundle(b)
|
|
L = ["#usda 1.0", "(", " customLayerData = {"]
|
|
if b["audio"] is not None:
|
|
L.append(f' string "wai:audio" = "{b["audio"].hex()}"')
|
|
if b["media"]:
|
|
L.append(f' string "wai:media" = {usd_string(media_text(b["media"]))}')
|
|
L.append(f' string "wai:profile" = "{PROFILE}"')
|
|
if b["world"] is not None:
|
|
L.append(f' string "wai:world" = "{b["world"].hex()}"')
|
|
L += [" }", ' defaultPrim = "World"']
|
|
if a:
|
|
if a["duration"] > 1 << 53 or a["rate"] > 1 << 53:
|
|
raise Refuse("not_presentable", "duration")
|
|
L += [f" endTimeCode = {a['duration']}", " startTimeCode = 0", f" timeCodesPerSecond = {a['rate']}"]
|
|
L += [")", "", 'def Xform "World"', "{"]
|
|
blocks = []
|
|
if w:
|
|
for nid, st in w["nodes"]:
|
|
p = position(w["sim"], st)
|
|
blk = [f' def Xform "node_{nid.hex()}"', " {"]
|
|
if p is not None:
|
|
blk += [f" double3 xformOp:translate = {tuple3(p)}", ' uniform token[] xformOpOrder = ["xformOp:translate"]']
|
|
blocks.append(blk + [" }"])
|
|
if w["assets"]:
|
|
blk = [' def Scope "Assets"', " {"]
|
|
for i, (h, role, name) in enumerate(w["assets"]):
|
|
if i:
|
|
blk.append("")
|
|
blk += [f' def "asset_{i}" (', f" prepend references = @{URN}{h.hex()}@", " )", " {",
|
|
f" custom string wai:name = {usd_string(name)}", f" custom int wai:role = {role}", " }"]
|
|
blocks.append(blk + [" }"])
|
|
if a and a["objects"]:
|
|
blk = [' def Scope "Audio"', " {"]
|
|
for i, (src, keys) in enumerate(a["objects"]):
|
|
if i:
|
|
blk.append("")
|
|
if any(keys[j + 1][0] <= keys[j][0] for j in range(len(keys) - 1)):
|
|
raise Refuse("not_presentable", "order")
|
|
if any(s > 1 << 53 for s, _ in keys):
|
|
raise Refuse("not_presentable", "time code")
|
|
blk += [f' def Xform "object_{i}"', " {", f' custom string wai:source = "{src.hex()}"']
|
|
if keys:
|
|
blk.append(f" double3 xformOp:translate = {tuple3(keys[0][1])}")
|
|
if len(keys) >= 2:
|
|
blk.append(" double3 xformOp:translate.timeSamples = {")
|
|
blk += [f" {s}: {tuple3(p)}," for s, p in keys]
|
|
blk.append(" }")
|
|
blk.append(' uniform token[] xformOpOrder = ["xformOp:translate"]')
|
|
blk.append(" }")
|
|
blocks.append(blk + [" }"])
|
|
if b["media"]:
|
|
blk = [' def Scope "Media"', " {"]
|
|
for i, m in enumerate(b["media"]):
|
|
if i:
|
|
blk.append("")
|
|
blk += [f' def "media_{i}" (', f" prepend references = @{URN}{m['asset'].hex()}@", " )", " {",
|
|
f" custom string wai:capability = {usd_string(m['capability'])}"]
|
|
if m["width"] is not None:
|
|
blk.append(f" custom int wai:height = {m['height']}")
|
|
blk.append(f" custom string wai:name = {usd_string(m['name'])}")
|
|
if m["width"] is not None:
|
|
blk.append(f" custom int wai:width = {m['width']}")
|
|
blk.append(" }")
|
|
blocks.append(blk + [" }"])
|
|
for i, blk in enumerate(blocks):
|
|
if i:
|
|
L.append("")
|
|
L += blk
|
|
L.append("}")
|
|
return ("\n".join(L) + "\n").encode()
|
|
|
|
|
|
# ------------------------------------------------------------------ §7 reading USD
|
|
|
|
def NR(why):
|
|
return Refuse("usd_not_read", why)
|
|
|
|
|
|
def lex(t):
|
|
out, i, n = [], 0, len(t)
|
|
while i < n:
|
|
c = t[i]
|
|
if c in " \t\r\n":
|
|
i += 1
|
|
elif c == "#" or t.startswith("//", i):
|
|
while i < n and t[i] not in "\r\n":
|
|
i += 1
|
|
elif t.startswith("/*", i):
|
|
j = t.find("*/", i + 2)
|
|
if j < 0:
|
|
raise DM("comment")
|
|
i = j + 2
|
|
elif c == "-" or c.isdigit() and c.isascii() or (c == "." and i + 1 < n and t[i + 1].isdigit() and t[i + 1].isascii()):
|
|
s0 = i
|
|
if c == "-":
|
|
i += 1
|
|
if t.startswith("inf", i):
|
|
i += 3
|
|
else:
|
|
while i < n and ((t[i].isascii() and t[i].isdigit()) or t[i] in ".eE" or (t[i] in "+-" and t[i - 1] in "eE")):
|
|
i += 1
|
|
out.append(("num", t[s0:i]))
|
|
elif c in "()[]{}=:,;.":
|
|
out.append(("p", c))
|
|
i += 1
|
|
elif c in "\"'":
|
|
if t.startswith(c * 3, i):
|
|
raise NR("triple quotes")
|
|
i += 1
|
|
s = ""
|
|
while True:
|
|
if i >= n:
|
|
raise DM("string")
|
|
x = t[i]
|
|
i += 1
|
|
if x == c:
|
|
break
|
|
if x in "\r\n":
|
|
raise DM("line break")
|
|
if x != "\\":
|
|
s += x
|
|
continue
|
|
if i >= n:
|
|
raise DM("escape")
|
|
e = t[i]
|
|
i += 1
|
|
simple = {"n": "\n", "r": "\r", "t": "\t", "a": "\a", "b": "\b", "f": "\f", "v": "\v", "\\": "\\", "'": "'", '"': '"'}
|
|
if e in simple:
|
|
s += simple[e]
|
|
elif e == "x":
|
|
v = d = 0
|
|
while d < 2 and i < n and t[i] in "0123456789ABCDEF":
|
|
v = v * 16 + int(t[i], 16)
|
|
i += 1
|
|
d += 1
|
|
if d == 0:
|
|
raise DM("empty \\x")
|
|
if v >= 0x80:
|
|
raise NR("\\x past 0x7F")
|
|
s += chr(v)
|
|
else:
|
|
raise NR("escape")
|
|
out.append(("str", s))
|
|
elif c == "@":
|
|
if t.startswith("@@", i + 1):
|
|
raise NR("@@@")
|
|
if t.startswith("@", i + 1):
|
|
raise NR("@@@")
|
|
i += 1
|
|
s = ""
|
|
while True:
|
|
if i >= n:
|
|
raise DM("asset path")
|
|
x = t[i]
|
|
i += 1
|
|
if x == "@":
|
|
break
|
|
if x in "\r\n":
|
|
raise DM("line break")
|
|
s += x
|
|
out.append(("asset", s))
|
|
elif c == "<":
|
|
j = t.find(">", i + 1)
|
|
if j < 0:
|
|
raise DM("path")
|
|
out.append(("path", t[i + 1:j]))
|
|
i = j + 1
|
|
elif (c.isascii() and c.isalpha()) or c == "_":
|
|
s0 = i
|
|
while i < n and ((t[i].isascii() and (t[i].isalnum() or t[i] == "_")) or (t[i] == ":" and i + 1 < n and ((t[i + 1].isascii() and t[i + 1].isalpha()) or t[i + 1] == "_"))):
|
|
i += 1
|
|
out.append(("id", t[s0:i]))
|
|
else:
|
|
raise NR(f"character {c!r}")
|
|
return out
|
|
|
|
|
|
class UP:
|
|
def __init__(self, toks):
|
|
self.t, self.i = toks, 0
|
|
|
|
def peek(self):
|
|
return self.t[self.i] if self.i < len(self.t) else None
|
|
|
|
def next(self):
|
|
if self.i >= len(self.t):
|
|
raise DM("ends early")
|
|
self.i += 1
|
|
return self.t[self.i - 1]
|
|
|
|
def is_(self, c):
|
|
return self.peek() == ("p", c)
|
|
|
|
def punct(self, c):
|
|
if self.next() != ("p", c):
|
|
raise DM(f"{c} due")
|
|
|
|
def ident(self):
|
|
k, v = self.next()
|
|
if k != "id":
|
|
raise DM("name due")
|
|
return v
|
|
|
|
def skip(self):
|
|
while self.is_(";"):
|
|
self.i += 1
|
|
|
|
def value(self):
|
|
k, v = self.next()
|
|
if k in ("num", "str", "id", "asset", "path"):
|
|
return (k, v)
|
|
if (k, v) == ("p", "("):
|
|
items = []
|
|
while not self.is_(")"):
|
|
items.append(self.value())
|
|
if not self.is_(")"):
|
|
self.punct(",")
|
|
self.i += 1
|
|
return ("tuple", items)
|
|
if (k, v) == ("p", "["):
|
|
items = []
|
|
while not self.is_("]"):
|
|
items.append(self.value())
|
|
if not self.is_("]"):
|
|
self.punct(",")
|
|
self.i += 1
|
|
return ("list", items)
|
|
if (k, v) == ("p", "{"):
|
|
return self.dict_body()
|
|
raise DM("value due")
|
|
|
|
def dict_body(self):
|
|
items = []
|
|
while True:
|
|
self.skip()
|
|
if self.is_("}"):
|
|
self.i += 1
|
|
return ("dict", items)
|
|
ty = self.ident()
|
|
ty2 = ty
|
|
if self.is_("["):
|
|
self.i += 1
|
|
self.punct("]")
|
|
ty2 += "[]"
|
|
k, key = self.next()
|
|
if k not in ("str", "id"):
|
|
raise DM("key")
|
|
self.punct("=")
|
|
if ty == "dictionary":
|
|
self.punct("{")
|
|
v = self.dict_body()
|
|
else:
|
|
v = self.value()
|
|
items.append((ty2, key, v))
|
|
|
|
def samples(self):
|
|
self.punct("{")
|
|
v = []
|
|
while not self.is_("}"):
|
|
k, tcode = self.next()
|
|
if k != "num":
|
|
raise DM("time code")
|
|
self.punct(":")
|
|
v.append((tcode, self.value()))
|
|
if not self.is_("}"):
|
|
self.punct(",")
|
|
self.i += 1
|
|
return ("samples", v)
|
|
|
|
def prim(self):
|
|
s = self.ident()
|
|
if s in ("over", "class"):
|
|
raise NR(s)
|
|
if s != "def":
|
|
raise DM("prim due")
|
|
p = {"ty": "", "name": None, "refs": [], "props": [], "children": []}
|
|
if self.peek() and self.peek()[0] == "id":
|
|
p["ty"] = self.next()[1]
|
|
k, name = self.next()
|
|
if k != "str":
|
|
raise DM("prim name")
|
|
p["name"] = name
|
|
if self.is_("("):
|
|
self.i += 1
|
|
while True:
|
|
self.skip()
|
|
if self.is_(")"):
|
|
self.i += 1
|
|
break
|
|
if self.peek() and self.peek()[0] == "str":
|
|
self.i += 1
|
|
continue
|
|
key = self.ident()
|
|
op = None
|
|
if key in ("prepend", "append", "delete", "add", "reorder"):
|
|
op, key = key, self.ident()
|
|
if key == "references":
|
|
if op not in (None, "prepend", "append"):
|
|
raise NR("references op")
|
|
self.punct("=")
|
|
v = self.value()
|
|
its = v[1] if v[0] == "list" else ([] if v == ("id", "None") else [v])
|
|
for it in its:
|
|
if it[0] != "asset":
|
|
raise NR("not an asset path")
|
|
if (self.peek() and self.peek()[0] == "path") or self.is_("("):
|
|
raise NR("prim path or parameters")
|
|
p["refs"].append(it[1])
|
|
elif key in ("payload", "inherits", "specializes", "variants", "variantSets", "relocates"):
|
|
raise NR(key)
|
|
else:
|
|
if op is not None:
|
|
raise NR("list op")
|
|
self.punct("=")
|
|
self.value()
|
|
self.punct("{")
|
|
while True:
|
|
self.skip()
|
|
if self.is_("}"):
|
|
self.i += 1
|
|
break
|
|
pk = self.peek()
|
|
if pk and pk[0] == "id" and pk[1] in ("def", "over", "class"):
|
|
c = self.prim()
|
|
if any(x["name"] == c["name"] for x in p["children"]):
|
|
raise DM("prim twice")
|
|
p["children"].append(c)
|
|
elif pk and pk[0] == "id" and pk[1] in ("rel", "variantSet", "reorder", "prepend", "append", "delete", "add"):
|
|
raise NR(pk[1])
|
|
else:
|
|
name, ty, default, smp, special = self.prop()
|
|
e = next((x for x in p["props"] if x[0] == name), None)
|
|
if e is None:
|
|
e = [name, "", None, None]
|
|
p["props"].append(e)
|
|
if e[1] and e[1] != ty:
|
|
raise DM("two types")
|
|
e[1] = ty
|
|
if special:
|
|
if e[3] is not None:
|
|
raise DM("samples twice")
|
|
e[3] = smp
|
|
else:
|
|
if e[2] is not None:
|
|
raise DM("default twice")
|
|
e[2] = default
|
|
return p
|
|
|
|
def prop(self):
|
|
t = self.ident()
|
|
if t == "custom":
|
|
t = self.ident()
|
|
if t in ("uniform", "varying"):
|
|
t = self.ident()
|
|
ty = t
|
|
if self.is_("["):
|
|
self.i += 1
|
|
self.punct("]")
|
|
ty += "[]"
|
|
name = self.ident()
|
|
default = smp = None
|
|
special = False
|
|
if self.is_("."):
|
|
self.i += 1
|
|
if self.ident() != "timeSamples":
|
|
raise NR("not timeSamples")
|
|
self.punct("=")
|
|
smp = self.samples()
|
|
special = True
|
|
elif self.is_("="):
|
|
self.i += 1
|
|
default = self.value()
|
|
if self.is_("("):
|
|
self.i += 1
|
|
while True:
|
|
self.skip()
|
|
if self.is_(")"):
|
|
self.i += 1
|
|
break
|
|
if self.peek() and self.peek()[0] == "str":
|
|
self.i += 1
|
|
continue
|
|
self.ident()
|
|
self.punct("=")
|
|
self.value()
|
|
return name, ty, default, smp, special
|
|
|
|
|
|
def parse_usda(b):
|
|
try:
|
|
text = b.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise DM("not UTF-8")
|
|
if not text.startswith("#usda"):
|
|
raise DM("header")
|
|
rest = text[5:]
|
|
m = re.search(r"[\r\n]", rest)
|
|
le = m.start() if m else len(rest)
|
|
if rest[:le].strip(" \t") != "1.0":
|
|
raise NR("version")
|
|
p = UP(lex(rest[le:]))
|
|
custom, default_prim, prims = [], None, []
|
|
if p.is_("("):
|
|
p.i += 1
|
|
while True:
|
|
p.skip()
|
|
if p.is_(")"):
|
|
p.i += 1
|
|
break
|
|
if p.peek() and p.peek()[0] == "str":
|
|
p.i += 1
|
|
continue
|
|
key = p.ident()
|
|
if key in ("subLayers", "relocates", "prepend", "append", "delete"):
|
|
raise NR(key)
|
|
p.punct("=")
|
|
v = p.value()
|
|
if key == "customLayerData":
|
|
if v[0] != "dict":
|
|
raise DM("customLayerData")
|
|
custom += [(k, x) for ty, k, x in v[1] if ty == "string"]
|
|
elif key == "defaultPrim":
|
|
if v[0] != "str":
|
|
raise DM("defaultPrim")
|
|
default_prim = v[1]
|
|
while p.peek() is not None:
|
|
c = p.prim()
|
|
if any(x["name"] == c["name"] for x in prims):
|
|
raise DM("prim twice")
|
|
prims.append(c)
|
|
return custom, default_prim, prims
|
|
|
|
|
|
def import_usda(b, params):
|
|
custom, default_prim, prims = parse_usda(b)
|
|
wai = [(k, v) for k, v in custom if k.startswith("wai:")]
|
|
if wai:
|
|
bundle = {"world": None, "audio": None, "media": []}
|
|
prof = False
|
|
for k, v in wai:
|
|
if v[0] != "str":
|
|
raise Refuse("carried_malformed", "not a string")
|
|
s = v[1]
|
|
if k == "wai:profile" and s == PROFILE:
|
|
prof = True
|
|
elif k in ("wai:world", "wai:audio"):
|
|
x = unhex(s)
|
|
if x is None:
|
|
raise Refuse("carried_malformed", k)
|
|
bundle[k[4:]] = x
|
|
elif k == "wai:media":
|
|
try:
|
|
m = media_from(strict_json(s.encode()))
|
|
except Refuse:
|
|
m = None
|
|
if m is None:
|
|
raise Refuse("carried_malformed", "media")
|
|
bundle["media"] = m
|
|
else:
|
|
raise Refuse("carried_malformed", k)
|
|
if not prof:
|
|
raise Refuse("carried_malformed", "profile")
|
|
try:
|
|
again = export_usda(bundle)
|
|
except Refuse as r:
|
|
raise Refuse("carried_malformed", r.code)
|
|
if again != b:
|
|
raise Refuse("presentation_mismatch")
|
|
return bundle, []
|
|
# §7.3
|
|
if default_prim is not None:
|
|
root = next((x for x in prims if x["name"] == default_prim), None)
|
|
if root is None:
|
|
raise DM("defaultPrim")
|
|
else:
|
|
if not prims:
|
|
raise DM("no prim")
|
|
root = prims[0]
|
|
nodes, assets, dropped = [], [], []
|
|
stack = [(root, [0, 0, 0], "/" + root["name"])]
|
|
while stack:
|
|
pr, base, path = stack.pop()
|
|
off = [0, 0, 0]
|
|
for name, ty, default, smp in pr["props"]:
|
|
if name == "xformOpOrder":
|
|
if not (default is not None and default[0] == "list" and default[1] == [("str", "xformOp:translate")]) or smp is not None:
|
|
raise Refuse("transform_not_carried", "order")
|
|
elif name == "xformOp:translate":
|
|
if smp is not None or default is None or default[0] != "tuple":
|
|
raise Refuse("transform_not_carried", "translate")
|
|
if len(default[1]) != 3:
|
|
raise DM("three values")
|
|
for i, x in enumerate(default[1]):
|
|
if x[0] != "num":
|
|
raise DM("numbers")
|
|
try:
|
|
f = float(x[1])
|
|
except ValueError:
|
|
raise DM("number")
|
|
if f != f or f in (float("inf"), float("-inf")):
|
|
raise Refuse("position_not_exact")
|
|
q = to_fx(f)
|
|
if q is None:
|
|
raise Refuse("position_not_exact")
|
|
off[i] = q
|
|
elif name.startswith("xformOp:"):
|
|
raise Refuse("transform_not_carried", name)
|
|
at = [base[i] + off[i] for i in range(3)]
|
|
if any(not -(1 << 63) <= x < (1 << 63) for x in at):
|
|
raise Refuse("position_not_exact")
|
|
if pr["refs"]:
|
|
for r in pr["refs"]:
|
|
if not (r.startswith(URN) and re.fullmatch(r"[0-9a-f]{64}", r[len(URN):])):
|
|
raise Refuse("asset_not_wai", r)
|
|
assets.append((bytes.fromhex(r[len(URN):]), 0, pr["name"]))
|
|
elif pr["ty"] == "Xform" and any(x[0] == "xformOp:translate" for x in pr["props"]):
|
|
nodes.append(at)
|
|
elif pr["ty"] not in ("", "Xform", "Scope"):
|
|
dropped.append(f"{path}: {pr['ty']}")
|
|
for c in reversed(pr["children"]):
|
|
stack.append((c, at, path + "/" + c["name"]))
|
|
bundle = {"world": None, "audio": None, "media": []}
|
|
if nodes or assets:
|
|
j = contract_json(params.get("world_contract"))
|
|
if not isinstance(j, Obj) or j.get("sim") != KIN:
|
|
raise Refuse("sim_not_importable")
|
|
ns = [(mint(params["actor"], i + 1), kin_state(p)) for i, p in enumerate(nodes)]
|
|
bundle["world"] = world_bytes(params["world_contract"], ns, assets)
|
|
return bundle, dropped
|
|
|
|
|
|
# ------------------------------------------------------------------ the corpus
|
|
|
|
def run(corpus):
|
|
doc = json.load(open(os.path.join(corpus, "vectors.json"), encoding="utf-8"))
|
|
known = codes()
|
|
fails = []
|
|
for inp in doc["inputs"]:
|
|
b = open(os.path.join(corpus, inp["path"]), "rb").read()
|
|
if hashlib.sha256(b).hexdigest() != inp["sha256"]:
|
|
fails.append(f"input {inp['path']} does not match its pin")
|
|
files = {}
|
|
for f in doc["files"]:
|
|
b = open(os.path.join(corpus, f["file"]), "rb").read()
|
|
if hashlib.sha256(b).hexdigest() != f["sha256"]:
|
|
fails.append(f"{f['file']} does not match its digest")
|
|
files[f["file"]] = b
|
|
listed = set(files) | {"vectors.json", "README.md", ".gitattributes"}
|
|
for root, _, fs in os.walk(corpus):
|
|
for x in fs:
|
|
rel = os.path.relpath(os.path.join(root, x), corpus).replace(os.sep, "/")
|
|
if rel not in listed:
|
|
fails.append(f"{rel} is not listed")
|
|
|
|
def load(path):
|
|
return files[path] if path in files else open(os.path.join(corpus, path), "rb").read()
|
|
|
|
def bundle_of(bv):
|
|
media = []
|
|
for m in bv["media"]:
|
|
media.append({"name": m["name"], "asset": bytes.fromhex(m["asset"]), "capability": m["capability"], "width": m.get("width"), "height": m.get("height")})
|
|
return {"world": load(bv["world"]) if bv["world"] else None, "audio": load(bv["audio"]) if bv["audio"] else None, "media": media}
|
|
|
|
exported = []
|
|
for v in doc["vectors"]:
|
|
name, op, fmt = v["name"], v["op"], v["format"]
|
|
if v["outcome"] == "refuse" and v["code"] not in known:
|
|
fails.append(f"{name}: {v['code']} is not in §8")
|
|
try:
|
|
if op == "export":
|
|
got = (export_gltf if fmt == "gltf" else export_usda)(bundle_of(v["bundle"]))
|
|
else:
|
|
pv = v["params"] or {}
|
|
params = {"actor": bytes.fromhex(pv["actor"]) if pv.get("actor") else bytes(32),
|
|
"world_contract": pv["world_contract"].encode() if pv.get("world_contract") is not None else None,
|
|
"audio_contract": pv["audio_contract"].encode() if pv.get("audio_contract") is not None else None}
|
|
got = (import_gltf if fmt == "gltf" else import_usda)(files[v["file"]], params)
|
|
outcome = ("accept", got)
|
|
except Refuse as r:
|
|
outcome = ("refuse", r.code)
|
|
if v["outcome"] == "refuse":
|
|
if outcome != ("refuse", v["code"]):
|
|
fails.append(f"{name}: wanted {v['code']}, got {outcome[0]} {outcome[1] if outcome[0] == 'refuse' else ''}")
|
|
continue
|
|
if outcome[0] != "accept":
|
|
fails.append(f"{name}: wanted accept, got {outcome[1]}")
|
|
continue
|
|
if op == "export":
|
|
if outcome[1] != files[v["file"]]:
|
|
fails.append(f"{name}: the document differs from {v['file']}")
|
|
else:
|
|
exported.append((fmt, files[v["file"]], bundle_of(v["bundle"])))
|
|
else:
|
|
want = bundle_of(v["bundle"])
|
|
b, dropped = outcome[1]
|
|
if (b["world"], b["audio"]) != (want["world"], want["audio"]) or b["media"] != want["media"]:
|
|
fails.append(f"{name}: read to another bundle")
|
|
if dropped != v["dropped"]:
|
|
fails.append(f"{name}: reported {dropped}, wanted {v['dropped']}")
|
|
missing = set(known) - {v.get("code") for v in doc["vectors"]}
|
|
if missing:
|
|
fails.append(f"codes no vector gives: {sorted(missing)}")
|
|
caught = 0
|
|
for fmt, b, want in exported:
|
|
for at in sorted({0, len(b) // 4, len(b) // 2, 3 * len(b) // 4, len(b) - 2}):
|
|
m = bytearray(b)
|
|
m[at] ^= 0x04
|
|
try:
|
|
got, _ = (import_gltf if fmt == "gltf" else import_usda)(bytes(m), {"actor": bytes(32), "world_contract": None, "audio_contract": None})
|
|
ok = got != want
|
|
except Refuse:
|
|
ok = True
|
|
if ok:
|
|
caught += 1
|
|
else:
|
|
fails.append(f"a flip at byte {at} of a {fmt} export read to the same bundle")
|
|
return fails, len(doc["vectors"]), caught
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print(__doc__)
|
|
sys.exit(2)
|
|
fails, n, caught = run(sys.argv[1])
|
|
for f in fails:
|
|
print("FAIL", f)
|
|
if fails:
|
|
sys.exit(1)
|
|
print(f"scene: every vector verifies ({n}), and every mutated copy is caught ({caught})")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|