util ¶

@classproperty.cached
@classmethod
def default_scaler(cls) -> type[DefaultScalerT]:
    """
    Retrieve the default scaler type associated with this class.

    This method looks up the type parameter that matches `cls.__scaler_type_param_name__`
    in the class's type parameters (`__type_params__`) or, if not found, in the MRO.

    If no default scaler type is defined, an error is raised.

    Raises:
        CustomRuntimeError: If neither the class nor any of its bases define a type parameter matching
            `__scaler_type_param_name__`.
        CustomRuntimeError: If the type parameter exists but does not specify a default scaler type.

    Returns:
        The default scaler type defined for the class.
    """
    type_param = next(
        (t for t in cls.__dict__.get("__type_params__", ()) if t.__name__ == cls.__scaler_type_param_name__), None
    )

    if type_param is None:
        for base in cls.__mro__:
            type_param = next(
                (t for t in base.__type_params__ if t.__name__ == cls.__scaler_type_param_name__), None
            )

            if type_param is not None:
                break
        else:
            raise CustomRuntimeError(
                f"At least one base class of '{cls.__name__}' must define a "
                f"default scaler type parameter named '{cls.__scaler_type_param_name__}'.",
                cls,
            )

    if not isinstance(type_param, TypeVar):
        raise CustomTypeError(f"Expected a TypeVar for scaler type parameter, got {type(type_param).__name__}", cls)

    if type_param.has_default():
        return type_param.__default__

    raise CustomRuntimeError(
        f"'{cls.__name__}' does not define a default scaler type and cannot be instantiated.", cls
    )

@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)

@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", ()))

@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.
    """
    ...

@classproperty.cached
@classmethod
def default_scaler(cls) -> type[DefaultScalerT]:
    """
    Retrieve the default scaler type associated with this class.

    This method looks up the type parameter that matches `cls.__scaler_type_param_name__`
    in the class's type parameters (`__type_params__`) or, if not found, in the MRO.

    If no default scaler type is defined, an error is raised.

    Raises:
        CustomRuntimeError: If neither the class nor any of its bases define a type parameter matching
            `__scaler_type_param_name__`.
        CustomRuntimeError: If the type parameter exists but does not specify a default scaler type.

    Returns:
        The default scaler type defined for the class.
    """
    type_param = next(
        (t for t in cls.__dict__.get("__type_params__", ()) if t.__name__ == cls.__scaler_type_param_name__), None
    )

    if type_param is None:
        for base in cls.__mro__:
            type_param = next(
                (t for t in base.__type_params__ if t.__name__ == cls.__scaler_type_param_name__), None
            )

            if type_param is not None:
                break
        else:
            raise CustomRuntimeError(
                f"At least one base class of '{cls.__name__}' must define a "
                f"default scaler type parameter named '{cls.__scaler_type_param_name__}'.",
                cls,
            )

    if not isinstance(type_param, TypeVar):
        raise CustomTypeError(f"Expected a TypeVar for scaler type parameter, got {type(type_param).__name__}", cls)

    if type_param.has_default():
        return type_param.__default__

    raise CustomRuntimeError(
        f"'{cls.__name__}' does not define a default scaler type and cannot be instantiated.", cls
    )

@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)

@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", ()))

@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.
    """
    ...

@classmethod
def from_func[**P](
    cls,
    func: Callable[Concatenate[vs.VideoNode, P], vs.VideoNode] | None = None,
    /,
    sigmoid: bool | tuple[Slope, Center] = False,
    resampler: ResamplerLike = Catrom,
    out_fmt: int | VideoFormatLike | HoldsVideoFormat | None = None,
) -> (
    Callable[Concatenate[vs.VideoNode, P], vs.VideoNode]
    | Callable[
        [Callable[Concatenate[vs.VideoNode, P], vs.VideoNode]], Callable[Concatenate[vs.VideoNode, P], vs.VideoNode]
    ]
):
    """
    Decorator version of LinearLight.
    """

    if func is None:
        return partial(cls.from_func, sigmoid=sigmoid, resampler=resampler, out_fmt=out_fmt)

    @wraps(func)
    def _wrapped(clip: vs.VideoNode, *args: P.args, **kwargs: P.kwargs) -> vs.VideoNode:
        with cls(clip, sigmoid, resampler, out_fmt) as ll:
            ll.linear = func(clip, *args, **kwargs)
        return ll.out

    return _wrapped

def get_linear(self) -> vs.VideoNode:
    """
    Getter for `linear` cached property.
    """
    if (
        self.ll._wclip.format.sample_type,
        self.ll._wclip.format.subsampling_w,
        self.ll._wclip.format.subsampling_h,
    ) != (vs.FLOAT, 0, 0):
        fmt = vs.RGBS if self.ll._wclip.format.color_family in (vs.YUV, vs.RGB) else vs.GRAYS
    else:
        fmt = self.ll._wclip

    wclip = self.ll._resampler.resample(
        self.ll._wclip, fmt, matrix_in=self.ll._matrix, transfer_in=self.ll._curve, transfer=Transfer.LINEAR
    )

    if self.ll.sigmoid:
        from vsexprtools import norm_expr

        wclip = norm_expr(
            wclip,
            "{center} 1 {slope} / 1 x 0 1 clamp {scale} * {offset} + / 1 - log * -",
            center=self.ll._scenter,
            slope=self.ll._sslope,
            scale=self.ll._sscale,
            offset=self.ll._soffset,
            func=self.__class__,
        )

    return wclip

