Skip to content

API Reference

FFmpeg

Bases: EventEmitter

Source code in ffmpeg/ffmpeg.py 
 20
 21
 22
 23
 24
 25
 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
223
224
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
257
258
259
class FFmpeg(EventEmitter):
    def __init__(self, executable: str = "ffmpeg"):
        """Initialize an `FFmpeg` instance.

        Args:
            executable: The path to the ffmpeg executable. Defaults to "ffmpeg".
        """
        super().__init__()

        self._executable: str = executable
        self._options: Options = Options()

        self._process: subprocess.Popen[bytes]
        self._executed: bool = False
        self._terminated: bool = False

        self._tracker = Tracker(self)  # type: ignore

    @property
    def arguments(self) -> list[str]:
        """Return a list of arguments to be used when executing FFmpeg.

        Returns:
            A lit of arguments to be used when executing FFmpeg.
        """
        return [self._executable, *self._options.build()]

    def option(self, key: str, value: Optional[types.Option] = None) -> Self:
        """Add a global option `-key` or `-key value`.

        Args:
            key: A key of the global option
            value: A value of the global option. If the option does not require a value, use None. Defaults to None.

        Returns:
            An instance of `FFmpeg` itself, so that calls can be chained.
        """
        self._options.option(key, value)
        return self

    def input(
        self,
        url: Union[str, os.PathLike],
        options: Optional[dict[str, Optional[types.Option]]] = None,
        **kwargs: Optional[types.Option],
    ) -> Self:
        """Add an input file with specified options.
           By calling this method multiple times, an arbitrary number of input files can be added.

        Args:
            url: URL for the input file.
            options: Options for the input file. Defaults to None.
            kwargs: Additional options for the input file.

        Note:
            Options for an input file can be specified in two ways:

            Using `options`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4", {"codec:v": "libx264"}).output("output.mp4")
            # Corresponds to `ffmpeg -codec:v libx264 -i input.mp4 output.mp4`
            ```

            Using `**kwargs`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4", vcodec="libx264").output("output.mp4")
            # Corresponds to `ffmpeg -vcodec libx264 -i input.mp4 output.mp4`
            ```

        Note:
            If an option does not require a value, use `None` for its value.

            ```python
            ffmpeg = FFmpeg().input("input.mp4", ignore_unknown=None).output("output.mp4")
            # Corresponds to `ffmpeg -ignore_unknown -i input.mp4 output.mp4`
            ```

        Returns:
            An instance of `FFmpeg` itself, so that calls can be chained.
        """
        self._options.input(url, options, **kwargs)
        return self

    def output(
        self,
        url: Union[str, os.PathLike],
        options: Optional[dict[str, Optional[types.Option]]] = None,
        **kwargs: Optional[types.Option],
    ) -> Self:
        """Add an output file with specified options.
           By calling this method multiple times, an arbitrary number of output files can be specified.

        Args:
            url: URL for the output file.
            options: Options for the output file. Defaults to None.
            kwargs: Additional options for the output file.

        Note:
            Options for an output file can be specified in two ways:

            Using `options`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", {"codec:v": "libx264"})
            # Corresponds to `ffmpeg -i input.mp4 -codec:v libx264 output.mp4`
            ```

            Using `**kwargs`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", vcodec="libx264")
            # Corresponds to `ffmpeg -i input.mp4 -vcodec libx264 output.mp4`
            ```

        Note:
            If an option does not require a value, use `None` for its value.

            ```python
            ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", ignore_unknown=None)
            # Corresponds to `ffmpeg -i input.mp4 -ignore_unknown output.mp4`
            ```

        Returns:
            An instance of `FFmpeg` itself, so that calls can be chained.
        """
        self._options.output(url, options, **kwargs)
        return self

    def execute(self, stream: Optional[Union[bytes, IO[bytes]]] = None, timeout: Optional[float] = None) -> bytes:
        """Execute FFmpeg using specified global options and files.

        Args:
            stream: A stream to input to the standard input. Defaults to None.
            timeout: The maximum number of seconds to wait before returning. Defaults to None.

        Raises:
            FFmpegAlreadyExecuted: If FFmpeg is already executed.
            FFmpegError: If FFmpeg process returns non-zero exit status.
            subprocess.TimeoutExpired: If FFmpeg process does not terminate after `timeout` seconds.

        Returns:
            The output to the standard output.
        """
        if self._executed:
            raise FFmpegAlreadyExecuted("FFmpeg is already executed", arguments=self.arguments)

        self._executed = False
        self._terminated = False

        if stream is not None:
            stream = ensure_io(stream)

        self.emit("start", self.arguments)

        self._process = create_subprocess(
            self.arguments,
            bufsize=0,
            stdin=subprocess.PIPE if stream is not None else None,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
            self._executed = True
            futures = [
                executor.submit(self._write_stdin, stream),
                executor.submit(self._read_stdout),
                executor.submit(self._handle_stderr),
                executor.submit(self._process.wait, timeout),
            ]
            done, pending = concurrent.futures.wait(futures, return_when=concurrent.futures.FIRST_EXCEPTION)
            self._executed = False

            for future in done:
                exception = future.exception()
                if exception is not None:
                    self._process.terminate()
                    concurrent.futures.wait(pending)

                    raise exception

        if self._process.returncode == 0:
            self.emit("completed")
        elif self._terminated:
            self.emit("terminated")
        else:
            raise FFmpegError.create(message=futures[2].result(), arguments=self.arguments)

        return futures[1].result()

    def terminate(self):
        """Gracefully terminate the running FFmpeg process.

        Raises:
            FFmpegError: If FFmpeg is not executed
        """
        if not self._executed:
            raise FFmpegError("FFmpeg is not executed", arguments=self.arguments)

        sigterm = signal.SIGTERM
        if is_windows():
            # On Windows, SIGTERM is an alias for TerminateProcess().
            # To gracefully terminate the FFmpeg process, we should use CTRL_BREAK_EVENT signal.
            # References:
            # - https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
            # - https://github.com/FFmpeg/FFmpeg/blob/release/5.1/fftools/ffmpeg.c#L371
            sigterm = signal.CTRL_BREAK_EVENT  # type: ignore

        self._terminated = True
        self._process.send_signal(sigterm)

    def _write_stdin(self, stream: Optional[IO[bytes]]):
        if stream is None:
            return

        assert self._process.stdin is not None

        for chunk in read_stream(stream, size=io.DEFAULT_BUFFER_SIZE):
            self._process.stdin.write(chunk)

        self._process.stdin.flush()
        self._process.stdin.close()

    def _read_stdout(self) -> bytes:
        assert self._process.stdout is not None

        buffer = bytearray()
        for chunk in read_stream(self._process.stdout, size=io.DEFAULT_BUFFER_SIZE):
            buffer.extend(chunk)

        self._process.stdout.close()
        return bytes(buffer)

    def _handle_stderr(self) -> str:
        assert self._process.stderr is not None

        line = b""
        for line in readlines(self._process.stderr):
            self.emit("stderr", line.decode())

        self._process.stderr.close()
        return line.decode()

__init__(executable='ffmpeg')

Initialize an FFmpeg instance.

Parameters:

Name Type Description Default
executable str

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

 'ffmpeg'
Source code in ffmpeg/ffmpeg.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, executable: str = "ffmpeg"):
    """Initialize an `FFmpeg` instance.

    Args:
        executable: The path to the ffmpeg executable. Defaults to "ffmpeg".
    """
    super().__init__()

    self._executable: str = executable
    self._options: Options = Options()

    self._process: subprocess.Popen[bytes]
    self._executed: bool = False
    self._terminated: bool = False

    self._tracker = Tracker(self)  # type: ignore

