Skip to content

Nodes

This module contains all node types that make up the FFmpeg processing graph.

Base Node Classes

pyffmpeg.node.Node

Abstract base class for all components in the FFmpeg processing graph.

pyffmpeg.node.ProcessableNode

Bases: Node

Nodes that can be further processed with filters.

Source code in src/pyffmpeg/node.py 
17
18
19
20
21
22
23
class ProcessableNode(Node):
    """Nodes that can be further processed with filters."""

    def __init__(self, num_output_streams: int = 1):
        self.output_streams: list[Stream] = [
            Stream(self) for i in range(num_output_streams)
        ]

__init__(num_output_streams=1)

Source code in src/pyffmpeg/node.py 
20
21
22
23
def __init__(self, num_output_streams: int = 1):
    self.output_streams: list[Stream] = [
        Stream(self) for i in range(num_output_streams)
    ]

pyffmpeg.node.RunnableNode

Bases: Node

Source code in src/pyffmpeg/node.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class RunnableNode(Node):
    def __init__(self):
        self.global_options: list[str] = []

    def global_args(self, *args) -> "RunnableNode":
        """Adds global options"""
        self.global_options.extend(args)
        return self

    def overwrite_output(self) -> "RunnableNode":
        """Adds global overwrite option"""
        self.global_options.append("-y")
        return self

    def _compile_global_kwargs(self, options_dict: dict) -> list[str]:
        """Converts kwargs to list"""
        args = []
        key_map = {
            "overwrite_output": "y",
            "log_level": "loglevel",
            "quiet": "loglevel",
        }
        for key, value in options_dict.items():
            if key == "quiet" and value is True:
                args.extend(["-loglevel", "quiet"])
                continue

            flag_name = key_map.get(key, key)

            if value is True:
                # -y with overwrite_output=True
                args.append(f"-{flag_name}")
            elif value is not None and value is not False:
                # handles for example log_level="error"
                args.extend([f"-{flag_name}", str(value)])
        return args

    def get_args(self, overwrite_output: bool = False, **global_options) -> list[str]:
        """Builds a list of command arguments based on the current graph.

        Args:
            overwrite_output (bool): If True, adds the '-y' flag to overwrite output files without asking.
                Defaults to False.
            **global_options: Arbitrary global options passed to FFmpeg (e.g., v="quiet" becomes -v quiet).

        Returns:
            list[str]: A list of command-line arguments (excluding the executable name).
        """
        global_options["overwrite_output"] = overwrite_output
        kwargs_args = self._compile_global_kwargs(global_options)
        sorter = GraphSorter(self)
        command_builder = CommandBuilder(
            sorter.sort(), self.global_options + kwargs_args
        )
        return command_builder.build_args()

    def compile(
        self, cmd: str = "ffmpeg", overwrite_output: bool = False, **global_options
    ) -> list[str]:
        """Builds the full command line arguments for invoking FFmpeg, including the executable.

        Args:
            cmd (str): The path to the FFmpeg executable or command list.
                Defaults to "ffmpeg".
            overwrite_output (bool): If True, adds the '-y' flag to overwrite output files.
                Defaults to False.
            **global_options: Additional global options.

        Returns:
            list[str]: The complete command line arguments ready for execution.
        """
        if isinstance(cmd, str):
            cmd = [cmd]
        elif not isinstance(cmd, list):
            cmd = list(cmd)
        return cmd + self.get_args(overwrite_output=overwrite_output, **global_options)

    def run(
        self,
        cmd: str | list[str] = "ffmpeg",
        capture_stdout: bool = False,
        capture_stderr: bool = False,
        input: bytes | None = None,
        quiet: bool = False,
        overwrite_output: bool = False,
        cwd: str | None = None,
        compile_function=None,
    ) -> tuple[bytes | None, bytes | None]:
        """Execute the ffmpeg command represented by a RunnableNode synchronously.

        This method compiles the graph into an FFmpeg command and runs it using
        subprocess.run. It waits for the command to finish.

        Args:
            cmd (str | list[str]): The path to the FFmpeg executable or a command list.
                Defaults to "ffmpeg".
            capture_stdout (bool): If True, captures standard output and returns it.
                Defaults to False.
            capture_stderr (bool): If True, captures standard error and returns it.
                Defaults to False.
            input (bytes | None): Input data to be passed to the process's stdin.
                Defaults to None.
            quiet (bool): If True, passes a quiet flag to the compilation step to suppress logs.
                Defaults to False.
            overwrite_output (bool): If True, adds the '-y' flag to overwrite output files.
                Defaults to False.
            cwd (str | None): Sets the current working directory for the process.
                Defaults to None.
            compile_function (Callable | None): A custom function to compile the arguments.
                If None, uses self.__class__.compile. Defaults to None.

        Returns:
            tuple[bytes | None, bytes | None]: A tuple containing (stdout, stderr).
            If capture_stdout/stderr is False, the corresponding value will be None.

        Raises:
            TypeError: If the current instance is not a RunnableNode.
            Error: If the FFmpeg process returns a non-zero exit code (wraps CalledProcessError).
        """
        if not isinstance(self, RunnableNode):
            raise TypeError(f"Expected RunnableNode, got {type(self)}")

        compile_function = compile_function or self.__class__.compile
        cmdline = compile_function(
            self,
            cmd=cmd,
            overwrite_output=overwrite_output,
            quiet=quiet,
        )

        stdout = subprocess.PIPE if capture_stdout else None
        stderr = subprocess.PIPE if capture_stderr else None

        try:
            process = subprocess.run(
                cmdline,
                input=input,
                stdout=stdout,
                stderr=stderr,
                cwd=cwd,
                check=True,
            )
            return process.stdout, process.stderr
        except subprocess.CalledProcessError as e:
            raise Error(
                "ffmpeg error (see stderr output for detail)",
                stdout=e.stdout,
                stderr=e.stderr,
            )

    def run_async(
        self,
        cmd: str | list[str] = "ffmpeg",
        pipe_stdin: bool = False,
        pipe_stdout: bool = False,
        pipe_stderr: bool = False,
        quiet: bool = False,
        overwrite_output: bool = False,
        cwd: str | None = None,
    ) -> subprocess.Popen:
        """Runs the ffmpeg process asynchronously and returns a Popen object.

        This method compiles the graph and starts the process using subprocess.Popen.
        It does not wait for the process to finish.

        Args:
            cmd (str | list[str]): The path to the FFmpeg executable. Defaults to "ffmpeg".
            pipe_stdin (bool): If True, opens a pipe for standard input (stdin=subprocess.PIPE).
                Defaults to False.
            pipe_stdout (bool): If True, opens a pipe for standard output (stdout=subprocess.PIPE).
                Defaults to False.
            pipe_stderr (bool): If True, opens a pipe for standard error (stderr=subprocess.PIPE).
                Defaults to False.
            quiet (bool): If True, redirects stderr to stdout and stdout to DEVNULL.
                Overrides pipe configurations to silence output. Defaults to False.
            overwrite_output (bool): If True, adds the '-y' flag to overwrite output files.
                Defaults to False.
            cwd (str | None): Sets the current working directory for the process.
                Defaults to None.

        Returns:
            subprocess.Popen: A handle to the running FFmpeg process.
        """
        args = self.compile(cmd, overwrite_output=overwrite_output)
        stdin_stream = subprocess.PIPE if pipe_stdin else None
        stdout_stream = subprocess.PIPE if pipe_stdout else None
        stderr_stream = subprocess.PIPE if pipe_stderr else None
        if quiet:
            stderr_stream = subprocess.STDOUT
            stdout_stream = subprocess.DEVNULL
        return subprocess.Popen(
            args,
            stdin=stdin_stream,
            stdout=stdout_stream,
            stderr=stderr_stream,
            cwd=cwd,
        )

