diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 03a6e5e..576df90 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,6 +22,7 @@ jobs: - run: poetry run flake8 . - run: poetry run black --check --diff . - run: poetry run isort --check --diff . + - run: poetry run mypy . test: runs-on: ubuntu-latest diff --git a/docs/conf.py b/docs/conf.py index 5d88663..bbd927c 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -4,6 +4,7 @@ import datetime import os import sys +from typing import Dict sys.path.insert(0, os.path.abspath("..")) # noqa import ffmpy # noqa @@ -37,7 +38,7 @@ htmlhelp_basename = "ffmpydoc" -latex_elements = {} +latex_elements: Dict = {} latex_documents = [ (master_doc, "ffmpy.tex", "ffmpy Documentation", "Andrii Yurchuk", "manual"), ] diff --git a/docs/examples.rst b/docs/examples.rst index 6a827f1..28bbbfc 100644 --- a/docs/examples.rst +++ b/docs/examples.rst @@ -99,6 +99,10 @@ To multiplex video and audio back into an MPEG transport stream with re-encoding 'ffmpeg -i audio.mp4 -i video.mp4 -c:v h264 -c:a ac3 output.ts' >>> ff.run() +.. note:: + + Since Python 3.7 dictionaries preserve order. Using OrderedDict is no longer necessary. + There are cases where the order of inputs and outputs must be preserved (e.g. when using FFmpeg `-map `_ option). In these cases the use of regular Python dictionary will not work because it does not preserve order. Instead, use `OrderedDict `_. For example we want to multiplex one video and two audio streams into an MPEG transport streams re-encoding both audio streams using different codecs. Here we use an OrderedDict to preserve the order of inputs so they match the order of streams in output options: .. code:: python diff --git a/ffmpy.py b/ffmpy.py index 81243ae..c6ce8a3 100644 --- a/ffmpy.py +++ b/ffmpy.py @@ -1,6 +1,8 @@ import errno +import itertools import shlex import subprocess +from typing import IO, Any, List, Mapping, Optional, Sequence, Tuple, Union __version__ = "0.3.3" @@ -10,7 +12,13 @@ class FFmpeg(object): ffprobe). """ - def __init__(self, executable="ffmpeg", global_options=None, inputs=None, outputs=None): + def __init__( + self, + executable: str = "ffmpeg", + global_options: Optional[Union[Sequence[str], str]] = None, + inputs: Optional[Mapping[str, Optional[Union[Sequence[str], str]]]] = None, + outputs: Optional[Mapping[str, Optional[Union[Sequence[str], str]]]] = None, + ) -> None: """Initialize FFmpeg command line wrapper. Compiles FFmpeg command line from passed arguments (executable path, options, inputs and @@ -39,26 +47,28 @@ def __init__(self, executable="ffmpeg", global_options=None, inputs=None, output """ self.executable = executable self._cmd = [executable] + self._cmd += _normalize_options(global_options, split_mixed=True) - global_options = global_options or [] - if _is_sequence(global_options): - normalized_global_options = [] - for opt in global_options: - normalized_global_options += shlex.split(opt) - else: - normalized_global_options = shlex.split(global_options) + if inputs is not None: + self._cmd += _merge_args_opts(inputs, add_minus_i_option=True) - self._cmd += normalized_global_options - self._cmd += _merge_args_opts(inputs, add_input_option=True) - self._cmd += _merge_args_opts(outputs) + if outputs is not None: + self._cmd += _merge_args_opts(outputs) self.cmd = subprocess.list2cmdline(self._cmd) - self.process = None + self.process: Optional[subprocess.Popen] = None - def __repr__(self): + def __repr__(self) -> str: return "<{0!r} {1!r}>".format(self.__class__.__name__, self.cmd) - def run(self, input_data=None, stdout=None, stderr=None, env=None, **kwargs): + def run( + self, + input_data: Optional[bytes] = None, + stdout: Optional[Union[IO, int]] = None, + stderr: Optional[Union[IO, int]] = None, + env: Optional[Mapping[str, str]] = None, + **kwargs: Any + ) -> Tuple[Optional[bytes], Optional[bytes]]: """Execute FFmpeg command line. ``input_data`` can contain input for FFmpeg in case ``pipe`` protocol is used for input. @@ -99,17 +109,22 @@ def run(self, input_data=None, stdout=None, stderr=None, env=None, **kwargs): else: raise - out = self.process.communicate(input=input_data) + o_stdout, o_stderr = self.process.communicate(input=input_data) if self.process.returncode != 0: - raise FFRuntimeError(self.cmd, self.process.returncode, out[0], out[1]) + raise FFRuntimeError(self.cmd, self.process.returncode, o_stdout, o_stderr) - return out + return o_stdout, o_stderr class FFprobe(FFmpeg): """Wrapper for `ffprobe `_.""" - def __init__(self, executable="ffprobe", global_options="", inputs=None): + def __init__( + self, + executable: str = "ffprobe", + global_options: Optional[Union[Sequence[str], str]] = None, + inputs: Optional[Mapping[str, Optional[Union[Sequence[str], str]]]] = None, + ) -> None: """Create an instance of FFprobe. Compiles FFprobe command line from passed arguments (executable path, options, inputs). @@ -139,7 +154,7 @@ class FFRuntimeError(Exception): ``cmd``, ``exit_code``, ``stdout``, ``stderr``. """ - def __init__(self, cmd, exit_code, stdout, stderr): + def __init__(self, cmd: str, exit_code: int, stdout: bytes, stderr: bytes) -> None: self.cmd = cmd self.exit_code = exit_code self.stdout = stdout @@ -152,17 +167,10 @@ def __init__(self, cmd, exit_code, stdout, stderr): super(FFRuntimeError, self).__init__(message) -def _is_sequence(obj): - """Check if the object is a sequence (list, tuple etc.). - - :param object obj: an object to be checked - :return: True if the object is iterable but is not a string, False otherwise - :rtype: bool - """ - return hasattr(obj, "__iter__") and not isinstance(obj, str) - - -def _merge_args_opts(args_opts_dict, **kwargs): +def _merge_args_opts( + args_opts_dict: Mapping[str, Optional[Union[Sequence[str], str]]], + add_minus_i_option: bool = False, +) -> List[str]: """Merge options with their corresponding arguments. Iterates over the dictionary holding arguments (keys) and options (values). Merges each @@ -173,22 +181,32 @@ def _merge_args_opts(args_opts_dict, **kwargs): :return: merged list of strings with arguments and their corresponding options :rtype: list """ - merged = [] - - if not args_opts_dict: - return merged + merged: List[str] = [] for arg, opt in args_opts_dict.items(): - if not _is_sequence(opt): - opt = shlex.split(opt or "") - merged += opt + merged += _normalize_options(opt) if not arg: continue - if "add_input_option" in kwargs: + if add_minus_i_option: merged.append("-i") merged.append(arg) return merged + + +def _normalize_options( + options: Optional[Union[Sequence[str], str]], split_mixed: bool = False +) -> List[str]: + """Normalize options string or list of strings.""" + if options is None: + return [] + elif isinstance(options, str): + return shlex.split(options) + else: + if split_mixed: + return list(itertools.chain(*[shlex.split(o) for o in options])) + else: + return list(options) diff --git a/py.typed b/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml index c7f09a5..c78fb3d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,7 @@ classifiers = [ "Operating System :: POSIX :: Linux", "Operating System :: MacOS :: MacOS X", ] -packages = [{ include = "ffmpy.py" }] +packages = [{ include = "ffmpy.py" }, { include = "py.typed" }] [tool.poetry.dependencies] python = "^3.8.1" # flake8 requires Python 3.8.1 or higher diff --git a/tests/test_cmd_compilation.py b/tests/test_cmd_compilation.py index 63b35fb..a1c5939 100644 --- a/tests/test_cmd_compilation.py +++ b/tests/test_cmd_compilation.py @@ -1,4 +1,4 @@ -from collections import OrderedDict +from typing import List import pytest @@ -14,12 +14,18 @@ ["-hide_banner -y", "-v debug"], ], ) -def test_global_options(global_options): +def test_global_options(global_options: List) -> None: ff = FFmpeg(global_options=global_options) assert ff._cmd == ["ffmpeg", "-hide_banner", "-y", "-v", "debug"] assert ff.cmd == "ffmpeg -hide_banner -y -v debug" +def test_global_options_none() -> None: + ff = FFmpeg(global_options=None) + assert ff._cmd == ["ffmpeg"] + assert ff.cmd == "ffmpeg" + + @pytest.mark.parametrize( "input_options", [ @@ -28,7 +34,7 @@ def test_global_options(global_options): ("-f", "rawvideo", "-pix_fmt", "rgb24", "-s:v", "640x480"), ], ) -def test_input_options(input_options): +def test_input_options(input_options: List) -> None: ff = FFmpeg(inputs={"/tmp/rawvideo": input_options}) assert ff._cmd == [ "ffmpeg", @@ -52,7 +58,7 @@ def test_input_options(input_options): ("-f", "rawvideo", "-pix_fmt", "rgb24", "-s:v", "640x480"), ], ) -def test_output_options(output_options): +def test_output_options(output_options: List) -> None: ff = FFmpeg(outputs={"/tmp/rawvideo": output_options}) assert ff._cmd == [ "ffmpeg", @@ -67,10 +73,34 @@ def test_output_options(output_options): assert ff.cmd == "ffmpeg -f rawvideo -pix_fmt rgb24 -s:v 640x480 /tmp/rawvideo" -def test_input_output_none(): - inputs = OrderedDict(((None, ["-f", "rawvideo"]), ("/tmp/video.mp4", ["-f", "mp4"]))) - outputs = OrderedDict((("/tmp/rawvideo", ["-f", "rawvideo"]), (None, ["-f", "mp4"]))) - ff = FFmpeg(inputs=inputs, outputs=outputs) +# This kind of usage would be invalid, but it's tested to ensure the correct behavior +def test_input_output_options_split_mixed() -> None: + ff = FFmpeg( + inputs={"/tmp/rawvideo": ["-f rawvideo", "-pix_fmt rgb24", "-s:v 640x480"]}, + outputs={"/tmp/rawvideo": ["-f rawvideo", "-pix_fmt rgb24", "-s:v 640x480"]}, + ) + assert ff._cmd == [ + "ffmpeg", + "-f rawvideo", + "-pix_fmt rgb24", + "-s:v 640x480", + "-i", + "/tmp/rawvideo", + "-f rawvideo", + "-pix_fmt rgb24", + "-s:v 640x480", + "/tmp/rawvideo", + ] + assert ff.cmd == ( + 'ffmpeg "-f rawvideo" "-pix_fmt rgb24" "-s:v 640x480" -i /tmp/rawvideo ' + '"-f rawvideo" "-pix_fmt rgb24" "-s:v 640x480" /tmp/rawvideo' + ) + + +def test_input_output_none() -> None: + inputs = {None: ["-f", "rawvideo"], "/tmp/video.mp4": ["-f", "mp4"]} + outputs = {"/tmp/rawvideo": ["-f", "rawvideo"], None: ["-f", "mp4"]} + ff = FFmpeg(inputs=inputs, outputs=outputs) # type: ignore[arg-type] assert ff._cmd == [ "ffmpeg", "-f", @@ -88,9 +118,9 @@ def test_input_output_none(): assert ff.cmd == ("ffmpeg -f rawvideo -f mp4 -i /tmp/video.mp4 -f rawvideo /tmp/rawvideo -f mp4") -def test_input_options_output_options_none(): - inputs = OrderedDict((("/tmp/rawvideo", None), ("/tmp/video.mp4", ["-f", "mp4"]))) - outputs = OrderedDict((("/tmp/rawvideo", ["-f", "rawvideo"]), ("/tmp/video.mp4", None))) +def test_input_options_output_options_none() -> None: + inputs = {"/tmp/rawvideo": None, "/tmp/video.mp4": ["-f", "mp4"]} + outputs = {"/tmp/rawvideo": ["-f", "rawvideo"], "/tmp/video.mp4": None} ff = FFmpeg(inputs=inputs, outputs=outputs) assert ff._cmd == [ "ffmpeg", @@ -110,7 +140,7 @@ def test_input_options_output_options_none(): ) -def test_quoted_option(): +def test_quoted_option() -> None: inputs = {"input.ts": None} quoted_option = ( "drawtext=" @@ -135,7 +165,7 @@ def test_quoted_option(): assert ff.cmd == 'ffmpeg -i input.ts -vf "{0}" -an output.ts'.format(quoted_option) -def test_path_with_spaces(): +def test_path_with_spaces() -> None: inputs = {"/home/ay/file with spaces": None} outputs = {"output.ts": None} @@ -144,14 +174,14 @@ def test_path_with_spaces(): assert ff.cmd == 'ffmpeg -i "/home/ay/file with spaces" output.ts' -def test_repr(): +def test_repr() -> None: inputs = {"input.ts": "-f rawvideo"} outputs = {"output.ts": "-f rawvideo"} ff = FFmpeg(inputs=inputs, outputs=outputs) assert repr(ff) == "<'FFmpeg' 'ffmpeg -f rawvideo -i input.ts -f rawvideo output.ts'>" -def test_ffprobe(): +def test_ffprobe() -> None: inputs = {"input.ts": "-f rawvideo"} ff = FFprobe(global_options="--verbose", inputs=inputs) assert repr(ff) == "<'FFprobe' 'ffprobe --verbose -f rawvideo -i input.ts'>" diff --git a/tests/test_cmd_execution.py b/tests/test_cmd_execution.py index e52c977..80170c5 100644 --- a/tests/test_cmd_execution.py +++ b/tests/test_cmd_execution.py @@ -12,14 +12,14 @@ os.environ["PATH"] = FFMPEG_PATH + os.pathsep + os.environ["PATH"] -def test_invalid_executable_path(): +def test_invalid_executable_path() -> None: ff = FFmpeg(executable="/tmp/foo/bar/ffmpeg") with pytest.raises(FFExecutableNotFoundError) as exc_info: ff.run() assert str(exc_info.value) == "Executable '/tmp/foo/bar/ffmpeg' not found" -def test_other_oserror(): +def test_other_oserror() -> None: executable = os.path.join(FFMPEG_PATH, "ffmpeg.go") ff = FFmpeg(executable=executable) with pytest.raises(PermissionError) as exc_info: @@ -27,14 +27,14 @@ def test_other_oserror(): assert str(exc_info.value).startswith("[Errno 13] Permission denied") -def test_executable_full_path(): +def test_executable_full_path() -> None: executable = os.path.join(FFMPEG_PATH, "ffmpeg") ff = FFmpeg(executable=executable) ff.run() assert ff.cmd == executable -def test_no_redirection(): +def test_no_redirection() -> None: global_options = "--stdin none --stdout oneline --stderr multiline --exit-code 0" ff = FFmpeg(global_options=global_options) stdout, stderr = ff.run() @@ -42,7 +42,7 @@ def test_no_redirection(): assert stderr is None -def test_redirect_to_devnull(): +def test_redirect_to_devnull() -> None: global_options = "--stdin none --stdout oneline --stderr multiline --exit-code 0" ff = FFmpeg(global_options=global_options) devnull = open(os.devnull, "wb") @@ -51,7 +51,7 @@ def test_redirect_to_devnull(): assert stderr is None -def test_redirect_to_pipe(): +def test_redirect_to_pipe() -> None: global_options = "--stdin none --stdout oneline --stderr multiline --exit-code 0" ff = FFmpeg(global_options=global_options) stdout, stderr = ff.run(stdout=subprocess.PIPE, stderr=subprocess.PIPE) @@ -59,7 +59,7 @@ def test_redirect_to_pipe(): assert stderr == b"These are\nmultiple lines\nprinted to stderr" -def test_input(): +def test_input() -> None: global_options = "--stdin pipe --stdout oneline --stderr multiline --exit-code 0" ff = FFmpeg(global_options=global_options) stdout, stderr = ff.run( @@ -69,7 +69,7 @@ def test_input(): assert stderr == b"These are\nmultiple lines\nprinted to stderr" -def test_non_zero_exitcode(): +def test_non_zero_exitcode() -> None: global_options = "--stdin none --stdout multiline --stderr multiline --exit-code 42" ff = FFmpeg(global_options=global_options) with pytest.raises(FFRuntimeError) as exc_info: @@ -95,7 +95,7 @@ def test_non_zero_exitcode(): ) -def test_non_zero_exitcode_no_stderr(): +def test_non_zero_exitcode_no_stderr() -> None: global_options = "--stdin none --stdout multiline --stderr none --exit-code 42" ff = FFmpeg(global_options=global_options) with pytest.raises(FFRuntimeError) as exc_info: @@ -118,7 +118,7 @@ def test_non_zero_exitcode_no_stderr(): ) -def test_non_zero_exitcode_no_stdout(): +def test_non_zero_exitcode_no_stdout() -> None: global_options = "--stdin none --stdout none --stderr multiline --exit-code 42" ff = FFmpeg(global_options=global_options) with pytest.raises(FFRuntimeError) as exc_info: @@ -142,7 +142,7 @@ def test_non_zero_exitcode_no_stdout(): ) -def test_non_zero_exitcode_no_stdout_and_stderr(): +def test_non_zero_exitcode_no_stdout_and_stderr() -> None: global_options = "--stdin none --stdout none --stderr none --exit-code 42" ff = FFmpeg(global_options=global_options) with pytest.raises(FFRuntimeError) as exc_info: @@ -161,7 +161,7 @@ def test_non_zero_exitcode_no_stdout_and_stderr(): ) -def test_raise_exception_with_stdout_stderr_none(): +def test_raise_exception_with_stdout_stderr_none() -> None: global_options = "--stdin none --stdout none --stderr none --exit-code 42" ff = FFmpeg(global_options=global_options) with pytest.raises(FFRuntimeError) as exc_info: @@ -176,7 +176,7 @@ def test_raise_exception_with_stdout_stderr_none(): ) -def test_terminate_process(): +def test_terminate_process() -> None: global_options = "--long-run" ff = FFmpeg(global_options=global_options) @@ -201,18 +201,18 @@ def test_terminate_process(): @mock.patch("ffmpy.subprocess.Popen") -def test_custom_env(popen_mock): +def test_custom_env(popen_mock: mock.MagicMock) -> None: ff = FFmpeg() popen_mock.return_value.communicate.return_value = ("output", "error") popen_mock.return_value.returncode = 0 - ff.run(env="customenv") + ff.run(env={"FOO": "BAR"}) popen_mock.assert_called_with( - mock.ANY, stdin=mock.ANY, stdout=mock.ANY, stderr=mock.ANY, env="customenv" + mock.ANY, stdin=mock.ANY, stdout=mock.ANY, stderr=mock.ANY, env={"FOO": "BAR"} ) @mock.patch("ffmpy.subprocess.Popen") -def test_arbitraty_popen_kwargs(popen_mock): +def test_arbitraty_popen_kwargs(popen_mock: mock.MagicMock) -> None: ff = FFmpeg() popen_mock.return_value.communicate.return_value = ("output", "error") popen_mock.return_value.returncode = 0