API Reference — Async¶

The asynchronous mirror under ffmpeg_wrap.aio, generated from its Google-style docstrings. Requires the optional [async] extra (uv add "ffmpeg-wrap[async]", or pip install "ffmpeg-wrap[async]").

ffmpeg_wrap.aio ¶

Asynchronous mirror of the public :mod:ffmpeg_wrap API, powered by AnyIO.

Importing this module requires the optional [async] extra::

pip install ffmpeg-wrap[async]

The coroutine mirrors live here so the core package stays dependency-free: import ffmpeg_wrap never imports :mod:anyio. Each coroutine is a thin imperative shell over the same pure helpers the sync API uses — command building, output parsing, decoding and error construction are written once in the private modules and reused here; the only async-specific line is the await anyio.run_process(...) exec call.

probe `async` ¶

probe(filename: str | PathLike[str], ffprobe_path: str = 'ffprobe') -> ProbeResult

Run ffprobe asynchronously and return typed output.

Async mirror of :func:ffmpeg_wrap.probe. Reuses the shared command builder and output parser; only the exec call is async.

Parameters:

Name	Type	Description	Default
`filename`	`str \| PathLike[str]`	Path to the file to probe.	required
`ffprobe_path`	`str`	Path to the ffprobe executable.	`'ffprobe'`

Returns:

Type	Description
`ProbeResult`	Parsed and typed output from ffprobe.

Raises:

Type	Description
`FFmpegError`	If ffprobe fails or output cannot be parsed.

Example

import anyio
from ffmpeg_wrap import aio

async def main():
    result = await aio.probe("video.mkv")
    print(result.format.format_name if result.format else None)
    for stream in result.streams:
        print(stream.index, stream.codec_type, stream.codec_name)

anyio.run(main)

Source code in src/ffmpeg_wrap/aio.py

async def probe(filename: str | os.PathLike[str], ffprobe_path: str = "ffprobe") -> ProbeResult:
    """Run ffprobe asynchronously and return typed output.

    Async mirror of :func:`ffmpeg_wrap.probe`. Reuses the shared command builder
    and output parser; only the exec call is async.

    Args:
        filename: Path to the file to probe.
        ffprobe_path: Path to the ffprobe executable.

    Returns:
        Parsed and typed output from ffprobe.

    Raises:
        FFmpegError: If ffprobe fails or output cannot be parsed.

    Example:
        ```python
        import anyio
        from ffmpeg_wrap import aio

        async def main():
            result = await aio.probe("video.mkv")
            print(result.format.format_name if result.format else None)
            for stream in result.streams:
                print(stream.index, stream.codec_type, stream.codec_name)

        anyio.run(main)
        ```
    """
    cmd = _build_probe_cmd(filename, ffprobe_path)
    try:
        result = await anyio.run_process(cmd, check=False)
    except OSError as e:
        logger.error(f"ffprobe could not be executed: {e}")
        raise _build_ffmpeg_error(f"ffprobe could not be executed: {e}", cmd=cmd) from e
    if result.returncode != 0:
        stderr_text = result.stderr.decode("utf-8", errors="replace") if result.stderr else None
        # Empty-stderr fallback mirrors sync ``probe`` (``stderr_text or str(e)``
        # where ``e`` is the ``CalledProcessError``) so ``str(FFmpegError)`` is
        # identical across sync/async.
        err_msg = stderr_text or str(CalledProcessError(result.returncode, cmd))
        logger.error(f"ffprobe failed: {err_msg}")
        raise _build_ffmpeg_error(
            f"ffprobe error: {err_msg}",
            stderr=stderr_text,
            returncode=result.returncode,
            cmd=cmd,
        )
    return _parse_probe_output(result.stdout, cmd)

validate `async` ¶

validate(filename: str | PathLike[str], ffprobe_path: str = 'ffprobe', loglevel: str = 'warning', extra_args: tuple[str, ...] = ()) -> tuple[bool, str]

Run ffprobe in validation mode asynchronously and report diagnostics.

Async mirror of :func:ffmpeg_wrap.validate. The loglevel precondition lives in the shared command builder, so an invalid loglevel raises ValueError exactly as in the sync path.

Parameters:

Name	Type	Description	Default
`filename`	`str \| PathLike[str]`	Path to the media file (str or PathLike).	required
`ffprobe_path`	`str`	Path to the ffprobe executable.	`'ffprobe'`
`loglevel`	`str`	Value passed to ffprobe's `-v` flag (default `"warning"`).	`'warning'`
`extra_args`	`tuple[str, ...]`	Additional raw arguments forwarded to ffprobe before the filename.	`()`

