fffw.wrapper package

Submodules

fffw.wrapper.base module

class fffw.wrapper.base.BaseWrapper

Bases: CommandMixin, Params

Base class for generating command line arguments from params.

Values meanings: * True: flag presence * False/None: flag absence * List/Tuple: param name is repeated multiple times with values * Callable: function call result is added to result * All other: param name and value are added to result

get_args() → List[Any]
get_cmd() → str
handle_stderr(line: str) → str
handle_stdout(line: str) → str
run(stdin: None | str | TextIO = None, timeout: float | None = None) → Tuple[int, str, str]
runner(command: str | bytes, *args: Any, stdin: None | str | TextIO = None, stdout: Callable[[str], str] | None = None, stderr: Callable[[str], str] | None = None, timeout: int | float | None = None) → Runner

Initializer runner instance.

command: str
class fffw.wrapper.base.CommandMixin

Bases: object

runner_class

alias of Runner

command: str
key_prefix: str = '-'
key_suffix: str = ' '
stderr: bool = True
stdout: bool = True
timeout: float | None = None
class fffw.wrapper.base.Runner(command: str | bytes, *args: Any, stdin: None | str | TextIO = None, stdout: Callable[[str], str] | None = None, stderr: Callable[[str], str] | None = None, timeout: int | float | None = None)

Bases: object

Wrapper for Popen process for non-blocking streams handling.

async static read(reader: StreamReader | None, callback: Callable[[str], str] | None, output: StringIO) → None

Handles read from stdout/stderr.

Reads lines from stream and feeds it to callback. Values, filtered by callback, are written to output buffer.

Parameters:

  • reader – Process.stdout or Process.stderr instance

  • callback – callback for handling read lines

  • output – output biffer

async run() → Tuple[int, str, str]
async start_process() → Process

Starts child process.

async static write(writer: StreamWriter | None, stream: TextIO | None) → None

Handle write to stdin.

Parameters:

  • writer – Process.stdin instance

  • stream – stream to read lines from

class fffw.wrapper.base.UniversalLineReader(reader: StreamReader, bufsize: int = 81920, blocksize: int = 8192, encoding: str = 'utf8')

Bases: object

Reads bytes from asyncio.StreamReader and splits it to lines with either CR or LF, or even CRLF.

https://docs.python.org/3/glossary.html#term-universal-newlines

>>> # noinspection PyUnresolvedReferences
... line_iter = UniversalLineReader(process.stderr)
>>> async for line in line_iter:
...    print(line)
async readlines() → AsyncIterator[str]

fffw.wrapper.helpers module

fffw.wrapper.helpers.ensure_binary(x: List[Any]) → List[bytes]
fffw.wrapper.helpers.ensure_binary(x: int | float | str | bytes | None) → bytes
fffw.wrapper.helpers.ensure_binary(x: Callable[[...], Tuple[Any]]) → Callable[[...], Tuple[bytes]]
fffw.wrapper.helpers.ensure_binary(x: Callable[[...], List[Any]]) → Callable[[...], List[bytes]]
fffw.wrapper.helpers.ensure_binary(x: Callable[[...], int | float | str | bytes | None]) → Callable[[...], bytes]

Recursively ensures that all values except collections are bytes.

  • tuples and lists are encoded recursively

  • strings are encoded with utf-8

  • bytes are left intact

  • functions are decorated with ensuring binary results

  • all other types are converted to string and encoded

fffw.wrapper.helpers.ensure_text(x: Tuple[Any, ...]) → Tuple[str, ...]
fffw.wrapper.helpers.ensure_text(x: List[Any]) → List[str]
fffw.wrapper.helpers.ensure_text(x: int | float | str | bytes | None) → str

Recursively ensures that all values except collections are strings.

fffw.wrapper.helpers.quote(token: Any) → str

Escapes a token for command line.

fffw.wrapper.params module

class fffw.wrapper.params.Params

Bases: object

Base class for parametrized objects.

as_pairs() → List[Tuple[str | None, str | None]]
Returns:

a list or pairs (key, value), where key is optional ffmpeg parameter name and value is parameter value

  • parameters are all dataclass fields without skip flag.

  • by default field name is used as ffmpeg parameter name, but this can be overridden by name metadata field.

  • if name is empty, key is omitted.

  • if stream_suffix metadata flag is set, :v or :a modifier is appended to key depending of current object stream kind.

  • if value is callable, real value is computed via function call

  • if value is iterable, corresponding parameter is added to a result multiple times

  • if value is True, parameter is added as a flag without a value

ALLOWED = ()

List of attributes that are allowed to be set after instance initialization.

fffw.wrapper.params.param(default: Any | None = None, name: None | str = None, stream_suffix: bool = False, init: bool = True, skip: bool = False, render: Callable[[Any], Any] | None = None) → Any

Command line and filter parameters constructor.

Parameters:

  • default – default value for parameter, or a callable (see dataclass field default_factory)

  • name – command line argument name

  • stream_suffix – flag to add stream specifier suffix to parameter name (like ‘-b:v’ or ‘-c:a’)

  • init – add parameter to dataclass constructor.

  • skip – skip parameter while generating command line args list.

  • render – a callable used to format output value.

Returns:

dataclass field definition with extra metadata.

>>> from fffw.encoding.filters import VideoFilter
>>> @dataclass
... class Deinterlace(VideoFilter):
...     filter = 'yadif'
...     mode: int = param(default=0)
...
>>>

Module contents

class fffw.wrapper.BaseWrapper

Bases: CommandMixin, Params

Base class for generating command line arguments from params.

Values meanings: * True: flag presence * False/None: flag absence * List/Tuple: param name is repeated multiple times with values * Callable: function call result is added to result * All other: param name and value are added to result

get_args() → List[Any]
get_cmd() → str
handle_stderr(line: str) → str
handle_stdout(line: str) → str
run(stdin: None | str | TextIO = None, timeout: float | None = None) → Tuple[int, str, str]
runner(command: str | bytes, *args: Any, stdin: None | str | TextIO = None, stdout: Callable[[str], str] | None = None, stderr: Callable[[str], str] | None = None, timeout: int | float | None = None) → Runner

Initializer runner instance.

command: str
fffw.wrapper.ensure_binary(x: Any) → Any

Recursively ensures that all values except collections are bytes.

  • tuples and lists are encoded recursively

  • strings are encoded with utf-8

  • bytes are left intact

  • functions are decorated with ensuring binary results

  • all other types are converted to string and encoded

fffw.wrapper.param(default: Any | None = None, name: None | str = None, stream_suffix: bool = False, init: bool = True, skip: bool = False, render: Callable[[Any], Any] | None = None) → Any

Command line and filter parameters constructor.

Parameters:

  • default – default value for parameter, or a callable (see dataclass field default_factory)

  • name – command line argument name

  • stream_suffix – flag to add stream specifier suffix to parameter name (like ‘-b:v’ or ‘-c:a’)

  • init – add parameter to dataclass constructor.

  • skip – skip parameter while generating command line args list.

  • render – a callable used to format output value.

Returns:

dataclass field definition with extra metadata.

>>> from fffw.encoding.filters import VideoFilter
>>> @dataclass
... class Deinterlace(VideoFilter):
...     filter = 'yadif'
...     mode: int = param(default=0)
...
>>>