render ¶

def clip_async_render[T](
    clip: vs.VideoNode,
    outfile: BinaryIO | SPathLike | None = None,
    progress: str | Callable[[int, int], None] | None = None,
    callback: Callable[[int, vs.VideoFrame], T] | None = None,
    prefetch: int | None = None,
    backlog: int | None = None,
    y4m: bool = False,
    async_requests: int | Literal[False] | AsyncRenderConf = False,
) -> list[T] | None:
    """
    Iterate over an entire clip and optionally write results to a file or gather data.

    This is mostly useful for metric gathering that must be performed before any other processing.
    This could be for example gathering scenechanges, per-frame heuristics, etc.

    It's highly recommended to perform as little filtering as possible on the input clip for speed purposes.

    Example usage:
        ```py
        # Gather scenechanges.
        scenechanges = clip_async_render(
            clip, None, "Searching for scenechanges...", lambda n, f: get_prop(f, "_SceneChange", int)
        )

        # Gather average planes stats.
        avg_planes = clip_async_render(
            clip, None, "Calculating average planes...", lambda n, f: get_prop(f, "PlaneStatsAverage", float)
        )
        ```

    Args:
        clip: Clip to render.
        outfile: Optional binary output or path to write to.
        progress: Progress reporting configuration. Can be:

               - `str`: A description message displayed next to the rendering progress bar.
               - `Callable[[int, int], None]`: A custom progress callback function
                 that receives the current frame index and total frames `(current_frame, total_frames)`.
               - `None`: Disables progress reporting completely.
        callback: Callback function invoked for every frame during rendering.
            Must accept `n` and `f` (like a frameeval would) and return some value of type `T`.
            The returned values are collected in order of frame index and returned as a `list[T]`.
            Only active when `outfile` is not provided.
        prefetch: Defines how many frames are rendered concurrently.
        backlog: Defines how many unconsumed frames vapoursynth buffers at most
            before it stops rendering additional frames.
        y4m: Whether to add YUV4MPEG2 headers to the rendered output.
        async_requests: Whether to render frames non-consecutively. If int, determines the number of requests.
    """
    if isinstance(outfile, (str, PathLike)) and outfile is not None:
        with open(outfile, "wb") as f:
            return clip_async_render(clip, f, progress, callback, prefetch, backlog, y4m, async_requests)  # type: ignore[misc,arg-type]

    async_conf: AsyncRenderConf | Literal[False]

    if isinstance(async_requests, AsyncRenderConf):
        async_conf = async_requests
    elif async_requests is False:
        async_conf = False
    else:
        async_conf = AsyncRenderConf(async_requests)

    if async_conf and (not callback or outfile):
        raise CustomValueError(
            "You cannot use async requests without a callback and with an outfile", clip_async_render
        )

    total_frames = clip.num_frames

    if not progress:
        pr_ctx = contextlib.nullcontext()
        pr_up = lambda *_: None  # noqa: E731
    elif callable(progress):
        pr_ctx = contextlib.nullcontext()
        pr_up = progress
    else:
        from .progress import get_render_progress

        pr_ctx = get_render_progress()
        task = pr_ctx.add_task(progress)
        pr_up = lambda n, total: pr_ctx.update(task, total=total, completed=n)  # noqa: E731

    with pr_ctx:
        if outfile:
            clip.output(outfile, y4m, pr_up, prefetch or 0, fallback(backlog, -1))
        elif not callback:
            for i, _ in enumerate(clip.frames(prefetch, backlog, True)):
                pr_up(i, total_frames)
        else:
            result = dict[int, T]()

            if not async_conf:
                clip = core.std.ModifyFrame(
                    clip.std.BlankClip(keep=True),
                    clip,
                    _get_render_callback(callback, result, lambda n: pr_up(n, total_frames)),
                )
            else:
                clip = async_conf.get_clip(clip, callback, result, lambda n: pr_up(n, total_frames))

            deque(clip.frames(prefetch, backlog, True), 0)

            try:
                return [result[i] for i in range(total_frames)]
            except KeyError:
                raise CustomRuntimeError(
                    "There was an error with the rendering and one frame request was rejected!", clip_async_render
                )

    return None