Returns:

Type	Description
`bool`	`(ok, stderr_text)` — see :func:`ffmpeg_wrap.validate`. Does NOT raise
`str`	on bad media — that is a normal outcome for a validator.

Raises:

Type	Description
`FFmpegError`	Only when the ffprobe executable cannot be run.
`ValueError`	On an invalid `loglevel`.

Example

import anyio
from ffmpeg_wrap import aio

async def main():
    ok, diag = await aio.validate("recording.mkv")
    if not ok:
        print("Validation failed:", diag)

anyio.run(main)

Source code in src/ffmpeg_wrap/aio.py

async def validate(
    filename: str | os.PathLike[str],
    ffprobe_path: str = "ffprobe",
    loglevel: str = "warning",
    extra_args: tuple[str, ...] = (),
) -> tuple[bool, str]:
    """Run ffprobe in validation mode asynchronously and report diagnostics.

    Async mirror of :func:`ffmpeg_wrap.validate`. The loglevel precondition
    lives in the shared command builder, so an invalid ``loglevel`` raises
    ``ValueError`` exactly as in the sync path.

    Args:
        filename: Path to the media file (str or PathLike).
        ffprobe_path: Path to the ffprobe executable.
        loglevel: Value passed to ffprobe's ``-v`` flag (default ``"warning"``).
        extra_args: Additional raw arguments forwarded to ffprobe before the
            filename.

    Returns:
        ``(ok, stderr_text)`` — see :func:`ffmpeg_wrap.validate`. Does NOT raise
        on bad media — that is a normal outcome for a validator.

    Raises:
        FFmpegError: Only when the ffprobe executable cannot be run.
        ValueError: On an invalid ``loglevel``.

    Example:
        ```python
        import anyio
        from ffmpeg_wrap import aio

        async def main():
            ok, diag = await aio.validate("recording.mkv")
            if not ok:
                print("Validation failed:", diag)

        anyio.run(main)
        ```
    """
    cmd = _build_validate_cmd(filename, ffprobe_path, loglevel, extra_args)
    try:
        result = await anyio.run_process(cmd, check=False)
    except OSError as e:
        logger.error(f"ffprobe could not be executed: {e}")
        raise _build_ffmpeg_error(f"ffprobe could not be executed: {e}", cmd=cmd) from e
    return _interpret_validate(result.returncode, result.stderr)

encoders `async` ¶

encoders(ffmpeg_path: str = 'ffmpeg') -> frozenset[str]

Return the set of encoder names reported by ffmpeg -encoders.

Async mirror of :func:ffmpeg_wrap.encoders. Shares the same per-path cache (_ENCODERS_CACHE) as the sync API, so a value resolved by either path populates one cache.

AnyIO's run_process always returns raw bytes; we decode stdout as UTF-8 before parsing so the result is frozenset[str] identical to sync (ffmpeg -encoders names are ASCII-only). The redundant-spawn race under concurrent first-use is benign: the only await is the subprocess itself, and the dict write is atomic.

Parameters:

Name	Type	Description	Default
`ffmpeg_path`	`str`	Path to the ffmpeg executable.	`'ffmpeg'`

Returns:

Type	Description
`frozenset[str]`	A frozenset of encoder names (e.g. `"libx264"`, `"h264_nvenc"`).

Raises:

Type	Description
`FFmpegError`	If ffmpeg cannot be run or exits non-zero.

Example

import anyio
from ffmpeg_wrap import aio

async def main():
    available = await aio.encoders()
    print("libx264" in available)

anyio.run(main)

Source code in src/ffmpeg_wrap/aio.py