arguments: list[str] property

Return a list of arguments to be used when executing FFmpeg.

Returns:

Type Description
list[str]

A lit of arguments to be used when executing FFmpeg.

option(key, value=None)

Add a global option -key or -key value.

Parameters:

Name Type Description Default
key str

A key of the global option

 required
value Optional[Option]

A value of the global option. If the option does not require a value, use None. Defaults to None.

 None

Returns:

Type Description
Self

An instance of FFmpeg itself, so that calls can be chained.
Source code in ffmpeg/ffmpeg.py 
47
48
49
50
51
52
53
54
55
56
57
58
def option(self, key: str, value: Optional[types.Option] = None) -> Self:
    """Add a global option `-key` or `-key value`.

    Args:
        key: A key of the global option
        value: A value of the global option. If the option does not require a value, use None. Defaults to None.

    Returns:
        An instance of `FFmpeg` itself, so that calls can be chained.
    """
    self._options.option(key, value)
    return self

input(url, options=None, **kwargs)

Add an input file with specified options. By calling this method multiple times, an arbitrary number of input files can be added.

Parameters:

Name Type Description Default
url Union[str, PathLike]

URL for the input file.

 required
options Optional[dict[str, Optional[Option]]]

Options for the input file. Defaults to None.

 None
kwargs Optional[Option]

Additional options for the input file.

 {}
Note

Options for an input file can be specified in two ways:

Using options: 

ffmpeg = FFmpeg().input("input.mp4", {"codec:v": "libx264"}).output("output.mp4")
# Corresponds to `ffmpeg -codec:v libx264 -i input.mp4 output.mp4`

Using **kwargs: 

ffmpeg = FFmpeg().input("input.mp4", vcodec="libx264").output("output.mp4")
# Corresponds to `ffmpeg -vcodec libx264 -i input.mp4 output.mp4`

