Operations

Every editing primitive in videopython is an Operation subclass — a Pydantic BaseModel whose fields ARE the JSON wire format. Subclasses auto-register via __pydantic_init_subclass__, so importing videopython.editing (or videopython.ai) populates the registry. The registry is what VideoEdit.json_schema() uses to build the discriminated-union schema for LLM-driven plan generation.

Subclass Contract

from typing import ClassVar, Literal

import numpy as np
from pydantic import Field

from videopython.editing import Operation, OpCategory, FilterCtx
from videopython.base.video import VideoMetadata


class Resize(Operation):
    """Resize the video.

    Args:
        width: Target width in pixels.
        height: Target height in pixels.
    """

    op: Literal["resize"] = "resize"            # discriminator + registry key
    category: ClassVar[OpCategory] = OpCategory.TRANSFORM

    width: int | None = Field(None, gt=0)
    height: int | None = Field(None, gt=0)

    def predict_metadata(self, meta: VideoMetadata) -> VideoMetadata: ...
    def to_ffmpeg_filter(self, ctx: FilterCtx) -> str | None: ...   # filter-compiled transforms

There is no apply(). Operations execute only through VideoEdit's streaming engine (run_to_file); they never run against a Video directly. A subclass implements:

predict_metadata(self, meta) -> VideoMetadata — predict the output VideoMetadata and fail fast on plans that would crash at run time. Defaults to identity (override on the base Operation; on Effect it is identity, since effects preserve shape and frame count).
either to_ffmpeg_filter(self, ctx) (and to_ffmpeg_audio_filter for a duration-changing transform's audio twin) — for ops compiled into the ffmpeg filter chain — or streaming_init(self, total_frames, fps, width, height, **context) + process_frame(self, frame, frame_index) — for per-frame Python effects.

Notes:

op is a one-value Literal field (not a ClassVar). It flows into the JSON wire as the discriminator and is also the registry key.
category is OpCategory.TRANSFORM, OpCategory.EFFECT, or OpCategory.SPECIAL.
Every registered op is streamable, decided structurally by op.streams() (there is no streamable flag): a transform streams iff it implements to_ffmpeg_filter; an effect iff it implements process_frame + streaming_init (a frame effect) or to_ffmpeg_filter + compiles_to_filter (a filter effect).
internal_only: ClassVar[bool] = False, when True, keeps an op OUT of the registry — constructed directly by the engine, never a chain op. cut/ cut_frames use it, since trimming is the segment's own start/end.
Context-dependent ops declare requires: ClassVar[tuple[str, ...]] = ("transcription",). The runner picks the matching keys out of the context dict passed to run_to_file(..., context=...), re-bases any time-based values onto the segment's local timeline, and threads them into the effect's streaming_init (and predict_metadata) as keyword arguments — or onto the FilterCtx.context for a filter-compiled op.

Effects

Effect(Operation) adds a window: TimeRange | None field and preserves shape and frame count (so its predict_metadata is identity). The streaming engine resolves window against the segment timeline, leaving frames outside the window untouched. A frame effect implements the streaming_init / process_frame pair:

class Glitch(Effect):  # a frame effect: no faithful ffmpeg form
    op: Literal["glitch"] = "glitch"
    # ... fields ...

    def process_frame(self, frame: np.ndarray, frame_index: int) -> np.ndarray: ...

The window field on the wire:

{"op": "blur_effect", "mode": "constant", "iterations": 2, "window": {"start": 1.0, "stop": 3.0}}

The two text-rendering effects instead compile to a native filter (no per-frame Python) by setting the compiles_to_filter property and implementing to_ffmpeg_filter: add_subtitles (libass subtitles=) and text_overlay (drawtext). Audio-coupled effects (Fade, VolumeAdjust) add to_ffmpeg_audio_filter for their audio twin while their video runs per-frame. Every other (pixel) effect runs vectorised numpy/cv2 in process_frame: benchmarks showed compiling them to ffmpeg filters bought at best ~1.1–1.4x (from skipping the rawvideo round-trip, not faster compute) and sometimes lost, so the engine reserves filters for geometry/timing transforms and text rendering.

Registry API

from videopython.editing import Operation

# Snapshot of {op_id: subclass} for every registered operation:
Operation.registry()

# LLM-safe subset: only ops with llm_exposed=True (omits server-only ops):
Operation.llm_registry()

# Look up by op_id (raises KeyError if unknown):
cls = Operation.get("resize")

# Discriminated-union JSON Schema over the LLM-exposed ops:
schema = Operation.json_schema()
# ...or over every registered op (worker / from_dict path):
full = Operation.json_schema(include_server_only=True)

AI operations register lazily, so call import videopython.ai before inspecting the registry if you need face_crop and friends.

LLM-exposed vs server-only ops

Every Operation carries llm_exposed: ClassVar[bool] = True. Set it to False for ops the model must never emit — typically ops that need a server-resolved source path (image_overlay, full_image_overlay). Operation.llm_registry() and the default Operation.json_schema() / VideoEdit.json_schema() cover only llm_exposed ops, while Operation.registry() and from_dict still see all ops so a stored plan continues to execute.

The same idea applies at the field level: a field declared with Field(json_schema_extra={"llm_hidden": True}) is a valid wire field (it still parses and runs) but is dropped from the LLM-facing schema. This hides advanced overrides the model shouldn't fill in — e.g. the raw font_filename path on text_overlay/add_subtitles, whose LLM-facing counterpart is the font name enum. The default Operation.json_schema() and cls.llm_json_schema() (below) strip these; cls.model_json_schema() keeps them.

Discovering Operations

from videopython.editing import Operation, OpCategory

for op_id, cls in Operation.registry().items():
    print(f"{op_id}: {cls.__doc__.splitlines()[0]}")

transforms = {k: v for k, v in Operation.registry().items()
              if v.category is OpCategory.TRANSFORM}

Per-Operation JSON Schema

Every subclass exposes cls.model_json_schema() (standard Pydantic), returning the JSON Schema for that specific op's fields. For an LLM-facing single-op schema, use cls.llm_json_schema() — identical but with llm_hidden fields stripped:

from videopython.editing import Operation

cls = Operation.get("blur_effect")
schema = cls.model_json_schema()         # full (all fields)
llm_schema = cls.llm_json_schema()       # LLM-facing (llm_hidden dropped)
# {
#   "properties": {
#     "op": {"const": "blur_effect", ...},
#     "mode": {"enum": ["constant", "ascending", "descending"], ...},
#     "iterations": {"type": "integer", "minimum": 1, ...},
#     "window": {"anyOf": [{"$ref": "..."}, {"type": "null"}], ...},
#     ...
#   },
#   ...
# }

Operation.json_schema() is the union over the LLM-exposed ops (pass include_server_only=True for all of them), and that's the schema VideoEdit.json_schema() embeds for the operations field.

Registered Operations

Base (no AI dependencies)

cut/cut_frames are internal-only: the engine trims each segment via its start/end, so they are not chain ops and do not appear here. Every registered op below is streamable (it compiles to an ffmpeg filter or is a per-frame effect).

ID	Class	Category	Streamable
`resize`	`Resize`	transform	yes
`resample_fps`	`ResampleFPS`	transform	yes
`crop`	`Crop`	transform	yes
`speed_change`	`SpeedChange`	transform	yes — compiles to `setpts` + CFR resample; audio time-stretched in sync
`freeze_frame`	`FreezeFrame`	transform	yes — compiles to a `loop`-based chain; silence inserted in the audio
`silence_removal`	`SilenceRemoval`	transform	yes — `select` keep-window cut (requires `transcription` context)
`blur_effect`	`Blur`	effect	yes
`zoom_effect`	`Zoom`	effect	yes
`color_adjust`	`ColorGrading`	effect	yes
`vignette`	`Vignette`	effect	yes
`ken_burns`	`KenBurns`	effect	yes
`full_image_overlay` †	`FullImageOverlay`	effect	yes
`image_overlay` †	`ImageOverlay`	effect	yes
`fade`	`Fade`	effect	yes
`volume_adjust`	`VolumeAdjust`	effect	yes
`text_overlay`	`TextOverlay`	effect	yes
`add_subtitles`	`TranscriptionOverlay`	effect	yes — compiles to a libass `subtitles=` filter (requires `transcription` context)
`shake`	`Shake`	effect	yes
`punch_in`	`PunchIn`	effect	yes
`flash`	`Flash`	effect	yes
`chromatic_aberration`	`ChromaticAberration`	effect	yes
`glitch`	`Glitch`	effect	yes
`film_grain`	`FilmGrain`	effect	yes
`sharpen`	`Sharpen`	effect	yes
`pixelate`	`Pixelate`	effect	yes
`mirror_flip`	`MirrorFlip`	effect	yes
`kaleidoscope`	`Kaleidoscope`	effect	yes

† Server-only (llm_exposed=False): excluded from Operation.llm_registry() and the default LLM-facing schema because they need a server-resolved source path. Still executable via from_dict / Operation.registry().

AI (require `import videopython.ai`)

ID	Class	Category	Streamable
`face_crop`	`FaceTrackingCrop`	transform	yes — compile-time detection pass drives a per-frame crop track
`object_detection_overlay`	`ObjectDetectionOverlay`	effect	yes — per-frame box overlay; D-FINE detection on a `detection_interval` cadence; bounded memory, not bounded compute

API Reference

Operation

Bases: BaseModel

Pydantic base for every editing primitive.

Concrete subclasses MUST declare an op field with a single-value Literal[str] annotation; that value is the discriminator on the JSON wire and the registry key. Subclasses may override the category and requires ClassVars.

predict_metadata defaults to identity; to_ffmpeg_filter defaults to None (no filter compilation).

Source code in src/videopython/editing/operation.py

class Operation(BaseModel):
    """Pydantic base for every editing primitive.

    Concrete subclasses MUST declare an ``op`` field with a single-value
    ``Literal[str]`` annotation; that value is the discriminator on the JSON
    wire and the registry key. Subclasses may override the ``category`` and
    ``requires`` ClassVars.

    ``predict_metadata`` defaults to identity; ``to_ffmpeg_filter`` defaults to
    ``None`` (no filter compilation).
    """

    model_config = ConfigDict(extra="forbid", validate_assignment=True)

    op: str

    category: ClassVar[OpCategory] = OpCategory.SPECIAL
    requires: ClassVar[tuple[str, ...]] = ()
    compiles_from_source: ClassVar[bool] = False
    """Whether the op's filter compile decodes the source itself (face_crop's
    detection pass). Such ops cannot sit at the encode stage -- the frames
    behind per-frame Python effects are not reproducible at compile time --
    so both the plan builder and the streamability report reject them as
    UNSTREAMABLE there."""
    changes_duration: ClassVar[bool] = False
    """Whether the op's output duration differs from its input (speed, freeze).

    The streaming plan builder folds ``predict_metadata`` through the chain
    either way; this flag additionally gates time-based *context*: a
    context-consuming op scheduled after a duration-changing transform would
    receive timestamps on the wrong timeline, so such plans are rejected as
    UNSTREAMABLE until context re-mapping exists. The streamability report
    mirrors the same rule.
    """
    llm_exposed: ClassVar[bool] = True
    internal_only: ClassVar[bool] = False
    """Whether this op is engine-internal and must NOT be a chain op.

    ``CutSeconds``/``CutFrames`` trim a segment, but trimming is the segment's own
    ``start``/``end`` mechanism -- the engine constructs them directly. They have
    no ffmpeg filter and no ``process_frame``, so this flag keeps them OUT of the
    registry: they cannot appear in a plan's ``operations`` list or the LLM
    schema, while direct construction (``CutSeconds(start=..., end=...)``) still
    works. Default False (a normal chain op)."""
    time_fields: ClassVar[tuple[BoundedTimeField, ...]] = ()
    """Time-valued (seconds) fields :meth:`VideoEdit.repair` may clamp into range.

    Declaring a :class:`BoundedTimeField` here lets ``repair`` clamp an
    out-of-range timestamp (e.g. ``freeze_frame.timestamp`` past the clip end)
    without per-op special-casing -- the repair pass reads the declaration,
    clamps to ``[0, bound]``, and records a :class:`PlanRepair`. Empty by
    default; ops with no time-valued params declare nothing.
    """

    _registry: ClassVar[dict[str, type[Operation]]] = {}

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)

        op_field = cls.model_fields.get("op")
        if op_field is None:
            return
        annotation = op_field.annotation
        if get_origin(annotation) is not Literal:
            # Abstract intermediate (e.g. Effect) -- no concrete op_id yet.
            return
        literal_values = get_args(annotation)
        if len(literal_values) != 1 or not isinstance(literal_values[0], str):
            raise TypeError(f"{cls.__name__}.op must be Literal of a single str, got {literal_values!r}")
        op_id = literal_values[0]

        if cls.internal_only:
            # Engine-internal op (CutSeconds/CutFrames): the op Literal is still
            # validated above, but the op is kept out of the registry so it
            # cannot be resolved as a chain op or exposed to the LLM.
            return

        existing = Operation._registry.get(op_id)
        if existing is not None and existing is not cls:
            raise ValueError(
                f"Duplicate op_id '{op_id}': "
                f"{cls.__module__}.{cls.__qualname__} vs "
                f"{existing.__module__}.{existing.__qualname__}"
            )
        Operation._registry[op_id] = cls

    @property
    def op_id(self) -> str:
        """Wire / registry identifier. Mirrors ``self.op``."""
        return self.op

    @classmethod
    def registry(cls) -> dict[str, type[Operation]]:
        """Snapshot of ``{op_id: subclass}`` for every registered Operation."""
        return dict(Operation._registry)

    @classmethod
    def llm_registry(cls) -> dict[str, type[Operation]]:
        """Snapshot of ``{op_id: subclass}`` for LLM-exposed Operations only.

        A subset of :meth:`registry` filtered to subclasses with
        ``llm_exposed`` True. Server-only ops (e.g. those needing a
        server-resolved ``source`` path) are excluded so they never leak into
        the LLM-facing schema.
        """
        return {op_id: sub for op_id, sub in Operation._registry.items() if sub.llm_exposed}

    @classmethod
    def get(cls, op_id: str) -> type[Operation]:
        """Look up the Operation subclass for ``op_id``."""
        try:
            return Operation._registry[op_id]
        except KeyError as exc:
            known = ", ".join(sorted(Operation._registry)) or "(none)"
            raise KeyError(f"Unknown op_id {op_id!r}. Known ops: [{known}]") from exc

    @classmethod
    def json_schema(cls, include_server_only: bool = False, *, strict: bool = False) -> dict[str, Any]:
        """Discriminated-union JSON schema over registered Operations.

        ``op`` is the discriminator tag. This is the LLM-facing schema for
        validating a single operation payload. By default the union covers only
        LLM-exposed ops (:meth:`llm_registry`); pass ``include_server_only=True``
        to build the union from the full :meth:`registry`. Fields marked
        ``llm_hidden`` (advanced overrides like raw font paths) are stripped.

        With ``strict=True`` the schema is rewritten for use as a provider
        structured-output **grammar** (OpenAI/OpenRouter ``json_schema`` strict
        mode): every object is closed (``additionalProperties: false``), every
        property is listed in ``required`` with its optionality kept exactly as
        Pydantic emitted it (an ``Optional`` field keeps its nullable branch; a
        defaulted non-``Optional`` field -- including the ``op`` discriminator --
        stays required and non-nullable), and the discriminated union is
        expressed as a plain ``anyOf`` of closed variants (``discriminator``,
        ``default``, custom ``format``, and ``$schema`` -- all unsupported or moot
        in strict mode -- are dropped). Numeric constraints
        (``minimum``/``maximum``/``exclusiveMinimum``) are preserved, so an
        entire class of bound violations becomes impossible at decode time.

        Note: the strict result is a *root-level* ``anyOf`` union -- an embeddable
        schema fragment, not a submittable strict root (providers require the root
        to be a closed object). It is consumed inside
        :meth:`VideoEdit.json_schema(strict=True) <VideoEdit.json_schema>`, which
        *is* a submittable object root; use that to constrain a whole plan.
        """
        source = Operation._registry if include_server_only else cls.llm_registry()
        if not source:
            return {"type": "object"}
        ops = sorted(source.values(), key=lambda c: c.__name__)
        annotated = Annotated[Union[tuple(ops)], Discriminator("op")]  # type: ignore[valid-type]  # noqa: UP007
        schema = _strip_llm_hidden(TypeAdapter(annotated).json_schema())
        return _to_strict_schema(schema) if strict else schema

    @classmethod
    def llm_json_schema(cls) -> dict[str, Any]:
        """Per-op JSON schema with ``llm_hidden`` fields removed.

        Like ``cls.model_json_schema()`` but drops advanced / non-LLM fields
        (e.g. raw font paths) so a single op can be exposed to an LLM directly
        without leaking a field the model shouldn't fill in.
        """
        return _strip_llm_hidden(cls.model_json_schema())

    def predict_metadata(self, meta: VideoMetadata) -> VideoMetadata:
        """Predict output metadata from input metadata. Default: identity.

        Run during ``VideoEdit.validate()``'s dry-run, before any frames are
        decoded. Beyond predicting shape, this is the fail-fast gate, and it
        has one contract: **reject exactly the plans that would otherwise crash
        or do unrecoverable / expensive work in** ``run_to_file()``;
        anything ``run_to_file()`` can absorb by graceful degradation is NOT rejected.
        ``TranscriptionOverlay`` rejects un-fittable subtitles (they used to
        crash mid-render); ``TextOverlay``/``ImageOverlay`` do not reject
        off-frame geometry (it clips to a valid no-op). Keep the check
        metadata-cheap -- no frame decode.

        Duration bounds checks use the shared
        :data:`videopython.editing.transforms.DURATION_EPS` tolerance: a value
        is rejected only when it exceeds the limit by more than ``DURATION_EPS``
        seconds, so sub-millisecond float drift at an exact boundary passes
        consistently across the editing layer.
        """
        return meta

    def to_ffmpeg_filter(self, ctx: FilterCtx) -> str | None:
        """Compile to an ffmpeg ``-vf`` filter expression, or ``None`` for no filter compilation.

        Streamable transforms override this. Effects use ``process_frame``
        instead -- they do not go through ffmpeg filters.
        """
        return None

    def to_ffmpeg_audio_filter(self, ctx: FilterCtx) -> str | None:
        """Compile the op's audio-domain twin to an ffmpeg audio-filter expression.

        The audio analogue of :meth:`to_ffmpeg_filter`: segment audio now
        streams through the SAME ffmpeg process as the video (a second
        ``-i source`` input routed through ``-filter_complex``), so a
        duration-changing transform expresses its audio effect as a filter on
        that graph instead of mutating an in-memory ``Audio`` array
        (``speed_change`` -> ``atempo``, ``freeze_frame`` -> silence splice,
        ``silence_removal`` -> ``aselect`` keep windows, ``fade`` -> ``afade``,
        ``volume_adjust`` -> ``volume``).

        ``ctx`` is the SAME :class:`FilterCtx` the video side builds at this
        op's plan position -- ``ctx.fps``/``ctx.frame_count`` are the
        already-folded values, ``ctx.context`` carries the resolved,
        segment-local ``requires`` -- so the audio chain stays in lockstep
        with the video chain. The returned expression is a comma-joined
        single-input/single-output filter sub-chain (e.g. ``"atempo=2.0"``);
        the plan builder appends it to the segment's labeled audio graph at
        the same stage (decode/encode) it appends the video filter. ``None``
        means "no audio effect" -- the default, so the builder only emits a
        filter for the four audio-affecting ops.
        """
        return None

    def streams(self) -> bool:
        """Whether this op streams in O(1) memory at its plan position.

        Structural replacement for the former ``streamable`` ClassVar: a transform
        streams iff it overrides :meth:`to_ffmpeg_filter`. :class:`Effect` widens
        this (a frame effect streams via ``process_frame``; a filter effect via
        ``compiles_to_filter``). The one case structure cannot express -- the
        override exists but the filter compiles to ``None`` at this position -- is
        caught at runtime by the ``STREAMING_UNSUPPORTED`` raise in
        ``VideoEdit._compile_streaming_plans``.
        """
        return type(self).to_ffmpeg_filter is not Operation.to_ffmpeg_filter