async def encoders(ffmpeg_path: str = "ffmpeg") -> frozenset[str]:
    """Return the set of encoder names reported by ``ffmpeg -encoders``.

    Async mirror of :func:`ffmpeg_wrap.encoders`. Shares the same per-path cache
    (``_ENCODERS_CACHE``) as the sync API, so a value resolved by either path
    populates one cache.

    AnyIO's ``run_process`` always returns raw ``bytes``; we decode stdout as
    UTF-8 before parsing so the result is ``frozenset[str]`` identical to sync
    (``ffmpeg -encoders`` names are ASCII-only). The redundant-spawn race under
    concurrent first-use is benign: the only ``await`` is the subprocess itself,
    and the dict write is atomic.

    Args:
        ffmpeg_path: Path to the ffmpeg executable.

    Returns:
        A frozenset of encoder names (e.g. ``"libx264"``, ``"h264_nvenc"``).

    Raises:
        FFmpegError: If ffmpeg cannot be run or exits non-zero.

    Example:
        ```python
        import anyio
        from ffmpeg_wrap import aio

        async def main():
            available = await aio.encoders()
            print("libx264" in available)

        anyio.run(main)
        ```
    """
    cached = _ENCODERS_CACHE.get(ffmpeg_path)
    if cached is not None:
        return cached

    cmd = _build_encoders_cmd(ffmpeg_path)
    try:
        result = await anyio.run_process(cmd, check=False)
    except OSError as e:
        logger.error(f"ffmpeg could not be executed: {e}")
        raise _build_ffmpeg_error(f"ffmpeg could not be executed: {e}", cmd=cmd) from e
    if result.returncode != 0:
        stderr_text = result.stderr.decode("utf-8", errors="replace") if result.stderr else None
        # Empty-stderr fallback mirrors sync ``encoders`` (``stderr_text or str(e)``).
        err_msg = stderr_text or str(CalledProcessError(result.returncode, cmd))
        logger.error(f"ffmpeg -encoders failed: {err_msg}")
        raise _build_ffmpeg_error(
            f"ffmpeg -encoders error: {err_msg}",
            stderr=stderr_text,
            returncode=result.returncode,
            cmd=cmd,
        )

    parsed = _parse_encoders(result.stdout.decode("utf-8", errors="replace"))
    _ENCODERS_CACHE[ffmpeg_path] = parsed
    return parsed

has_encoder `async` ¶

has_encoder(name: str, ffmpeg_path: str = 'ffmpeg') -> bool

Return whether name is an available ffmpeg encoder.

Async mirror of :func:ffmpeg_wrap.has_encoder.

Parameters:

Name	Type	Description	Default
`name`	`str`	Encoder name to check (e.g. `"h264_nvenc"`).	required
`ffmpeg_path`	`str`	Path to the ffmpeg executable.	`'ffmpeg'`

Returns:

Type	Description
`bool`	`True` iff `name` appears in :func:`encoders`.

Raises:

Type	Description
`FFmpegError`	If ffmpeg cannot be run or exits non-zero.

Example

import anyio
from ffmpeg_wrap import aio

async def main():
    if await aio.has_encoder("h264_nvenc"):
        print("NVENC is available")

anyio.run(main)

Source code in src/ffmpeg_wrap/aio.py

async def has_encoder(name: str, ffmpeg_path: str = "ffmpeg") -> bool:
    """Return whether ``name`` is an available ffmpeg encoder.

    Async mirror of :func:`ffmpeg_wrap.has_encoder`.

    Args:
        name: Encoder name to check (e.g. ``"h264_nvenc"``).
        ffmpeg_path: Path to the ffmpeg executable.

    Returns:
        ``True`` iff ``name`` appears in :func:`encoders`.

    Raises:
        FFmpegError: If ffmpeg cannot be run or exits non-zero.

    Example:
        ```python
        import anyio
        from ffmpeg_wrap import aio

        async def main():
            if await aio.has_encoder("h264_nvenc"):
                print("NVENC is available")

        anyio.run(main)
        ```
    """
    return name in await encoders(ffmpeg_path)

run `async` ¶

run(ffmpeg: FFmpeg, capture_stdout: bool = ..., capture_stderr: bool = ..., *, text: Literal[False] = ...) -> tuple[bytes | None, bytes | None]

run(ffmpeg: FFmpeg, capture_stdout: bool = ..., capture_stderr: bool = ..., *, text: Literal[True]) -> tuple[str | None, str | None]

run(ffmpeg: FFmpeg, capture_stdout: bool = False, capture_stderr: bool = False, *, text: bool = False) -> tuple[bytes | None, bytes | None] | tuple[str | None, str | None]

Build and execute an :class:~ffmpeg_wrap.FFmpeg command asynchronously.

Async mirror of :meth:ffmpeg_wrap.FFmpeg.run, powered by AnyIO (runs on both the asyncio and trio backends). The return contract is identical to the sync run: a (stdout, stderr) tuple of bytes (or str when text=True), with None for any stream that was not captured.

