Streams

Stream objects represent the flow of multimedia data through the processing pipeline.

Stream Class

pyffmpeg.node.Stream

Bases: GeneratedFiltersMixin

Represents a single output stream from a node.

Source code in src/pyffmpeg/node.py

class Stream(GeneratedFiltersMixin):
    """Represents a single output stream from a node."""

    def __init__(self, source_node: Node):
        self.source_node: Node = source_node
        self.index: str | None = None
        self.elementary_streams: dict[str, TypedStream | IndexedStream] = {}

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Stream):
            return NotImplemented
        return self.source_node == other.source_node

    def __hash__(self) -> int:
        return hash(id(self))

    @property
    def audio(self) -> "TypedStream":
        """Accesses the audio elementary stream component.

        Returns:
            TypedStream: An object representing the audio component of this stream.
        """
        return self.elementary_streams.setdefault("a", TypedStream("audio", self))

    @property
    def video(self) -> "TypedStream":
        """Accesses the video elementary stream component.

        If the video stream wrapper has not been accessed before for this node,
        creates and caches a new 'TypedStream' instance representing video.

        Returns:
            TypedStream: An object representing the video component of this stream.
        """
        return self.elementary_streams.setdefault("v", TypedStream("video", self))

    def __getitem__(self, key: str) -> "TypedStream | IndexedStream":
        """Allows accessing elementary stream contained in this Stream"""
        mapping = {"a": "audio", "v": "video"}
        if not isinstance(key, str) or len(key) != 1:
            raise TypeError("Expected string index (e.g. 'a')")

        if key in mapping:
            stream_type = mapping[key]
            return self.elementary_streams.setdefault(
                key, TypedStream(stream_type, self)
            )

        if key.isdigit():
            return self.elementary_streams.setdefault(key, IndexedStream(self))

        raise KeyError(
            f"Invalid stream key: {key!r}. Expected 'a', 'v', or a numeric index like '1'."
        )

    @property
    def node(self) -> list["Stream"]:
        """Retrieves all output streams produced by the source filter node.

        This property allows access to the complete list of outputs generated
        by the filter that created this specific stream instance (accessing sibling streams).

        Returns:
            list[Stream]: A list of all output streams from the parent FilterNode.

        Raises:
            AttributeError: If the source node is not a FilterNode (e.g., it is
                an InputNode), as it does not perform filtering operations.
        """
        if not isinstance(self.source_node, FilterNode):
            raise AttributeError(".node is only available on filter outputs")
        return self.source_node.output_streams

    def _apply_filter(
        self,
        filter_name: str,
        positional_arguments: tuple = (),
        named_arguments: dict[str, Any] = {},
        inputs: list["Stream"] | None = None,
        num_output_streams: int = 1,
    ) -> list["Stream"]:
        """Creates a FilterNode and returns its output streams."""
        node = FilterNode(
            filter_name,
            positional_arguments,
            named_arguments,
            inputs or [self],
            num_output_streams,
        )
        return node.output_streams

    def _apply_dynamic_outputs_filter(
        self,
        filter_name: str,
        positional_arguments: tuple = (),
        named_arguments: dict[str, Any] = {},
        inputs: list["Stream"] | None = None,
    ) -> "FilterMultiOutput":
        """Creates a FilterNode and returns FilterMultiOutput to allow dynamic outputs."""
        filter_node = FilterNode(
            filter_name,
            positional_arguments,
            named_arguments,
            inputs or [self],
            num_output_streams=0,
        )
        return FilterMultiOutput(filter_node)

    def _apply_sink_filter(
        self,
        filter_name: str,
        named_arguments: dict = {},
        inputs: list["Stream"] | None = None,
    ) -> SinkNode:
        filter_node = FilterNode(
            filter_name,
            positional_arguments=(),
            named_arguments=named_arguments,
            inputs=inputs or [self],
            num_output_streams=0,
        )
        return SinkNode(filter_node)

    def split(self, outputs: int = 2) -> list["Stream"]:
        """Pass on the input to N video outputs.

        Args:
            outputs (int): Set number of outputs (from 1 to INT_MAX).
                Defaults to 2.

        Returns:
            list[Stream]: A list of output streams produced by the split filter.
        """
        return self._apply_filter(
            filter_name="split",
            named_arguments={"outputs": outputs},
            num_output_streams=outputs,
        )

    def asplit(self, outputs: int = 2) -> list["Stream"]:
        """Pass on the audio input to N audio outputs.

        Args:
            outputs (int): Set number of outputs (from 1 to INT_MAX).
                Defaults to 2.

        Returns:
            list[Stream]: A list of output audio streams produced by the asplit filter.
        """
        return self._apply_filter(
            filter_name="asplit",
            named_arguments={"outputs": outputs},
            num_output_streams=outputs,
        )

    def filter(self, filter_name: str, *args, **kwargs) -> "Stream":
        """Custom filter with a single input and a single output"""
        return self._apply_filter(filter_name, args, kwargs)[0]

    def filter_multi_output(
        self, filter_name: str, *args, **kwargs
    ) -> "FilterMultiOutput":
        """Creates a custom filter allowing dynamic creation of output streams"""
        node = FilterNode(
            filter_name=filter_name,
            positional_arguments=args,
            named_arguments=kwargs,
            inputs=[self],
            num_output_streams=0,
        )
        return FilterMultiOutput(node)

    def output(
        self,
        filename: str,
        streams: list["Stream"] | None = None,
        format: str | None = None,
        vcodec: str | None = None,
        acodec: str | None = None,
        video_bitrate: str | int | None = None,
        audio_bitrate: str | int | None = None,
        aspect: str | float | None = None,
        frames: int | None = None,
        shortest: bool = False,
        **kwargs,
    ) -> "OutputNode":
        """
        Creates an output node for this stream, optionally muxing other streams.

        Args:
            filename (str): The output file path or URL.
            streams (list[Stream] | None): Additional streams to include in the output
                (e.g., audio tracks, subtitles) alongside the current stream.
            format (str | None): Force output format (ffmpeg flag: ``-f``).
            vcodec (str | None): Video codec (ffmpeg flag: ``-c:v``).
            acodec (str | None): Audio codec (ffmpeg flag: ``-c:a``).
            video_bitrate (str | int | None): Video bitrate (ffmpeg flag: ``-b:v``).
            audio_bitrate (str | int | None): Audio bitrate (ffmpeg flag: ``-b:a``).
            aspect (str | float | None): Set aspect ratio (ffmpeg flag: ``-aspect``).
            frames (int | None): Number of video frames to output (ffmpeg flag: ``-frames:v``).
            shortest (bool): Finish encoding when the shortest input stream ends.
            **kwargs: Additional output options.

        Returns:
            OutputNode: The created output node.
        """

        output_streams = [self]

        if streams is not None:
            output_streams.extend(streams)

        options = kwargs.copy()

        if format is not None:
            options["f"] = format
        if vcodec is not None:
            options["c:v"] = vcodec
        if acodec is not None:
            options["c:a"] = acodec
        if video_bitrate is not None:
            options["b:v"] = video_bitrate
        if audio_bitrate is not None:
            options["b:a"] = audio_bitrate
        if aspect is not None:
            options["aspect"] = aspect
        if frames is not None:
            options["frames:v"] = frames
        if shortest:
            options["shortest"] = None

        return OutputNode(filename, output_streams, output_options=options)