__init__()

Source code in src/pyffmpeg/node.py 
27
28
def __init__(self):
    self.global_options: list[str] = []

global_args(*args)

Adds global options

Source code in src/pyffmpeg/node.py 
30
31
32
33
def global_args(self, *args) -> "RunnableNode":
    """Adds global options"""
    self.global_options.extend(args)
    return self

overwrite_output()

Adds global overwrite option

Source code in src/pyffmpeg/node.py 
35
36
37
38
def overwrite_output(self) -> "RunnableNode":
    """Adds global overwrite option"""
    self.global_options.append("-y")
    return self

get_args(overwrite_output=False, **global_options)

Builds a list of command arguments based on the current graph.

Parameters:

Name Type Description Default
overwrite_output bool

If True, adds the '-y' flag to overwrite output files without asking. Defaults to False.

 False
**global_options

Arbitrary global options passed to FFmpeg (e.g., v="quiet" becomes -v quiet).

 {}

Returns:

Type Description
list[str]

list[str]: A list of command-line arguments (excluding the executable name).
Source code in src/pyffmpeg/node.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def get_args(self, overwrite_output: bool = False, **global_options) -> list[str]:
    """Builds a list of command arguments based on the current graph.

    Args:
        overwrite_output (bool): If True, adds the '-y' flag to overwrite output files without asking.
            Defaults to False.
        **global_options: Arbitrary global options passed to FFmpeg (e.g., v="quiet" becomes -v quiet).

    Returns:
        list[str]: A list of command-line arguments (excluding the executable name).
    """
    global_options["overwrite_output"] = overwrite_output
    kwargs_args = self._compile_global_kwargs(global_options)
    sorter = GraphSorter(self)
    command_builder = CommandBuilder(
        sorter.sort(), self.global_options + kwargs_args
    )
    return command_builder.build_args()

