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 RingMove, Scale, ScaleRing, Transition, scale_at


@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(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": [{"file": t.file, "model": t.model} for t in ring.transitions],
    }


def ring_move_to_dict(
    move: RingMove, ring: ScaleRing, chosen_clip_id: str | None = None
) -> dict:
    """JSON form of an encoder move: the landing scale's clip and the ordered
    transition clips to play (with direction). `fast` flags a collapsed fast-spin
    pass; the single step then carries `blended` so the renderer plays it quick.

    `target_clip_id` is the pool member the player should load on landing. The
    caller passes the random pick (`pick_clip_id(landed_scale, random.random())`,
    content-pipeline §11.1); when omitted it falls back to the scale's primary
    `clip_id` (deterministic — keeps pre-pool callers/tests working)."""
    return {
        "from_index": move.from_index,
        "to_index": move.to_index,
        "wrapped": move.wrapped,
        "fast": move.fast,
        "target_clip_id": chosen_clip_id or scale_at(ring, move.to_index).clip_id,
        "steps": [
            {
                "edge": st.edge,
                "reversed": st.reversed,
                "file": st.file,
                "to_index": st.to_index,
                "blended": st.blended,
            }
            for st in move.steps
        ],
    }