onnx ¶

def calc_tilesize(self, clip: vs.VideoNode) -> tuple[tuple[int, int], tuple[int, int]]:
    """
    Reimplementation of vsmlrt.calc_tilesize helper function
    """

    def calc_size(dimension: int, tiles: int, overlap: int, multiple: int) -> int:
        return math.ceil((dimension + 2 * overlap * (tiles - 1)) / (tiles * multiple)) * multiple

    overlap_w = self.overlap_w
    overlap_h = self.overlap_h

    if isinstance(self.tilesize, tuple):
        tile_w, tile_h = self.tilesize
    elif isinstance(self.tilesize, int):
        tile_w, tile_h = self.tilesize, self.tilesize
    else:
        if self.tiles is None:
            overlap_w = 0
            overlap_h = 0
            tile_w = clip.width
            tile_h = clip.height
        elif isinstance(self.tiles, int):
            tile_w = calc_size(clip.width, self.tiles, self.overlap_w, self.multiple)
            tile_h = calc_size(clip.height, self.tiles, self.overlap_h, self.multiple)
        else:
            tile_w = calc_size(clip.width, self.tiles[0], self.overlap_w, self.multiple)
            tile_h = calc_size(clip.height, self.tiles[1], self.overlap_h, self.multiple)

        if tile_w % self.multiple != 0 or tile_h % self.multiple != 0:
            raise CustomValueError(
                f"Tile size ({tile_w}, {tile_h}) must be divisible by {self.multiple}",
                self.__class__,
            )

    return (tile_w, tile_h), (overlap_w, overlap_h)

@classmethod
def ensure_obj(
    cls,
    scaler: str | type[Self] | Self | None = None,
    /,
    func_except: FuncExcept | None = None,
) -> Self:
    """
    Ensure that the input is a scaler instance, resolving it if necessary.

    Args:
        scaler: Scaler identifier (string, class, or instance).
        func_except: Function returned for custom error handling.

    Returns:
        Scaler instance.
    """
    return _base_ensure_obj(cls, scaler, func_except)

@classmethod
def from_param(
    cls,
    scaler: str | type[Self] | Self | None = None,
    /,
    func_except: FuncExcept | None = None,
) -> type[Self]:
    """
    Resolve and return a scaler type from a given input (string, type, or instance).

    Args:
        scaler: Scaler identifier (string, class, or instance).
        func_except: Function returned for custom error handling.

    Returns:
        Resolved scaler type.
    """
    return _base_from_param(cls, scaler, cls._err_class, func_except)

