#!/usr/bin/env python3 """Load the transition library and turn one recipe into ffmpeg filter fragments.""" from __future__ import annotations import json import re from dataclasses import dataclass from functools import lru_cache from pathlib import Path from skill_config import TRANSITIONS LIBRARY_PATH = TRANSITIONS / "transition_library.json" @dataclass(frozen=True) class Recipe: id: str label: str family: str energy: int frames_min: int frames_max: int frames_default: int when: str avoid: str ffmpeg: dict sfx: tuple[dict, ...] hyperframes: dict max_per_minute: int needs_speech_pause: bool @property def is_cut(self) -> bool: return self.ffmpeg.get("kind") == "cut" def duration(self, fps: int, frames: int | None = None) -> float: count = frames if frames is not None else self.frames_default count = max(self.frames_min, min(self.frames_max, count)) return count / max(1, fps) def as_dict(self) -> dict: return { "id": self.id, "label": self.label, "family": self.family, "energy": self.energy, "frames": {"min": self.frames_min, "max": self.frames_max, "default": self.frames_default}, "when": self.when, "avoid": self.avoid, "sfx": list(self.sfx), "max_per_minute": self.max_per_minute, "needs_speech_pause": self.needs_speech_pause, } @lru_cache(maxsize=1) def load_library(path: str | None = None) -> dict: target = Path(path) if path else LIBRARY_PATH if not target.exists(): raise SystemExit(f"Transition library not found: {target}") return json.loads(target.read_text(encoding="utf-8")) @lru_cache(maxsize=1) def recipes(path: str | None = None) -> dict[str, Recipe]: library = load_library(path) out: dict[str, Recipe] = {} for item in library["transitions"]: frames = item["frames"] out[item["id"]] = Recipe( id=item["id"], label=item["label"], family=item["family"], energy=int(item["energy"]), frames_min=int(frames["min"]), frames_max=int(frames["max"]), frames_default=int(frames["default"]), when=item["when"], avoid=item["avoid"], ffmpeg=item["ffmpeg"], sfx=tuple(item["sfx"]), hyperframes=item["hyperframes"], max_per_minute=int(item["max_per_minute"]), needs_speech_pause=bool(item.get("needs_speech_pause")), ) return out def recipe(transition_id: str) -> Recipe: table = recipes() if transition_id not in table: raise SystemExit( f"Unknown transition '{transition_id}'. Known: {', '.join(sorted(table))}" ) return table[transition_id] # Slot 0 of the expression state holds forward progress; recipes never touch it. PROGRESS_SLOT = 0 def expand_samples(expression: str) -> str: """Expand `A@(x, y)` / `B@(x, y)` into plane-correct accessor chains. Two traps in ffmpeg's `xfade=transition=custom` that this hides: 1. `a0(x, y)` always reads *plane 0*. Used verbatim, the chroma planes get luma values and the whole transition comes out green and magenta. The accessor has to be chosen from `PLANE`. 2. `P` counts **down** from 1 to 0 over the transition — it is the weight of the outgoing frame, not elapsed progress. Recipes are written with `Q` as forward progress (0 at the cut's start, 1 at its end) and `Q` is substituted for `1-P` here, so nobody has to remember the inversion. """ # Scanned rather than matched with a regex: the arguments contain nested # parentheses ("A@((X-W/2)/ld(1)+W/2, ...)") and a lazy pattern stops at the # first inner ")", producing a one-argument call that fails at render time. out = expression index = 0 while True: found = -1 for marker in ("A@(", "B@("): position = out.find(marker, index) if position >= 0 and (found < 0 or position < found): found = position if found < 0: break source = out[found].lower() depth = 0 end = -1 for scan in range(found + 2, len(out)): if out[scan] == "(": depth += 1 elif out[scan] == ")": depth -= 1 if depth == 0: end = scan break if end < 0: raise SystemExit(f"Unbalanced parentheses after {source.upper()}@ in: {expression}") parts = split_arguments(out[found + 3:end]) if len(parts) != 2: raise SystemExit( f"{source.upper()}@ needs exactly two arguments, got {len(parts)}: " f"{out[found:end + 1]}" ) x, y = (part.strip() for part in parts) replacement = ( f"if(eq(PLANE,0),{source}0({x},{y})," f"if(eq(PLANE,1),{source}1({x},{y}),{source}2({x},{y})))" ) out = out[:found] + replacement + out[end + 1:] # Resume just past the start of the rewritten call. The replacement holds # only a0/a1/a2 calls, so nothing here is re-expanded, while any A@/B@ that # was nested inside the arguments is still found on the next pass. index = found + 1 expanded = re.sub(r"\bQ\b", f"ld({PROGRESS_SLOT})", out) return f"st({PROGRESS_SLOT},1-P);" + expanded def split_arguments(text: str) -> list[str]: """Split on the top-level comma only — arguments contain nested calls.""" depth = 0 parts: list[str] = [] current = "" for char in text: if char == "(": depth += 1 elif char == ")": depth -= 1 if char == "," and depth == 0: parts.append(current) current = "" continue current += char parts.append(current) return parts def xfade_expression(spec: dict) -> str | None: """The ready-to-use `expr` for a custom xfade, if the recipe uses one.""" if spec.get("kind") != "xfade_custom": return None return expand_samples(spec.get("expr", "if(gt(Q,0.5), B@(X,Y), A@(X,Y))")) def needs_full_chroma(spec: dict) -> bool: """Custom expressions must run at 4:4:4. `X`/`Y` iterate over the current plane while `W`/`H` stay the frame size, so any geometry written against `W`/`H` samples far outside a subsampled chroma plane and clamps to its edge. Upsampling for the duration of the transition removes the discrepancy instead of scattering /2 factors through every recipe. """ return spec.get("kind") == "xfade_custom" def xfade_arguments(spec: dict, duration: float, offset: float) -> str: """Build the `xfade=...` filter arguments for one transition.""" kind = spec.get("kind") if kind == "xfade": name = spec.get("transition", "fade") return f"xfade=transition={name}:duration={duration:.4f}:offset={offset:.4f}" if kind == "xfade_custom": expression = xfade_expression(spec) return ( f"xfade=transition=custom:expr='{expression}'" f":duration={duration:.4f}:offset={offset:.4f}" ) raise SystemExit(f"Recipe kind '{kind}' has no xfade form")