benstull/human-experience-filter-art
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(ring,api): pick destination clip then resolve matching chained morph

Browse Source 
Tasks 3-6. Transition is now a directed clip-pair morph; ScaleRing.morph_for
looks up (from,to)->file. advance_ring returns structural steps (no file);
resolve_move picks each crossed altitude's clip THEN its morph, chained, and
reports the locked target. Fast spin keeps the whole chain (blended) instead
of collapsing. /api/ring/advance takes from_clip_id, draws picks, returns the
resolved chain. Dropped ring_move_to_dict + _rev_file.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
BenStullsBets
2026-06-27 07:12:25 -07:00
parent cc469b5298
commit 7855718f74
6 changed files with 326 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
player/ring.py
+88 -39
View File
@@ -74,57 +74,61 @@ class Scale:
@dataclass(frozen=True)
class Transition:
"""A pre-baked zoom/warp morph along ONE ring edge.
"""A pre-baked zoom/warp morph between TWO specific clips in adjacent
altitudes.
`file` plays FORWARD when zooming inward (scales[i] -> scales[i+1] in ring
order); the renderer reverses it for the outward direction, so one clip
covers both directions of an edge.
`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-edge transitions.
"""A closed ring of scales joined by per-clip-pair morphs.
Invariant: a ring of N>=2 scales has exactly N transitions (one per edge,
including the small->large wrap seam). A degenerate single-scale ring has no
edges. An empty ring is rejected.
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:
n = len(self.scales)
if n == 0:
if len(self.scales) == 0:
raise RingError("a ScaleRing needs at least one scale")
expected = 0 if n == 1 else n
if len(self.transitions) != expected:
raise RingError(
f"a {n}-scale ring needs {expected} transitions, "
f"got {len(self.transitions)}"
)
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 transition to play during a move: which edge clip, in which direction,
and the scale index it lands on.
"""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 the single collapsed step of a fast-spin pass (see
`advance_ring`'s `fast_spin_threshold`): the renderer plays it as one quick
accelerated morph straight to the destination instead of a full transition.
`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
file: str
to_index: int
blended: bool = False
@@ -207,31 +211,24 @@ def advance_ring(
if edge == n - 1:
wrapped = True
steps.append(
TransitionStep(
edge=edge,
reversed=reversed_,
file=ring.transitions[edge].file,
to_index=nxt,
)
TransitionStep(edge=edge, reversed=reversed_, to_index=nxt)
)
index = nxt
# Fast spin: collapse the whole chain to a single blended arrival pass. The
# landing index, seam-crossing, and arrival edge/direction are exactly those
# of the full chain — only the in-between transitions are dropped.
# 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:
arrival = steps[-1]
blended = TransitionStep(
edge=arrival.edge,
reversed=arrival.reversed,
file=arrival.file,
to_index=arrival.to_index,
blended=True,
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=(blended,),
steps=fast_steps,
wrapped=wrapped,
fast=True,
)
@@ -239,3 +236,55 @@ def advance_ring(
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)
simulator/app.py
+27 -17
View File
@@ -29,11 +29,18 @@ from player.alteration import (
from player.audio import AUDIO_SOURCES, VISUAL_POSITIONS, resolve_audio, resolve_visual
from player.ring import (
DEFAULT_FAST_SPIN_THRESHOLD,
ResolvedMove,
advance_ring,
pick_clip_id,
resolve_move,
scale_at,
)
from simulator.clips import load_manifest, load_ring, ring_move_to_dict, ring_to_dict
from simulator.clips import (
load_manifest,
load_ring,
resolved_move_to_dict,
ring_to_dict,
)
STATIC_DIR = Path(__file__).parent / "static"
MEDIA_DIR = Path(__file__).parent / "sample_media"
@@ -87,12 +94,6 @@ def _media_version(rel: str) -> Optional[str]:
return token
def _rev_file(file: str) -> str:
"""The baked zoom-out companion path for a transition file (mirrors the
client's `reverseFile`): `<edge>.mp4` -> `<edge>.rev.mp4`."""
return file[:-4] + ".rev.mp4" if file.endswith(".mp4") else file
class ControlsModel(BaseModel):
visual: str
audio: str
@@ -119,6 +120,7 @@ class AlterationRequest(BaseModel):
class RingAdvanceRequest(BaseModel):
from_index: int = 0
delta: int
from_clip_id: str = ""
class AuthorTrackRequest(BaseModel):
@@ -213,14 +215,14 @@ def create_app(manifest_path: Optional[Path] = None) -> FastAPI:
@app.get("/api/media-versions")
def api_media_versions():
"""Per-file content-hash tokens the client appends to /media URLs as
`?v=<hash>`. Covers every served file: each clip's base footage plus each
ring transition and its baked reverse. A re-baked clip's hash changes, so
its URL changes and the browser refetches — a permanent cache-bust."""
`?v=<hash>`. Covers every served file: each clip's base footage plus every
ring transition morph (both directions are explicit entries). A re-baked
clip's hash changes, so its URL changes and the browser refetches — a
permanent cache-bust."""
files = {c.base_file for c in app.state.clips}
if app.state.ring is not None:
for t in app.state.ring.transitions:
files.add(t.file)
files.add(_rev_file(t.file))
versions = {}
for f in sorted(files):
v = _media_version(f)
@@ -244,12 +246,20 @@ def create_app(manifest_path: Optional[Path] = None) -> FastAPI:
req.delta,
fast_spin_threshold=DEFAULT_FAST_SPIN_THRESHOLD,
)
# Rotating pool: pick a random member of the LANDED scale (content-pipeline
# §11.1). A delta=0 advance is the initial / re-roll pick (a no-op move that
# still yields a fresh random clip). The pure pick takes injected randomness.
landed = scale_at(app.state.ring, move.to_index)
chosen = pick_clip_id(landed, random.random())
return ring_move_to_dict(move, app.state.ring, chosen)
if move.steps:
# Choose the destination member of each crossed altitude FIRST (one
# random draw per step — content-pipeline §11.1), then resolve the
# morph from the prior clip to it, so the footage matches what we land
# on. The currently-shown clip threads in as `from_clip_id`.
src = req.from_clip_id or scale_at(app.state.ring, move.from_index).clip_id
picks = tuple(random.random() for _ in move.steps)
resolved = resolve_move(app.state.ring, move, src, picks)
else:
# A no-op move (delta=0) is the initial / re-roll pick: a fresh random
# member of the current scale, no morph.
landed = scale_at(app.state.ring, move.to_index)
resolved = ResolvedMove(steps=(), target_clip_id=pick_clip_id(landed, random.random()))
return resolved_move_to_dict(move, resolved)
@app.post("/api/author/track")
def api_author_track(req: AuthorTrackRequest):
simulator/clips.py
+25 -20
View File
@@ -14,7 +14,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any
from player.ring import RingMove, Scale, ScaleRing, Transition, scale_at
from player.ring import ResolvedMove, RingMove, Scale, ScaleRing, Transition
@dataclass(frozen=True)
@@ -99,7 +99,12 @@ def load_ring(path: str | Path) -> ScaleRing | None:
return None
scales = tuple(_scale_from_dict(s) for s in ring.get("scales", []))
transitions = tuple(
Transition(file=t["file"], model=t.get("model", ""))
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)
@@ -127,35 +132,35 @@ def ring_to_dict(ring: ScaleRing, clips: list[Clip]) -> dict:
}
for s in ring.scales
],
"transitions": [{"file": t.file, "model": t.model} for t in ring.transitions],
"transitions": [
{"from": t.from_clip, "to": t.to_clip, "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.
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 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)."""
`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": chosen_clip_id or scale_at(ring, move.to_index).clip_id,
"target_clip_id": resolved.target_clip_id,
"steps": [
{
"edge": st.edge,
"reversed": st.reversed,
"file": st.file,
"to_index": st.to_index,
"blended": st.blended,
"from_clip": s.from_clip,
"to_clip": s.to_clip,
"file": s.file,
"to_index": s.to_index,
"blended": s.blended,
}
for st in move.steps
for s in resolved.steps
],
}
tests/test_clips.py
+26 -15
View File
@@ -2,12 +2,12 @@ import json
import pytest
from player.ring import RingMove, ScaleRing, TransitionStep, advance_ring
from player.ring import RingMove, ScaleRing, advance_ring, resolve_move
from simulator.clips import (
Clip,
load_manifest,
load_ring,
ring_move_to_dict,
resolved_move_to_dict,
ring_to_dict,
)
@@ -86,9 +86,12 @@ def _ring_manifest_dict():
{"id": "abyss", "clip_id": "abyss"},
],
"transitions": [
{"file": "transitions/cosmos-forest.mp4", "model": "placeholder"},
{"file": "transitions/forest-abyss.mp4", "model": "placeholder"},
{"file": "transitions/abyss-cosmos.mp4", "model": "placeholder"},
{"from": "cosmos", "to": "forest", "file": "transitions/cosmos__forest.mp4", "model": "placeholder"},
{"from": "forest", "to": "cosmos", "file": "transitions/cosmos__forest.rev.mp4", "model": "placeholder"},
{"from": "forest", "to": "abyss", "file": "transitions/forest__abyss.mp4", "model": "placeholder"},
{"from": "abyss", "to": "forest", "file": "transitions/forest__abyss.rev.mp4", "model": "placeholder"},
{"from": "abyss", "to": "cosmos", "file": "transitions/abyss__cosmos.mp4", "model": "placeholder"},
{"from": "cosmos", "to": "abyss", "file": "transitions/abyss__cosmos.rev.mp4", "model": "placeholder"},
],
},
}
@@ -102,7 +105,8 @@ def test_load_ring_builds_a_scale_ring(tmp_path):
assert len(ring) == 3
assert ring.scales[0].id == "cosmos"
assert ring.scales[0].clip_id == "cosmos"
assert ring.transitions[2].file == "transitions/abyss-cosmos.mp4"
assert ring.morph_for("abyss", "cosmos") == "transitions/abyss__cosmos.mp4" # zoom in (wrap)
assert ring.morph_for("cosmos", "forest") == "transitions/cosmos__forest.mp4"
assert ring.transitions[0].model == "placeholder"
@@ -127,32 +131,39 @@ def test_ring_to_dict_carries_scale_titles_from_clips(tmp_path):
assert [s["id"] for s in d["scales"]] == ["cosmos", "forest", "abyss"]
assert d["scales"][0]["title"] == "NASA cosmos"
assert d["scales"][0]["clip_id"] == "cosmos"
assert d["transitions"][2]["file"] == "transitions/abyss-cosmos.mp4"
assert d["transitions"][0] == {
"from": "cosmos", "to": "forest",
"file": "transitions/cosmos__forest.mp4", "model": "placeholder",
}
def test_ring_move_to_dict_serializes_steps(tmp_path):
def test_resolved_move_to_dict_serializes_chained_morphs(tmp_path):
p = tmp_path / "manifest.json"
p.write_text(json.dumps(_ring_manifest_dict()))
ring = load_ring(p)
move = advance_ring(ring, from_index=2, delta=1) # abyss -> wrap -> cosmos
d = ring_move_to_dict(move, ring)
resolved = resolve_move(ring, move, from_clip_id="abyss", picks=(0.0,))
d = resolved_move_to_dict(move, resolved)
assert d["from_index"] == 2
assert d["to_index"] == 0
assert d["wrapped"] is True
assert d["target_clip_id"] == "cosmos"
assert d["steps"][0]["file"] == "transitions/abyss-cosmos.mp4"
assert d["steps"][0]["reversed"] is False
assert d["steps"][0]["from_clip"] == "abyss"
assert d["steps"][0]["to_clip"] == "cosmos"
assert d["steps"][0]["file"] == "transitions/abyss__cosmos.mp4" # zoom in (wrap)
assert d["steps"][0]["to_index"] == 0
assert d["fast"] is False
assert d["steps"][0]["blended"] is False
def test_ring_move_to_dict_marks_fast_blended_pass(tmp_path):
def test_resolved_move_to_dict_marks_fast_blended_chain(tmp_path):
p = tmp_path / "manifest.json"
p.write_text(json.dumps(_ring_manifest_dict()))
ring = load_ring(p)
move = advance_ring(ring, from_index=0, delta=4, fast_spin_threshold=3)
d = ring_move_to_dict(move, ring)
resolved = resolve_move(ring, move, from_clip_id="cosmos",
picks=tuple(0.0 for _ in move.steps))
d = resolved_move_to_dict(move, resolved)
assert d["fast"] is True
assert len(d["steps"]) == 1
assert d["steps"][0]["blended"] is True
assert len(d["steps"]) == 4 # whole chain kept, played fast
assert all(s["blended"] for s in d["steps"])
tests/test_player_ring.py
+125 -55
View File
@@ -9,12 +9,20 @@ from player.ring import (
TransitionStep,
advance_ring,
pick_clip_id,
resolve_move,
scale_at,
)
def _pair(a, b):
"""Both directed morphs for an adjacent clip pair (zoom-in + zoom-out)."""
return (Transition(a, b, f"t/{a}__{b}.mp4"),
Transition(b, a, f"t/{a}__{b}.rev.mp4"))
def _ring():
# cosmos (largest, idx 0) -> forest -> abyss (smallest) -> wraps to cosmos
# cosmos (largest, idx 0) -> forest -> abyss (smallest) -> wraps to cosmos.
# Pool-of-one scales, so each adjacent pair is a single clip pair.
return ScaleRing(
scales=(
Scale(id="cosmos", clip_id="cosmos"),
@@ -22,19 +30,20 @@ def _ring():
Scale(id="abyss", clip_id="abyss"),
),
transitions=(
Transition(file="t/cosmos-forest.mp4"), # edge 0: cosmos->forest
Transition(file="t/forest-abyss.mp4"), # edge 1: forest->abyss
Transition(file="t/abyss-cosmos.mp4"), # edge 2: abyss->cosmos (wrap)
*_pair("cosmos", "forest"),
*_pair("forest", "abyss"),
*_pair("abyss", "cosmos"),
),
)
def test_ring_needs_one_transition_per_scale():
with pytest.raises(RingError):
ScaleRing(
scales=(Scale("a", "a"), Scale("b", "b")),
transitions=(Transition("t/ab.mp4"),), # 1 edge for 2 scales
)
def test_morph_for_returns_directed_file():
r = _ring()
assert r.morph_for("cosmos", "forest") == "t/cosmos__forest.mp4" # zoom in
assert r.morph_for("forest", "cosmos") == "t/cosmos__forest.rev.mp4" # zoom out
# cosmos<->abyss is adjacent via the wrap edge, so it resolves too:
assert r.morph_for("cosmos", "abyss") == "t/abyss__cosmos.rev.mp4"
assert r.morph_for("cosmos", "nope") is None # no such pair baked
def test_ring_rejects_empty():
@@ -42,6 +51,13 @@ def test_ring_rejects_empty():
ScaleRing(scales=(), transitions=())
def test_ring_allows_any_transition_count():
# The old "N scales need N transitions" rule is gone — the per-clip-pair model
# has a variable count, and a ring with no morphs is valid (plain cuts).
r = ScaleRing(scales=(Scale("a", "a"), Scale("b", "b")), transitions=())
assert r.morph_for("a", "b") is None
def test_scale_at_wraps_modulo():
r = _ring()
assert scale_at(r, 0).id == "cosmos"
@@ -50,27 +66,23 @@ def test_scale_at_wraps_modulo():
assert scale_at(r, 4).id == "forest"
def test_advance_one_step_inward_plays_edge_forward():
def test_advance_one_step_inward_crosses_edge_forward():
r = _ring()
move = advance_ring(r, from_index=0, delta=1)
assert isinstance(move, RingMove)
assert move.from_index == 0
assert move.to_index == 1
assert move.wrapped is False
assert move.steps == (
TransitionStep(edge=0, reversed=False, file="t/cosmos-forest.mp4", to_index=1),
)
assert move.steps == (TransitionStep(edge=0, reversed=False, to_index=1),)
def test_advance_one_step_outward_plays_edge_reversed():
def test_advance_one_step_outward_crosses_edge_reversed():
r = _ring()
move = advance_ring(r, from_index=1, delta=-1)
assert move.to_index == 0
assert move.wrapped is False
# going forest(1) -> cosmos(0) is edge 0 played in reverse
assert move.steps == (
TransitionStep(edge=0, reversed=True, file="t/cosmos-forest.mp4", to_index=0),
)
# going forest(1) -> cosmos(0) is edge 0 crossed in reverse
assert move.steps == (TransitionStep(edge=0, reversed=True, to_index=0),)
def test_advance_inward_past_smallest_wraps_to_largest():
@@ -79,9 +91,7 @@ def test_advance_inward_past_smallest_wraps_to_largest():
move = advance_ring(r, from_index=2, delta=1)
assert move.to_index == 0
assert move.wrapped is True
assert move.steps == (
TransitionStep(edge=2, reversed=False, file="t/abyss-cosmos.mp4", to_index=0),
)
assert move.steps == (TransitionStep(edge=2, reversed=False, to_index=0),)
def test_advance_outward_past_largest_wraps_to_smallest():
@@ -90,19 +100,17 @@ def test_advance_outward_past_largest_wraps_to_smallest():
move = advance_ring(r, from_index=0, delta=-1)
assert move.to_index == 2
assert move.wrapped is True
assert move.steps == (
TransitionStep(edge=2, reversed=True, file="t/abyss-cosmos.mp4", to_index=2),
)
assert move.steps == (TransitionStep(edge=2, reversed=True, to_index=2),)
def test_advance_multi_detent_chains_transitions():
def test_advance_multi_detent_chains_steps():
r = _ring()
move = advance_ring(r, from_index=0, delta=2)
assert move.to_index == 2
assert move.wrapped is False
assert move.steps == (
TransitionStep(edge=0, reversed=False, file="t/cosmos-forest.mp4", to_index=1),
TransitionStep(edge=1, reversed=False, file="t/forest-abyss.mp4", to_index=2),
TransitionStep(edge=0, reversed=False, to_index=1),
TransitionStep(edge=1, reversed=False, to_index=2),
)
@@ -125,7 +133,6 @@ def test_advance_zero_is_a_noop():
def test_advance_normalizes_from_index():
r = _ring()
# from_index out of range is taken modulo N first
move = advance_ring(r, from_index=3, delta=1)
assert move.from_index == 0
assert move.to_index == 1
@@ -142,10 +149,10 @@ def test_degenerate_single_scale_ring_is_a_noop():
# --- fast-spin: a faster blended pass past a speed threshold (scales design §3) ---
#
# A fast encoder spin batches many detents into one advance() call, so |delta| is
# the input adapter's proxy for spin speed. Past a configured threshold the move
# collapses to a SINGLE blended pass (the arrival-edge clip, played fast) instead
# of chaining N full transitions. The policy is opt-in (default off) because the
# pure function cannot know wall-clock spin speed — the input layer enables it.
# the input adapter's proxy for spin speed. Past a configured threshold every step
# is marked `blended` so the renderer plays the WHOLE chain fast (each crossed
# altitude still morphs). The policy is opt-in (default off) because the pure
# function cannot know wall-clock spin speed — the input layer enables it.
def test_fast_spin_disabled_by_default_chains_every_detent():
@@ -156,46 +163,40 @@ def test_fast_spin_disabled_by_default_chains_every_detent():
assert all(not s.blended for s in move.steps)
def test_fast_spin_below_threshold_still_chains():
def test_fast_spin_below_threshold_still_chains_unblended():
r = _ring()
move = advance_ring(r, from_index=0, delta=2, fast_spin_threshold=3)
assert move.fast is False
assert len(move.steps) == 2
assert move.steps[0].edge == 0 and move.steps[1].edge == 1
assert all(not s.blended for s in move.steps)
def test_fast_spin_at_or_above_threshold_collapses_to_one_blended_step():
def test_fast_spin_at_or_above_threshold_keeps_all_steps_blended():
r = _ring()
# 4 inward detents from cosmos(0): 0->1->2->0->1, lands on forest(1),
# crosses the abyss->cosmos seam, last edge crossed is edge 0 (forward).
# crosses the abyss->cosmos seam. Every step is kept, marked blended.
move = advance_ring(r, from_index=0, delta=4, fast_spin_threshold=3)
assert move.fast is True
assert move.to_index == 1
assert move.wrapped is True
assert move.steps == (
TransitionStep(
edge=0, reversed=False, file="t/cosmos-forest.mp4", to_index=1, blended=True
),
)
assert len(move.steps) == 4
assert all(s.blended for s in move.steps)
# structural path preserved (not collapsed)
assert [s.edge for s in move.steps] == [0, 1, 2, 0]
def test_fast_spin_outward_blended_step_is_reversed():
def test_fast_spin_outward_keeps_all_steps_blended_and_reversed():
r = _ring()
# 4 outward detents from cosmos(0): 0->2->1->0->2, lands on abyss(2);
# outward arrival crosses edge 2 (abyss->cosmos) reversed.
move = advance_ring(r, from_index=0, delta=-4, fast_spin_threshold=3)
assert move.fast is True
assert move.to_index == 2
assert move.wrapped is True
assert move.steps == (
TransitionStep(
edge=2, reversed=True, file="t/abyss-cosmos.mp4", to_index=2, blended=True
),
)
assert len(move.steps) == 4
assert all(s.blended and s.reversed for s in move.steps)
def test_fast_spin_threshold_below_two_is_ignored():
# A threshold of 0 or 1 would blend even single deliberate steps — treat as off.
r = _ring()
move = advance_ring(r, from_index=0, delta=3, fast_spin_threshold=1)
assert move.fast is False
@@ -215,7 +216,6 @@ def test_fast_spin_noop_on_degenerate_ring():
def test_scale_without_pool_is_a_pool_of_one():
s = Scale(id="cosmos", clip_id="cosmos")
assert s.members == ("cosmos",)
# any r maps to the single member
assert pick_clip_id(s, 0.0) == "cosmos"
assert pick_clip_id(s, 0.99) == "cosmos"
@@ -227,7 +227,6 @@ def test_scale_with_pool_exposes_all_members():
def test_pick_clip_id_selects_member_by_injected_r():
s = Scale(id="coast", clip_id="a", pool=("a", "b", "c", "d"))
# r in [0,1) partitions the pool into len equal buckets
assert pick_clip_id(s, 0.0) == "a"
assert pick_clip_id(s, 0.24) == "a"
assert pick_clip_id(s, 0.25) == "b"
@@ -237,13 +236,84 @@ def test_pick_clip_id_selects_member_by_injected_r():
def test_pick_clip_id_clamps_out_of_range_r():
s = Scale(id="coast", clip_id="a", pool=("a", "b", "c"))
assert pick_clip_id(s, -5.0) == "a" # below 0 clamps to first
assert pick_clip_id(s, 1.0) == "c" # 1.0 must not index past the end
assert pick_clip_id(s, -5.0) == "a"
assert pick_clip_id(s, 1.0) == "c"
assert pick_clip_id(s, 999.0) == "c"
def test_scale_carries_an_optional_audio_path():
s = Scale(id="coast", clip_id="coast_birdrock", audio="coast/waves.loop.mp3")
assert s.audio == "coast/waves.loop.mp3"
# back-compat: audio defaults to empty
assert Scale(id="x", clip_id="x").audio == ""
# --- resolve_move: pick the destination clip THEN the matching morph, chained ---
def _chain_ring():
# 3 scales, pools of 2/1/2; descending edges cosmos->coast->abyss + wrap.
scales = (Scale("cosmos", "c1", ("c1", "c2")),
Scale("coast", "o1", ("o1",)),
Scale("abyss", "a1", ("a1", "a2")))
t = []
for h in ("c1", "c2"):
t += list(_pair(h, "o1")) # cosmos<->coast
for l in ("a1", "a2"):
t += list(_pair("o1", l)) # coast<->abyss
for h in ("a1", "a2"):
t += list(_pair(h, "c1")) # abyss<->cosmos wrap (to c1)
t += list(_pair(h, "c2")) # abyss<->cosmos wrap (to c2)
return ScaleRing(scales=scales, transitions=tuple(t))
def test_resolve_single_step_zoom_in():
r = _chain_ring()
move = advance_ring(r, from_index=0, delta=1) # cosmos -> coast
res = resolve_move(r, move, from_clip_id="c2", picks=(0.0,))
assert len(res.steps) == 1
s = res.steps[0]
assert (s.from_clip, s.to_clip) == ("c2", "o1")
assert s.file == "t/c2__o1.mp4"
assert res.target_clip_id == "o1"
def test_resolve_chains_through_intermediate_altitude():
r = _chain_ring()
move = advance_ring(r, from_index=0, delta=2) # cosmos -> coast -> abyss
# coast member (only o1) then abyss member: 0.6*2 -> index 1 -> a2
res = resolve_move(r, move, from_clip_id="c1", picks=(0.0, 0.6))
files = [s.file for s in res.steps]
assert files == ["t/c1__o1.mp4", "t/o1__a2.mp4"]
assert res.target_clip_id == "a2"
def test_resolve_zoom_out_uses_reverse_file():
r = _chain_ring()
move = advance_ring(r, from_index=2, delta=-1) # abyss -> coast (ascend)
res = resolve_move(r, move, from_clip_id="a1", picks=(0.0,))
assert res.steps[0].file == "t/o1__a1.rev.mp4"
assert res.target_clip_id == "o1"
def test_resolve_empty_move_keeps_source():
r = _chain_ring()
move = advance_ring(r, from_index=0, delta=0)
res = resolve_move(r, move, from_clip_id="c2", picks=())
assert res.steps == () and res.target_clip_id == "c2"
def test_resolve_requires_one_pick_per_step():
r = _chain_ring()
move = advance_ring(r, from_index=0, delta=2)
with pytest.raises(ValueError):
resolve_move(r, move, from_clip_id="c1", picks=(0.0,))
def test_resolve_missing_morph_yields_none_file():
# A pair with no baked morph resolves to file=None (renderer plain-cuts).
r = ScaleRing(scales=(Scale("a", "a1", ("a1",)), Scale("b", "b1", ("b1",))),
transitions=())
move = advance_ring(r, from_index=0, delta=1)
res = resolve_move(r, move, from_clip_id="a1", picks=(0.0,))
assert res.steps[0].file is None
assert res.target_clip_id == "b1"
tests/test_simulator_api.py
+35 -14
View File
@@ -200,9 +200,12 @@ def ring_manifest_path(tmp_path):
{"id": "abyss", "clip_id": "abyss", "audio": "abyss/whale.loop.mp3"},
],
"transitions": [
{"file": "transitions/cosmos-forest.mp4", "model": "placeholder"},
{"file": "transitions/forest-abyss.mp4", "model": "placeholder"},
{"file": "transitions/abyss-cosmos.mp4", "model": "placeholder"},
{"from": "cosmos", "to": "forest", "file": "transitions/cosmos__forest.mp4", "model": "placeholder"},
{"from": "forest", "to": "cosmos", "file": "transitions/cosmos__forest.rev.mp4", "model": "placeholder"},
{"from": "forest", "to": "abyss", "file": "transitions/forest__abyss.mp4", "model": "placeholder"},
{"from": "abyss", "to": "forest", "file": "transitions/forest__abyss.rev.mp4", "model": "placeholder"},
{"from": "abyss", "to": "cosmos", "file": "transitions/abyss__cosmos.mp4", "model": "placeholder"},
{"from": "cosmos", "to": "abyss", "file": "transitions/abyss__cosmos.rev.mp4", "model": "placeholder"},
],
},
}))
@@ -220,7 +223,7 @@ def test_ring_returns_scales_and_transitions(ring_client):
data = resp.json()
assert [s["id"] for s in data["scales"]] == ["cosmos", "forest", "abyss"]
assert data["scales"][0]["title"] == "NASA cosmos"
assert len(data["transitions"]) == 3
assert len(data["transitions"]) == 6 # 3 edges x 2 directions (clip-pair morphs)
def test_ring_exposes_each_scales_audio(ring_client):
@@ -233,14 +236,27 @@ def test_ring_exposes_each_scales_audio(ring_client):
def test_ring_advance_inward_one_step(ring_client):
resp = ring_client.post("/api/ring/advance", json={"from_index": 0, "delta": 1})
resp = ring_client.post("/api/ring/advance",
json={"from_index": 0, "delta": 1, "from_clip_id": "cosmos"})
assert resp.status_code == 200
data = resp.json()
assert data["to_index"] == 1
assert data["target_clip_id"] == "forest"
assert data["wrapped"] is False
assert data["steps"][0]["file"] == "transitions/cosmos-forest.mp4"
assert data["steps"][0]["reversed"] is False
step = data["steps"][0]
assert step["from_clip"] == "cosmos" # threaded current clip
assert step["to_clip"] == "forest" # picked destination
assert step["file"] == "transitions/cosmos__forest.mp4" # morph matches landing
def test_ring_advance_picks_then_returns_matching_morph(ring_client):
# The morph's `to_clip` always equals the locked target, and its file is the
# directed morph for (from_clip -> to_clip).
data = ring_client.post("/api/ring/advance",
json={"from_index": 0, "delta": 1, "from_clip_id": "cosmos"}).json()
step = data["steps"][-1]
assert step["to_clip"] == data["target_clip_id"]
assert step["file"] == f"transitions/cosmos__{data['target_clip_id']}.mp4"
def test_ring_advance_wraps_smallest_to_largest(ring_client):
@@ -251,11 +267,15 @@ def test_ring_advance_wraps_smallest_to_largest(ring_client):
assert data["wrapped"] is True
def test_ring_advance_outward_reverses_edge(ring_client):
resp = ring_client.post("/api/ring/advance", json={"from_index": 1, "delta": -1})
def test_ring_advance_outward_plays_reverse_morph(ring_client):
resp = ring_client.post("/api/ring/advance",
json={"from_index": 1, "delta": -1, "from_clip_id": "forest"})
data = resp.json()
assert data["to_index"] == 0
assert data["steps"][0]["reversed"] is True
# forest(1) -> cosmos(0) is a zoom-out: the directed (forest, cosmos) morph
# resolves to the .rev companion.
assert data["steps"][0]["file"] == "transitions/cosmos__forest.rev.mp4"
assert data["target_clip_id"] == "cosmos"
def test_ring_advance_small_delta_chains_not_fast(ring_client):
@@ -266,13 +286,14 @@ def test_ring_advance_small_delta_chains_not_fast(ring_client):
assert len(data["steps"]) == 2
def test_ring_advance_fast_spin_collapses_to_one_blended_pass(ring_client):
# A fast spin (4 detents batched) crosses the threshold -> one blended pass.
def test_ring_advance_fast_spin_plays_whole_chain_blended(ring_client):
# A fast spin (4 detents batched) crosses the threshold -> the whole chain is
# kept and played fast (each step blended), not collapsed to one pass.
resp = ring_client.post("/api/ring/advance", json={"from_index": 0, "delta": 4})
data = resp.json()
assert data["fast"] is True
assert len(data["steps"]) == 1
assert data["steps"][0]["blended"] is True
assert len(data["steps"]) == 4
assert all(s["blended"] for s in data["steps"])
assert data["to_index"] == 1