compiles_from_source `class-attribute`

compiles_from_source: bool = False

Whether the op's filter compile decodes the source itself (face_crop's detection pass). Such ops cannot sit at the encode stage -- the frames behind per-frame Python effects are not reproducible at compile time -- so both the plan builder and the streamability report reject them as UNSTREAMABLE there.

changes_duration `class-attribute`

changes_duration: bool = False

Whether the op's output duration differs from its input (speed, freeze).

The streaming plan builder folds predict_metadata through the chain either way; this flag additionally gates time-based context: a context-consuming op scheduled after a duration-changing transform would receive timestamps on the wrong timeline, so such plans are rejected as UNSTREAMABLE until context re-mapping exists. The streamability report mirrors the same rule.

internal_only `class-attribute`

internal_only: bool = False

Whether this op is engine-internal and must NOT be a chain op.

CutSeconds/CutFrames trim a segment, but trimming is the segment's own start/end mechanism -- the engine constructs them directly. They have no ffmpeg filter and no process_frame, so this flag keeps them OUT of the registry: they cannot appear in a plan's operations list or the LLM schema, while direct construction (CutSeconds(start=..., end=...)) still works. Default False (a normal chain op).

time_fields `class-attribute`

time_fields: tuple[BoundedTimeField, ...] = ()