Note

If an option does not require a value, use None for its value.

ffmpeg = FFmpeg().input("input.mp4", ignore_unknown=None).output("output.mp4")
# Corresponds to `ffmpeg -ignore_unknown -i input.mp4 output.mp4`

Returns:

Type Description
Self

An instance of FFmpeg itself, so that calls can be chained.
Source code in ffmpeg/ffmpeg.py 
 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
def input(
    self,
    url: Union[str, os.PathLike],
    options: Optional[dict[str, Optional[types.Option]]] = None,
    **kwargs: Optional[types.Option],
) -> Self:
    """Add an input file with specified options.
       By calling this method multiple times, an arbitrary number of input files can be added.

    Args:
        url: URL for the input file.
        options: Options for the input file. Defaults to None.
        kwargs: Additional options for the input file.

    Note:
        Options for an input file can be specified in two ways:

        Using `options`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4", {"codec:v": "libx264"}).output("output.mp4")
        # Corresponds to `ffmpeg -codec:v libx264 -i input.mp4 output.mp4`
        ```

        Using `**kwargs`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4", vcodec="libx264").output("output.mp4")
        # Corresponds to `ffmpeg -vcodec libx264 -i input.mp4 output.mp4`
        ```

    Note:
        If an option does not require a value, use `None` for its value.

        ```python
        ffmpeg = FFmpeg().input("input.mp4", ignore_unknown=None).output("output.mp4")
        # Corresponds to `ffmpeg -ignore_unknown -i input.mp4 output.mp4`
        ```

    Returns:
        An instance of `FFmpeg` itself, so that calls can be chained.
    """
    self._options.input(url, options, **kwargs)
    return self

output(url, options=None, **kwargs)

Add an output file with specified options. By calling this method multiple times, an arbitrary number of output files can be specified.

Parameters:

Name Type Description Default
url Union[str, PathLike]

URL for the output file.

 required
options Optional[dict[str, Optional[Option]]]

Options for the output file. Defaults to None.

 None
kwargs Optional[Option]

Additional options for the output file.

 {}
Note

Options for an output file can be specified in two ways:

Using options: 

ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", {"codec:v": "libx264"})
# Corresponds to `ffmpeg -i input.mp4 -codec:v libx264 output.mp4`

Using **kwargs: 

ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", vcodec="libx264")
# Corresponds to `ffmpeg -i input.mp4 -vcodec libx264 output.mp4`

Note

If an option does not require a value, use None for its value.

ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", ignore_unknown=None)
# Corresponds to `ffmpeg -i input.mp4 -ignore_unknown output.mp4`

Returns:

Type Description
Self

An instance of FFmpeg itself, so that calls can be chained.
Source code in ffmpeg/ffmpeg.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
def output(
    self,
    url: Union[str, os.PathLike],
    options: Optional[dict[str, Optional[types.Option]]] = None,
    **kwargs: Optional[types.Option],
) -> Self:
    """Add an output file with specified options.
       By calling this method multiple times, an arbitrary number of output files can be specified.

    Args:
        url: URL for the output file.
        options: Options for the output file. Defaults to None.
        kwargs: Additional options for the output file.

    Note:
        Options for an output file can be specified in two ways:

        Using `options`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", {"codec:v": "libx264"})
        # Corresponds to `ffmpeg -i input.mp4 -codec:v libx264 output.mp4`
        ```

        Using `**kwargs`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", vcodec="libx264")
        # Corresponds to `ffmpeg -i input.mp4 -vcodec libx264 output.mp4`
        ```

    Note:
        If an option does not require a value, use `None` for its value.

        ```python
        ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", ignore_unknown=None)
        # Corresponds to `ffmpeg -i input.mp4 -ignore_unknown output.mp4`
        ```

    Returns:
        An instance of `FFmpeg` itself, so that calls can be chained.
    """
    self._options.output(url, options, **kwargs)
    return self

execute(stream=None, timeout=None)

Execute FFmpeg using specified global options and files.

Parameters:

Name Type Description Default
stream Optional[Union[bytes, IO[bytes]]]

A stream to input to the standard input. Defaults to None.

 None
timeout Optional[float]

The maximum number of seconds to wait before returning. Defaults to None.

 None

Raises:

Type Description
FFmpegAlreadyExecuted

If FFmpeg is already executed.
FFmpegError

If FFmpeg process returns non-zero exit status.
TimeoutExpired

If FFmpeg process does not terminate after timeout seconds.

Returns:

Type Description
bytes