compile(cmd='ffmpeg', overwrite_output=False, **global_options)

Builds the full command line arguments for invoking FFmpeg, including the executable.

Parameters:

Name Type Description Default
cmd str

The path to the FFmpeg executable or command list. Defaults to "ffmpeg".

 'ffmpeg'
overwrite_output bool

If True, adds the '-y' flag to overwrite output files. Defaults to False.

 False
**global_options

Additional global options.

 {}

Returns:

Type Description
list[str]

list[str]: The complete command line arguments ready for execution.
Source code in src/pyffmpeg/node.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def compile(
    self, cmd: str = "ffmpeg", overwrite_output: bool = False, **global_options
) -> list[str]:
    """Builds the full command line arguments for invoking FFmpeg, including the executable.

    Args:
        cmd (str): The path to the FFmpeg executable or command list.
            Defaults to "ffmpeg".
        overwrite_output (bool): If True, adds the '-y' flag to overwrite output files.
            Defaults to False.
        **global_options: Additional global options.

    Returns:
        list[str]: The complete command line arguments ready for execution.
    """
    if isinstance(cmd, str):
        cmd = [cmd]
    elif not isinstance(cmd, list):
        cmd = list(cmd)
    return cmd + self.get_args(overwrite_output=overwrite_output, **global_options)

run(cmd='ffmpeg', capture_stdout=False, capture_stderr=False, input=None, quiet=False, overwrite_output=False, cwd=None, compile_function=None)

Execute the ffmpeg command represented by a RunnableNode synchronously.

This method compiles the graph into an FFmpeg command and runs it using subprocess.run. It waits for the command to finish.

Parameters:

Name Type Description Default
cmd str | list[str]

The path to the FFmpeg executable or a command list. Defaults to "ffmpeg".

 'ffmpeg'
capture_stdout bool

If True, captures standard output and returns it. Defaults to False.

 False
capture_stderr bool

If True, captures standard error and returns it. Defaults to False.

 False
input bytes | None

Input data to be passed to the process's stdin. Defaults to None.

 None
quiet bool

If True, passes a quiet flag to the compilation step to suppress logs. Defaults to False.

 False
overwrite_output bool

If True, adds the '-y' flag to overwrite output files. Defaults to False.

 False
cwd str | None

Sets the current working directory for the process. Defaults to None.

 None
compile_function Callable | None

A custom function to compile the arguments. If None, uses self.class.compile. Defaults to None.

 None

Returns:

Type Description
bytes | None

tuple[bytes | None, bytes | None]: A tuple containing (stdout, stderr).
bytes | None

If capture_stdout/stderr is False, the corresponding value will be None.

Raises:

Type Description
TypeError

If the current instance is not a RunnableNode.
Error