Time-valued (seconds) fields :meth:VideoEdit.repair may clamp into range.

Declaring a :class:BoundedTimeField here lets repair clamp an out-of-range timestamp (e.g. freeze_frame.timestamp past the clip end) without per-op special-casing -- the repair pass reads the declaration, clamps to [0, bound], and records a :class:PlanRepair. Empty by default; ops with no time-valued params declare nothing.

op_id `property`

op_id: str

Wire / registry identifier. Mirrors self.op.

registry `classmethod`

registry() -> dict[str, type[Operation]]

Snapshot of {op_id: subclass} for every registered Operation.

Source code in src/videopython/editing/operation.py

@classmethod
def registry(cls) -> dict[str, type[Operation]]:
    """Snapshot of ``{op_id: subclass}`` for every registered Operation."""
    return dict(Operation._registry)

llm_registry `classmethod`

llm_registry() -> dict[str, type[Operation]]

Snapshot of {op_id: subclass} for LLM-exposed Operations only.

A subset of :meth:registry filtered to subclasses with llm_exposed True. Server-only ops (e.g. those needing a server-resolved source path) are excluded so they never leak into the LLM-facing schema.

Source code in src/videopython/editing/operation.py

@classmethod
def llm_registry(cls) -> dict[str, type[Operation]]:
    """Snapshot of ``{op_id: subclass}`` for LLM-exposed Operations only.

    A subset of :meth:`registry` filtered to subclasses with
    ``llm_exposed`` True. Server-only ops (e.g. those needing a
    server-resolved ``source`` path) are excluded so they never leak into
    the LLM-facing schema.
    """
    return {op_id: sub for op_id, sub in Operation._registry.items() if sub.llm_exposed}