@cachedproperty
def out(self) -> vs.VideoNode:
    if not self.ll._exited:
        raise CustomRuntimeError(
            "You can't get .out while still inside of the context manager!", func=self.__class__
        )

    if not hasattr(self, "_linear"):
        raise CustomValueError("You need to set .linear before getting .out!", self.__class__)

    if self.ll.sigmoid:
        from vsexprtools import norm_expr

        processed = norm_expr(
            self._linear,
            "1 1 {slope} {center} x 0 1 clamp - * exp + / {offset} - {scale} /",
            slope=self.ll._sslope,
            center=self.ll._scenter,
            offset=self.ll._soffset,
            scale=self.ll._sscale,
            func=self.__class__,
        )
    else:
        processed = self._linear

    processed = vs.core.resize2.Point(processed, transfer_in=Transfer.LINEAR, transfer=self.ll._curve)

    return resample_to(processed, self.ll._fmt, self.ll._matrix, self.ll._resampler)

def set_linear(self, processed: vs.VideoNode) -> None:
    """
    Setter for `linear` cached property.
    """
    if self.ll._exited:
        raise CustomRuntimeError(
            "You can't set .linear after going out of the context manager!", func=self.__class__
        )
    self._linear = processed

@classproperty.cached
@classmethod
def default_scaler(cls) -> type[DefaultScalerT]:
    """
    Retrieve the default scaler type associated with this class.

    This method looks up the type parameter that matches `cls.__scaler_type_param_name__`
    in the class's type parameters (`__type_params__`) or, if not found, in the MRO.

    If no default scaler type is defined, an error is raised.

    Raises:
        CustomRuntimeError: If neither the class nor any of its bases define a type parameter matching
            `__scaler_type_param_name__`.
        CustomRuntimeError: If the type parameter exists but does not specify a default scaler type.

    Returns:
        The default scaler type defined for the class.
    """
    type_param = next(
        (t for t in cls.__dict__.get("__type_params__", ()) if t.__name__ == cls.__scaler_type_param_name__), None
    )

    if type_param is None:
        for base in cls.__mro__:
            type_param = next(
                (t for t in base.__type_params__ if t.__name__ == cls.__scaler_type_param_name__), None
            )

            if type_param is not None:
                break
        else:
            raise CustomRuntimeError(
                f"At least one base class of '{cls.__name__}' must define a "
                f"default scaler type parameter named '{cls.__scaler_type_param_name__}'.",
                cls,
            )

    if not isinstance(type_param, TypeVar):
        raise CustomTypeError(f"Expected a TypeVar for scaler type parameter, got {type(type_param).__name__}", cls)

    if type_param.has_default():
        return type_param.__default__

    raise CustomRuntimeError(
        f"'{cls.__name__}' does not define a default scaler type and cannot be instantiated.", cls
    )

@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", ()))

@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 scale(
    self,
    clip: vs.VideoNode,
    width: int | None = None,
    height: int | None = None,
    shift: tuple[TopShift, LeftShift] = (0, 0),
    **kwargs: Any,
) -> vs.VideoNode:
    """
    Scale a clip to a specified resolution.

    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.
        width: Target width (defaults to clip width if None).
        height: Target height (defaults to clip height if None).
        shift: Subpixel shift (top, left) applied during scaling.
        **kwargs: Additional arguments forwarded to the scale function.

    Returns:
        The scaled clip.
    """
    width, height = self._wh_norm(clip, width, height)

    scale_args = self.get_scale_args(clip, shift, width, height, **kwargs)

    logger.debug("%s: Passing clip: %r; arguments: %s", self.scale, clip, scale_args)

    return self.scale_function(clip, **scale_args)

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)

@classproperty.cached
@classmethod
def default_scaler(cls) -> type[DefaultScalerT]:
    """
    Retrieve the default scaler type associated with this class.

    This method looks up the type parameter that matches `cls.__scaler_type_param_name__`
    in the class's type parameters (`__type_params__`) or, if not found, in the MRO.

    If no default scaler type is defined, an error is raised.

    Raises:
        CustomRuntimeError: If neither the class nor any of its bases define a type parameter matching
            `__scaler_type_param_name__`.
        CustomRuntimeError: If the type parameter exists but does not specify a default scaler type.

    Returns:
        The default scaler type defined for the class.
    """
    type_param = next(
        (t for t in cls.__dict__.get("__type_params__", ()) if t.__name__ == cls.__scaler_type_param_name__), None
    )

    if type_param is None:
        for base in cls.__mro__:
            type_param = next(
                (t for t in base.__type_params__ if t.__name__ == cls.__scaler_type_param_name__), None
            )

            if type_param is not None:
                break
        else:
            raise CustomRuntimeError(
                f"At least one base class of '{cls.__name__}' must define a "
                f"default scaler type parameter named '{cls.__scaler_type_param_name__}'.",
                cls,
            )

    if not isinstance(type_param, TypeVar):
        raise CustomTypeError(f"Expected a TypeVar for scaler type parameter, got {type(type_param).__name__}", cls)

    if type_param.has_default():
        return type_param.__default__

    raise CustomRuntimeError(
        f"'{cls.__name__}' does not define a default scaler type and cannot be instantiated.", cls
    )