The output to the standard output.
Source code in ffmpeg/ffmpeg.py 
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
def execute(self, stream: Optional[Union[bytes, IO[bytes]]] = None, timeout: Optional[float] = None) -> bytes:
    """Execute FFmpeg using specified global options and files.

    Args:
        stream: A stream to input to the standard input. Defaults to None.
        timeout: The maximum number of seconds to wait before returning. Defaults to None.

    Raises:
        FFmpegAlreadyExecuted: If FFmpeg is already executed.
        FFmpegError: If FFmpeg process returns non-zero exit status.
        subprocess.TimeoutExpired: If FFmpeg process does not terminate after `timeout` seconds.

    Returns:
        The output to the standard output.
    """
    if self._executed:
        raise FFmpegAlreadyExecuted("FFmpeg is already executed", arguments=self.arguments)

    self._executed = False
    self._terminated = False

    if stream is not None:
        stream = ensure_io(stream)

    self.emit("start", self.arguments)

    self._process = create_subprocess(
        self.arguments,
        bufsize=0,
        stdin=subprocess.PIPE if stream is not None else None,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        self._executed = True
        futures = [
            executor.submit(self._write_stdin, stream),
            executor.submit(self._read_stdout),
            executor.submit(self._handle_stderr),
            executor.submit(self._process.wait, timeout),
        ]
        done, pending = concurrent.futures.wait(futures, return_when=concurrent.futures.FIRST_EXCEPTION)
        self._executed = False

        for future in done:
            exception = future.exception()
            if exception is not None:
                self._process.terminate()
                concurrent.futures.wait(pending)

                raise exception

    if self._process.returncode == 0:
        self.emit("completed")
    elif self._terminated:
        self.emit("terminated")
    else:
        raise FFmpegError.create(message=futures[2].result(), arguments=self.arguments)

    return futures[1].result()

terminate()

Gracefully terminate the running FFmpeg process.

Raises:

Type Description
FFmpegError

If FFmpeg is not executed
Source code in ffmpeg/ffmpeg.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def terminate(self):
    """Gracefully terminate the running FFmpeg process.

    Raises:
        FFmpegError: If FFmpeg is not executed
    """
    if not self._executed:
        raise FFmpegError("FFmpeg is not executed", arguments=self.arguments)

    sigterm = signal.SIGTERM
    if is_windows():
        # On Windows, SIGTERM is an alias for TerminateProcess().
        # To gracefully terminate the FFmpeg process, we should use CTRL_BREAK_EVENT signal.
        # References:
        # - https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
        # - https://github.com/FFmpeg/FFmpeg/blob/release/5.1/fftools/ffmpeg.c#L371
        sigterm = signal.CTRL_BREAK_EVENT  # type: ignore

    self._terminated = True
    self._process.send_signal(sigterm)

ffmpeg.asyncio

FFmpeg

Bases: AsyncIOEventEmitter

Source code in ffmpeg/asyncio/ffmpeg.py 
 21
 22
 23
 24
 25
 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
223
224
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
257
258
259
260
261
262
263
264
265
class FFmpeg(AsyncIOEventEmitter):
    def __init__(self, executable: str = "ffmpeg"):
        """Initialize an `FFmpeg` instance using `asyncio`

        Args:
            executable: The path to the ffmpeg executable. Defaults to "ffmpeg".
        """
        super().__init__()

        self._executable: str = executable
        self._options: Options = Options()

        self._process: asyncio.subprocess.Process
        self._executed: bool = False
        self._terminated: bool = False

        self._tracker = Tracker(self)  # type: ignore

        self.once("error", self._reraise_exception)

    @property
    def arguments(self) -> list[str]:
        """Return a list of arguments to be used when executing FFmpeg.

        Returns:
            A lit of arguments to be used when executing FFmpeg.
        """
        return [self._executable, *self._options.build()]

    def option(self, key: str, value: Optional[types.Option] = None) -> Self:
        """Add a global option `-key` or `-key value`.

        Args:
            key: A key of the global option
            value: A value of the global option. If the option does not require a value, use None. Defaults to None.

        Returns:
            An instance of `FFmpeg` itself, so that calls can be chained.
        """
        self._options.option(key, value)
        return self

    def input(
        self,
        url: Union[str, os.PathLike],
        options: Optional[dict[str, Optional[types.Option]]] = None,
        **kwargs: Optional[types.Option],
    ) -> Self:
        """Add an input file with specified options.
           By calling this method multiple times, an arbitrary number of input files can be added.

        Args:
            url: URL for the input file.
            options: Options for the input file. Defaults to None.
            kwargs: Additional options for the input file.

        Note:
            Options for an input file can be specified in two ways:

            Using `options`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4", {"codec:v": "libx264"}).output("output.mp4")
            # Corresponds to `ffmpeg -codec:v libx264 -i input.mp4 output.mp4`
            ```

            Using `**kwargs`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4", vcodec="libx264").output("output.mp4")
            # Corresponds to `ffmpeg -vcodec libx264 -i input.mp4 output.mp4`
            ```

        Note:
            If an option does not require a value, use `None` for its value.

            ```python
            ffmpeg = FFmpeg().input("input.mp4", ignore_unknown=None).output("output.mp4")
            # Corresponds to `ffmpeg -ignore_unknown -i input.mp4 output.mp4`
            ```

        Returns:
            An instance of `FFmpeg` itself, so that calls can be chained.
        """
        self._options.input(url, options, **kwargs)
        return self

    def output(
        self,
        url: Union[str, os.PathLike],
        options: Optional[dict[str, Optional[types.Option]]] = None,
        **kwargs: Optional[types.Option],
    ) -> Self:
        """Add an output file with specified options.
           By calling this method multiple times, an arbitrary number of output files can be specified.

        Args:
            url: URL for the output file.
            options: Options for the output file. Defaults to None.
            kwargs: Additional options for the output file.

        Note:
            Options for an output file can be specified in two ways:

            Using `options`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", {"codec:v": "libx264"})
            # Corresponds to `ffmpeg -i input.mp4 -codec:v libx264 output.mp4`
            ```

            Using `**kwargs`:
            ```python
            ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", vcodec="libx264")
            # Corresponds to `ffmpeg -i input.mp4 -vcodec libx264 output.mp4`
            ```

        Note:
            If an option does not require a value, use `None` for its value.

            ```python
            ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", ignore_unknown=None)
            # Corresponds to `ffmpeg -i input.mp4 -ignore_unknown output.mp4`
            ```

        Returns:
            An instance of `FFmpeg` itself, so that calls can be chained.
        """
        self._options.output(url, options, **kwargs)
        return self

    async def execute(
        self, stream: Optional[Union[bytes, asyncio.StreamReader]] = None, timeout: Optional[float] = None
    ) -> bytes:
        """Execute FFmpeg using specified global options and files.

        Args:
            stream: A stream to input to the standard input. Defaults to None.
            timeout: The maximum number of seconds to wait before returning. Defaults to None.

        Raises:
            FFmpegAlreadyExecuted: If FFmpeg is already executed.
            FFmpegError: If FFmpeg process returns non-zero exit status.
            asyncio.TimeoutError: If FFmpeg process does not terminate after `timeout` seconds.

        Returns:
            The output to the standard output.
        """
        if self._executed:
            raise FFmpegAlreadyExecuted("FFmpeg is already executed", arguments=self.arguments)

        self._executed = False
        self._terminated = False

        if stream is not None:
            stream = ensure_stream_reader(stream)

        self.emit("start", self.arguments)

        self._process = await create_subprocess(
            *self.arguments,
            stdin=subprocess.PIPE if stream is not None else None,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

        self._executed = True
        tasks = [
            asyncio.create_task(self._write_stdin(stream)),
            asyncio.create_task(self._read_stdout()),
            asyncio.create_task(self._handle_stderr()),
            asyncio.create_task(asyncio.wait_for(self._process.wait(), timeout=timeout)),
        ]
        done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
        self._executed = False

        for task in done:
            exception = task.exception()
            if exception is not None:
                self._process.terminate()
                for task in pending:
                    await task

                raise exception

        if self._process.returncode == 0:
            self.emit("completed")
        elif self._terminated:
            self.emit("terminated")
        else:
            raise FFmpegError.create(message=tasks[2].result(), arguments=self.arguments)

        return tasks[1].result()

    def terminate(self):
        """Gracefully terminate the running FFmpeg process.

        Raises:
            FFmpegError: If FFmpeg is not executed
        """
        if not self._executed:
            raise FFmpegError("FFmpeg is not executed", arguments=self.arguments)

        sigterm = signal.SIGTERM
        if is_windows():
            # On Windows, SIGTERM is an alias for TerminateProcess().
            # To gracefully terminate the FFmpeg process, we should use CTRL_BREAK_EVENT signal.
            # References:
            # - https://docs.python.org/3.10/library/subprocess.html#subprocess.Popen.send_signal
            # - https://github.com/FFmpeg/FFmpeg/blob/release/5.1/fftools/ffmpeg.c#L371
            sigterm = signal.CTRL_BREAK_EVENT  # type: ignore

        self._terminated = True
        self._process.send_signal(sigterm)

    async def _write_stdin(self, stream: Optional[asyncio.StreamReader]):
        if stream is None:
            return

        assert self._process.stdin is not None

        async for chunk in read_stream(stream, size=io.DEFAULT_BUFFER_SIZE):
            self._process.stdin.write(chunk)
            await self._process.stdin.drain()

        self._process.stdin.close()
        await self._process.stdin.wait_closed()

    async def _read_stdout(self) -> bytes:
        assert self._process.stdout is not None

        buffer = bytearray()
        async for chunk in read_stream(self._process.stdout, size=io.DEFAULT_BUFFER_SIZE):
            buffer.extend(chunk)

        return bytes(buffer)

    async def _handle_stderr(self) -> str:
        assert self._process.stderr is not None

        line = b""
        async for line in readlines(self._process.stderr):
            self.emit("stderr", line.decode())

        return line.decode()

    def _reraise_exception(self, exception: Exception):
        raise exception

__init__(executable='ffmpeg')

Initialize an FFmpeg instance using asyncio

Parameters:

Name Type Description Default
executable str

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

 'ffmpeg'
Source code in ffmpeg/asyncio/ffmpeg.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(self, executable: str = "ffmpeg"):
    """Initialize an `FFmpeg` instance using `asyncio`

    Args:
        executable: The path to the ffmpeg executable. Defaults to "ffmpeg".
    """
    super().__init__()

    self._executable: str = executable
    self._options: Options = Options()

    self._process: asyncio.subprocess.Process
    self._executed: bool = False
    self._terminated: bool = False

    self._tracker = Tracker(self)  # type: ignore

    self.once("error", self._reraise_exception)

arguments: list[str] property

Return a list of arguments to be used when executing FFmpeg.

Returns:

Type Description
list[str]

A lit of arguments to be used when executing FFmpeg.

option(key, value=None)

Add a global option -key or -key value.

Parameters:

Name Type Description Default
key str

A key of the global option

 required
value Optional[Option]

A value of the global option. If the option does not require a value, use None. Defaults to None.

 None

Returns:

Type Description
Self

An instance of FFmpeg itself, so that calls can be chained.
Source code in ffmpeg/asyncio/ffmpeg.py 
50
51
52
53
54
55
56
57
58
59
60
61
def option(self, key: str, value: Optional[types.Option] = None) -> Self:
    """Add a global option `-key` or `-key value`.

    Args:
        key: A key of the global option
        value: A value of the global option. If the option does not require a value, use None. Defaults to None.

    Returns:
        An instance of `FFmpeg` itself, so that calls can be chained.
    """
    self._options.option(key, value)
    return self

input(url, options=None, **kwargs)

Add an input file with specified options. By calling this method multiple times, an arbitrary number of input files can be added.

Parameters:

Name Type Description Default
url Union[str, PathLike]

URL for the input file.

 required
options Optional[dict[str, Optional[Option]]]

Options for the input file. Defaults to None.

 None
kwargs Optional[Option]

Additional options for the input file.

 {}
Note

Options for an input file can be specified in two ways:

Using options: 

ffmpeg = FFmpeg().input("input.mp4", {"codec:v": "libx264"}).output("output.mp4")
# Corresponds to `ffmpeg -codec:v libx264 -i input.mp4 output.mp4`

Using **kwargs: 

ffmpeg = FFmpeg().input("input.mp4", vcodec="libx264").output("output.mp4")
# Corresponds to `ffmpeg -vcodec libx264 -i input.mp4 output.mp4`

Note

If an option does not require a value, use None for its value.

ffmpeg = FFmpeg().input("input.mp4", ignore_unknown=None).output("output.mp4")
# Corresponds to `ffmpeg -ignore_unknown -i input.mp4 output.mp4`

Returns:

Type Description
Self

An instance of FFmpeg itself, so that calls can be chained.
Source code in ffmpeg/asyncio/ffmpeg.py 
 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
def input(
    self,
    url: Union[str, os.PathLike],
    options: Optional[dict[str, Optional[types.Option]]] = None,
    **kwargs: Optional[types.Option],
) -> Self:
    """Add an input file with specified options.
       By calling this method multiple times, an arbitrary number of input files can be added.

    Args:
        url: URL for the input file.
        options: Options for the input file. Defaults to None.
        kwargs: Additional options for the input file.

    Note:
        Options for an input file can be specified in two ways:

        Using `options`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4", {"codec:v": "libx264"}).output("output.mp4")
        # Corresponds to `ffmpeg -codec:v libx264 -i input.mp4 output.mp4`
        ```

        Using `**kwargs`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4", vcodec="libx264").output("output.mp4")
        # Corresponds to `ffmpeg -vcodec libx264 -i input.mp4 output.mp4`
        ```

    Note:
        If an option does not require a value, use `None` for its value.

        ```python
        ffmpeg = FFmpeg().input("input.mp4", ignore_unknown=None).output("output.mp4")
        # Corresponds to `ffmpeg -ignore_unknown -i input.mp4 output.mp4`
        ```

    Returns:
        An instance of `FFmpeg` itself, so that calls can be chained.
    """
    self._options.input(url, options, **kwargs)
    return self

output(url, options=None, **kwargs)

Add an output file with specified options. By calling this method multiple times, an arbitrary number of output files can be specified.

Parameters:

Name Type Description Default
url Union[str, PathLike]

URL for the output file.

 required
options Optional[dict[str, Optional[Option]]]

Options for the output file. Defaults to None.

 None
kwargs Optional[Option]

Additional options for the output file.

 {}
Note

Options for an output file can be specified in two ways:

Using options: 

ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", {"codec:v": "libx264"})
# Corresponds to `ffmpeg -i input.mp4 -codec:v libx264 output.mp4`

Using **kwargs: 

ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", vcodec="libx264")
# Corresponds to `ffmpeg -i input.mp4 -vcodec libx264 output.mp4`

Note

If an option does not require a value, use None for its value.

ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", ignore_unknown=None)
# Corresponds to `ffmpeg -i input.mp4 -ignore_unknown output.mp4`

Returns:

Type Description
Self

An instance of FFmpeg itself, so that calls can be chained.
Source code in ffmpeg/asyncio/ffmpeg.py 
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
def output(
    self,
    url: Union[str, os.PathLike],
    options: Optional[dict[str, Optional[types.Option]]] = None,
    **kwargs: Optional[types.Option],
) -> Self:
    """Add an output file with specified options.
       By calling this method multiple times, an arbitrary number of output files can be specified.

    Args:
        url: URL for the output file.
        options: Options for the output file. Defaults to None.
        kwargs: Additional options for the output file.

    Note:
        Options for an output file can be specified in two ways:

        Using `options`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", {"codec:v": "libx264"})
        # Corresponds to `ffmpeg -i input.mp4 -codec:v libx264 output.mp4`
        ```

        Using `**kwargs`:
        ```python
        ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", vcodec="libx264")
        # Corresponds to `ffmpeg -i input.mp4 -vcodec libx264 output.mp4`
        ```

    Note:
        If an option does not require a value, use `None` for its value.

        ```python
        ffmpeg = FFmpeg().input("input.mp4").output("output.mp4", ignore_unknown=None)
        # Corresponds to `ffmpeg -i input.mp4 -ignore_unknown output.mp4`
        ```

    Returns:
        An instance of `FFmpeg` itself, so that calls can be chained.
    """
    self._options.output(url, options, **kwargs)
    return self

execute(stream=None, timeout=None) async

Execute FFmpeg using specified global options and files.

Parameters:

Name Type Description Default
stream Optional[Union[bytes, StreamReader]]

A stream to input to the standard input. Defaults to None.

 None
timeout Optional[float]

The maximum number of seconds to wait before returning. Defaults to None.

 None

Raises:

Type Description
FFmpegAlreadyExecuted

If FFmpeg is already executed.
FFmpegError

If FFmpeg process returns non-zero exit status.
TimeoutError

If FFmpeg process does not terminate after timeout seconds.

Returns:

Type Description
bytes

The output to the standard output.
Source code in ffmpeg/asyncio/ffmpeg.py 
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
async def execute(
    self, stream: Optional[Union[bytes, asyncio.StreamReader]] = None, timeout: Optional[float] = None
) -> bytes:
    """Execute FFmpeg using specified global options and files.

    Args:
        stream: A stream to input to the standard input. Defaults to None.
        timeout: The maximum number of seconds to wait before returning. Defaults to None.

    Raises:
        FFmpegAlreadyExecuted: If FFmpeg is already executed.
        FFmpegError: If FFmpeg process returns non-zero exit status.
        asyncio.TimeoutError: If FFmpeg process does not terminate after `timeout` seconds.

    Returns:
        The output to the standard output.
    """
    if self._executed:
        raise FFmpegAlreadyExecuted("FFmpeg is already executed", arguments=self.arguments)

    self._executed = False
    self._terminated = False

    if stream is not None:
        stream = ensure_stream_reader(stream)

    self.emit("start", self.arguments)

    self._process = await create_subprocess(
        *self.arguments,
        stdin=subprocess.PIPE if stream is not None else None,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    self._executed = True
    tasks = [
        asyncio.create_task(self._write_stdin(stream)),
        asyncio.create_task(self._read_stdout()),
        asyncio.create_task(self._handle_stderr()),
        asyncio.create_task(asyncio.wait_for(self._process.wait(), timeout=timeout)),
    ]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
    self._executed = False

    for task in done:
        exception = task.exception()
        if exception is not None:
            self._process.terminate()
            for task in pending:
                await task

            raise exception

    if self._process.returncode == 0:
        self.emit("completed")
    elif self._terminated:
        self.emit("terminated")
    else:
        raise FFmpegError.create(message=tasks[2].result(), arguments=self.arguments)

    return tasks[1].result()

terminate()

Gracefully terminate the running FFmpeg process.

Raises:

Type Description
FFmpegError

If FFmpeg is not executed
Source code in ffmpeg/asyncio/ffmpeg.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def terminate(self):
    """Gracefully terminate the running FFmpeg process.

    Raises:
        FFmpegError: If FFmpeg is not executed
    """
    if not self._executed:
        raise FFmpegError("FFmpeg is not executed", arguments=self.arguments)

    sigterm = signal.SIGTERM
    if is_windows():
        # On Windows, SIGTERM is an alias for TerminateProcess().
        # To gracefully terminate the FFmpeg process, we should use CTRL_BREAK_EVENT signal.
        # References:
        # - https://docs.python.org/3.10/library/subprocess.html#subprocess.Popen.send_signal
        # - https://github.com/FFmpeg/FFmpeg/blob/release/5.1/fftools/ffmpeg.c#L371
        sigterm = signal.CTRL_BREAK_EVENT  # type: ignore

    self._terminated = True
    self._process.send_signal(sigterm)

Progress dataclass

Represents a progress of FFmpeg operation.

Attributes:

Name Type Description
frame int

The number of processed frames.
fps float

The processing speed in frame per seconds.
size int

The current size of the media in bytes.
time timedelta

The current time of the media.
bitrate float

The processing speed in kilobits per second.
speed float

The processing speed
Source code in ffmpeg/progress.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@dataclass(frozen=True)
class Progress:
    """Represents a progress of `FFmpeg` operation.

    Attributes:
        frame: The number of processed frames.
        fps: The processing speed in frame per seconds.
        size: The current size of the media in bytes.
        time: The current time of the media.
        bitrate: The processing speed in kilobits per second.
        speed: The processing speed
    """

    frame: int
    fps: float
    size: int
    time: timedelta
    bitrate: float
    speed: float

Exceptions

FFmpegError

Bases: Exception

Represents an error that has occurred during FFmpeg operation.

Attributes:

Name Type Description
message

An error message providing details about the error.
arguments

Arguments passed to FFmpeg.
Source code in ffmpeg/errors.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class FFmpegError(Exception):
    """Represents an error that has occurred during `FFmpeg` operation.

    Attributes:
        message: An error message providing details about the error.
        arguments: Arguments passed to FFmpeg.
    """

    _patterns: ClassVar[Optional[list[str]]] = None

    def __init__(self, message: str, arguments: list[str]):
        super().__init__(message)

        self.message = message
        self.arguments = arguments

    @classmethod
    def create(cls, message: str, arguments: list[str]) -> Self:
        for subclass in cls.__subclasses__():
            if subclass._patterns is None:
                continue

            for pattern in subclass._patterns:
                if re.search(pattern, message, flags=re.IGNORECASE) is not None:
                    return subclass(message, arguments)

        return cls(message, arguments)

FFmpegAlreadyExecuted

Bases: FFmpegError

Represents FFmpeg is being executed

Source code in ffmpeg/errors.py 
38
39
class FFmpegAlreadyExecuted(FFmpegError):
    "Represents FFmpeg is being executed"

FFmpegFileNotFound

Bases: FFmpegError

Represents an input file was not found

Source code in ffmpeg/errors.py 
42
43
44
45
46
47
class FFmpegFileNotFound(FFmpegError):
    "Represents an input file was not found"
    _patterns = [
        r"no such file",
        r"could not open file",
    ]

FFmpegInvalidCommand

Bases: FFmpegError

Represents FFmpeg was passed invalid options or arguments

Source code in ffmpeg/errors.py 
50
51
52
53
54
55
56
57
58
class FFmpegInvalidCommand(FFmpegError):
    "Represents FFmpeg was passed invalid options or arguments"
    _patterns = [
        r"option .* ?not found",
        r"unrecognized option",
        r"trailing options were found on the commandline",
        r"invalid encoder type",
        r"codec not currently supported in container",
    ]

FFmpegUnsupportedCodec

Bases: FFmpegError

Represents FFmpeg attempted to use an unsupported codec

Source code in ffmpeg/errors.py 
61
62
63
64
65
66
67
68
class FFmpegUnsupportedCodec(FFmpegError):
    "Represents FFmpeg attempted to use an unsupported codec"
    _patterns = [
        r"unknown encoder",
        r"encoder not found",
        r"unknown decoder",
        r"decoder not found",
    ]

Events

start

This event is emitted just before FFmpeg is executed.

Parameters:

Name Type Description
arguments list[str] A list of arguments to execute FFmpeg

stderr

This event is emitted when a line is output to the standard error by FFmpeg.

Parameters:

Name Type Description
line str A line from the standard error

progress

This event is emitted when FFmpeg reports progress.

Parameters:

Name Type Description
progress Progress A progress of FFmpeg operation

completed

This event is emitted when FFmpeg is successfully exited.

terminated

This event is emitted when FFmpeg is gracefully terminated by calling FFmpeg.terminate().