get `classmethod`

get(op_id: str) -> type[Operation]

Look up the Operation subclass for op_id.

Source code in src/videopython/editing/operation.py

@classmethod
def get(cls, op_id: str) -> type[Operation]:
    """Look up the Operation subclass for ``op_id``."""
    try:
        return Operation._registry[op_id]
    except KeyError as exc:
        known = ", ".join(sorted(Operation._registry)) or "(none)"
        raise KeyError(f"Unknown op_id {op_id!r}. Known ops: [{known}]") from exc

json_schema `classmethod`

json_schema(
    include_server_only: bool = False,
    *,
    strict: bool = False,
) -> dict[str, Any]

Discriminated-union JSON schema over registered Operations.

op is the discriminator tag. This is the LLM-facing schema for validating a single operation payload. By default the union covers only LLM-exposed ops (:meth:llm_registry); pass include_server_only=True to build the union from the full :meth:registry. Fields marked llm_hidden (advanced overrides like raw font paths) are stripped.

With strict=True the schema is rewritten for use as a provider structured-output grammar (OpenAI/OpenRouter json_schema strict mode): every object is closed (additionalProperties: false), every property is listed in required with its optionality kept exactly as Pydantic emitted it (an Optional field keeps its nullable branch; a defaulted non-Optional field -- including the op discriminator -- stays required and non-nullable), and the discriminated union is expressed as a plain anyOf of closed variants (discriminator, default, custom format, and $schema -- all unsupported or moot in strict mode -- are dropped). Numeric constraints (minimum/maximum/exclusiveMinimum) are preserved, so an entire class of bound violations becomes impossible at decode time.

