human-experience-filter-art/player/ring.py

"""Scale-ring navigation: the endless-encoder closed loop of nature scales (§3).

Design: docs/superpowers/specs/2026-06-07-scales-library-and-right-axis-pipeline-
design.md §3 (scale navigation & zoom transitions).

The scales-of-nature library is navigated as a CLOSED RING, not a line. A
dedicated endless rotary encoder — relative, vs the absolute 0..4 experience
knobs — walks the ring one step per detent; diving past the smallest scale wraps
around to the largest (the infinite-zoom payoff). Between each adjacent pair of
scales a short pre-baked AI zoom/warp transition plays.

This is the canonical navigation engine (Python-canonical, shared with the Pi
player/firmware); the simulator browser only renders what `advance_ring` returns.

Ordering convention:
  - index 0 = the LARGEST scale (cosmos); increasing index zooms INWARD toward
    the smallest.
  - +1 detent zooms in, -1 zooms out; BOTH wrap (past the smallest wraps to the
    largest, and vice versa).
  - edge i connects scales[i] -> scales[(i+1) % N]; the last edge is the
    small->large wrap seam. One transition clip per edge plays FORWARD when
    zooming inward and REVERSED when zooming outward, so N scales need only N
    transition clips.

Fast spin (§3, "transitions chain, or past a speed threshold a faster blended
pass is used"): a quick encoder spin batches many detents into one advance, so
`abs(delta)` is the input adapter's proxy for spin speed. Past an opt-in
`fast_spin_threshold` the move collapses to a SINGLE blended pass (the
arrival-edge clip, played fast) rather than chaining N full transitions, so a
fast spin stays responsive instead of grinding through every morph. The policy
is opt-in (default off) because this pure function cannot observe wall-clock
spin speed — the input layer (simulator/firmware), which can, enables it.
"""

from __future__ import annotations

from dataclasses import dataclass


# Default spin-speed cutoff for the input layer (simulator/firmware) to enable
# the fast-spin blended pass: 3+ detents batched into one advance() is a fast
# spin crossing several scales; 1-2 are deliberate single steps that chain. Tune
# by eye — it is a UX policy, not canonical ring structure.
DEFAULT_FAST_SPIN_THRESHOLD = 3


class RingError(ValueError):
    """Raised when a ScaleRing is structurally invalid."""


@dataclass(frozen=True)
class Scale:
    """A node on the ring: a scale of nature and the clip(s) it can play.

    A scale carries a rotating POOL of vetted clips (content-pipeline design
    §11.1): when the viewer lands on the scale, the player picks one member at
    random (`pick_clip_id`). `clip_id` is the PRIMARY member (the deterministic
    default / pool[0]); `pool` is the full set. A scale written with only
    `clip_id` and no `pool` is a pool of one — exactly the Increment-1 behavior,
    so old manifests are unchanged.
    """

    id: str
    clip_id: str
    pool: tuple[str, ...] = ()
    audio: str = ""              # per-scale soundtrack, relative to /media/audio/ (audio spec §5.1)

    @property
    def members(self) -> tuple[str, ...]:
        """The clip ids this scale can play — the pool, or just `clip_id` when
        no pool was authored (a pool of one)."""
        return self.pool if self.pool else (self.clip_id,)


@dataclass(frozen=True)
class Transition:
    """A pre-baked zoom/warp morph between TWO specific clips in adjacent
    altitudes.

    `file` plays as-is: the manifest holds both directions as separate, directed
    entries — a forward zoom-in (`<src>__<dst>.mp4`, descending) and its `.rev`
    zoom-out companion (ascending). So a directed `(from_clip, to_clip)` pair maps
    to exactly one file and the renderer needs no reverse logic at play time.
    """

    from_clip: str
    to_clip: str
    file: str
    model: str = ""


@dataclass(frozen=True)
class ScaleRing:
    """A closed ring of scales joined by per-clip-pair morphs.

    Each `Transition` is a directed morph between two clips in adjacent altitudes
    (both directions stored explicitly). A single-scale (or empty-pool) ring has
    no morphs. An empty ring is rejected. The directed `(from_clip, to_clip) ->
    file` lookup is built once for `morph_for`.
    """

    scales: tuple[Scale, ...]
    transitions: tuple[Transition, ...]

    def __post_init__(self) -> None:
        if len(self.scales) == 0:
            raise RingError("a ScaleRing needs at least one scale")
        lookup = {(t.from_clip, t.to_clip): t.file for t in self.transitions}
        object.__setattr__(self, "_morphs", lookup)

    def __len__(self) -> int:
        return len(self.scales)

    def morph_for(self, from_clip: str, to_clip: str) -> str | None:
        """The morph file for the directed clip pair, or None if none was baked."""
        return self._morphs.get((from_clip, to_clip))


@dataclass(frozen=True)
class TransitionStep:
    """One STRUCTURAL step of a move: which edge is crossed, in which direction,
    and the scale index it lands on. The actual morph FILE is resolved later by
    `resolve_move` (it depends on the clip picked for the destination), so this
    step carries no file.

    `blended` marks a fast-spin step (see `advance_ring`'s `fast_spin_threshold`):
    the renderer plays it as one quick accelerated morph rather than at 1x.
    """

    edge: int
    reversed: bool
    to_index: int
    blended: bool = False