Two execution paths mirror sync:

Capture path (capture_stderr=True): anyio.run_process captures stderr (and stdout when capture_stdout is set), building FFmpegError on a non-zero exit.
Tee path (capture_stderr=False): ffmpeg's stderr is forwarded live to the inherited terminal via the shared :class:~ffmpeg_wrap._textio.TeePump while only a bounded tail is retained for the failure path. This uses an AnyIO task group instead of the sync version's OS pump thread — so on the trio backend no per-process reaping thread is created either.

Parameters:

Name	Type	Description	Default
`ffmpeg`	`FFmpeg`	The built :class:`~ffmpeg_wrap.FFmpeg` instance to execute.	required
`capture_stdout`	`bool`	Whether to capture stdout (else inherit to terminal).	`False`
`capture_stderr`	`bool`	Whether to capture stderr (else tee it live).	`False`
`text`	`bool`	When True, decode stdout/stderr (and `FFmpegError.stderr`) as text leniently using the platform default encoding.	`False`

Returns:

Type	Description
`tuple[bytes \| None, bytes \| None] \| tuple[str \| None, str \| None]`	Tuple of (stdout, stderr); see :meth:`ffmpeg_wrap.FFmpeg.run`.

Raises:

Type	Description
`FFmpegError`	If ffmpeg fails or cannot be executed.

Example

import anyio
from ffmpeg_wrap import aio, input

async def main():
    _, stderr = await aio.run(
        input("in.mkv").output("out.mp4").overwrite_output(),
        capture_stderr=True,
        text=True,
    )
    print(stderr)

anyio.run(main)

Source code in src/ffmpeg_wrap/aio.py

async def run(
    ffmpeg: FFmpeg,
    capture_stdout: bool = False,
    capture_stderr: bool = False,
    *,
    text: bool = False,
) -> tuple[bytes | None, bytes | None] | tuple[str | None, str | None]:
    """Build and execute an :class:`~ffmpeg_wrap.FFmpeg` command asynchronously.

    Async mirror of :meth:`ffmpeg_wrap.FFmpeg.run`, powered by AnyIO (runs on
    both the asyncio and trio backends). The return contract is identical to the
    sync ``run``: a ``(stdout, stderr)`` tuple of ``bytes`` (or ``str`` when
    ``text=True``), with ``None`` for any stream that was not captured.

    Two execution paths mirror sync:

    * **Capture path** (``capture_stderr=True``): ``anyio.run_process`` captures
      stderr (and stdout when ``capture_stdout`` is set), building ``FFmpegError``
      on a non-zero exit.
    * **Tee path** (``capture_stderr=False``): ffmpeg's stderr is forwarded live
      to the inherited terminal via the shared :class:`~ffmpeg_wrap._textio.TeePump`
      while only a bounded tail is retained for the failure path. This uses an
      AnyIO task group instead of the sync version's OS pump thread — so on the
      trio backend no per-process reaping thread is created either.

    Args:
        ffmpeg: The built :class:`~ffmpeg_wrap.FFmpeg` instance to execute.
        capture_stdout: Whether to capture stdout (else inherit to terminal).
        capture_stderr: Whether to capture stderr (else tee it live).
        text: When True, decode stdout/stderr (and ``FFmpegError.stderr``) as
            text leniently using the platform default encoding.

    Returns:
        Tuple of (stdout, stderr); see :meth:`ffmpeg_wrap.FFmpeg.run`.

    Raises:
        FFmpegError: If ffmpeg fails or cannot be executed.

    Example:
        ```python
        import anyio
        from ffmpeg_wrap import aio, input

        async def main():
            _, stderr = await aio.run(
                input("in.mkv").output("out.mp4").overwrite_output(),
                capture_stderr=True,
                text=True,
            )
            print(stderr)

        anyio.run(main)
        ```
    """
    cmd = ffmpeg.compile()
    stdout_dest = PIPE if capture_stdout else None

    logger.debug(f"Running ffmpeg command (async): {' '.join(cmd)}")

    if capture_stderr:
        return await _run_capture(cmd, stdout_dest, text=text)
    return await _run_tee(cmd, stdout_dest, text=text)

API Reference — Async¶

ffmpeg_wrap.aio ¶

probe async ¶

validate async ¶

encoders async ¶

has_encoder async ¶

run async ¶

probe `async` ¶

validate `async` ¶

encoders `async` ¶

has_encoder `async` ¶

run `async` ¶