Note: the strict result is a root-level anyOf union -- an embeddable schema fragment, not a submittable strict root (providers require the root to be a closed object). It is consumed inside :meth:VideoEdit.json_schema(strict=True) <VideoEdit.json_schema>, which is a submittable object root; use that to constrain a whole plan.

Source code in src/videopython/editing/operation.py

@classmethod
def json_schema(cls, include_server_only: bool = False, *, strict: bool = False) -> dict[str, Any]:
    """Discriminated-union JSON schema over registered Operations.

    ``op`` is the discriminator tag. This is the LLM-facing schema for
    validating a single operation payload. By default the union covers only
    LLM-exposed ops (:meth:`llm_registry`); pass ``include_server_only=True``
    to build the union from the full :meth:`registry`. Fields marked
    ``llm_hidden`` (advanced overrides like raw font paths) are stripped.

    With ``strict=True`` the schema is rewritten for use as a provider
    structured-output **grammar** (OpenAI/OpenRouter ``json_schema`` strict
    mode): every object is closed (``additionalProperties: false``), every
    property is listed in ``required`` with its optionality kept exactly as
    Pydantic emitted it (an ``Optional`` field keeps its nullable branch; a
    defaulted non-``Optional`` field -- including the ``op`` discriminator --
    stays required and non-nullable), and the discriminated union is
    expressed as a plain ``anyOf`` of closed variants (``discriminator``,
    ``default``, custom ``format``, and ``$schema`` -- all unsupported or moot
    in strict mode -- are dropped). Numeric constraints
    (``minimum``/``maximum``/``exclusiveMinimum``) are preserved, so an
    entire class of bound violations becomes impossible at decode time.

    Note: the strict result is a *root-level* ``anyOf`` union -- an embeddable
    schema fragment, not a submittable strict root (providers require the root
    to be a closed object). It is consumed inside
    :meth:`VideoEdit.json_schema(strict=True) <VideoEdit.json_schema>`, which
    *is* a submittable object root; use that to constrain a whole plan.
    """
    source = Operation._registry if include_server_only else cls.llm_registry()
    if not source:
        return {"type": "object"}
    ops = sorted(source.values(), key=lambda c: c.__name__)
    annotated = Annotated[Union[tuple(ops)], Discriminator("op")]  # type: ignore[valid-type]  # noqa: UP007
    schema = _strip_llm_hidden(TypeAdapter(annotated).json_schema())
    return _to_strict_schema(schema) if strict else schema

llm_json_schema `classmethod`

llm_json_schema() -> dict[str, Any]

Per-op JSON schema with llm_hidden fields removed.

Like cls.model_json_schema() but drops advanced / non-LLM fields (e.g. raw font paths) so a single op can be exposed to an LLM directly without leaking a field the model shouldn't fill in.

Source code in src/videopython/editing/operation.py

@classmethod
def llm_json_schema(cls) -> dict[str, Any]:
    """Per-op JSON schema with ``llm_hidden`` fields removed.

    Like ``cls.model_json_schema()`` but drops advanced / non-LLM fields
    (e.g. raw font paths) so a single op can be exposed to an LLM directly
    without leaking a field the model shouldn't fill in.
    """
    return _strip_llm_hidden(cls.model_json_schema())

predict_metadata

predict_metadata(meta: VideoMetadata) -> VideoMetadata

Predict output metadata from input metadata. Default: identity.

Run during VideoEdit.validate()'s dry-run, before any frames are decoded. Beyond predicting shape, this is the fail-fast gate, and it has one contract: reject exactly the plans that would otherwise crash or do unrecoverable / expensive work in run_to_file(); anything run_to_file() can absorb by graceful degradation is NOT rejected. TranscriptionOverlay rejects un-fittable subtitles (they used to crash mid-render); TextOverlay/ImageOverlay do not reject off-frame geometry (it clips to a valid no-op). Keep the check metadata-cheap -- no frame decode.

Duration bounds checks use the shared :data:videopython.editing.transforms.DURATION_EPS tolerance: a value is rejected only when it exceeds the limit by more than DURATION_EPS seconds, so sub-millisecond float drift at an exact boundary passes consistently across the editing layer.

Source code in src/videopython/editing/operation.py

def predict_metadata(self, meta: VideoMetadata) -> VideoMetadata:
    """Predict output metadata from input metadata. Default: identity.

    Run during ``VideoEdit.validate()``'s dry-run, before any frames are
    decoded. Beyond predicting shape, this is the fail-fast gate, and it
    has one contract: **reject exactly the plans that would otherwise crash
    or do unrecoverable / expensive work in** ``run_to_file()``;
    anything ``run_to_file()`` can absorb by graceful degradation is NOT rejected.
    ``TranscriptionOverlay`` rejects un-fittable subtitles (they used to
    crash mid-render); ``TextOverlay``/``ImageOverlay`` do not reject
    off-frame geometry (it clips to a valid no-op). Keep the check
    metadata-cheap -- no frame decode.

    Duration bounds checks use the shared
    :data:`videopython.editing.transforms.DURATION_EPS` tolerance: a value
    is rejected only when it exceeds the limit by more than ``DURATION_EPS``
    seconds, so sub-millisecond float drift at an exact boundary passes
    consistently across the editing layer.
    """
    return meta

to_ffmpeg_filter

to_ffmpeg_filter(ctx: FilterCtx) -> str | None

Compile to an ffmpeg -vf filter expression, or None for no filter compilation.

Streamable transforms override this. Effects use process_frame instead -- they do not go through ffmpeg filters.

Source code in src/videopython/editing/operation.py

def to_ffmpeg_filter(self, ctx: FilterCtx) -> str | None:
    """Compile to an ffmpeg ``-vf`` filter expression, or ``None`` for no filter compilation.

    Streamable transforms override this. Effects use ``process_frame``
    instead -- they do not go through ffmpeg filters.
    """
    return None

to_ffmpeg_audio_filter

to_ffmpeg_audio_filter(ctx: FilterCtx) -> str | None

