vsmuxtools.utils.source

def src(
    filePath: PathLike,
    preview_sourcefilter: SourceFilter | None = SourceFilter.AUTO,
    sourcefilter: SourceFilter = SourceFilter.BESTSOURCE,
    **kwargs: KwargsT,
) -> vs.VideoNode:
    """
    Uses lsmas for previewing and bestsource otherwise.
    Still supports dgi files directly if dgdecodenv is installed to not break existing scripts.

    :param filepath:                Path to video or dgi file
    :param preview_sourcefilter:    Source filter to be used when previewing using vspreview.
                                    `None` will make it fall back to `sourcefilter`.
    :param sourcefilter:            Source filter to be used otherwise.
    :param kwargs:                  Other arguments you may or may not wanna pass to the indexer.
    :return:                        Video Node
    """
    filePath = ensure_path_exists(filePath, src)
    dgiFile = filePath.with_suffix(".dgi")
    if filePath.suffix.lower() == ".dgi" or dgiFile.exists():
        if not hasattr(core, "dgdecodenv"):
            raise error("Trying to use a dgi file without dgdecodenv installed.", src)
        return core.lazy.dgdecodenv.DGSource(str(filePath.resolve()) if not dgiFile.exists() else str(dgiFile.resolve()), **kwargs)

    force_lsmas = kwargs.pop("force_lsmas", False)
    if force_lsmas:
        warn("force_lsmas is deprecated!\nPlease switch to using the explicit sourcefilter params.", src, 5)
        preview_sourcefilter = None
        sourcefilter = SourceFilter.LSMASH

    force_bs = kwargs.pop("force_bs", False)
    if force_bs:
        warn("force_bs is deprecated!\nPlease switch to using the explicit sourcefilter params.", src, 5)
        preview_sourcefilter = None
        sourcefilter = SourceFilter.BESTSOURCE

    if preview_sourcefilter is SourceFilter.AUTO:
        preview_sourcefilter = SourceFilter.LSMASH if "m2ts" in filePath.name.lower() else SourceFilter.FFMS2

    if sourcefilter is SourceFilter.AUTO:
        sourcefilter = SourceFilter.LSMASH if "m2ts" in filePath.name.lower() else SourceFilter.BESTSOURCE

    if is_previewing() and preview_sourcefilter is not None:
        return _call_sourcefilter(filePath.resolve(), preview_sourcefilter, **kwargs)
    else:
        return _call_sourcefilter(filePath.resolve(), sourcefilter, **kwargs)

def frames_to_samples(frame: int | None, sample_rate: vs.AudioNode | int = 48000, fps: vs.VideoNode | Fraction = Fraction(24000, 1001)) -> int:
    """
    Converts a frame number to a sample number

    :param frame:           The frame number
    :param sample_rate:     Can be a flat number like 48000 (=48 kHz) or an AudioNode to get the sample rate from
    :param fps:             Can be a Fraction or a VideoNode to get the fps from

    :return:                The sample number
    """
    if not frame:
        return 0
    sample_rate = sample_rate.sample_rate if isinstance(sample_rate, vs.AudioNode) else sample_rate
    fps = Fraction(fps.fps_num, fps.fps_den) if isinstance(fps, vs.VideoNode) else fps
    return int(Fraction(sample_rate * frame, 1) / fps)