def get_scale_args(
    self,
    clip: vs.VideoNode,
    shift: tuple[TopShift, LeftShift] = (0, 0),
    width: int | None = None,
    height: int | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """
    Generate the keyword arguments used for scaling.

    Args:
        clip: The source clip.
        shift: Subpixel shift (top, left).
        width: Target width.
        height: Target height.
        **kwargs: Extra parameters to merge.

    Returns:
        Final dictionary of keyword arguments for the scale function.
    """
    return {"width": width, "height": height, "src_top": shift[0], "src_left": shift[1]} | self.kwargs | kwargs

@classproperty.cached
@classmethod
def implemented_funcs(cls) -> frozenset[str]:
    """
    Returns a set of function names that are implemented in the current class and the parent classes.

    These functions determine which keyword arguments will be extracted from the `__init__` method.

    Returns:
        Frozen set of function names.
    """
    return frozenset(func for klass in cls.mro() for func in getattr(klass, "_implemented_funcs", ()))

def inference(self, clip: vs.VideoNode, **kwargs: Any) -> vs.VideoNode:
    """
    Runs inference on the given video clip using the configured model and backend.
    """

    tilesize, overlaps = self.calc_tilesize(clip)

    logger.debug("%s: Passing clip to inference: %r", self.inference, clip.format)
    logger.debug("%s: Passing model: %s", self.inference, self.model)
    logger.debug("%s: Passing tiles size: %s", self.inference, tilesize)
    logger.debug("%s: Passing overlaps: %s", self.inference, overlaps)
    logger.debug("%s: Passing extra kwargs: %s", self.inference, kwargs)

    return self.backend.inference(clip, self.model, overlaps, tilesize, flexible=False, **kwargs)

@classproperty
@classmethod
def is_abstract(cls) -> bool:
    """Return True if this class can't be instantiated."""
    return _is_base_scaler_abstract(cls)

@BaseScalerMeta.cachedproperty
def kernel_radius(self) -> int:
    """
    Return the effective kernel radius for the scaler.

    Raises:
        CustomNotImplementedError: If no kernel radius is defined.

    Returns:
        Kernel radius.
    """
    ...

def postprocess_clip(self, clip: vs.VideoNode, input_clip: vs.VideoNode, **kwargs: Any) -> vs.VideoNode:
    """
    Handles postprocessing of the model's output after inference.
    """
    return clip

def preprocess_clip(self, clip: vs.VideoNode, **kwargs: Any) -> vs.VideoNode:
    return super().preprocess_clip(get_y(clip), **kwargs)

def scale(
    self,
    clip: vs.VideoNode,
    width: int | None = None,
    height: int | None = None,
    shift: tuple[float, float] = (0, 0),
    **kwargs: Any,
) -> vs.VideoNode:
    """
    Scale the given clip using the ONNX model.

    Args:
        clip: The input clip to be scaled.
        width: The target width for scaling. If None, the width of the input clip will be used.
        height: The target height for scaling. If None, the height of the input clip will be used.
        shift: A tuple representing the shift values for the x and y axes.
        **kwargs: Additional arguments to be passed to the `preprocess_clip`, `postprocess_clip`, `inference`, and
            `_final_scale` methods. Use the prefix `preprocess_` or `postprocess_` to pass an argument to the
            respective method. Use the prefix `inference_` to pass an argument to the inference method.

    Returns:
        The scaled clip.
    """
    if clip.format.sample_type != vs.SampleType.FLOAT:
        raise CustomRuntimeError("Only FLOAT formats are supported.", self.scale)

    width, height = self._wh_norm(clip, width, height)

    preprocess_kwargs = dict[str, Any]()
    postprocess_kwargs = dict[str, Any]()
    inference_kwargs = dict[str, Any]()

    for k in kwargs.copy():
        for prefix, ckwargs in zip(
            ("preprocess_", "postprocess_", "inference_"), (preprocess_kwargs, postprocess_kwargs, inference_kwargs)
        ):
            if k.startswith(prefix):
                ckwargs[k.removeprefix(prefix)] = kwargs.pop(k)
                break

    logger.debug("%s: Preprocess kwargs: %s", self.scale, preprocess_kwargs)
    logger.debug("%s: Postprocess kwargs: %s", self.scale, postprocess_kwargs)
    logger.debug("%s: Inference kwargs: %s", self.scale, inference_kwargs)

    wclip = self.preprocess_clip(clip, **preprocess_kwargs)

    if 0 not in {clip.width, clip.height}:
        scaled = self.inference(wclip, **inference_kwargs)
    else:
        logger.debug("%s: Variable resolution clip detected...", self.scale)

        if not isinstance(self.backend, Backend.TRT) or self.backend.static_shape:
            raise CustomValueError("Only TRT backends support static_shape=False", self.__class__, self.backend)

        scaled = ProcessVariableResClip.from_func(
            wclip, lambda c: self.inference(c, **inference_kwargs), False, wclip.format, self.max_instances
        )

    scaled = self.postprocess_clip(scaled, clip, **postprocess_kwargs)

    return self._finish_scale(scaled, clip, width, height, shift, **kwargs)

def supersample(
    self, clip: vs.VideoNode, rfactor: float = 2.0, shift: tuple[TopShift, LeftShift] = (0, 0), **kwargs: Any
) -> vs.VideoNode:
    """
    Supersample a clip by a given scaling factor.

    Keyword arguments passed during initialization are automatically injected here,
    unless explicitly overridden by the arguments provided at call time.
    Only arguments that match named parameters in this method are injected.

    Args:
        clip: The source clip.
        rfactor: Scaling factor for supersampling.
        shift: Subpixel shift (top, left) applied during scaling.
        **kwargs: Additional arguments forwarded to the scale function.

    Raises:
        CustomValueError: If resulting resolution is non-positive.

    Returns:
        The supersampled clip.
    """
    assert check_variable_resolution(clip, self.supersample)

    dst_width, dst_height = ceil(clip.width * rfactor), ceil(clip.height * rfactor)

    if max(dst_width, dst_height) <= 0.0:
        raise CustomValueError(
            'Multiplying the resolution by "rfactor" must result in a positive resolution!',
            self.supersample,
            rfactor,
        )

    return self.scale(clip, dst_width, dst_height, shift, **kwargs)

def calc_tilesize(self, clip: vs.VideoNode) -> tuple[tuple[int, int], tuple[int, int]]:
    """
    Reimplementation of vsmlrt.calc_tilesize helper function
    """

    def calc_size(dimension: int, tiles: int, overlap: int, multiple: int) -> int:
        return math.ceil((dimension + 2 * overlap * (tiles - 1)) / (tiles * multiple)) * multiple

    overlap_w = self.overlap_w
    overlap_h = self.overlap_h

    if isinstance(self.tilesize, tuple):
        tile_w, tile_h = self.tilesize
    elif isinstance(self.tilesize, int):
        tile_w, tile_h = self.tilesize, self.tilesize
    else:
        if self.tiles is None:
            overlap_w = 0
            overlap_h = 0
            tile_w = clip.width
            tile_h = clip.height
        elif isinstance(self.tiles, int):
            tile_w = calc_size(clip.width, self.tiles, self.overlap_w, self.multiple)
            tile_h = calc_size(clip.height, self.tiles, self.overlap_h, self.multiple)
        else:
            tile_w = calc_size(clip.width, self.tiles[0], self.overlap_w, self.multiple)
            tile_h = calc_size(clip.height, self.tiles[1], self.overlap_h, self.multiple)

        if tile_w % self.multiple != 0 or tile_h % self.multiple != 0:
            raise CustomValueError(
                f"Tile size ({tile_w}, {tile_h}) must be divisible by {self.multiple}",
                self.__class__,
            )

    return (tile_w, tile_h), (overlap_w, overlap_h)

@classmethod
def ensure_obj(
    cls,
    scaler: str | type[Self] | Self | None = None,
    /,
    func_except: FuncExcept | None = None,
) -> Self:
    """
    Ensure that the input is a scaler instance, resolving it if necessary.

    Args:
        scaler: Scaler identifier (string, class, or instance).
        func_except: Function returned for custom error handling.

    Returns:
        Scaler instance.
    """
    return _base_ensure_obj(cls, scaler, func_except)

@classmethod
def from_param(
    cls,
    scaler: str | type[Self] | Self | None = None,
    /,
    func_except: FuncExcept | None = None,
) -> type[Self]:
    """
    Resolve and return a scaler type from a given input (string, type, or instance).

    Args:
        scaler: Scaler identifier (string, class, or instance).
        func_except: Function returned for custom error handling.

    Returns:
        Resolved scaler type.
    """
    return _base_from_param(cls, scaler, cls._err_class, func_except)

def get_scale_args(
    self,
    clip: vs.VideoNode,
    shift: tuple[TopShift, LeftShift] = (0, 0),
    width: int | None = None,
    height: int | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """
    Generate the keyword arguments used for scaling.

    Args:
        clip: The source clip.
        shift: Subpixel shift (top, left).
        width: Target width.
        height: Target height.
        **kwargs: Extra parameters to merge.

    Returns:
        Final dictionary of keyword arguments for the scale function.
    """
    return {"width": width, "height": height, "src_top": shift[0], "src_left": shift[1]} | self.kwargs | kwargs

@classproperty.cached
@classmethod
def implemented_funcs(cls) -> frozenset[str]:
    """
    Returns a set of function names that are implemented in the current class and the parent classes.

    These functions determine which keyword arguments will be extracted from the `__init__` method.

    Returns:
        Frozen set of function names.
    """
    return frozenset(func for klass in cls.mro() for func in getattr(klass, "_implemented_funcs", ()))

def inference(self, clip: vs.VideoNode, **kwargs: Any) -> vs.VideoNode:
    """
    Runs inference on the given video clip using the configured model and backend.
    """

    tilesize, overlaps = self.calc_tilesize(clip)

    logger.debug("%s: Passing clip to inference: %r", self.inference, clip.format)
    logger.debug("%s: Passing model: %s", self.inference, self.model)
    logger.debug("%s: Passing tiles size: %s", self.inference, tilesize)
    logger.debug("%s: Passing overlaps: %s", self.inference, overlaps)
    logger.debug("%s: Passing extra kwargs: %s", self.inference, kwargs)

    return self.backend.inference(clip, self.model, overlaps, tilesize, flexible=False, **kwargs)

@classproperty
@classmethod
def is_abstract(cls) -> bool:
    """Return True if this class can't be instantiated."""
    return _is_base_scaler_abstract(cls)

@BaseScalerMeta.cachedproperty
def kernel_radius(self) -> int:
    """
    Return the effective kernel radius for the scaler.

    Raises:
        CustomNotImplementedError: If no kernel radius is defined.

    Returns:
        Kernel radius.
    """
    ...

def postprocess_clip(self, clip: vs.VideoNode, input_clip: vs.VideoNode, **kwargs: Any) -> vs.VideoNode:
    """
    Handles postprocessing of the model's output after inference.
    """
    return clip

def preprocess_clip(self, clip: vs.VideoNode, **kwargs: Any) -> vs.VideoNode:
    return super().preprocess_clip(get_y(clip), **kwargs)

def scale(
    self,
    clip: vs.VideoNode,
    width: int | None = None,
    height: int | None = None,
    shift: tuple[float, float] = (0, 0),
    **kwargs: Any,
) -> vs.VideoNode:
    """
    Scale the given clip using the ONNX model.

    Args:
        clip: The input clip to be scaled.
        width: The target width for scaling. If None, the width of the input clip will be used.
        height: The target height for scaling. If None, the height of the input clip will be used.
        shift: A tuple representing the shift values for the x and y axes.
        **kwargs: Additional arguments to be passed to the `preprocess_clip`, `postprocess_clip`, `inference`, and
            `_final_scale` methods. Use the prefix `preprocess_` or `postprocess_` to pass an argument to the
            respective method. Use the prefix `inference_` to pass an argument to the inference method.

    Returns:
        The scaled clip.
    """
    if clip.format.sample_type != vs.SampleType.FLOAT:
        raise CustomRuntimeError("Only FLOAT formats are supported.", self.scale)

    width, height = self._wh_norm(clip, width, height)

    preprocess_kwargs = dict[str, Any]()
    postprocess_kwargs = dict[str, Any]()
    inference_kwargs = dict[str, Any]()

    for k in kwargs.copy():
        for prefix, ckwargs in zip(
            ("preprocess_", "postprocess_", "inference_"), (preprocess_kwargs, postprocess_kwargs, inference_kwargs)
        ):
            if k.startswith(prefix):
                ckwargs[k.removeprefix(prefix)] = kwargs.pop(k)
                break

    logger.debug("%s: Preprocess kwargs: %s", self.scale, preprocess_kwargs)
    logger.debug("%s: Postprocess kwargs: %s", self.scale, postprocess_kwargs)
    logger.debug("%s: Inference kwargs: %s", self.scale, inference_kwargs)

    wclip = self.preprocess_clip(clip, **preprocess_kwargs)

    if 0 not in {clip.width, clip.height}:
        scaled = self.inference(wclip, **inference_kwargs)
    else:
        logger.debug("%s: Variable resolution clip detected...", self.scale)

        if not isinstance(self.backend, Backend.TRT) or self.backend.static_shape:
            raise CustomValueError("Only TRT backends support static_shape=False", self.__class__, self.backend)

        scaled = ProcessVariableResClip.from_func(
            wclip, lambda c: self.inference(c, **inference_kwargs), False, wclip.format, self.max_instances
        )

    scaled = self.postprocess_clip(scaled, clip, **postprocess_kwargs)

    return self._finish_scale(scaled, clip, width, height, shift, **kwargs)