If the FFmpeg process returns a non-zero exit code (wraps CalledProcessError).
Source code in src/pyffmpeg/node.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def run(
    self,
    cmd: str | list[str] = "ffmpeg",
    capture_stdout: bool = False,
    capture_stderr: bool = False,
    input: bytes | None = None,
    quiet: bool = False,
    overwrite_output: bool = False,
    cwd: str | None = None,
    compile_function=None,
) -> tuple[bytes | None, bytes | None]:
    """Execute the ffmpeg command represented by a RunnableNode synchronously.

    This method compiles the graph into an FFmpeg command and runs it using
    subprocess.run. It waits for the command to finish.

    Args:
        cmd (str | list[str]): The path to the FFmpeg executable or a command list.
            Defaults to "ffmpeg".
        capture_stdout (bool): If True, captures standard output and returns it.
            Defaults to False.
        capture_stderr (bool): If True, captures standard error and returns it.
            Defaults to False.
        input (bytes | None): Input data to be passed to the process's stdin.
            Defaults to None.
        quiet (bool): If True, passes a quiet flag to the compilation step to suppress logs.
            Defaults to False.
        overwrite_output (bool): If True, adds the '-y' flag to overwrite output files.
            Defaults to False.
        cwd (str | None): Sets the current working directory for the process.
            Defaults to None.
        compile_function (Callable | None): A custom function to compile the arguments.
            If None, uses self.__class__.compile. Defaults to None.

    Returns:
        tuple[bytes | None, bytes | None]: A tuple containing (stdout, stderr).
        If capture_stdout/stderr is False, the corresponding value will be None.

    Raises:
        TypeError: If the current instance is not a RunnableNode.
        Error: If the FFmpeg process returns a non-zero exit code (wraps CalledProcessError).
    """
    if not isinstance(self, RunnableNode):
        raise TypeError(f"Expected RunnableNode, got {type(self)}")

    compile_function = compile_function or self.__class__.compile
    cmdline = compile_function(
        self,
        cmd=cmd,
        overwrite_output=overwrite_output,
        quiet=quiet,
    )

    stdout = subprocess.PIPE if capture_stdout else None
    stderr = subprocess.PIPE if capture_stderr else None

    try:
        process = subprocess.run(
            cmdline,
            input=input,
            stdout=stdout,
            stderr=stderr,
            cwd=cwd,
            check=True,
        )
        return process.stdout, process.stderr
    except subprocess.CalledProcessError as e:
        raise Error(
            "ffmpeg error (see stderr output for detail)",
            stdout=e.stdout,
            stderr=e.stderr,
        )

run_async(cmd='ffmpeg', pipe_stdin=False, pipe_stdout=False, pipe_stderr=False, quiet=False, overwrite_output=False, cwd=None)

Runs the ffmpeg process asynchronously and returns a Popen object.

This method compiles the graph and starts the process using subprocess.Popen. It does not wait for the process to finish.

Parameters:

Name Type Description Default
cmd str | list[str]

The path to the FFmpeg executable. Defaults to "ffmpeg".

 'ffmpeg'
pipe_stdin bool

If True, opens a pipe for standard input (stdin=subprocess.PIPE). Defaults to False.

 False
pipe_stdout bool

If True, opens a pipe for standard output (stdout=subprocess.PIPE). Defaults to False.

 False
pipe_stderr bool

If True, opens a pipe for standard error (stderr=subprocess.PIPE). Defaults to False.

 False
quiet bool

If True, redirects stderr to stdout and stdout to DEVNULL. Overrides pipe configurations to silence output. Defaults to False.

 False
overwrite_output bool

If True, adds the '-y' flag to overwrite output files. Defaults to False.

 False
cwd str | None

Sets the current working directory for the process. Defaults to None.

 None

Returns:

Type Description
Popen

