human-experience-filter-art/simulator/clips.py

"""The base-clip + variant + annotation manifest the simulator renders.

Replaces simulator/fixtures.py (the selection-era synthetic catalog). Each base
clip carries: the raw base file, a map of pre-baked Right-strength variant files
(strength 0 is always the raw base), an authored Left annotation track (box +
label key + the minimum Left level at which it appears), and per-language string
tables. See the reconciled-simulator-alteration-slice design §3.2.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from player.ring import ResolvedMove, RingMove, Scale, ScaleRing, Transition


@dataclass(frozen=True)
class Clip:
    id: str
    title: str
    base_file: str
    license: str
    source: str
    right_variants: dict          # {"1": {"file": ...}, "4": {...}} (no "0")
    annotations: list             # [{"key", "box":[x,y,w,h], "min_level"}, ...]
    affect: list                  # [{"key", "at":[x,y], "min_level"}, ...] (Left x Right)
    strings: dict                 # {"en": {key: text}}

    def variant_file(self, strength: int) -> str:
        """The video file for a Right strength; 0 and any unauthored strength
        fall back to the raw base file."""
        entry = self.right_variants.get(str(strength))
        return entry["file"] if entry else self.base_file

    def to_dict(self) -> dict:
        variants = {"0": {"file": self.base_file, "raw": True}}
        for k, v in self.right_variants.items():
            variants[k] = v
        return {
            "id": self.id,
            "title": self.title,
            "base_file": self.base_file,
            "license": self.license,
            "source": self.source,
            "right_variants": variants,
            "annotations": self.annotations,
            "affect": self.affect,
            "strings": self.strings,
        }


def _clip_from_dict(d: dict[str, Any]) -> Clip:
    return Clip(
        id=d["id"],
        title=d["title"],
        base_file=d["base_file"],
        license=d.get("license", ""),
        source=d.get("source", ""),
        right_variants=d.get("right_variants", {}),
        annotations=d.get("annotations", []),
        affect=d.get("affect", []),
        strings=d.get("strings", {}),
    )


def load_manifest(path: str | Path) -> list[Clip]:
    """Load the base-clip manifest. Raises FileNotFoundError if missing."""
    path = Path(path)
    data = json.loads(path.read_text())
    return [_clip_from_dict(c) for c in data["clips"]]


def _scale_from_dict(s: dict[str, Any]) -> Scale:
    """A ring scale node, with its rotating clip pool (content-pipeline §11.1).

    Accepts a `pool` list (the rotating model) and/or a legacy single `clip_id`.
    The primary `clip_id` is the explicit one if given, else the first pool
    member; `pool` defaults to `(clip_id,)` for a pre-pool manifest.
    """
    pool = tuple(s.get("pool", []))
    clip_id = s.get("clip_id") or (pool[0] if pool else "")
    return Scale(id=s["id"], clip_id=clip_id, pool=pool, audio=s.get("audio", ""))


def load_ring(path: str | Path) -> ScaleRing | None:
    """Load the optional `ring` section as a `ScaleRing`, or None if absent.

    The ring closes the scales-of-nature library into the navigable loop the
    endless encoder walks (scales design §3); the navigation math itself lives
    in player.ring.
    """
    path = Path(path)
    data = json.loads(path.read_text())
    ring = data.get("ring")
    if not ring:
        return None
    scales = tuple(_scale_from_dict(s) for s in ring.get("scales", []))
    transitions = tuple(
        Transition(
            from_clip=t.get("from", ""),
            to_clip=t.get("to", ""),
            file=t["file"],
            model=t.get("model", ""),
        )
        for t in ring.get("transitions", [])
    )
    return ScaleRing(scales=scales, transitions=transitions)


def ring_to_dict(ring: ScaleRing, clips: list[Clip]) -> dict:
    """JSON form of the ring for the API: ordered scales (with their clip title)
    and the per-edge transitions.

    Each scale carries its full rotating `pool` (clip id + title per member) plus
    the primary `clip_id`/`title` for back-compat. The client lands on a member
    chosen at the API boundary (`/api/ring/advance` → `target_clip_id`)."""
    titles = {c.id: c.title for c in clips}
    return {
        "scales": [
            {
                "id": s.id,
                "clip_id": s.clip_id,
                "title": titles.get(s.clip_id, s.id),
                "audio": s.audio,
                "pool": [
                    {"clip_id": cid, "title": titles.get(cid, cid)}
                    for cid in s.members
                ],
            }
            for s in ring.scales
        ],
        "transitions": [
            {"from": t.from_clip, "to": t.to_clip, "file": t.file, "model": t.model}
            for t in ring.transitions
        ],
    }


def resolved_move_to_dict(move: RingMove, resolved: ResolvedMove) -> dict:
    """JSON form of a RESOLVED encoder move: where it lands, the locked clip, and
    the ordered chained morphs. Each step carries the clip it morphs FROM (the
    clip currently shown), the chosen clip it morphs TO, and the matching morph
    file (None if no morph was baked — the renderer plain-cuts). `fast` marks a
    fast spin; each step then carries `blended` so the renderer accelerates it.

    `target_clip_id` is the final clip the viewer locks onto."""
    return {
        "from_index": move.from_index,
        "to_index": move.to_index,
        "wrapped": move.wrapped,
        "fast": move.fast,
        "target_clip_id": resolved.target_clip_id,
        "steps": [
            {
                "from_clip": s.from_clip,
                "to_clip": s.to_clip,
                "file": s.file,
                "to_index": s.to_index,
                "blended": s.blended,
            }
            for s in resolved.steps
        ],
    }