def clip_data_gather[T](
    clip: vs.VideoNode,
    progress: str | Callable[[int, int], None] | None,
    callback: Callable[[int, vs.VideoFrame], SentinelT | T],
    async_requests: int | bool | AsyncRenderConf = False,
    prefetch: int = 0,
    backlog: int = -1,
) -> list[T]:
    frames = clip_async_render(clip, None, progress, callback, prefetch, backlog, False, async_requests)

    return list(Sentinel.filter(frames))

def find_prop(
    src: vs.VideoNode,
    prop: str,
    op: str | Callable[[float, float], bool] | None,
    ref: float | bool,
    range_length: int = 0,
    async_requests: int | Literal[False] = False,
) -> list[int] | list[tuple[int, int]]:
    """
    Find specific frame props in the clip and return a list of frame ranges that meets the conditions.

    Example usage:
        ```py
        # Return a list of all frames that were marked as combed.
        find_prop(clip, "_Combed", None, True, 0)
        ```

    Args:
        src: Input clip.
        prop: Frame prop to perform checks on.
        op: Conditional operator to apply between prop and ref ("<", "<=", "==", "!=", ">" or ">="). If None, check
            whether a prop is truthy.
        ref: Value to be compared with prop.
        range_length: Amount of frames to finish a sequence, to avoid false negatives. This will create ranges with a
            sequence of start-end tuples.
        async_requests: Whether to render frames non-consecutively.

    Returns:
        Frame ranges at the specified conditions.
    """

    prop_src, callback = prop_compare_cb(src, prop, op, ref, return_frame_n=True)

    aconf = False if async_requests is False else AsyncRenderConf(async_requests, False)
    frames = clip_data_gather(prop_src, f"Searching {prop} {op} {ref}...", callback, aconf)

    if range_length > 0:
        return normalize_list_to_ranges(frames, range_length)

    return frames

def find_prop_rfs(
    clip_a: vs.VideoNode,
    clip_b: vs.VideoNode,
    prop: str,
    op: str | Callable[[float, float], bool] | None,
    prop_ref: float | bool,
    ref: vs.VideoNode | None = None,
    mismatch: bool = False,
) -> vs.VideoNode:
    """
    Conditional replace frames from the original clip with a replacement clip by comparing frame properties.

    Example usage:
        ```py
        # Replace a rescaled clip with the original clip for frames where the error
        # (defined on another clip) is equal to or greater than 0.025.
        find_prop_rfs(scaled, src, "PlaneStatsAverage", ">=", 0.025, err_clip)
        ```

    Args:
        clip_a: Original clip.
        clip_b: Replacement clip.
        prop: Frame prop to perform checks on.
        op: Conditional operator to apply between prop and ref ("<", "<=", "==", "!=", ">" or ">="). If None, check
            whether a prop is truthy. Default: None.
        prop_ref: Value to be compared with prop.
        ref: Optional reference clip to read frame properties from. Default: None.
        mismatch: Accept format or resolution mismatch between clips. Default: False.

    Returns:
        Clip where frames that meet the specified criteria were replaced with a different clip.
    """
    prop_src, callback = prop_compare_cb(ref or clip_a, prop, op, prop_ref, False)

    return replace_ranges(clip_a, clip_b, callback, False, mismatch, prop_src=prop_src)

def prop_compare_cb(
    src: vs.VideoNode,
    prop: str,
    op: str | Callable[[float, float], bool] | None,
    ref: float | bool,
    return_frame_n: bool = False,
) -> (
    tuple[vs.VideoNode, Callable[[int, vs.VideoFrame], bool]]
    | tuple[vs.VideoNode, Callable[[int, vs.VideoFrame], int | SentinelT]]
):
    bool_check = isinstance(ref, bool)
    assert (op is None) if bool_check else (op is not None)

    if isinstance(op, str):
        assert op in _operators

    op_ = _operators[op][0] if isinstance(op, str) else op

    def _cb_return_frame_n(n: int, f: vs.VideoFrame) -> int | SentinelT:
        assert op_
        return Sentinel.check(n, op_(get_prop(f, prop, (float, bool)), ref))

    def _cb_not_return_frame_n(n: int, f: vs.VideoFrame) -> bool:
        assert op_
        return op_(get_prop(f, prop, (float, bool)), ref)

    callback = _cb_return_frame_n if return_frame_n else _cb_not_return_frame_n

    return src, callback