subprocess.Popen: A handle to the running FFmpeg process.
Source code in src/pyffmpeg/node.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def run_async(
    self,
    cmd: str | list[str] = "ffmpeg",
    pipe_stdin: bool = False,
    pipe_stdout: bool = False,
    pipe_stderr: bool = False,
    quiet: bool = False,
    overwrite_output: bool = False,
    cwd: str | None = None,
) -> subprocess.Popen:
    """Runs the ffmpeg process asynchronously and returns a Popen object.

    This method compiles the graph and starts the process using subprocess.Popen.
    It does not wait for the process to finish.

    Args:
        cmd (str | list[str]): The path to the FFmpeg executable. Defaults to "ffmpeg".
        pipe_stdin (bool): If True, opens a pipe for standard input (stdin=subprocess.PIPE).
            Defaults to False.
        pipe_stdout (bool): If True, opens a pipe for standard output (stdout=subprocess.PIPE).
            Defaults to False.
        pipe_stderr (bool): If True, opens a pipe for standard error (stderr=subprocess.PIPE).
            Defaults to False.
        quiet (bool): If True, redirects stderr to stdout and stdout to DEVNULL.
            Overrides pipe configurations to silence output. Defaults to False.
        overwrite_output (bool): If True, adds the '-y' flag to overwrite output files.
            Defaults to False.
        cwd (str | None): Sets the current working directory for the process.
            Defaults to None.

    Returns:
        subprocess.Popen: A handle to the running FFmpeg process.
    """
    args = self.compile(cmd, overwrite_output=overwrite_output)
    stdin_stream = subprocess.PIPE if pipe_stdin else None
    stdout_stream = subprocess.PIPE if pipe_stdout else None
    stderr_stream = subprocess.PIPE if pipe_stderr else None
    if quiet:
        stderr_stream = subprocess.STDOUT
        stdout_stream = subprocess.DEVNULL
    return subprocess.Popen(
        args,
        stdin=stdin_stream,
        stdout=stdout_stream,
        stderr=stderr_stream,
        cwd=cwd,
    )

Input/Output Nodes

pyffmpeg.node.InputNode

Bases: ProcessableNode

Nodes representing input files.

Source code in src/pyffmpeg/node.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
class InputNode(ProcessableNode):
    """Nodes representing input files."""

    def __init__(self, filename: str, options: dict[str, Any] = None):
        super().__init__()
        self.filename: str = filename
        self.options: dict[str, Any] = options

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

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

    def get_input_args(self) -> list[str]:
        """Returns command args for this input"""
        options = self.options.copy()
        args = []

        if format := options.pop("format", None):
            args.extend(["-f", str(format)])
        if video_size := options.pop("video_size", None):
            if isinstance(video_size, (tuple, list)) and len(video_size) == 2:
                video_size = f"{video_size[0]}x{video_size[1]}"
            args.extend(["-video_size", str(video_size)])

        args.extend(convert_kwargs_to_cmd_line_args(options))
        args.extend(["-i", self.filename])

        return args

__init__(filename, options=None)

Source code in src/pyffmpeg/node.py 
228
229
230
231
def __init__(self, filename: str, options: dict[str, Any] = None):
    super().__init__()
    self.filename: str = filename
    self.options: dict[str, Any] = options

get_input_args()

Returns command args for this input

Source code in src/pyffmpeg/node.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def get_input_args(self) -> list[str]:
    """Returns command args for this input"""
    options = self.options.copy()
    args = []

    if format := options.pop("format", None):
        args.extend(["-f", str(format)])
    if video_size := options.pop("video_size", None):
        if isinstance(video_size, (tuple, list)) and len(video_size) == 2:
            video_size = f"{video_size[0]}x{video_size[1]}"
        args.extend(["-video_size", str(video_size)])

    args.extend(convert_kwargs_to_cmd_line_args(options))
    args.extend(["-i", self.filename])

    return args

pyffmpeg.node.OutputNode

Bases: RunnableNode

Nodes representing output files.

Source code in src/pyffmpeg/node.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
class OutputNode(RunnableNode):
    """Nodes representing output files."""

    def __init__(
        self,
        filename: str,
        inputs: list["Stream"],
        output_options: dict[str, str | list[str]] = {},
    ):
        super().__init__()
        self.inputs: list[Stream] = inputs
        self.filename: str = filename
        self.output_options: dict[str, str | list[str]] = output_options

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, OutputNode):
            return NotImplemented
        return self.filename == other.filename and self.inputs == other.inputs

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

    def _normalize_output_options(self):
        """Replaces keys names in output_options from human readable (passed by the user)"""
        """to names existing in ffmpeg docs, prepares for later use to build command"""
        """Casts option values to string"""
        keys_names_mapping = {
            "video_bitrate": "b:v",
            "audio_bitrate": "b:a",
            "format": "f",
        }
        self.output_options = {
            keys_names_mapping.get(k, k): v for k, v in self.output_options.items()
        }

        video_size = self.output_options.get("video_size")
        if isinstance(video_size, Sequence) and not isinstance(video_size, str):
            try:
                width, height = video_size
            except ValueError:
                raise ValueError(
                    "video_size must contain exactly two elements: (width, height)"
                )
            self.output_options["video_size"] = f"{width}x{height}"

    def get_output_args(self, enforce_output_mapping) -> list[str]:
        """Builds command args representing the output"""
        """Generates args for output options"""
        """Generates args for mapping streams to the output if neccessary"""
        args: list[str] = []
        self._normalize_output_options()
        options = self.output_options.copy()
        format = options.pop("f", None)
        args.extend(convert_kwargs_to_cmd_line_args(options, sort=False))

        if (
            len(self.inputs) == 1
            and isinstance(self.inputs[0].source_node, InputNode)
            and not isinstance(self.inputs[0], (TypedStream, IndexedStream))
            and not enforce_output_mapping
        ):
            args.append(self.filename)
            return args

        for input in self.inputs:
            args.append("-map")
            args.append(
                f"[{input.index}]"
                if isinstance(input.source_node, FilterNode)
                else input.index
            )

        if format:
            args.extend(["-f", str(format)])

        args.append(self.filename)

        return args

