fft ¶

def __call__(
    self, clip: vs.VideoNode, sloc: SLocT | None = None,
    ftype: FilterTypeT | None = None,
    block_size: int | None = None, overlap: int | None = None,
    tr: int | None = None, tr_overlap: int | None = None,
    swin: SynthesisTypeT | None = None,
    twin: SynthesisTypeT | None = None,
    zmean: bool | None = None, alpha: float | None = None, ssystem: int | None = None,
    blockwise: bool = True, planes: PlanesT = None, *, func: FuncExceptT | None = None,
    default_args: KwargsT | None = None, **dkwargs: Any
) -> vs.VideoNode:
    func = func or self.__class__

    assert check_variable(clip, func)

    Backend = DFTTest.Backend

    kwargs: KwargsT = KwargsT(
        ftype=ftype, swin=swin, sbsize=block_size, sosize=overlap, tbsize=((tr or 0) * 2) + 1,
        tosize=tr_overlap, twin=twin, zmean=zmean, alpha=alpha, ssystem=ssystem,
        planes=planes, smode=int(blockwise)
    )

    if isinstance(sloc, SLocation.MultiDim):
        kwargs |= KwargsT(ssx=sloc._horizontal, ssy=sloc._vertical, sst=sloc._temporal)
    else:
        kwargs |= KwargsT(slocation=SLocation.from_param(sloc))

    clean_dft_args = KwargsNotNone(default_args or {}) | KwargsNotNone(kwargs)

    dft_args: KwargsT = KwargsT()

    for key, value in clean_dft_args.items():
        if isinstance(value, Enum) and callable(value):
            value = value()

        if isinstance(value, SynthesisTypeWithInfo):
            value = value.to_dict(key[0])

        if isinstance(value, dict):
            dft_args |= value
            continue

        if key == 'nlocation' and value:
            value = cast(Sequence[NLocation | int], value)
            value = list[int](flatten(value))

        if isinstance(value, SLocation):
            value = list(value)

        if isinstance(value, Enum):
            value = value.value

        dft_args[key] = value

    backend = self.resolved_backend

    if backend.is_dfttest2:
        from dfttest2 import Backend as DFTBackend
        from dfttest2 import DFTTest as DFTTest2  # noqa

        dft2_backend = None

        if backend is Backend.NVRTC:
            num_streams = dkwargs.pop('num_streams', self.num_streams if hasattr(self, 'num_streams') else 1)
            dft2_backend = DFTBackend.NVRTC(**(dict(**self) | dict(num_streams=num_streams)))
        elif backend is Backend.cuFFT:
            dft2_backend = DFTBackend.cuFFT(**self)
        elif backend is Backend.CPU:
            dft2_backend = DFTBackend.CPU(**self)
        elif backend is Backend.GCC:
            dft2_backend = DFTBackend.GCC(**self)

        if dft_args.get('tmode') is not None:
            raise CustomValueError('{backend} doesn\'t support tmode', func, backend=backend)

        if dft_args.pop('tosize'):
            raise CustomValueError('{backend} doesn\'t support tosize', func, backend=backend)

        return DFTTest2(clip, **dft_args | dkwargs, backend=dft2_backend)  # type: ignore[no-any-return]

    dft_args |= self

    if backend is Backend.OLD:
        return core.dfttest.DFTTest(clip, **dft_args)

    if backend is Backend.NEO:
        return core.neo_dfttest.DFTTest(clip, **dft_args)  # type: ignore

    if any(hasattr(core, x) for x in ('dfttest2_cpu', 'dfttest2_cuda', 'dfttest2_nvrtc')):
        raise CustomRuntimeError(
            'dfttest2 plugin is installed but is missing the python package, please install it.', self.__class__
        )

    raise CustomRuntimeError(
        'No implementation of DFTTest could be found, please install one. dfttest2 is recommended.', self.__class__
    )

@classmethod
def from_param(cls, value: DFTTest.Backend | BackendInfo) -> BackendInfo:
    return value() if isinstance(value, DFTTest.Backend) else value

@inject_self
def denoise(
    self, ref_or_sloc: vs.VideoNode | SLocT, sloc: SLocT | None = None, /,
    ftype: FilterTypeT = FilterType.WIENER,
    tr: int = 0, tr_overlap: int = 0,
    swin: SynthesisTypeT = SynthesisType.HANNING,
    twin: SynthesisTypeT = SynthesisType.RECTANGULAR,
    block_size: int = 16, overlap: int = 12,
    zmean: bool = True, alpha: float | None = None, ssystem: int = 0,
    blockwise: bool = True, planes: PlanesT = None, func: FuncExceptT | None = None, **kwargs: Any
) -> vs.VideoNode:
    func = func or self.denoise

    clip = self.clip
    nsloc = self.default_slocation

    if isinstance(ref_or_sloc, vs.VideoNode):
        clip = ref_or_sloc
    else:
        nsloc = ref_or_sloc

    if sloc is not None:
        nsloc = sloc

    if clip is None:
        raise CustomValueError('You must pass a clip!', func)

    if (fb := FieldBased.from_video(clip, False, func)).is_inter:
        raise UnsupportedFieldBasedError('Interlaced input is not supported!', func, fb)

    return self.plugin(
        clip, nsloc, func=func, **(self.default_args | dict(
            ftype=ftype, block_size=block_size, overlap=overlap, tr=tr, tr_overlap=tr_overlap, swin=swin,
            twin=twin, zmean=zmean, alpha=alpha, ssystem=ssystem, blockwise=blockwise, planes=planes
        ) | kwargs)
    )

