80dabeb126
The same pure-i64 Rust that decodes the integer inter-frame video natively compiles to wasm32 and produces byte-identical frames in the browser — a clip played in-tab with every frame re-checked against the native reference hash, no ML runtime. - video_int feature added to the wai dep; src/video.rs binds decodeIntVideo (bitstream -> DecodedVideo: all frames concatenated + nFrames/width/height). fnv1aHex (the generate binding) does the per-frame receipt check. - Drives the live demo at wai.transaction.science/video (Astro page + a 16-frame 128x128 bouncing-discs clip from tools/wai_int_video_demo_encode.py): decodes in ~20 ms, verifies all 16 frames byte-for-byte against the native hashes, and plays the clip on a canvas. Frame-state-equivalence, in-browser. Co-Authored-By: Claude Opus 4.8 <[email protected]>
123 lines
4.7 KiB
Python
123 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Encode a nicer DEMO clip with the deterministic integer inter-frame codec
|
|
(`wai.video.int_motion`) for the in-browser player: two bouncing discs over a
|
|
gradient, 128x128. Same codec as the committed conformance vector (integer block
|
|
MC + quantized residual + rANS); this is just a bigger, livelier clip to serve.
|
|
|
|
python tools/wai_int_video_demo_encode.py <out-dir>
|
|
"""
|
|
import sys, os, json, base64
|
|
import numpy as np
|
|
from compressai.ans import RansEncoder
|
|
|
|
H = W = 128
|
|
NF, BLOCK, SEARCH, Q = 16, 16, 6, 6
|
|
RMAX = (255 + Q - 1) // Q
|
|
|
|
|
|
def fnv1a(buf):
|
|
h = 14695981039346656037
|
|
for b in buf:
|
|
h = ((h ^ b) * 1099511628211) & 0xFFFFFFFFFFFFFFFF
|
|
return h
|
|
|
|
|
|
def laplacian_cdf():
|
|
n = 2 * RMAX + 1
|
|
pmf = np.exp(-np.abs(np.arange(n) - RMAX) / 6.0)
|
|
total = 1 << 16
|
|
f = np.maximum((pmf / pmf.sum() * (total - 1)).round().astype(np.int64), 1)
|
|
f[f.argmax()] += (total - 1) - int(f.sum())
|
|
cdf = [0]
|
|
for v in f:
|
|
cdf.append(cdf[-1] + int(v))
|
|
cdf.append(total)
|
|
return cdf
|
|
|
|
|
|
def make_clip():
|
|
yy, xx = np.mgrid[0:H, 0:W]
|
|
grad = ((xx * 80 // W) + (yy * 60 // H)).astype(np.int64)
|
|
# two discs with integer position/velocity, reflecting off the walls
|
|
p1, v1, r1, col1 = [28, 30], [6, 4], 16, [250, 205, 70]
|
|
p2, v2, r2, col2 = [96, 80], [-5, 6], 10, [70, 215, 230]
|
|
frames = []
|
|
for _ in range(NF):
|
|
img = np.empty((H, W, 3), np.int64)
|
|
img[..., 0] = np.clip(35 + grad, 0, 255)
|
|
img[..., 1] = np.clip(55 + grad // 2, 0, 255)
|
|
img[..., 2] = np.clip(85 + grad // 3, 0, 255)
|
|
for (p, r, col) in ((p1, r1, col1), (p2, r2, col2)):
|
|
img[((xx - p[0]) ** 2 + (yy - p[1]) ** 2) <= r * r] = col
|
|
frames.append(img.astype(np.int64))
|
|
for (p, v, r) in ((p1, v1, r1), (p2, v2, r2)):
|
|
for i in (0, 1):
|
|
p[i] += v[i]
|
|
if p[i] < r or p[i] >= (W if i == 0 else H) - r:
|
|
v[i] = -v[i]
|
|
p[i] += 2 * v[i]
|
|
return frames
|
|
|
|
|
|
def mc_predict(ref, mvs):
|
|
pred = np.empty_like(ref)
|
|
nbx = (W + BLOCK - 1) // BLOCK
|
|
for by in range(0, H, BLOCK):
|
|
for bx in range(0, W, BLOCK):
|
|
dy, dx = mvs[(by // BLOCK) * nbx + (bx // BLOCK)]
|
|
ys = np.clip(np.arange(by, min(by + BLOCK, H)) + dy, 0, H - 1)
|
|
xs = np.clip(np.arange(bx, min(bx + BLOCK, W)) + dx, 0, W - 1)
|
|
pred[by:by + len(ys), bx:bx + len(xs)] = ref[np.ix_(ys, xs)]
|
|
return pred
|
|
|
|
|
|
def search_mvs(cur, ref):
|
|
mvs = []
|
|
for by in range(0, H, BLOCK):
|
|
for bx in range(0, W, BLOCK):
|
|
cb = cur[by:by + BLOCK, bx:bx + BLOCK]
|
|
best, bmv = None, (0, 0)
|
|
for dy in range(-SEARCH, SEARCH + 1):
|
|
for dx in range(-SEARCH, SEARCH + 1):
|
|
ys = np.clip(np.arange(by, by + cb.shape[0]) + dy, 0, H - 1)
|
|
xs = np.clip(np.arange(bx, bx + cb.shape[1]) + dx, 0, W - 1)
|
|
sad = int(np.abs(cb - ref[np.ix_(ys, xs)]).sum())
|
|
if best is None or sad < best:
|
|
best, bmv = sad, (dy, dx)
|
|
mvs.append([int(bmv[0]), int(bmv[1])])
|
|
return mvs
|
|
|
|
|
|
def main(outdir):
|
|
os.makedirs(outdir, exist_ok=True)
|
|
frames = make_clip()
|
|
cdf, offset = laplacian_cdf(), -RMAX
|
|
enc = RansEncoder()
|
|
syms, mvs_all, recon, hashes = [], [], [], []
|
|
for f in range(NF):
|
|
if f == 0:
|
|
pred, mvs = np.full((H, W, 3), 128, np.int64), []
|
|
else:
|
|
mvs = search_mvs(frames[f], recon[-1])
|
|
pred = np.stack([mc_predict(recon[-1][..., c], mvs) for c in range(3)], -1)
|
|
rq = np.round((frames[f] - pred) / Q).astype(np.int64).clip(-RMAX, RMAX)
|
|
syms.extend(rq.reshape(-1).tolist())
|
|
rec = np.clip(pred + rq * Q, 0, 255).astype(np.int64)
|
|
recon.append(rec)
|
|
mvs_all.append(mvs)
|
|
hashes.append(format(fnv1a(np.ascontiguousarray(rec.astype(np.uint8)).tobytes()), '016x'))
|
|
|
|
bits = enc.encode_with_indexes(syms, [0] * len(syms), [cdf], [len(cdf)], [offset])
|
|
clip_hash = format(fnv1a(b''.join(bytes.fromhex(h) for h in hashes)), '016x')
|
|
json.dump({'block': BLOCK, 'q': Q, 'rmax': RMAX, 'offset': offset, 'cdf': cdf,
|
|
'n_frames': NF, 'h': H, 'w': W, 'mvs': mvs_all,
|
|
'rans_b64': base64.b64encode(bytes(bits)).decode()},
|
|
open(f'{outdir}/video_demo_bitstream.json', 'w'))
|
|
json.dump({'n_frames': NF, 'h': H, 'w': W, 'frame_hashes': hashes, 'clip_hash': clip_hash},
|
|
open(f'{outdir}/video_demo.json', 'w'))
|
|
sz = os.path.getsize(f'{outdir}/video_demo_bitstream.json')
|
|
print(f'demo clip: {NF}x {H}x{W}, {len(bits)} rANS bytes (bitstream {sz} B) -> clip_hash {clip_hash}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main(sys.argv[1])
|