__init__(filename, inputs, output_options={})

Source code in src/pyffmpeg/node.py 
262
263
264
265
266
267
268
269
270
271
def __init__(
    self,
    filename: str,
    inputs: list["Stream"],
    output_options: dict[str, str | list[str]] = {},
):
    super().__init__()
    self.inputs: list[Stream] = inputs
    self.filename: str = filename
    self.output_options: dict[str, str | list[str]] = output_options

get_output_args(enforce_output_mapping)

Builds command args representing the output

Source code in src/pyffmpeg/node.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def get_output_args(self, enforce_output_mapping) -> list[str]:
    """Builds command args representing the output"""
    """Generates args for output options"""
    """Generates args for mapping streams to the output if neccessary"""
    args: list[str] = []
    self._normalize_output_options()
    options = self.output_options.copy()
    format = options.pop("f", None)
    args.extend(convert_kwargs_to_cmd_line_args(options, sort=False))

    if (
        len(self.inputs) == 1
        and isinstance(self.inputs[0].source_node, InputNode)
        and not isinstance(self.inputs[0], (TypedStream, IndexedStream))
        and not enforce_output_mapping
    ):
        args.append(self.filename)
        return args

    for input in self.inputs:
        args.append("-map")
        args.append(
            f"[{input.index}]"
            if isinstance(input.source_node, FilterNode)
            else input.index
        )

    if format:
        args.extend(["-f", str(format)])

    args.append(self.filename)

    return args

pyffmpeg.node.MergedOutputNode

Bases: RunnableNode

Node representing multiple outputs

Source code in src/pyffmpeg/node.py 
339
340
341
342
343
344
class MergedOutputNode(RunnableNode):
    """Node representing multiple outputs"""

    def __init__(self, outputs: Sequence[OutputNode]):
        super().__init__()
        self.outputs: tuple[OutputNode] = tuple(outputs)

__init__(outputs)

Source code in src/pyffmpeg/node.py 
342
343
344
def __init__(self, outputs: Sequence[OutputNode]):
    super().__init__()
    self.outputs: tuple[OutputNode] = tuple(outputs)

pyffmpeg.node.SinkNode

Bases: RunnableNode

Represents a graph terminal node that is a sink filter (e.g., nullsink, buffersink), not a file output.

Source code in src/pyffmpeg/node.py 
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
class SinkNode(RunnableNode):
    """
    Represents a graph terminal node that is a sink filter (e.g., nullsink, buffersink),
    not a file output.
    """

    def __init__(self, filter_node: "FilterNode"):
        super().__init__()
        self.filter_node = filter_node
        self.inputs = filter_node.inputs

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

    def __hash__(self) -> int:
        return hash(self.filter_node)

__init__(filter_node)

Source code in src/pyffmpeg/node.py 
353
354
355
356
def __init__(self, filter_node: "FilterNode"):
    super().__init__()
    self.filter_node = filter_node
    self.inputs = filter_node.inputs

Filter Nodes

pyffmpeg.node.FilterNode

Bases: ProcessableNode

Nodes representing filter operations.

