noise ¶

@inject_self.init_kwargs.clean
def grain(
    self, clip: vs.VideoNode, strength: float | tuple[float, float] | None = None,
    dynamic: bool | None = None, **kwargs: Any
) -> vs.VideoNode:
    assert clip.format

    kwargs = self._get_kw(kwargs)

    if strength is None:
        strength = self.strength

    dynamic = fallback(dynamic, self.dynamic)
    strength = strength if isinstance(strength, tuple) else (
        strength, strength if clip.format.num_planes > 1 else 0.0
    )

    if max(strength) <= 0.0:
        return clip

    if strength[0] <= 0.0 and strength[1] > 0.0:
        planes: PlanesT = [1, 2]
    elif strength[0] > 0.0 and strength[1] <= 0.0:
        planes = 0
    else:
        planes = None

    do_taverage = (
        dynamic
        and self.temporal_average > 0 and self.temporal_radius > 0
        and (clip.num_frames > self.temporal_radius * 2)
    )
    do_protect_chroma = self.protect_chroma and strength[1] > 0.0 and clip.format.color_family is vs.YUV
    input_dep = self._is_input_dependent(clip, **kwargs)

    def _wrap_implementation(clip: vs.VideoNode, neutral_out: bool) -> vs.VideoNode:
        if input_dep and do_taverage and not kwargs.get('unsafe_graining', False):
            raise CustomValueError(
                'You can\'t have temporal averaging with input dependent graining as it will create ghosting!'
            )

        if neutral_out and not input_dep:
            length = clip.num_frames + ((self.temporal_radius * 2) if do_taverage else 0)
            base_clip = clip.std.BlankClip(length=length, color=get_neutral_values(clip))
        elif do_taverage:
            base_clip = (clip[0] * self.temporal_radius) + clip + (clip[-1] * self.temporal_radius)
        else:
            base_clip = clip

        def _try_grain(src: vs.VideoNode, stre: tuple[float, float] = strength, **args: Any) -> vs.VideoNode:
            args = kwargs | dict(strength=stre, dynamic=dynamic) | args
            try:
                self._check_input(src, **args)
                grained = self._perform_graining(src, **args)
            except NotImplementedError as e:
                reason, *_ = map(str, e.args)

                if reason == 'dynamic-only':
                    grained = _try_grain(src[src.num_frames // 2], dynamic=True)
                elif reason.startswith('bad-depth'):
                    good_depth = int(reason.split('-')[-1])
                    grained = _try_grain(depth(src, good_depth))
                    grained = depth(grained, src)
                elif reason == 'single-plane':
                    str_luma, str_chroma = strength

                    if str_luma > 0 and str_chroma > 0:
                        return join(
                            _try_grain(plane(src, 0), str_luma),
                            _try_grain(plane(src, 1), str_chroma),
                            _try_grain(plane(src, 2), str_chroma)
                        )
                    elif str_luma > 0:
                        return join(
                            _try_grain(plane(src, 0), str_luma),
                            src
                        )
                    elif str_chroma > 0:
                        return join(
                            src,
                            _try_grain(plane(src, 1), str_chroma),
                            _try_grain(plane(src, 2), str_chroma)
                        )

                    return src
                else:
                    raise e

            return grained

        grained = _try_grain(base_clip)

        if input_dep and neutral_out:
            grained = clip.std.MakeDiff(grained)

        return grained

    if (
        self.size == (1.0, 1.0) and not do_taverage and not self.postprocess
        and not do_protect_chroma and self.luma_scaling is None and not self.fade_limits
    ):
        return _wrap_implementation(clip, self.neutral_out)

    (sizex, sizey), mod = self.size, max(clip.format.subsampling_w, clip.format.subsampling_h) << 1
    sx, sy = mod_x(clip.width / sizex, mod), mod_x(clip.height / sizey, mod)

    if (sx, sy) != (clip.width, clip.height):
        sxa, sya = mod_x((clip.width + sx) / 2, mod), mod_x((clip.height + sy) / 2, mod)

        grained = _wrap_implementation(self.scaler.scale(clip, sx, sy), True)

        # If the scale is too big, we need to scale it in two passes, else the window
        # will be too big and the grain will be dampened down too much
        if max(self.size) > 1.5:
            grained = self.scaler.scale(grained, sxa, sya)

        grained = self.scaler.scale(grained, clip.width, clip.height)
    else:
        grained = _wrap_implementation(clip, True)

    if do_taverage:
        average = BlurMatrix.MEAN(taps=self.temporal_radius, mode=ConvMode.TEMPORAL)(grained)
        grained = grained.std.Merge(average, self.temporal_average)
        grained = grained[self.temporal_radius:-self.temporal_radius]

    if self.fade_limits:
        low, high = (None, None) if self.fade_limits is True else self.fade_limits

        low = [
            scale_value(
                threshold, 8, clip.format.bits_per_sample,
                chroma=not not plane_index
            )
            for plane_index, threshold in enumerate(normalize_seq(fallback(low, 16)))
        ]

        high = [
            scale_value(
                threshold, 8, clip.format.bits_per_sample,
                chroma=not not plane_index
            )
            for plane_index, threshold in enumerate(normalize_seq(fallback(high, [235, 240])))
        ]

        if complexpr_available:
            limit_expr = 'y neutral - abs A! x A@ - {low} < x A@ + {high} > or neutral y ?'
        else:
            limit_expr = 'x y neutral - abs - {low} < x y neutral - abs + {high} > or neutral y ?'

        grained = norm_expr([clip, grained], limit_expr, planes, low=low, high=high, func=self.__class__.grain)

    if self.postprocess:
        for postprocess in cast(list[GrainPostProcessT], to_arr(self.postprocess)):
            if callable(postprocess):
                try:
                    postprocess = postprocess(grained, clip)
                except TypeError:
                    postprocess = postprocess(grained)

            if isinstance(postprocess, vs.VideoNode):
                grained = postprocess
            else:
                if isinstance(postprocess, GrainPP):
                    postprocess, ppkwargs = postprocess.value, postprocess.kwargs
                else:
                    ppkwargs = KwargsT()

                # fuck importing re
                uses_y = ' y ' in postprocess or postprocess.startswith('y ') or postprocess.endswith(' y')  #type: ignore
                grained = norm_expr(
                    [grained, clip] if uses_y else grained, postprocess, **ppkwargs, func=self.__class__.grain
                )

    neutral = get_neutral_value(clip)

    if self.neutral_out:
        merge_clip = grained.std.MakeDiff(grained)[0].std.Loop(grained.num_frames)
    else:
        merge_clip, grained = clip, clip.std.MergeDiff(grained, planes)

    if do_protect_chroma:
        neutral_mask = Lanczos.resample(clip, clip.format.replace(subsampling_h=0, subsampling_w=0))

        neutral_mask = norm_expr(
            split(neutral_mask), f'y {neutral} = z {neutral} = and {get_peak_value(clip, range_in=ColorRange.FULL)} 0 ?',
            func=self.__class__.grain
        )

        grained = grained.std.MaskedMerge(merge_clip, neutral_mask, [1, 2])

    if self.luma_scaling is not None:
        mask = adg_mask(clip, self.luma_scaling, func=self.grain)

        grained = merge_clip.std.MaskedMerge(grained, mask)

    return grained

@inject_self.init_kwargs.clean
def grain(
    self, clip: vs.VideoNode, strength: float | tuple[float, float] | None = None,
    dynamic: bool | None = None, **kwargs: Any
) -> vs.VideoNode:
    assert clip.format

    kwargs = self._get_kw(kwargs)

    if strength is None:
        strength = self.strength

    dynamic = fallback(dynamic, self.dynamic)
    strength = strength if isinstance(strength, tuple) else (
        strength, strength if clip.format.num_planes > 1 else 0.0
    )

    if max(strength) <= 0.0:
        return clip

    if strength[0] <= 0.0 and strength[1] > 0.0:
        planes: PlanesT = [1, 2]
    elif strength[0] > 0.0 and strength[1] <= 0.0:
        planes = 0
    else:
        planes = None

    do_taverage = (
        dynamic
        and self.temporal_average > 0 and self.temporal_radius > 0
        and (clip.num_frames > self.temporal_radius * 2)
    )
    do_protect_chroma = self.protect_chroma and strength[1] > 0.0 and clip.format.color_family is vs.YUV
    input_dep = self._is_input_dependent(clip, **kwargs)

    def _wrap_implementation(clip: vs.VideoNode, neutral_out: bool) -> vs.VideoNode:
        if input_dep and do_taverage and not kwargs.get('unsafe_graining', False):
            raise CustomValueError(
                'You can\'t have temporal averaging with input dependent graining as it will create ghosting!'
            )

        if neutral_out and not input_dep:
            length = clip.num_frames + ((self.temporal_radius * 2) if do_taverage else 0)
            base_clip = clip.std.BlankClip(length=length, color=get_neutral_values(clip))
        elif do_taverage:
            base_clip = (clip[0] * self.temporal_radius) + clip + (clip[-1] * self.temporal_radius)
        else:
            base_clip = clip

        def _try_grain(src: vs.VideoNode, stre: tuple[float, float] = strength, **args: Any) -> vs.VideoNode:
            args = kwargs | dict(strength=stre, dynamic=dynamic) | args
            try:
                self._check_input(src, **args)
                grained = self._perform_graining(src, **args)
            except NotImplementedError as e:
                reason, *_ = map(str, e.args)

                if reason == 'dynamic-only':
                    grained = _try_grain(src[src.num_frames // 2], dynamic=True)
                elif reason.startswith('bad-depth'):
                    good_depth = int(reason.split('-')[-1])
                    grained = _try_grain(depth(src, good_depth))
                    grained = depth(grained, src)
                elif reason == 'single-plane':
                    str_luma, str_chroma = strength

                    if str_luma > 0 and str_chroma > 0:
                        return join(
                            _try_grain(plane(src, 0), str_luma),
                            _try_grain(plane(src, 1), str_chroma),
                            _try_grain(plane(src, 2), str_chroma)
                        )
                    elif str_luma > 0:
                        return join(
                            _try_grain(plane(src, 0), str_luma),
                            src
                        )
                    elif str_chroma > 0:
                        return join(
                            src,
                            _try_grain(plane(src, 1), str_chroma),
                            _try_grain(plane(src, 2), str_chroma)
                        )

                    return src
                else:
                    raise e

            return grained

        grained = _try_grain(base_clip)

        if input_dep and neutral_out:
            grained = clip.std.MakeDiff(grained)

        return grained

    if (
        self.size == (1.0, 1.0) and not do_taverage and not self.postprocess
        and not do_protect_chroma and self.luma_scaling is None and not self.fade_limits
    ):
        return _wrap_implementation(clip, self.neutral_out)

    (sizex, sizey), mod = self.size, max(clip.format.subsampling_w, clip.format.subsampling_h) << 1
    sx, sy = mod_x(clip.width / sizex, mod), mod_x(clip.height / sizey, mod)

    if (sx, sy) != (clip.width, clip.height):
        sxa, sya = mod_x((clip.width + sx) / 2, mod), mod_x((clip.height + sy) / 2, mod)

        grained = _wrap_implementation(self.scaler.scale(clip, sx, sy), True)

        # If the scale is too big, we need to scale it in two passes, else the window
        # will be too big and the grain will be dampened down too much
        if max(self.size) > 1.5:
            grained = self.scaler.scale(grained, sxa, sya)

        grained = self.scaler.scale(grained, clip.width, clip.height)
    else:
        grained = _wrap_implementation(clip, True)

    if do_taverage:
        average = BlurMatrix.MEAN(taps=self.temporal_radius, mode=ConvMode.TEMPORAL)(grained)
        grained = grained.std.Merge(average, self.temporal_average)
        grained = grained[self.temporal_radius:-self.temporal_radius]

    if self.fade_limits:
        low, high = (None, None) if self.fade_limits is True else self.fade_limits

        low = [
            scale_value(
                threshold, 8, clip.format.bits_per_sample,
                chroma=not not plane_index
            )
            for plane_index, threshold in enumerate(normalize_seq(fallback(low, 16)))
        ]

        high = [
            scale_value(
                threshold, 8, clip.format.bits_per_sample,
                chroma=not not plane_index
            )
            for plane_index, threshold in enumerate(normalize_seq(fallback(high, [235, 240])))
        ]

        if complexpr_available:
            limit_expr = 'y neutral - abs A! x A@ - {low} < x A@ + {high} > or neutral y ?'
        else:
            limit_expr = 'x y neutral - abs - {low} < x y neutral - abs + {high} > or neutral y ?'

        grained = norm_expr([clip, grained], limit_expr, planes, low=low, high=high, func=self.__class__.grain)

    if self.postprocess:
        for postprocess in cast(list[GrainPostProcessT], to_arr(self.postprocess)):
            if callable(postprocess):
                try:
                    postprocess = postprocess(grained, clip)
                except TypeError:
                    postprocess = postprocess(grained)

            if isinstance(postprocess, vs.VideoNode):
                grained = postprocess
            else:
                if isinstance(postprocess, GrainPP):
                    postprocess, ppkwargs = postprocess.value, postprocess.kwargs
                else:
                    ppkwargs = KwargsT()

                # fuck importing re
                uses_y = ' y ' in postprocess or postprocess.startswith('y ') or postprocess.endswith(' y')  #type: ignore
                grained = norm_expr(
                    [grained, clip] if uses_y else grained, postprocess, **ppkwargs, func=self.__class__.grain
                )

    neutral = get_neutral_value(clip)

    if self.neutral_out:
        merge_clip = grained.std.MakeDiff(grained)[0].std.Loop(grained.num_frames)
    else:
        merge_clip, grained = clip, clip.std.MergeDiff(grained, planes)

    if do_protect_chroma:
        neutral_mask = Lanczos.resample(clip, clip.format.replace(subsampling_h=0, subsampling_w=0))

        neutral_mask = norm_expr(
            split(neutral_mask), f'y {neutral} = z {neutral} = and {get_peak_value(clip, range_in=ColorRange.FULL)} 0 ?',
            func=self.__class__.grain
        )

        grained = grained.std.MaskedMerge(merge_clip, neutral_mask, [1, 2])

    if self.luma_scaling is not None:
        mask = adg_mask(clip, self.luma_scaling, func=self.grain)

        grained = merge_clip.std.MaskedMerge(grained, mask)

    return grained

@inject_self.init_kwargs.clean
def grain(
    self, clip: vs.VideoNode, strength: float | tuple[float, float] | None = None,
    dynamic: bool | None = None, **kwargs: Any
) -> vs.VideoNode:
    assert clip.format

    kwargs = self._get_kw(kwargs)

    if strength is None:
        strength = self.strength

    dynamic = fallback(dynamic, self.dynamic)
    strength = strength if isinstance(strength, tuple) else (
        strength, strength if clip.format.num_planes > 1 else 0.0
    )

    if max(strength) <= 0.0:
        return clip

    if strength[0] <= 0.0 and strength[1] > 0.0:
        planes: PlanesT = [1, 2]
    elif strength[0] > 0.0 and strength[1] <= 0.0:
        planes = 0
    else:
        planes = None

    do_taverage = (
        dynamic
        and self.temporal_average > 0 and self.temporal_radius > 0
        and (clip.num_frames > self.temporal_radius * 2)
    )
    do_protect_chroma = self.protect_chroma and strength[1] > 0.0 and clip.format.color_family is vs.YUV
    input_dep = self._is_input_dependent(clip, **kwargs)

    def _wrap_implementation(clip: vs.VideoNode, neutral_out: bool) -> vs.VideoNode:
        if input_dep and do_taverage and not kwargs.get('unsafe_graining', False):
            raise CustomValueError(
                'You can\'t have temporal averaging with input dependent graining as it will create ghosting!'
            )

        if neutral_out and not input_dep:
            length = clip.num_frames + ((self.temporal_radius * 2) if do_taverage else 0)
            base_clip = clip.std.BlankClip(length=length, color=get_neutral_values(clip))
        elif do_taverage:
            base_clip = (clip[0] * self.temporal_radius) + clip + (clip[-1] * self.temporal_radius)
        else:
            base_clip = clip

        def _try_grain(src: vs.VideoNode, stre: tuple[float, float] = strength, **args: Any) -> vs.VideoNode:
            args = kwargs | dict(strength=stre, dynamic=dynamic) | args
            try:
                self._check_input(src, **args)
                grained = self._perform_graining(src, **args)
            except NotImplementedError as e:
                reason, *_ = map(str, e.args)

                if reason == 'dynamic-only':
                    grained = _try_grain(src[src.num_frames // 2], dynamic=True)
                elif reason.startswith('bad-depth'):
                    good_depth = int(reason.split('-')[-1])
                    grained = _try_grain(depth(src, good_depth))
                    grained = depth(grained, src)
                elif reason == 'single-plane':
                    str_luma, str_chroma = strength

                    if str_luma > 0 and str_chroma > 0:
                        return join(
                            _try_grain(plane(src, 0), str_luma),
                            _try_grain(plane(src, 1), str_chroma),
                            _try_grain(plane(src, 2), str_chroma)
                        )
                    elif str_luma > 0:
                        return join(
                            _try_grain(plane(src, 0), str_luma),
                            src
                        )
                    elif str_chroma > 0:
                        return join(
                            src,
                            _try_grain(plane(src, 1), str_chroma),
                            _try_grain(plane(src, 2), str_chroma)
                        )

                    return src
                else:
                    raise e

            return grained

        grained = _try_grain(base_clip)

        if input_dep and neutral_out:
            grained = clip.std.MakeDiff(grained)

        return grained

    if (
        self.size == (1.0, 1.0) and not do_taverage and not self.postprocess
        and not do_protect_chroma and self.luma_scaling is None and not self.fade_limits
    ):
        return _wrap_implementation(clip, self.neutral_out)

    (sizex, sizey), mod = self.size, max(clip.format.subsampling_w, clip.format.subsampling_h) << 1
    sx, sy = mod_x(clip.width / sizex, mod), mod_x(clip.height / sizey, mod)

    if (sx, sy) != (clip.width, clip.height):
        sxa, sya = mod_x((clip.width + sx) / 2, mod), mod_x((clip.height + sy) / 2, mod)

        grained = _wrap_implementation(self.scaler.scale(clip, sx, sy), True)

        # If the scale is too big, we need to scale it in two passes, else the window
        # will be too big and the grain will be dampened down too much
        if max(self.size) > 1.5:
            grained = self.scaler.scale(grained, sxa, sya)

        grained = self.scaler.scale(grained, clip.width, clip.height)
    else:
        grained = _wrap_implementation(clip, True)

    if do_taverage:
        average = BlurMatrix.MEAN(taps=self.temporal_radius, mode=ConvMode.TEMPORAL)(grained)
        grained = grained.std.Merge(average, self.temporal_average)
        grained = grained[self.temporal_radius:-self.temporal_radius]

    if self.fade_limits:
        low, high = (None, None) if self.fade_limits is True else self.fade_limits

        low = [
            scale_value(
                threshold, 8, clip.format.bits_per_sample,
                chroma=not not plane_index
            )
            for plane_index, threshold in enumerate(normalize_seq(fallback(low, 16)))
        ]

        high = [
            scale_value(
                threshold, 8, clip.format.bits_per_sample,
                chroma=not not plane_index
            )
            for plane_index, threshold in enumerate(normalize_seq(fallback(high, [235, 240])))
        ]

        if complexpr_available:
            limit_expr = 'y neutral - abs A! x A@ - {low} < x A@ + {high} > or neutral y ?'
        else:
            limit_expr = 'x y neutral - abs - {low} < x y neutral - abs + {high} > or neutral y ?'

        grained = norm_expr([clip, grained], limit_expr, planes, low=low, high=high, func=self.__class__.grain)

    if self.postprocess:
        for postprocess in cast(list[GrainPostProcessT], to_arr(self.postprocess)):
            if callable(postprocess):
                try:
                    postprocess = postprocess(grained, clip)
                except TypeError:
                    postprocess = postprocess(grained)

            if isinstance(postprocess, vs.VideoNode):
                grained = postprocess
            else:
                if isinstance(postprocess, GrainPP):
                    postprocess, ppkwargs = postprocess.value, postprocess.kwargs
                else:
                    ppkwargs = KwargsT()

                # fuck importing re
                uses_y = ' y ' in postprocess or postprocess.startswith('y ') or postprocess.endswith(' y')  #type: ignore
                grained = norm_expr(
                    [grained, clip] if uses_y else grained, postprocess, **ppkwargs, func=self.__class__.grain
                )

    neutral = get_neutral_value(clip)

    if self.neutral_out:
        merge_clip = grained.std.MakeDiff(grained)[0].std.Loop(grained.num_frames)
    else:
        merge_clip, grained = clip, clip.std.MergeDiff(grained, planes)

    if do_protect_chroma:
        neutral_mask = Lanczos.resample(clip, clip.format.replace(subsampling_h=0, subsampling_w=0))

        neutral_mask = norm_expr(
            split(neutral_mask), f'y {neutral} = z {neutral} = and {get_peak_value(clip, range_in=ColorRange.FULL)} 0 ?',
            func=self.__class__.grain
        )

        grained = grained.std.MaskedMerge(merge_clip, neutral_mask, [1, 2])

    if self.luma_scaling is not None:
        mask = adg_mask(clip, self.luma_scaling, func=self.grain)

        grained = merge_clip.std.MaskedMerge(grained, mask)

    return grained