Compile the op's audio-domain twin to an ffmpeg audio-filter expression.

The audio analogue of :meth:to_ffmpeg_filter: segment audio now streams through the SAME ffmpeg process as the video (a second -i source input routed through -filter_complex), so a duration-changing transform expresses its audio effect as a filter on that graph instead of mutating an in-memory Audio array (speed_change -> atempo, freeze_frame -> silence splice, silence_removal -> aselect keep windows, fade -> afade, volume_adjust -> volume).

ctx is the SAME :class:FilterCtx the video side builds at this op's plan position -- ctx.fps/ctx.frame_count are the already-folded values, ctx.context carries the resolved, segment-local requires -- so the audio chain stays in lockstep with the video chain. The returned expression is a comma-joined single-input/single-output filter sub-chain (e.g. "atempo=2.0"); the plan builder appends it to the segment's labeled audio graph at the same stage (decode/encode) it appends the video filter. None means "no audio effect" -- the default, so the builder only emits a filter for the four audio-affecting ops.

Source code in src/videopython/editing/operation.py

def to_ffmpeg_audio_filter(self, ctx: FilterCtx) -> str | None:
    """Compile the op's audio-domain twin to an ffmpeg audio-filter expression.

    The audio analogue of :meth:`to_ffmpeg_filter`: segment audio now
    streams through the SAME ffmpeg process as the video (a second
    ``-i source`` input routed through ``-filter_complex``), so a
    duration-changing transform expresses its audio effect as a filter on
    that graph instead of mutating an in-memory ``Audio`` array
    (``speed_change`` -> ``atempo``, ``freeze_frame`` -> silence splice,
    ``silence_removal`` -> ``aselect`` keep windows, ``fade`` -> ``afade``,
    ``volume_adjust`` -> ``volume``).

    ``ctx`` is the SAME :class:`FilterCtx` the video side builds at this
    op's plan position -- ``ctx.fps``/``ctx.frame_count`` are the
    already-folded values, ``ctx.context`` carries the resolved,
    segment-local ``requires`` -- so the audio chain stays in lockstep
    with the video chain. The returned expression is a comma-joined
    single-input/single-output filter sub-chain (e.g. ``"atempo=2.0"``);
    the plan builder appends it to the segment's labeled audio graph at
    the same stage (decode/encode) it appends the video filter. ``None``
    means "no audio effect" -- the default, so the builder only emits a
    filter for the four audio-affecting ops.
    """
    return None

streams

streams() -> bool

Whether this op streams in O(1) memory at its plan position.

Structural replacement for the former streamable ClassVar: a transform streams iff it overrides :meth:to_ffmpeg_filter. :class:Effect widens this (a frame effect streams via process_frame; a filter effect via compiles_to_filter). The one case structure cannot express -- the override exists but the filter compiles to None at this position -- is caught at runtime by the STREAMING_UNSUPPORTED raise in VideoEdit._compile_streaming_plans.

Source code in src/videopython/editing/operation.py

def streams(self) -> bool:
    """Whether this op streams in O(1) memory at its plan position.

    Structural replacement for the former ``streamable`` ClassVar: a transform
    streams iff it overrides :meth:`to_ffmpeg_filter`. :class:`Effect` widens
    this (a frame effect streams via ``process_frame``; a filter effect via
    ``compiles_to_filter``). The one case structure cannot express -- the
    override exists but the filter compiles to ``None`` at this position -- is
    caught at runtime by the ``STREAMING_UNSUPPORTED`` raise in
    ``VideoEdit._compile_streaming_plans``.
    """
    return type(self).to_ffmpeg_filter is not Operation.to_ffmpeg_filter

Effect

Bases: Operation

Operation that preserves shape and frame count, driven by per-frame streaming.

Subclasses implement the streaming contract -- :meth:process_frame (and :meth:streaming_init for any precomputed per-stream state) -- which is the single source of truth for the effect's pixel logic. The streaming engine in editing/streaming.py drives that contract for bounded-memory execution, resolving :attr:window against the segment timeline so frames outside the window pass through untouched.

Effects that compile to a native ffmpeg filter instead set :attr:compiles_to_filter and implement :meth:to_ffmpeg_filter (and, for audio-coupled effects like Fade/VolumeAdjust, :meth:to_ffmpeg_audio_filter) so the window stays coherent across the decode/encode graph.

Source code in src/videopython/editing/operation.py

class Effect(Operation):
    """Operation that preserves shape and frame count, driven by per-frame streaming.

    Subclasses implement the streaming contract -- :meth:`process_frame` (and
    :meth:`streaming_init` for any precomputed per-stream state) -- which is the
    single source of truth for the effect's pixel logic. The streaming engine
    in ``editing/streaming.py`` drives that contract for bounded-memory
    execution, resolving :attr:`window` against the segment timeline so frames
    outside the window pass through untouched.

    Effects that compile to a native ffmpeg filter instead set
    :attr:`compiles_to_filter` and implement :meth:`to_ffmpeg_filter` (and, for
    audio-coupled effects like ``Fade``/``VolumeAdjust``,
    :meth:`to_ffmpeg_audio_filter`) so the window stays coherent across the
    decode/encode graph.
    """

    category: ClassVar[OpCategory] = OpCategory.EFFECT
    audio_coupled: ClassVar[bool] = False
    """Whether the effect mutates audio alongside pixels (``afade``/``volume``)."""

    window: TimeRange | None = Field(
        None,
        description="Time window for the effect in seconds. Omit to apply across the full duration.",
    )

    @property
    def compiles_to_filter(self) -> bool:
        """Whether this effect joins the decode filter chain instead of scheduling per-frame Python.

        When True, the streaming plan builder calls :meth:`to_ffmpeg_filter`
        (with the segment's resolved context on the :class:`FilterCtx`) and, if
        it compiles, appends the result to the vf chain at this op's plan
        position -- the Filter class of the streaming contract. Instance-level
        rather than a ClassVar because it may depend on field values (e.g.
        ``add_subtitles``'s ``renderer``). False by default: effects normally
        stream via ``streaming_init``/``process_frame``.
        """
        return False

    def predict_metadata(self, meta: VideoMetadata, **_context: Any) -> VideoMetadata:
        """Effects preserve shape and frame count, so the prediction is identity.

        Accepts ``**_context`` so requires-aware effects (``TranscriptionOverlay``)
        validate without subclasses needing to override just to widen the
        signature. Mirrors :meth:`Effect.streaming_init`'s ``**_context`` accept-all.
        """
        return meta

    def streaming_init(self, total_frames: int, fps: float, width: int, height: int, **_context: Any) -> None:
        """Hook for per-stream precomputation (per-frame alphas, sigma curves...).

        ``_context`` carries resolved ``requires`` values for context-aware
        effects (e.g. ``transcription=...`` for ``TranscriptionOverlay``),
        already re-based onto the local timeline by the runner. Effects that
        declare no ``requires`` are always called without context kwargs.

        Default: no-op. Override in subclasses that need it.
        """

    def process_frame(self, frame: np.ndarray, frame_index: int) -> np.ndarray:
        """Process one ``(H, W, 3) uint8`` frame in streaming mode.

        ``frame_index`` is 0-based within this effect's active window.
        """
        raise NotImplementedError(f"{type(self).__name__} does not support streaming")

    def streams(self) -> bool:
        """An effect streams via per-frame Python (``process_frame``) or a filter.

        Frame effects override :meth:`process_frame`; filter effects
        (``add_subtitles``, ``vignette``, ...) instead set
        :attr:`compiles_to_filter` and implement :meth:`to_ffmpeg_filter`.
        ``add_subtitles`` streams *only* via the filter path (it does not override
        ``process_frame``), so ``compiles_to_filter`` is consulted per-instance.
        """
        return type(self).process_frame is not Effect.process_frame or self.compiles_to_filter

