cuda ¶

def to_tuple(self) -> tuple[str, ...]:
    options = [] if self.options is None else list(self.options)

    if self.use_fast_math:
        options.append('--use_fast_math')

    if self.std:
        options.append(f'--std=c++{self.std:02}')

    if self.extra_vectorization:
        options.append('--extra-device-vectorization')

    return tuple(set(options))

def __call__(
    self, src: NDT_T, dst: NDT_T, *args: Any,
    kernel_size: tuple[int, ...] = ..., block_size: tuple[int, ...] = ..., shared_mem: int = ...
) -> Any:
    ...

def __call__(
    self, src: NDT_T, dst: NDT_T, *args: Any,
    kernel_size: tuple[int, ...] = ..., block_size: tuple[int, ...] = ..., shared_mem: int = ...
) -> Any:
    ...

def __call__(self, func: Callable[..., Any]) -> vs.VideoNode:
    this_args = {'self', 'f', 'src', 'dst', 'plane', 'n'}

    annotations = set(func.__annotations__.keys()) - {'return'}

    if not annotations:
        raise CustomTypeError(f'{self.__class__.__name__}: You must type hint the function!', self.__class__)

    if annotations - this_args:
        raise CustomTypeError(f'{self.__class__.__name__}: Unknown arguments specified!', self.__class__)

    miss_args = this_args - annotations

    if 'self' in annotations:
        func = partial(func, self)
        annotations.remove('self')

    if not miss_args:
        self.process_SingleSrcIPP = self.process_SingleSrcIPF = func
        self.process_MultiSrcIPP = self.process_MultiSrcIPF = func
    else:
        def _wrapper_ipf(src: Any, dst: Any, f: vs.VideoFrame, n: int) -> None:
            curr_locals = locals()
            func(**{name: curr_locals[name] for name in annotations})

        def _wrapper_ipp(src: Any, dst: Any, f: vs.VideoFrame, plane: int, n: int) -> None:
            curr_locals = locals()
            func(**{name: curr_locals[name] for name in annotations})

        self.process_SingleSrcIPF = self.process_MultiSrcIPF = _wrapper_ipf
        self.process_SingleSrcIPP = self.process_MultiSrcIPP = _wrapper_ipp

    return self.invoke()

@staticmethod
def ensure_output(func: F_VD) -> F_VD:
    @wraps(func)
    def _wrapper(self: CLS_T[FD_T], *args: Any, **kwargs: Any) -> Any:
        return self.options.ensure_output(self, func(self, *args, **kwargs))

    return cast(F_VD, _wrapper)

@PyPluginBackendBase.ensure_output
def invoke(self) -> vs.VideoNode:
    output_func = self._invoke_func()

    modify_frame_partial = partial(
        vs.core.std.ModifyFrame, self.ref_clip, (self.ref_clip, *self.clips), output_func
    )

    if self.filter_mode is FilterMode.Serial:
        output = modify_frame_partial()
    elif self.filter_mode is FilterMode.Parallel:
        output = self.ref_clip.std.FrameEval(lambda n: modify_frame_partial())
    else:
        if self.clips:
            output_func_multi = cast(Callable[[tuple[vs.VideoFrame, ...], int], vs.VideoFrame], output_func)

            @frame_eval_async(self.ref_clip)
            async def output(n: int) -> vs.VideoFrame:
                return output_func_multi(await get_frames(self.ref_clip, *self.clips, frame_no=n), n)
        else:
            output_func_single = cast(Callable[[vs.VideoFrame, int], vs.VideoFrame], output_func)

            @frame_eval_async(self.ref_clip)
            async def output(n: int) -> vs.VideoFrame:
                return output_func_single(await get_frame(self.ref_clip, n), n)

    return output

@staticmethod  # type: ignore
def process(mode_or_func: ProcessMode | ALL_PMODES_ST[DT_T] | None = None, /) -> (
    PassthroughC[ALL_PMODES_ST[DT_T]] | ProcessMode.Any_ST[DT_T]
):
    if mode_or_func is None:
        return PyPluginBackendBase.process  # type: ignore
    elif not isinstance(mode_or_func, ProcessMode):
        return PyPluginBackendBase.process(ProcessMode.Any)(mode_or_func)  # type: ignore

    def _wrapper(func: Callable[..., None]) -> Callable[..., None]:
        func.__dict__.update(
            __pybackend_overload__=True,
            __pybackend_mode__=mode_or_func
        )

        return func

    return _wrapper

