feat(badapple): cost model, frame selection and build CLI

bad-apple/pipeline/build.py

@@ -0,0 +1,105 @@
+"""CLI: video -> vector frames -> tile deltas -> movie.bin + manifest."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+
+import numpy as np
+
+from .cost import CostConstants, estimate_record_us, select_records
+from .frames import extract_binary
+from .movie import MovieHeader, write_movie
+from .preview import write_previews
+from .vectorize import vectorize_frame
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+
+
+def _stats(values) -> dict:
+    if not values:
+        return {"mean": 0.0, "p50": 0, "p95": 0, "max": 0}
+    arr = np.asarray(values, dtype=float)
+    return {
+        "mean": float(arr.mean()),
+        "p50": int(np.percentile(arr, 50)),
+        "p95": int(np.percentile(arr, 95)),
+        "max": int(arr.max()),
+    }
+
+
+def build_movie(video: str, out_dir: str, params: dict | None = None) -> dict:
+    with open(os.path.join(HERE, "params.json"), encoding="utf-8") as handle:
+        p = json.load(handle)
+    if params:
+        p.update(params)
+
+    frames = []
+    for index, frame in enumerate(extract_binary(
+            video, p["start"], p["duration"], p["fps"],
+            p["width"], p["height"], p)):
+        frames.append(vectorize_frame(frame, p))
+        if (index + 1) % 60 == 0:
+            print("  vectorised %d frames" % (index + 1))
+    if not frames:
+        raise RuntimeError("no frames extracted from %s" % video)
+
+    constants = CostConstants()
+    records = select_records(frames, constants, budget_us=13900.0, max_skip=2)
+
+    height, width = frames[0].shape
+    header = MovieHeader(
+        width=width, height=height, scale=int(p.get("scale", 2)),
+        tile=int(p.get("tile", 32)), tiles_x=width // int(p.get("tile", 32)),
+        tiles_y=height // int(p.get("tile", 32)), frame_count=len(records),
+        source_fps=int(p.get("fps", 60)), display_hz=72)
+
+    os.makedirs(out_dir, exist_ok=True)
+    data_bytes = write_movie(os.path.join(out_dir, "movie.bin"), records, header)
+
+    dirty = [len(record.tiles) for record in records]
+    rects = [sum(len(tile.rects) for tile in record.tiles) for record in records]
+    costs = [estimate_record_us(record, constants) for record in records]
+    manifest = {
+        "video": video,
+        "data_bytes": data_bytes,
+        "frame_count": len(records),
+        "source_frames": len(frames),
+        "dropped": len(frames) - len(records),
+        "width": width,
+        "height": height,
+        "scale": header.scale,
+        "tile": header.tile,
+        "source_fps": header.source_fps,
+        "display_hz": header.display_hz,
+        "dirty_tiles": _stats(dirty),
+        "rects": _stats(rects),
+        "cost_us": _stats(costs),
+    }
+    with open(os.path.join(out_dir, "manifest.json"), "w",
+              encoding="utf-8") as handle:
+        json.dump(manifest, handle, indent=2, sort_keys=True)
+    write_previews(os.path.join(out_dir, "preview"), frames,
+                   [0, 60, 300, 900, 1799])
+    return manifest
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--video", required=True)
+    parser.add_argument("--out", default="data")
+    parser.add_argument("--duration", type=float, default=None)
+    parser.add_argument("--start", type=float, default=None)
+    args = parser.parse_args(argv)
+    overrides = {}
+    if args.duration is not None:
+        overrides["duration"] = args.duration
+    if args.start is not None:
+        overrides["start"] = args.start
+    manifest = build_movie(args.video, args.out, overrides)
+    print(json.dumps(manifest, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

bad-apple/pipeline/cost.py

@@ -0,0 +1,70 @@
+"""Paint-cost model and build-time frame selection."""
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from typing import Sequence
+
+import numpy as np
+
+from .movie import FrameRecord, TileRecord
+from .tilepack import diff_tiles
+
+
+@dataclass(frozen=True)
+class CostConstants:
+    word_us: float = 0.40
+    call_us: float = 3.0
+    rect_us: float = 1.0
+
+
+def _words(width_px: int) -> int:
+    return max(1, int(math.ceil(width_px / 16.0)))
+
+
+def estimate_record_us(record: FrameRecord, constants: CostConstants,
+                       scale: int = 2, tile: int = 32) -> float:
+    """Unit-time estimate for one record, including call and decode overhead."""
+    tile_words = _words(tile * scale) * tile * scale
+    total = 0.0
+    for rec in record.tiles:
+        total += tile_words * constants.word_us + constants.call_us
+        for rect in rec.rects:
+            total += (_words(rect.w * scale) * rect.h * scale
+                      * constants.word_us)
+            total += constants.call_us + constants.rect_us
+    return total
+
+
+def select_records(frames: Sequence[np.ndarray], constants: CostConstants,
+                   budget_us: float = 13900.0,
+                   max_skip: int = 2) -> list[FrameRecord]:
+    """Build exact deltas, dropping source frames that cannot fit the gate.
+
+    The dropped frame is never emitted; the next kept frame is diffed from the
+    last *shown* frame and carries the accumulated ``ticks``.  A run of heavy
+    frames is eventually emitted anyway (a cut), so the movie still advances.
+    """
+    if not frames:
+        return []
+    prev = np.zeros_like(frames[0], dtype=bool)
+    out: list[FrameRecord] = []
+    pending = 0
+    skipped = 0
+    last = frames[0]
+    for current in frames:
+        pending += 1
+        last = current
+        rec_tiles = diff_tiles(prev, current, tile=32)
+        cost = estimate_record_us(FrameRecord(1, 0, rec_tiles), constants)
+        if cost <= budget_us or skipped >= max_skip:
+            out.append(FrameRecord(pending, 0, rec_tiles))
+            prev = current
+            pending = 0
+            skipped = 0
+        else:
+            skipped += 1
+    if pending:
+        rec_tiles = diff_tiles(prev, last, tile=32)
+        out.append(FrameRecord(pending, 0, rec_tiles))
+    return out

bad-apple/pipeline/preview.py

@@ -0,0 +1,25 @@
+"""Write human-viewable previews of the logical frames."""
+from __future__ import annotations
+
+import os
+from typing import Sequence
+
+import numpy as np
+from PIL import Image
+
+
+def write_previews(out_dir: str, frames: Sequence[np.ndarray],
+                   indices: Sequence[int], scale: int = 4) -> list[str]:
+    os.makedirs(out_dir, exist_ok=True)
+    written: list[str] = []
+    for index in indices:
+        if index < 0 or index >= len(frames):
+            continue
+        gray = np.where(frames[index], 0, 255).astype(np.uint8)
+        image = Image.fromarray(gray, mode="L")
+        image = image.resize((gray.shape[1] * scale, gray.shape[0] * scale),
+                             Image.Resampling.NEAREST)
+        path = os.path.join(out_dir, "frame_%04d.png" % index)
+        image.save(path)
+        written.append(path)
+    return written

bad-apple/tests/test_build.py

@@ -0,0 +1,21 @@
+import json
+
+import numpy as np
+
+import pipeline.build as build
+from pipeline.movie import read_movie
+
+
+def test_build_tiny_movie(tmp_path, monkeypatch):
+    frames = [np.zeros((64, 64), dtype=bool) for _ in range(3)]
+    frames[1][8:24, 8:24] = True
+    monkeypatch.setattr(build, "extract_binary", lambda *a, **k: iter(frames))
+    monkeypatch.setattr(build, "vectorize_frame", lambda m, p: m)
+    out = tmp_path / "data"
+    manifest = build.build_movie("unused.mp4", str(out),
+                                 {"duration": 3, "fps": 60})
+    header, recs = read_movie(str(out / "movie.bin"))
+    assert header.frame_count == 3
+    assert len(recs) == 3
+    assert (out / "manifest.json").exists()
+    assert json.loads((out / "manifest.json").read_text())["frame_count"] == 3

bad-apple/tests/test_cost.py

@@ -0,0 +1,26 @@
+import numpy as np
+
+from pipeline.cost import CostConstants, estimate_record_us, select_records
+from pipeline.movie import FrameRecord, TileRecord
+
+
+C = CostConstants()
+
+
+def test_one_solid_tile_cost():
+    rec = FrameRecord(1, 0, [TileRecord(0, 1, [])])
+    # 64x64 screen tile = 256 words; 256*0.40 + 3us call
+    assert abs(estimate_record_us(rec, C) - (256 * 0.40 + 3.0)) < 0.01
+
+
+def test_full_screen_is_over_budget():
+    recs = [TileRecord(i, 1, []) for i in range(320)]
+    assert estimate_record_us(FrameRecord(1, 0, recs), C) > 13900.0
+
+
+def test_select_records_holds_time_not_records():
+    quiet = np.zeros((32, 32), dtype=bool)
+    busy = np.ones((32, 32), dtype=bool)
+    out = select_records([quiet, busy, quiet, quiet], C, budget_us=50.0)
+    assert all(r.ticks >= 1 for r in out)
+    assert sum(r.ticks for r in out) == 4