@inject_self
def extract_freq(self, clip: vs.VideoNode, sloc: SLocT, **kwargs: Any) -> vs.VideoNode:
    kwargs = dict(func=self.extract_freq) | kwargs
    return clip.std.MakeDiff(self.denoise(clip, sloc, **kwargs))

@inject_self
def insert_freq(self, low: vs.VideoNode, high: vs.VideoNode, sloc: SLocT, **kwargs: Any) -> vs.VideoNode:
    return low.std.MergeDiff(self.extract_freq(high, sloc, **dict(func=self.insert_freq) | kwargs))

@inject_self
def merge_freq(self, low: vs.VideoNode, high: vs.VideoNode, sloc: SLocT, **kwargs: Any) -> vs.VideoNode:
    return self.insert_freq(
        self.denoise(low, sloc, **kwargs), high, sloc, **dict(func=self.merge_freq) | kwargs
    )

def __call__(self, **kwargs: Any) -> FilterTypeWithInfo:
    if self is FilterType.WIENER:
        def_kwargs = KwargsT(sigma=8.0, beta=1.0, nlocation=None)
    elif self in {FilterType.THR, FilterType.MULT}:
        def_kwargs = KwargsT(sigma=8.0, nlocation=None)
    elif self is FilterType.MULT_PSD:
        def_kwargs = KwargsT(sigma=8.0, pmin=0.0, sigma2=16.0, pmax=500.0)
    elif self is FilterType.MULT_RANGE:
        def_kwargs = KwargsT(sigma=8.0, pmin=0.0, pmax=500.0)
    else:
        def_kwargs = KwargsT()

    kwargs = def_kwargs | kwargs

    if 'beta' in kwargs:
        kwargs['f0beta'] = kwargs.pop('beta')

    return FilterTypeWithInfo(ftype=self.value, **kwargs)

def __call__(  # type: ignore
    self, *locations: SLocationT, res: int = 20, digits: int = 3
) -> SLocation | SLocation.MultiDim:
    if len(locations) == 1:
        return SLocation.from_param(locations[0]).interpolate(self, res, digits)

    return SLocation.MultiDim(*(self(x, res, digits) for x in locations))

@classmethod
def boundsCheck(
    cls, values: list[float], bounds: tuple[float | None, float | None], strict: bool = False
) -> list[float]:
    if not values:
        raise CustomValueError('"values" can\'t be empty!', cls)

    values = values.copy()

    bounds_str = iter('inf' if x is None else x for x in bounds)
    of_error = CustomOverflowError("Invalid value at index {i}, not in ({bounds})", cls, bounds=bounds_str)

    low_bound, up_bound = bounds

    for i, value in enumerate(values):
        if low_bound is not None and value < low_bound:
            if strict:
                raise of_error(i=i)

            values[i] = low_bound

        if up_bound is not None and value > up_bound:
            if strict:
                raise of_error(i=i)

            values[i] = up_bound

    return values

@classmethod
def from_param(cls: type[SLocBoundT], location: SLocationT | Literal[False] | None) -> SLocBoundT | None:
    if isinstance(location, SupportsFloatOrIndex) and location is not False:
        location = float(location)
        location = {0: location, 1: location}

    if location is None:
        return None

    if location is False:
        location = SLocation.NoProcess

    if isinstance(location, SLocation):
        return cls(list(location))

    return cls(location)

def interpolate(self, method: SInterMode = SInterMode.LINEAR, res: int = 20, digits: int = 3) -> SLocation:
    try:
        from scipy.interpolate import interp1d  # type: ignore
    except ModuleNotFoundError as e:
        raise DependencyNotFoundError(
            self.__class__, e, "scipy is required for interpolation. Use `pip install scipy`"
        )

    frequencies = list({round(x / (res - 1), digits) for x in range(res)})
    sigmas = interp1d(
        list(self.frequencies), list(self.sigmas), method.value, fill_value='extrapolate'
    )(frequencies)

    return SLocation(
        dict(zip(frequencies, sigmas)) | dict(zip(self.frequencies, self.sigmas)), strict=False
    )

def __call__(self, **kwargs: Any) -> SynthesisTypeWithInfo:
    if self is SynthesisType.KAISER_BESSEL and 'beta' not in kwargs:
        kwargs['beta'] = 2.5

    return SynthesisTypeWithInfo(win=self.value, **kwargs)

def to_dict(self, otype: str) -> KwargsT:
    value = self.copy()

    if 'beta' in value:
        value = value.copy()
        value[f'{otype}beta'] = value.pop('beta')

    value[f'{otype}win'] = value.pop('win')

    return value