Skip to content

base

Classes:

PyPlugin 

PyPlugin(
    ref_clip: VideoNode,
    clips: list[VideoNode] | None = None,
    *,
    filter_mode: FilterMode | None = None,
    options: PyPluginOptions | None = None,
    input_per_plane: bool | list[bool] | None = None,
    output_per_plane: bool | None = None,
    channels_last: bool | None = None,
    min_clips: int | None = None,
    max_clips: int | None = None,
    **kwargs: Any
)

Bases: PyPluginBase[FD_T, memoryview]

Methods:

Attributes:

Source code 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def __init__(
    self,
    ref_clip: vs.VideoNode,
    clips: list[vs.VideoNode] | None = None,
    *,
    filter_mode: FilterMode | None = None,
    options: PyPluginOptions | None = None,
    input_per_plane: bool | list[bool] | None = None,
    output_per_plane: bool | None = None,
    channels_last: bool | None = None,
    min_clips: int | None = None,
    max_clips: int | None = None,
    **kwargs: Any
) -> None:
    assert ref_clip.format

    arguments = [
        (filter_mode, 'filter_mode', FilterMode.Parallel),
        (options, 'options', PyPluginOptions()),
        (input_per_plane, 'input_per_plane', True),
        (output_per_plane, 'output_per_plane', True),
        (channels_last, 'channels_last', False),
        (min_clips, 'min_clips', 1),
        (max_clips, 'max_clips', -1)
    ]

    for value, name, default in arguments:
        if value is not None:
            setattr(self, name, value)
        elif not hasattr(self, name):
            setattr(self, name, default)

    self.out_format = ref_clip.format

    self.ref_clip = self.options.norm_clip(ref_clip)

    self.clips = [self.options.norm_clip(clip) for clip in clips] if clips else []

    self_annotations = self.__annotations__.keys()

    for name, value in list(kwargs.items()):
        if name in self_annotations:
            setattr(self, name, value)
            kwargs.pop(name)

    if callable(self.filter_data):
        self.fd = self.filter_data(**kwargs)  # type: ignore
    else:
        self.fd = None  # type: ignore

    n_clips = 1 + len(self.clips)

    inputs_per_plane = self.input_per_plane

    if not isinstance(inputs_per_plane, list):
        inputs_per_plane = [inputs_per_plane]

    for _ in range((1 + len(self.clips)) - len(inputs_per_plane)):
        inputs_per_plane.append(inputs_per_plane[-1])

    self._input_per_plane = inputs_per_plane

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

    if n_clips < self.min_clips or (self.max_clips > 0 and n_clips > self.max_clips):
        max_clips_str = 'inf' if self.max_clips == -1 else self.max_clips
        raise CustomIndexError(
            f'You must pass {self.min_clips} <= n clips <= {max_clips_str}!', self.__class__
        )

    if not self.output_per_plane and (ref_clip.format.subsampling_w or ref_clip.format.subsampling_h):
        raise CustomValueError(
            'You can\'t have output_per_plane=False with a subsampled clip!', self.__class__
        )

    for idx, clip, ipp in zip(count(-1), (self.ref_clip, *self.clips), self._input_per_plane):
        assert clip.format
        if not ipp and (clip.format.subsampling_w or clip.format.subsampling_h):
            raise InvalidSubsamplingError(
                self.__class__,
                'You can\'t have input_per_plane=False with a subsampled clip! ({clip_type})',
                clip_type='Ref Clip' if idx == -1 else f'Clip Index: {idx}'
            )

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 

backend: PyBackend = NONE

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

PyPluginBase 

PyPluginBase(
    ref_clip: VideoNode,
    clips: list[VideoNode] | None = None,
    *,
    filter_mode: FilterMode | None = None,
    options: PyPluginOptions | None = None,
    input_per_plane: bool | list[bool] | None = None,
    output_per_plane: bool | None = None,
    channels_last: bool | None = None,
    min_clips: int | None = None,
    max_clips: int | None = None,
    **kwargs: Any
)