def __call__(self, func: Callable[..., Any]) -> vs.VideoNode:
    this_args = {'self', 'f', 'src', 'dst', 'plane', 'n'}

    annotations = set(func.__annotations__.keys()) - {'return'}

    if not annotations:
        raise CustomTypeError(f'{self.__class__.__name__}: You must type hint the function!', self.__class__)

    if annotations - this_args:
        raise CustomTypeError(f'{self.__class__.__name__}: Unknown arguments specified!', self.__class__)

    miss_args = this_args - annotations

    if 'self' in annotations:
        func = partial(func, self)
        annotations.remove('self')

    if not miss_args:
        self.process_SingleSrcIPP = self.process_SingleSrcIPF = func
        self.process_MultiSrcIPP = self.process_MultiSrcIPF = func
    else:
        def _wrapper_ipf(src: Any, dst: Any, f: vs.VideoFrame, n: int) -> None:
            curr_locals = locals()
            func(**{name: curr_locals[name] for name in annotations})

        def _wrapper_ipp(src: Any, dst: Any, f: vs.VideoFrame, plane: int, n: int) -> None:
            curr_locals = locals()
            func(**{name: curr_locals[name] for name in annotations})

        self.process_SingleSrcIPF = self.process_MultiSrcIPF = _wrapper_ipf
        self.process_SingleSrcIPP = self.process_MultiSrcIPP = _wrapper_ipp

    return self.invoke()

@staticmethod
def ensure_output(func: F_VD) -> F_VD:
    @wraps(func)
    def _wrapper(self: CLS_T[FD_T], *args: Any, **kwargs: Any) -> Any:
        return self.options.ensure_output(self, func(self, *args, **kwargs))

    return cast(F_VD, _wrapper)

@PyPluginBackendBase.ensure_output
def invoke(self) -> vs.VideoNode:
    output_func = self._invoke_func()

    modify_frame_partial = partial(
        vs.core.std.ModifyFrame, self.ref_clip, (self.ref_clip, *self.clips), output_func
    )

    if self.filter_mode is FilterMode.Serial:
        output = modify_frame_partial()
    elif self.filter_mode is FilterMode.Parallel:
        output = self.ref_clip.std.FrameEval(lambda n: modify_frame_partial())
    else:
        if self.clips:
            output_func_multi = cast(Callable[[tuple[vs.VideoFrame, ...], int], vs.VideoFrame], output_func)

            @frame_eval_async(self.ref_clip)
            async def output(n: int) -> vs.VideoFrame:
                return output_func_multi(await get_frames(self.ref_clip, *self.clips, frame_no=n), n)
        else:
            output_func_single = cast(Callable[[vs.VideoFrame, int], vs.VideoFrame], output_func)

            @frame_eval_async(self.ref_clip)
            async def output(n: int) -> vs.VideoFrame:
                return output_func_single(await get_frame(self.ref_clip, n), n)

    return output

@staticmethod  # type: ignore
def process(mode_or_func: ProcessMode | ALL_PMODES_ST[DT_T] | None = None, /) -> (
    PassthroughC[ALL_PMODES_ST[DT_T]] | ProcessMode.Any_ST[DT_T]
):
    if mode_or_func is None:
        return PyPluginBackendBase.process  # type: ignore
    elif not isinstance(mode_or_func, ProcessMode):
        return PyPluginBackendBase.process(ProcessMode.Any)(mode_or_func)  # type: ignore

    def _wrapper(func: Callable[..., None]) -> Callable[..., None]:
        func.__dict__.update(
            __pybackend_overload__=True,
            __pybackend_mode__=mode_or_func
        )

        return func

    return _wrapper

def ensure_output(self, plugin: PyPlugin[FD_T], clip: vs.VideoNode) -> vs.VideoNode:
    assert plugin.ref_clip.format
    assert (fmt := clip.format)

    if plugin.out_format.id != plugin.ref_clip.format.id:
        clip = clip.resize.Bicubic(format=plugin.out_format.id, dither_type='none')

    if self.shift_chroma and fmt.num_planes == 3:
        clip = clip.std.Expr(['', 'x 0.5 -'])

    return clip

def norm_clip(self, clip: vs.VideoNode | None) -> vs.VideoNode | None:
    if not clip:
        return clip

    assert (fmt := clip.format)

    if self.force_precision:
        if fmt.sample_type is not vs.FLOAT or fmt.bits_per_sample != self.force_precision:
            clip = clip.resize.Point(
                format=fmt.replace(sample_type=vs.FLOAT, bits_per_sample=self.force_precision).id,
                dither_type='none'
            )

    if self.shift_chroma:
        if fmt.sample_type is not vs.FLOAT and self.force_precision != 32:
            raise CustomValueError(
                f'{self.__class__.__name__}: You need to have a clip with float sample type for shift_chroma=True!'
            )

        if fmt.num_planes == 3:
            clip = clip.std.Expr(['', 'x 0.5 +'])

    return clip