Add type hints and some refactoring (#79)

.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

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"),
 ]

docs/examples.rst

@@ -99,6 +99,11 @@ 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
+   <https://docs.python.org/3/library/collections.html#collections.OrderedDict>`_ is no longer necessary.
+
 There are cases where the order of inputs and outputs must be preserved (e.g. when using FFmpeg `-map <https://trac.ffmpeg.org/wiki/How%20to%20use%20-map%20option>`_ option). In these cases the use of regular Python dictionary will not work because it does not preserve order. Instead, use `OrderedDict <https://docs.python.org/3/library/collections.html#collections.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

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 <https://www.ffmpeg.org/ffprobe.html>`_."""
 
-    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)

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

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'>"