audio property

Accesses the audio elementary stream component.

Returns:

Name	Type	Description
TypedStream	TypedStream	An object representing the audio component of this stream.

video property

Accesses the video elementary stream component.

If the video stream wrapper has not been accessed before for this node, creates and caches a new 'TypedStream' instance representing video.

Returns:

Name	Type	Description
TypedStream	TypedStream	An object representing the video component of this stream.

node property

Retrieves all output streams produced by the source filter node.

This property allows access to the complete list of outputs generated by the filter that created this specific stream instance (accessing sibling streams).

Returns:

Type	Description
list[Stream]	list[Stream]: A list of all output streams from the parent FilterNode.

Raises:

Type	Description
AttributeError	If the source node is not a FilterNode (e.g., it is an InputNode), as it does not perform filtering operations.

__init__(source_node)

Source code in src/pyffmpeg/node.py

def __init__(self, source_node: Node):
    self.source_node: Node = source_node
    self.index: str | None = None
    self.elementary_streams: dict[str, TypedStream | IndexedStream] = {}

__getitem__(key)

Allows accessing elementary stream contained in this Stream

Source code in src/pyffmpeg/node.py

def __getitem__(self, key: str) -> "TypedStream | IndexedStream":
    """Allows accessing elementary stream contained in this Stream"""
    mapping = {"a": "audio", "v": "video"}
    if not isinstance(key, str) or len(key) != 1:
        raise TypeError("Expected string index (e.g. 'a')")

    if key in mapping:
        stream_type = mapping[key]
        return self.elementary_streams.setdefault(
            key, TypedStream(stream_type, self)
        )

    if key.isdigit():
        return self.elementary_streams.setdefault(key, IndexedStream(self))

    raise KeyError(
        f"Invalid stream key: {key!r}. Expected 'a', 'v', or a numeric index like '1'."
    )

filter(filter_name, *args, **kwargs)

Custom filter with a single input and a single output

Source code in src/pyffmpeg/node.py

def filter(self, filter_name: str, *args, **kwargs) -> "Stream":
    """Custom filter with a single input and a single output"""
    return self._apply_filter(filter_name, args, kwargs)[0]

filter_multi_output(filter_name, *args, **kwargs)

Creates a custom filter allowing dynamic creation of output streams

Source code in src/pyffmpeg/node.py

def filter_multi_output(
    self, filter_name: str, *args, **kwargs
) -> "FilterMultiOutput":
    """Creates a custom filter allowing dynamic creation of output streams"""
    node = FilterNode(
        filter_name=filter_name,
        positional_arguments=args,
        named_arguments=kwargs,
        inputs=[self],
        num_output_streams=0,
    )
    return FilterMultiOutput(node)