Source code in src/pyffmpeg/node.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
class FilterNode(ProcessableNode):
    """Nodes representing filter operations."""

    def __init__(
        self,
        filter_name: str,
        positional_arguments: tuple[str],
        named_arguments: dict[str, Any],
        inputs: list["Stream"],
        num_output_streams: int = 1,
    ):
        super().__init__(num_output_streams)
        self.filter_name: str = filter_name
        self.positional_arguments: tuple = positional_arguments
        self.named_arguments: dict = named_arguments
        self.inputs: list[Stream] = inputs

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, FilterNode):
            return NotImplemented
        return (
            self.filter_name == other.filter_name
            and self.positional_arguments == other.positional_arguments
            and self.named_arguments == other.named_arguments
            and self.inputs == other.inputs
        )

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

    def get_command_string(self) -> str:
        """Builds a command string based on this filter"""
        input_streams = [f"[{input.index}]" for input in self.inputs]
        output_streams = [f"[{output.index}]" for output in self.output_streams]

        positional_arguments = (str(arg) for arg in self.positional_arguments)

        named_arguments = []
        for name, value in sorted(self.named_arguments.items()):
            if value is None:
                continue

            if value is True:
                value = "true"
            elif value is False:
                value = "false"

            val_escaped = escape_filter_description(escape_filter_option(value))
            named_arguments.append(f"{name}={val_escaped}")

        all_arguments = [*positional_arguments, *named_arguments]
        arguments_string = ":".join(all_arguments)

        return f"{''.join(input_streams)}{self.filter_name}{f'=' if arguments_string else ''}{arguments_string}{''.join(output_streams)}"

__init__(filter_name, positional_arguments, named_arguments, inputs, num_output_streams=1)

Source code in src/pyffmpeg/node.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
def __init__(
    self,
    filter_name: str,
    positional_arguments: tuple[str],
    named_arguments: dict[str, Any],
    inputs: list["Stream"],
    num_output_streams: int = 1,
):
    super().__init__(num_output_streams)
    self.filter_name: str = filter_name
    self.positional_arguments: tuple = positional_arguments
    self.named_arguments: dict = named_arguments
    self.inputs: list[Stream] = inputs

get_command_string()

Builds a command string based on this filter

Source code in src/pyffmpeg/node.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def get_command_string(self) -> str:
    """Builds a command string based on this filter"""
    input_streams = [f"[{input.index}]" for input in self.inputs]
    output_streams = [f"[{output.index}]" for output in self.output_streams]

    positional_arguments = (str(arg) for arg in self.positional_arguments)

    named_arguments = []
    for name, value in sorted(self.named_arguments.items()):
        if value is None:
            continue

        if value is True:
            value = "true"
        elif value is False:
            value = "false"

        val_escaped = escape_filter_description(escape_filter_option(value))
        named_arguments.append(f"{name}={val_escaped}")

    all_arguments = [*positional_arguments, *named_arguments]
    arguments_string = ":".join(all_arguments)

    return f"{''.join(input_streams)}{self.filter_name}{f'=' if arguments_string else ''}{arguments_string}{''.join(output_streams)}"

pyffmpeg.node.FilterMultiOutput

Filter node wrapper which allows creating arbitrary outputs for the filter node dynamically

Source code in src/pyffmpeg/node.py 
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
class FilterMultiOutput:
    """Filter node wrapper which allows creating arbitrary outputs for the filter node dynamically"""

    def __init__(self, filter_node: FilterNode):
        self.filter_node = filter_node
        self.stream_cache: dict[str, Stream] = {}

    def __getitem__(self, key: str | int) -> "Stream":
        """Returns new or existing stream under label"""
        label = str(key)
        if label in self.stream_cache:
            return self.stream_cache[label]

        new_stream = Stream(self.filter_node)
        self.filter_node.output_streams.append(new_stream)
        self.stream_cache[label] = new_stream
        return new_stream

__getitem__(key)

Returns new or existing stream under label

Source code in src/pyffmpeg/node.py 
783
784
785
786
787
788
789
790
791
792
def __getitem__(self, key: str | int) -> "Stream":
    """Returns new or existing stream under label"""
    label = str(key)
    if label in self.stream_cache:
        return self.stream_cache[label]

    new_stream = Stream(self.filter_node)
    self.filter_node.output_streams.append(new_stream)
    self.stream_cache[label] = new_stream
    return new_stream