Bases: Generic[FD_T, DT_T], PyPluginBackendBase[DT_T]

Methods:

Attributes:

Source code 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def __init__(
    self,
    ref_clip: vs.VideoNode,
    clips: list[vs.VideoNode] | None = None,
    *,
    filter_mode: FilterMode | None = None,
    options: PyPluginOptions | None = None,
    input_per_plane: bool | list[bool] | None = None,
    output_per_plane: bool | None = None,
    channels_last: bool | None = None,
    min_clips: int | None = None,
    max_clips: int | None = None,
    **kwargs: Any
) -> None:
    assert ref_clip.format

    arguments = [
        (filter_mode, 'filter_mode', FilterMode.Parallel),
        (options, 'options', PyPluginOptions()),
        (input_per_plane, 'input_per_plane', True),
        (output_per_plane, 'output_per_plane', True),
        (channels_last, 'channels_last', False),
        (min_clips, 'min_clips', 1),
        (max_clips, 'max_clips', -1)
    ]

    for value, name, default in arguments:
        if value is not None:
            setattr(self, name, value)
        elif not hasattr(self, name):
            setattr(self, name, default)

    self.out_format = ref_clip.format

    self.ref_clip = self.options.norm_clip(ref_clip)

    self.clips = [self.options.norm_clip(clip) for clip in clips] if clips else []

    self_annotations = self.__annotations__.keys()

    for name, value in list(kwargs.items()):
        if name in self_annotations:
            setattr(self, name, value)
            kwargs.pop(name)

    if callable(self.filter_data):
        self.fd = self.filter_data(**kwargs)  # type: ignore
    else:
        self.fd = None  # type: ignore

    n_clips = 1 + len(self.clips)

    inputs_per_plane = self.input_per_plane

    if not isinstance(inputs_per_plane, list):
        inputs_per_plane = [inputs_per_plane]

    for _ in range((1 + len(self.clips)) - len(inputs_per_plane)):
        inputs_per_plane.append(inputs_per_plane[-1])

    self._input_per_plane = inputs_per_plane

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

    if n_clips < self.min_clips or (self.max_clips > 0 and n_clips > self.max_clips):
        max_clips_str = 'inf' if self.max_clips == -1 else self.max_clips
        raise CustomIndexError(
            f'You must pass {self.min_clips} <= n clips <= {max_clips_str}!', self.__class__
        )

    if not self.output_per_plane and (ref_clip.format.subsampling_w or ref_clip.format.subsampling_h):
        raise CustomValueError(
            'You can\'t have output_per_plane=False with a subsampled clip!', self.__class__
        )

    for idx, clip, ipp in zip(count(-1), (self.ref_clip, *self.clips), self._input_per_plane):
        assert clip.format
        if not ipp and (clip.format.subsampling_w or clip.format.subsampling_h):
            raise InvalidSubsamplingError(
                self.__class__,
                'You can\'t have input_per_plane=False with a subsampled clip! ({clip_type})',
                clip_type='Ref Clip' if idx == -1 else f'Clip Index: {idx}'
            )

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 

backend: PyBackend = NONE

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

PyPluginOptions dataclass 

PyPluginOptions(force_precision: int | None = None, shift_chroma: bool = False)

Methods:

Attributes:

force_precision class-attribute instance-attribute 

force_precision: int | None = None

shift_chroma class-attribute instance-attribute 

shift_chroma: bool = False

ensure_output 

ensure_output(plugin: PyPlugin[FD_T], clip: VideoNode) -> VideoNode
Source code 
61
62
63
64
65
66
67
68
69
70
71
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

norm_clip 

norm_clip(clip: VideoNode) -> VideoNode

norm_clip(clip: None) -> None

norm_clip(clip: VideoNode | None) -> VideoNode | None
Source code 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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

PyPluginUnavailableBackend 

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

Bases: PyPluginUnavailableBackendBase[FD_T, Any]

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 

backend: PyBackend = NONE

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

PyPluginUnavailableBackendBase 

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

Bases: PyPluginBase[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 

backend: PyBackend = NONE

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