@dataclass(frozen=True)
class RingMove:
    """The result of advancing the encoder: where you end up and the ordered
    transitions to play getting there (chained for a multi-detent spin).

    `fast` is True when the move was collapsed to a single blended pass because
    the spin crossed the speed threshold (see `advance_ring`).
    """

    from_index: int
    to_index: int
    steps: tuple[TransitionStep, ...]
    wrapped: bool
    fast: bool = False


def scale_at(ring: ScaleRing, index: int) -> Scale:
    """The scale at `index`, taken modulo the ring length (wraps both ways)."""
    return ring.scales[index % len(ring)]


def pick_clip_id(scale: Scale, r: float) -> str:
    """Pick one clip id from a scale's rotating pool, given an injected uniform
    `r` in [0, 1) (content-pipeline design §11.1).

    Randomness is INJECTED so this stays a pure function — testable and shared
    with the Pi player; the impure draw (`random.random()`) happens once at the
    API/runtime boundary. `r` is clamped into [0, 1) so an off-by-epsilon caller
    never indexes out of range; a pool of one always returns its single member.
    """
    members = scale.members
    r = min(max(r, 0.0), 0.999999)
    return members[int(r * len(members))]


def advance_ring(
    ring: ScaleRing,
    from_index: int,
    delta: int,
    *,
    fast_spin_threshold: int = 0,
) -> RingMove:
    """Walk the endless encoder `delta` detents from `from_index` (signed).

    Returns a `RingMove` with the landing index and the ordered `TransitionStep`s
    to play. Inward steps (+) play edge i forward; outward steps (-) play the
    crossed edge reversed. `wrapped` is True if any step crossed the small<->large
    seam. A degenerate single-scale ring (or delta 0) is a no-op.

    `fast_spin_threshold` (opt-in, default off): when >= 2 and `abs(delta)` meets
    it, the spin is treated as fast and the move collapses to a single blended
    pass — one `TransitionStep` (the arrival edge, marked `blended`) landing
    straight on the destination, with `RingMove.fast=True` — instead of chaining
    every transition. See the module docstring (§3).
    """
    n = len(ring)
    start = from_index % n
    if n == 1 or delta == 0:
        return RingMove(from_index=start, to_index=start, steps=(), wrapped=False)

    direction = 1 if delta > 0 else -1
    steps: list[TransitionStep] = []
    wrapped = False
    index = start
    for _ in range(abs(delta)):
        if direction > 0:
            # zoom inward: edge `index` forward, land at index+1 (mod n)
            edge = index
            nxt = (index + 1) % n
            reversed_ = False
        else:
            # zoom outward: cross edge `index-1` reversed, land at index-1 (mod n)
            edge = (index - 1) % n
            nxt = (index - 1) % n
            reversed_ = True
        if edge == n - 1:
            wrapped = True
        steps.append(
            TransitionStep(edge=edge, reversed=reversed_, to_index=nxt)
        )
        index = nxt

    # Fast spin: play the WHOLE chain quickly (every crossed altitude still
    # morphs, so the chained member->member morphs all show) — each step is just
    # marked `blended` so the renderer accelerates it, rather than collapsing the
    # chain to a single arrival pass.
    if fast_spin_threshold >= 2 and abs(delta) >= fast_spin_threshold:
        fast_steps = tuple(
            TransitionStep(edge=s.edge, reversed=s.reversed,
                           to_index=s.to_index, blended=True)
            for s in steps
        )
        return RingMove(
            from_index=start,
            to_index=index,
            steps=fast_steps,
            wrapped=wrapped,
            fast=True,
        )

    return RingMove(
        from_index=start, to_index=index, steps=tuple(steps), wrapped=wrapped
    )


@dataclass(frozen=True)
class MorphStep:
    """One resolved morph to play while advancing: from the clip currently shown
    to the chosen clip of the next altitude, the morph file, the index it lands
    on, and whether it plays fast (a fast-spin pass). `file` is None if no morph
    was baked for the pair (the renderer falls back to a plain cut)."""

    from_clip: str
    to_clip: str
    file: str | None
    to_index: int
    blended: bool


@dataclass(frozen=True)
class ResolvedMove:
    """A move with each step's destination clip chosen and its morph resolved.
    `target_clip_id` is the final clip the viewer locks onto."""

    steps: tuple[MorphStep, ...]
    target_clip_id: str


def resolve_move(
    ring: ScaleRing,
    move: RingMove,
    from_clip_id: str,
    picks: tuple[float, ...],
) -> ResolvedMove:
    """Choose the destination clip for each crossed altitude (injected `picks`,
    one per step) and resolve the morph from the previous clip to it. The
    destination clip is chosen BEFORE the morph, so the footage matches what we
    land on; the morph direction (zoom-in vs zoom-out) is encoded in the directed
    `(src, dst)` pair, which `morph_for` maps to the right baked file. The final
    `target_clip_id` is the clip the viewer locks onto."""
    if len(picks) != len(move.steps):
        raise ValueError(f"need one pick per step: {len(picks)} != {len(move.steps)}")
    steps: list[MorphStep] = []
    src = from_clip_id
    for st, r in zip(move.steps, picks):
        dst = pick_clip_id(scale_at(ring, st.to_index), r)
        steps.append(MorphStep(
            from_clip=src,
            to_clip=dst,
            file=ring.morph_for(src, dst),
            to_index=st.to_index,
            blended=st.blended,
        ))
        src = dst
    return ResolvedMove(steps=tuple(steps), target_clip_id=src)