output(filename, streams=None, format=None, vcodec=None, acodec=None, video_bitrate=None, audio_bitrate=None, aspect=None, frames=None, shortest=False, **kwargs)

Creates an output node for this stream, optionally muxing other streams.

Parameters:

Name	Type	Description	Default
filename	str	The output file path or URL.	required
streams	list[Stream] | None	Additional streams to include in the output (e.g., audio tracks, subtitles) alongside the current stream.	None
format	str | None	Force output format (ffmpeg flag: -f).	None
vcodec	str | None	Video codec (ffmpeg flag: -c:v).	None
acodec	str | None	Audio codec (ffmpeg flag: -c:a).	None
video_bitrate	str | int | None	Video bitrate (ffmpeg flag: -b:v).	None
audio_bitrate	str | int | None	Audio bitrate (ffmpeg flag: -b:a).	None
aspect	str | float | None	Set aspect ratio (ffmpeg flag: -aspect).	None
frames	int | None	Number of video frames to output (ffmpeg flag: -frames:v).	None
shortest	bool	Finish encoding when the shortest input stream ends.	False
**kwargs		Additional output options.	{}

Returns:

Name	Type	Description
OutputNode	OutputNode	The created output node.

Source code in src/pyffmpeg/node.py

def output(
    self,
    filename: str,
    streams: list["Stream"] | None = None,
    format: str | None = None,
    vcodec: str | None = None,
    acodec: str | None = None,
    video_bitrate: str | int | None = None,
    audio_bitrate: str | int | None = None,
    aspect: str | float | None = None,
    frames: int | None = None,
    shortest: bool = False,
    **kwargs,
) -> "OutputNode":
    """
    Creates an output node for this stream, optionally muxing other streams.

    Args:
        filename (str): The output file path or URL.
        streams (list[Stream] | None): Additional streams to include in the output
            (e.g., audio tracks, subtitles) alongside the current stream.
        format (str | None): Force output format (ffmpeg flag: ``-f``).
        vcodec (str | None): Video codec (ffmpeg flag: ``-c:v``).
        acodec (str | None): Audio codec (ffmpeg flag: ``-c:a``).
        video_bitrate (str | int | None): Video bitrate (ffmpeg flag: ``-b:v``).
        audio_bitrate (str | int | None): Audio bitrate (ffmpeg flag: ``-b:a``).
        aspect (str | float | None): Set aspect ratio (ffmpeg flag: ``-aspect``).
        frames (int | None): Number of video frames to output (ffmpeg flag: ``-frames:v``).
        shortest (bool): Finish encoding when the shortest input stream ends.
        **kwargs: Additional output options.

    Returns:
        OutputNode: The created output node.
    """

    output_streams = [self]

    if streams is not None:
        output_streams.extend(streams)

    options = kwargs.copy()

    if format is not None:
        options["f"] = format
    if vcodec is not None:
        options["c:v"] = vcodec
    if acodec is not None:
        options["c:a"] = acodec
    if video_bitrate is not None:
        options["b:v"] = video_bitrate
    if audio_bitrate is not None:
        options["b:a"] = audio_bitrate
    if aspect is not None:
        options["aspect"] = aspect
    if frames is not None:
        options["frames:v"] = frames
    if shortest:
        options["shortest"] = None

    return OutputNode(filename, output_streams, output_options=options)

Elementary Streams

pyffmpeg.node.TypedStream

Bases: Stream

Elementary stream representing specific type of media out of those contained within a Stream

Source code in src/pyffmpeg/node.py

class TypedStream(Stream):
    """Elementary stream representing specific type of media out of those contained within a Stream"""

    def __init__(self, type: str, source_stream: Stream):
        self.type = type
        self.source_node = source_stream.source_node
        self.index: str | None = None

    def __getitem__(self, _):
        """Raises ValueError"""
        mapping = {"audio": "a", "video": "v"}
        raise ValueError(f"Stream already has a selector: {mapping[self.type]}")

__getitem__(_)

Raises ValueError

Source code in src/pyffmpeg/node.py

def __getitem__(self, _):
    """Raises ValueError"""
    mapping = {"audio": "a", "video": "v"}
    raise ValueError(f"Stream already has a selector: {mapping[self.type]}")

pyffmpeg.node.IndexedStream

Bases: Stream

Elementary stream represented by index within the containing stream

Source code in src/pyffmpeg/node.py

class IndexedStream(Stream):
    """Elementary stream represented by index within the containing stream"""

    def __init__(self, source_stream: Stream):
        self.source_node = source_stream.source_node
        self.label: str
        self.index: str | None = None

    def __getitem__(self, _):
        """Raises ValueError"""
        raise ValueError(f"Stream already has a selector")

__getitem__(_)

Raises ValueError

Source code in src/pyffmpeg/node.py

def __getitem__(self, _):
    """Raises ValueError"""
    raise ValueError(f"Stream already has a selector")