y5gfunc.encode.video

video

Functions:

Name	Description
encode_video	Encode one or multiple VapourSynth video nodes using external encoders or output directly to stdout.

encode_video

encode_video(clip: Union[VideoNode, list[Union[VideoNode, tuple[VideoNode, int]]]], encoder: Union[list[Popen], Popen, IO, None] = None, multi: bool = False) -> None

Encode one or multiple VapourSynth video nodes using external encoders or output directly to stdout.

Parameters:

Name	Type	Description	Default
clip	Union[VideoNode, list[Union[VideoNode, tuple[VideoNode, int]]]]	A VapourSynth video node or a list of video nodes/tuples to encode.	required
encoder	Union[list[Popen], Popen, IO, None]	External encoder process(es) created with subprocess.Popen, a file-like object, or None to output to stdout.	None
multi	bool	If True, handle multiple input clips and multiple encoders. If False, handle a single clip.	False

Examples:

```python

# Output to an external encoder
encoder = subprocess.Popen(['x264', '--demuxer', 'y4m', '-', '-o', 'output.mp4'], stdin=subprocess.PIPE)
encode_video(clip, encoder)

# Output directly to stdout (like vspipe)
encode_video(clip, None)
# or
encode_video(clip, sys.stdout)

# Output to a file
with open('output.y4m', 'wb') as f:
    encode_video(clip, f)

# Example with multiple encoders
encoders = [
    subprocess.Popen(['x264', '--demuxer', 'y4m', '-', '-o', 'output1.mp4'], stdin=subprocess.PIPE),
    subprocess.Popen(['x264', '--demuxer', 'y4m', '-', '-o', 'output2.mp4'], stdin=subprocess.PIPE)
]
encode_video([clip1, clip2], encoders, multi=True)

```

Source code in y5gfunc/encode/video.py

def encode_video(
    clip: Union[vs.VideoNode, list[Union[vs.VideoNode, tuple[vs.VideoNode, int]]]],
    encoder: Union[list[Popen], Popen, IO, None] = None,
    multi: bool = False,
) -> None:
    """
    Encode one or multiple VapourSynth video nodes using external encoders or output directly to stdout.

    Args:
        clip: A VapourSynth video node or a list of video nodes/tuples to encode.
        encoder: External encoder process(es) created with subprocess.Popen, a file-like object, or None to output to stdout.
        multi: If True, handle multiple input clips and multiple encoders. If False, handle a single clip.

    Examples:

        ```python

        # Output to an external encoder
        encoder = subprocess.Popen(['x264', '--demuxer', 'y4m', '-', '-o', 'output.mp4'], stdin=subprocess.PIPE)
        encode_video(clip, encoder)

        # Output directly to stdout (like vspipe)
        encode_video(clip, None)
        # or
        encode_video(clip, sys.stdout)

        # Output to a file
        with open('output.y4m', 'wb') as f:
            encode_video(clip, f)

        # Example with multiple encoders
        encoders = [
            subprocess.Popen(['x264', '--demuxer', 'y4m', '-', '-o', 'output1.mp4'], stdin=subprocess.PIPE),
            subprocess.Popen(['x264', '--demuxer', 'y4m', '-', '-o', 'output2.mp4'], stdin=subprocess.PIPE)
        ]
        encode_video([clip1, clip2], encoders, multi=True)

        ```
    """

    if not multi:
        output_clip = None

        if isinstance(clip, vs.VideoNode):
            output_clip = clip
        elif isinstance(clip, list):
            for item in clip:
                if isinstance(item, tuple):
                    clip, index = item
                    if isinstance(clip, vs.VideoNode) and isinstance(index, int):
                        if index == 0:
                            output_clip = clip
                    else:
                        raise TypeError("encode_video: Tuple must be (VideoNode, int)")
            if not output_clip:
                output_clip = clip[0] if isinstance(clip[0], vs.VideoNode) else None
            if not output_clip:
                raise ValueError("encode_video: Couldn't parse clip!")

        assert isinstance(output_clip, vs.VideoNode)
        assert (
            output_clip.format.color_family == vs.YUV
        ), "encode_video: All clips must be YUV color family"
        assert output_clip.fps != 0, "encode_video: all clips must be CFR"

        # Allow direct output to stdout when encoder is None
        if encoder is None:
            _MIMO([output_clip], [sys.stdout])
        elif hasattr(encoder, "write") and callable(encoder.write):  # type: ignore[union-attr] # File-like object
            _MIMO([output_clip], [encoder])  # type: ignore[arg-type]
        else:  # Subprocess.Popen object
            _MIMO([output_clip], [encoder.stdin])  # type: ignore[arg-type]
            encoder.communicate()  # type: ignore[union-attr]
            encoder.wait()  # type: ignore[union-attr]
    else:
        assert isinstance(
            encoder, list
        ), "encode_video: encoder must be a list when multi=True"
        assert isinstance(
            clip, list
        ), "encode_video: clip must be a list when multi=True"
        assert len(encoder) == len(
            clip
        ), "encode_video: encoder and clip must have the same length"
        assert all(
            isinstance(item, vs.VideoNode) for item in clip
        ), "encode_video: all items in clip must be VideoNodes"
        assert all(
            clip.format.color_family == vs.YUV for clip in clip  # type: ignore[index]
        ), "encode_video: all clips must be YUV color family"
        assert all(
            clip.fps != 0 for clip in clip  # type: ignore[index]
        ), "encode_video: all clips must be CFR"

        output_clips = []
        stdins = []
        for i, clip in enumerate(clip):  # type: ignore[index]
            output_clips.append(clip)
            if hasattr(encoder[i], "write") and callable(encoder[i].write):  # type: ignore[union-attr] # File-like object
                stdins.append(encoder[i])
            else:  # Subprocess.Popen object
                stdins.append(encoder[i].stdin)

        _MIMO(output_clips, stdins)  # type: ignore[arg-type]

        # Only call communicate and wait for Popen objects
        for i, enc in enumerate(encoder):
            if not hasattr(enc, "write") or not callable(enc.write):  # type: ignore[union-attr] # Not a file-like object
                enc.communicate()
                enc.wait()