Skip to content

numpy

Classes:

Attributes:

NDT_T module-attribute 

NDT_T = TypeVar('NDT_T', bound=NDArray[Any])

this_backend module-attribute 

this_backend = NUMPY

PyPluginNumpy 

PyPluginNumpy(*args: Any, **kwargs: Any)

Bases: PyPluginNumpyBase[FD_T]

Methods:

Attributes:

Source code 
363
364
365
366
@copy_signature(PyPlugin.__init__)
def __init__(self, *args: Any, **kwargs: Any) -> None:
    from .exceptions import UnavailableBackend
    raise UnavailableBackend(self.backend, self.__class__)

DT class-attribute instance-attribute 

DT: TypeAlias = DT_T

DTA class-attribute instance-attribute 

DTA: TypeAlias = DT_T | SupportsIndexing[DT_T]

DTL class-attribute instance-attribute 

DTL: TypeAlias = SupportsIndexing[DT_T]

backend class-attribute instance-attribute 

backend = this_backend

channels_last instance-attribute 

channels_last: bool

clips instance-attribute 

clips: list[VideoNode] = [norm_clip(clip) for clip in clips] if clips else []

debug class-attribute instance-attribute 

debug: bool = False

fd instance-attribute 

fd: FD_T

filter_data instance-attribute 

filter_data: Type[FD_T]

filter_mode instance-attribute 

filter_mode: FilterMode

input_per_plane instance-attribute 

input_per_plane: bool | list[bool]

is_single_plane instance-attribute 

is_single_plane = [
    bool(format and num_planes == 1) for clip in (ref_clip, *clips)
]

max_clips instance-attribute 

max_clips: int

min_clips instance-attribute 

min_clips: int

options instance-attribute 

options: PyPluginOptions

out_format instance-attribute 

out_format: VideoFormat = format

output_per_plane instance-attribute 

output_per_plane: bool

process_MultiSrcIPF instance-attribute 

process_MultiSrcIPF: MultiSrcIPF_T[DT_T] | None

process_MultiSrcIPP instance-attribute 

process_MultiSrcIPP: MultiSrcIPP_T[DT_T] | None

process_SingleSrcIPF instance-attribute 

process_SingleSrcIPF: SingleSrcIPF_T[DT_T] | None

process_SingleSrcIPP instance-attribute 

process_SingleSrcIPP: SingleSrcIPP_T[DT_T] | None

ref_clip instance-attribute 

ref_clip: VideoNode = norm_clip(ref_clip)

__call__ 

__call__(func: Callable[..., Any]) -> VideoNode
Source code 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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()

ensure_output staticmethod 

ensure_output(func: F_VD) -> F_VD
Source code 
167
168
169
170
171
172
173
@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)

invoke 

invoke() -> VideoNode
Source code 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@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

process staticmethod 

process(mode: Literal[Any]) -> PassthroughC[Any_T[DT_T]]

process(mode: Literal[SingleSrcIPP]) -> PassthroughC[SingleSrcIPP_ST[DT_T]]

process(mode: Literal[MultiSrcIPP]) -> PassthroughC[MultiSrcIPP_ST[DT_T]]

process(mode: Literal[SingleSrcIPF]) -> PassthroughC[SingleSrcIPF_ST[DT_T]]

process(mode: Literal[MultiSrcIPF]) -> PassthroughC[MultiSrcIPF_ST[DT_T]]

process(func: Any_ST[DT_T]) -> Any_ST[DT_T]

process(func: None) -> PassthroughC[PassthroughC[Any_ST[DT_T]]]

process(
    mode_or_func: ProcessMode | ALL_PMODES_ST[DT_T] | None = None,
) -> PassthroughC[ALL_PMODES_ST[DT_T]] | Any_ST[DT_T]
Source code 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@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

PyPluginNumpyBase 

PyPluginNumpyBase(*args: Any, **kwargs: Any)

Bases: PyPluginUnavailableBackendBase[FD_T, DT_T]

Methods:

Attributes:

Source code 
363
364
365
366
@copy_signature(PyPlugin.__init__)
def __init__(self, *args: Any, **kwargs: Any) -> None:
    from .exceptions import UnavailableBackend
    raise UnavailableBackend(self.backend, self.__class__)

DT class-attribute instance-attribute 

DT: TypeAlias = DT_T

DTA class-attribute instance-attribute 

DTA: TypeAlias = DT_T | SupportsIndexing[DT_T]

DTL class-attribute instance-attribute 

DTL: TypeAlias = SupportsIndexing[DT_T]

backend class-attribute instance-attribute 

backend = this_backend

channels_last instance-attribute 

channels_last: bool

clips instance-attribute 

clips: list[VideoNode] = [norm_clip(clip) for clip in clips] if clips else []

debug class-attribute instance-attribute 

debug: bool = False

fd instance-attribute 

fd: FD_T

filter_data instance-attribute 

filter_data: Type[FD_T]

filter_mode instance-attribute 

filter_mode: FilterMode

input_per_plane instance-attribute 

input_per_plane: bool | list[bool]

is_single_plane instance-attribute 

is_single_plane = [
    bool(format and num_planes == 1) for clip in (ref_clip, *clips)
]

max_clips instance-attribute 

max_clips: int

min_clips instance-attribute 

min_clips: int

options instance-attribute 

options: PyPluginOptions

out_format instance-attribute 

out_format: VideoFormat = format

output_per_plane instance-attribute 

output_per_plane: bool

process_MultiSrcIPF instance-attribute 

process_MultiSrcIPF: MultiSrcIPF_T[DT_T] | None

process_MultiSrcIPP instance-attribute 

process_MultiSrcIPP: MultiSrcIPP_T[DT_T] | None

process_SingleSrcIPF instance-attribute 

process_SingleSrcIPF: SingleSrcIPF_T[DT_T] | None

process_SingleSrcIPP instance-attribute 

process_SingleSrcIPP: SingleSrcIPP_T[DT_T] | None

ref_clip instance-attribute 

ref_clip: VideoNode = norm_clip(ref_clip)

__call__ 

__call__(func: Callable[..., Any]) -> VideoNode
Source code 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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()

ensure_output staticmethod 

ensure_output(func: F_VD) -> F_VD
Source code 
167
168
169
170
171
172
173
@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)

invoke 

invoke() -> VideoNode
Source code 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@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

process staticmethod 

process(mode: Literal[Any]) -> PassthroughC[Any_T[DT_T]]

process(mode: Literal[SingleSrcIPP]) -> PassthroughC[SingleSrcIPP_ST[DT_T]]

process(mode: Literal[MultiSrcIPP]) -> PassthroughC[MultiSrcIPP_ST[DT_T]]

process(mode: Literal[SingleSrcIPF]) -> PassthroughC[SingleSrcIPF_ST[DT_T]]

process(mode: Literal[MultiSrcIPF]) -> PassthroughC[MultiSrcIPF_ST[DT_T]]

process(func: Any_ST[DT_T]) -> Any_ST[DT_T]

process(func: None) -> PassthroughC[PassthroughC[Any_ST[DT_T]]]

process(
    mode_or_func: ProcessMode | ALL_PMODES_ST[DT_T] | None = None,
) -> PassthroughC[ALL_PMODES_ST[DT_T]] | Any_ST[DT_T]
Source code 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@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