def generate_svt_av1_keyframes(
    clip: vs.VideoNode,
    start_frame: int = 0,
    min_scene_length: int = 129,
    min_still_scene_length: int = 193,
    max_scene_length: int = 321,
) -> list[int]:
    """
    Run `generate_keyframes`, and then filter the WWXD keyframe result for SVT-AV1 derived encoders.

    A huge contribution to the efficiency of SVT-AV1 derived encoders comes from its referencing system.
    SVT-AV1 derived encoders by default have a 32 frame hierarchical structure.
    It will first encode frame 0 as a key frame, followed by frame 32 referencing frame 0.
    Since later frames will all be referencing these frames, these frames of the lowest temporal layer will be given very good q.
    After that, the encoder will continue with frame 16, referencing both frame 0 and frame 32.
    And then after frame 16, it'll be frame 8, frame 4, frame 2, frame 1, in this order.
    The frames in higher temporal layer will be given a bad q, mostly relying on information in the frame they reference to get a good encoding quality.
    It's very efficient to allow SVT-AV1 derived encoders to build a full hierarchical structure like this, with lower level frames being given a good q, and higher level frames relying on referencing and saving bitrate.
    For scene detection, the scenes that gives the best efficiency should have a length of `(integer * 32) + 1`, followed by scenes that has a length of `(integer * 16) + 1`, and then `8`, and then `4`, et cetera.

    Additionally, since both frame 0 and frame 32 will be given very good q, if there were a real scene change in the middle of this 32 frame structure, it will be handled well because frame 32 will certainly be given a good enough q to encode the new scene.
    On the other hand, making too much scenecut in scene detection, resulting in a lot of smaller and incomplete hierarchical strctures in each scene is generally a very bad idea.
    Since WWXD often has the tendency to place way too much scenecuts in challenging sections, the purpose of this function is to filter the result from WWXD and create scenes that has more efficient hierarchical structure as much as possible.
    """
    import numpy as np
    from numpy.lib.stride_tricks import sliding_window_view

    frames = generate_keyframes(clip, start_frame)

    if start_frame:
        clip = clip[start_frame:]

    diff_clip = clip.std.PlaneStats(clip[0] + clip, plane=0, prop="Luma")

    frames.append(len(clip))
    head = -1  # Because the result from generate_keyframes doesn't have `0`
    current_frame = 0
    svt_av1_frames = [0]
    while head < len(frames) - 1:
        head += 1

        # Choosing between WWXD selected frames within the limit of min_scene_length and max_scene_length
        if frames[head] - current_frame < min_scene_length:
            if head != len(frames) - 1:
                continue

            else:
                current_frame = frames[head]
                svt_av1_frames.append(current_frame)  # Only to get popped

        elif frames[head] - current_frame <= max_scene_length:
            available_frames = []
            for looka_head in range(head, len(frames)):
                if frames[looka_head] - current_frame <= max_scene_length:
                    available_frames.append(frames[looka_head])
                else:
                    break

            selected_head = None
            for structure in [32, 16, 8, 4, 2]:
                for available_head in range(len(available_frames) - 1, -1, -1):
                    if (available_frames[available_head] - current_frame) % structure == 1:
                        selected_head = available_head
                        break
                if selected_head is not None:
                    break

            if selected_head is None:
                selected_head = len(available_frames) - 1

            head = head + selected_head
            current_frame = frames[head]
            svt_av1_frames.append(current_frame)

        # If WWXD doesn't select anything within max_scene_length, try finding good frames using diffs.
        else:
            selected_frame = None
            diffs = np.array(
                [
                    frame.props["LumaDiff"]
                    for frame in diff_clip[current_frame + min_still_scene_length : current_frame + max_scene_length + 1].frames()
                ]
            )
            windows = sliding_window_view(diffs, 25)
            median = np.median(windows, axis=1).reshape((-1, 1))
            mad = np.median(np.abs(windows - median), axis=1).reshape((-1, 1))
            thr = (median + 3.0 * mad).reshape((-1,))
            thr = np.concatenate((np.full((12,), thr[0]), thr, np.full((12,), thr[-1])))
            motion_frames = np.argwhere(diffs > thr).reshape((-1,))
            motion_frames += current_frame + min_still_scene_length

            if motion_frames.shape[0] != 0:
                for structure in [32, 16, 8]:
                    for frame in motion_frames[::-1]:
                        if (frame - current_frame) % structure == 1:
                            selected_frame = int(frame)
                            break
                    if selected_frame is not None:
                        break

            if selected_frame is None:
                selected_frame = current_frame + max_scene_length

            head -= 1
            current_frame = selected_frame
            svt_av1_frames.append(current_frame)

    svt_av1_frames.pop()

    return svt_av1_frames