audio_coupled `class-attribute`

audio_coupled: bool = False

Whether the effect mutates audio alongside pixels (afade/volume).

compiles_to_filter `property`

compiles_to_filter: bool

Whether this effect joins the decode filter chain instead of scheduling per-frame Python.

When True, the streaming plan builder calls :meth:to_ffmpeg_filter (with the segment's resolved context on the :class:FilterCtx) and, if it compiles, appends the result to the vf chain at this op's plan position -- the Filter class of the streaming contract. Instance-level rather than a ClassVar because it may depend on field values (e.g. add_subtitles's renderer). False by default: effects normally stream via streaming_init/process_frame.

predict_metadata

predict_metadata(
    meta: VideoMetadata, **_context: Any
) -> VideoMetadata

Effects preserve shape and frame count, so the prediction is identity.

Accepts **_context so requires-aware effects (TranscriptionOverlay) validate without subclasses needing to override just to widen the signature. Mirrors :meth:Effect.streaming_init's **_context accept-all.

Source code in src/videopython/editing/operation.py

def predict_metadata(self, meta: VideoMetadata, **_context: Any) -> VideoMetadata:
    """Effects preserve shape and frame count, so the prediction is identity.

    Accepts ``**_context`` so requires-aware effects (``TranscriptionOverlay``)
    validate without subclasses needing to override just to widen the
    signature. Mirrors :meth:`Effect.streaming_init`'s ``**_context`` accept-all.
    """
    return meta

streaming_init

streaming_init(
    total_frames: int,
    fps: float,
    width: int,
    height: int,
    **_context: Any,
) -> None

Hook for per-stream precomputation (per-frame alphas, sigma curves...).

_context carries resolved requires values for context-aware effects (e.g. transcription=... for TranscriptionOverlay), already re-based onto the local timeline by the runner. Effects that declare no requires are always called without context kwargs.

Default: no-op. Override in subclasses that need it.

Source code in src/videopython/editing/operation.py

def streaming_init(self, total_frames: int, fps: float, width: int, height: int, **_context: Any) -> None:
    """Hook for per-stream precomputation (per-frame alphas, sigma curves...).

    ``_context`` carries resolved ``requires`` values for context-aware
    effects (e.g. ``transcription=...`` for ``TranscriptionOverlay``),
    already re-based onto the local timeline by the runner. Effects that
    declare no ``requires`` are always called without context kwargs.

    Default: no-op. Override in subclasses that need it.
    """

process_frame

process_frame(
    frame: ndarray, frame_index: int
) -> np.ndarray

Process one (H, W, 3) uint8 frame in streaming mode.

frame_index is 0-based within this effect's active window.

Source code in src/videopython/editing/operation.py

def process_frame(self, frame: np.ndarray, frame_index: int) -> np.ndarray:
    """Process one ``(H, W, 3) uint8`` frame in streaming mode.

    ``frame_index`` is 0-based within this effect's active window.
    """
    raise NotImplementedError(f"{type(self).__name__} does not support streaming")

streams

streams() -> bool

An effect streams via per-frame Python (process_frame) or a filter.

Frame effects override :meth:process_frame; filter effects (add_subtitles, vignette, ...) instead set :attr:compiles_to_filter and implement :meth:to_ffmpeg_filter. add_subtitles streams only via the filter path (it does not override process_frame), so compiles_to_filter is consulted per-instance.

Source code in src/videopython/editing/operation.py

def streams(self) -> bool:
    """An effect streams via per-frame Python (``process_frame``) or a filter.

    Frame effects override :meth:`process_frame`; filter effects
    (``add_subtitles``, ``vignette``, ...) instead set
    :attr:`compiles_to_filter` and implement :meth:`to_ffmpeg_filter`.
    ``add_subtitles`` streams *only* via the filter path (it does not override
    ``process_frame``), so ``compiles_to_filter`` is consulted per-instance.
    """
    return type(self).process_frame is not Effect.process_frame or self.compiles_to_filter

TimeRange

Bases: BaseModel

Half-open time window in seconds: [start, stop).

Either endpoint may be None, meaning "from the beginning" / "to the end" respectively. Used by :class:Effect.window and elsewhere.

Parsing is deliberately permissive: start/stop are plain floats with no ge=0 or ordering constraint. The plan skeleton accepts the shape; the numeric bounds (>= 0, stop >= start, in-duration) are owned by :meth:VideoEdit.validate / :meth:VideoEdit.check, which report them as structured, collectable, repairable :class:PlanErrors instead of aborting at from_dict. The window is still clamped to min(stop, total_seconds) at run time, so a plan run without validation degrades rather than crashes.

Source code in src/videopython/editing/operation.py

class TimeRange(BaseModel):
    """Half-open time window in seconds: ``[start, stop)``.

    Either endpoint may be ``None``, meaning "from the beginning" / "to the
    end" respectively. Used by :class:`Effect.window` and elsewhere.

    Parsing is deliberately permissive: ``start``/``stop`` are plain floats
    with no ``ge=0`` or ordering constraint. The plan skeleton accepts the
    *shape*; the numeric bounds (``>= 0``, ``stop >= start``, in-duration) are
    owned by :meth:`VideoEdit.validate` / :meth:`VideoEdit.check`, which report
    them as structured, collectable, repairable :class:`PlanError`s instead of
    aborting at ``from_dict``. The window is still clamped to
    ``min(stop, total_seconds)`` at run time, so a plan run without validation
    degrades rather than crashes.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    start: float | None = Field(None, description="Start time in seconds. None means 0.")
    stop: float | None = Field(None, description="Stop time in seconds. None means end of video.")

OpCategory

Bases: str, Enum

Coarse execution category for an Operation subclass.

Source code in src/videopython/editing/operation.py

class OpCategory(str, Enum):
    """Coarse execution category for an Operation subclass."""

    TRANSFORM = "transform"
    EFFECT = "effect"
    SPECIAL = "special"

FilterCtx

FilterCtx `dataclass`

Current pipeline state (post-prior-ops) when compiling to ffmpeg.

frame_count is the number of frames entering the filter at this chain position (the plan builder folds predict_metadata through the chain), so duration-aware compilations (a speed ramp's time-warp expression, a freeze's frame indices) can be exact. 0 when unknown -- compilations that need it must return None (no filter compilation) in that case.

context carries the resolved, segment-local runtime context (the same re-based values streaming_init receives) so a context-consuming op can compile itself into the filter chain (e.g. add_subtitles consuming the transcription to write an .ass file). Empty when no context applies.

owned_files collects temp files a compilation creates (the .ass file a subtitles= entry references); the plan runner deletes them once streaming finishes or the plan is abandoned.

source_path/start_second/end_second locate the segment on disk, and decode_filters is the decode-stage filter prefix ahead of this op -- together they let a compilation run its own bounded decode pass over exactly the frames the filter will see (face_crop's detection). decode_filters is None when those frames are not reproducible at compile time (the op sits at the encode stage, behind per-frame Python effects); such compilations must return None.

Source code in src/videopython/editing/operation.py

@dataclass(frozen=True)
class FilterCtx:
    """Current pipeline state (post-prior-ops) when compiling to ffmpeg.

    ``frame_count`` is the number of frames entering the filter at this chain
    position (the plan builder folds ``predict_metadata`` through the chain),
    so duration-aware compilations (a speed ramp's time-warp expression, a
    freeze's frame indices) can be exact. ``0`` when unknown -- compilations
    that need it must return ``None`` (no filter compilation) in that case.

    ``context`` carries the resolved, segment-local runtime context (the same
    re-based values ``streaming_init`` receives) so a context-consuming op can
    compile itself into the filter chain (e.g. ``add_subtitles`` consuming the
    transcription to write an ``.ass`` file). Empty when no context applies.

    ``owned_files`` collects temp files a compilation creates (the ``.ass``
    file a ``subtitles=`` entry references); the plan runner deletes them once
    streaming finishes or the plan is abandoned.

    ``source_path``/``start_second``/``end_second`` locate the segment on
    disk, and ``decode_filters`` is the decode-stage filter prefix ahead of
    this op -- together they let a compilation run its own bounded decode
    pass over exactly the frames the filter will see (``face_crop``'s
    detection). ``decode_filters`` is ``None`` when those frames are not
    reproducible at compile time (the op sits at the encode stage, behind
    per-frame Python effects); such compilations must return ``None``.
    """

    width: int
    height: int
    fps: float
    frame_count: int = 0
    context: dict[str, Any] = field(default_factory=dict)
    owned_files: list[Path] = field(default_factory=list)
    source_path: Path | None = None
    start_second: float = 0.0
    end_second: float | None = None
    decode_filters: tuple[str, ...] | None = ()
    audio_label: str = "a"
    """A unique-within-the-graph prefix an audio-filter compilation can use to
    name internal ``filter_complex`` labels (``freeze_frame``'s split/concat
    splice). The plan builder sets a distinct value per op so a multi-statement
    audio fragment cannot collide with another op's internal labels. The
    surrounding ``[in]<chain>[out]`` wrapper labels are owned by the builder;
    this names only the op's *internal* intermediate streams."""

audio_label `class-attribute` `instance-attribute`

audio_label: str = 'a'

A unique-within-the-graph prefix an audio-filter compilation can use to name internal filter_complex labels (freeze_frame's split/concat splice). The plan builder sets a distinct value per op so a multi-statement audio fragment cannot collide with another op's internal labels. The surrounding [in]<chain>[out] wrapper labels are owned by the builder; this names only the op's internal intermediate streams.

Operations

Subclass Contract

Effects

Registry API

LLM-exposed vs server-only ops

Discovering Operations

Per-Operation JSON Schema

Registered Operations

Base (no AI dependencies)

AI (require import videopython.ai)

API Reference

Operation

Operation

compiles_from_source class-attribute

changes_duration class-attribute

internal_only class-attribute

time_fields class-attribute

op_id property

registry classmethod

llm_registry classmethod

get classmethod

json_schema classmethod

llm_json_schema classmethod

predict_metadata

to_ffmpeg_filter

to_ffmpeg_audio_filter

streams

Effect

Effect

audio_coupled class-attribute

compiles_to_filter property

predict_metadata

streaming_init

process_frame

streams

TimeRange

TimeRange

OpCategory

OpCategory

FilterCtx

FilterCtx dataclass

audio_label class-attribute instance-attribute

AI (require `import videopython.ai`)

compiles_from_source `class-attribute`

changes_duration `class-attribute`

internal_only `class-attribute`

time_fields `class-attribute`

op_id `property`

registry `classmethod`

llm_registry `classmethod`

get `classmethod`

json_schema `classmethod`

llm_json_schema `classmethod`

audio_coupled `class-attribute`

compiles_to_filter `property`

FilterCtx `dataclass`

audio_label `class-attribute` `instance-attribute`