py35 compat ffprobe

py35 ffplay

black, py35 throughout

vers

Author: scivision

.travis.yml

@@ -9,6 +9,8 @@ git:
 python:
 - 3.8-dev
 - 3.7
+- 3.6
+- 3.5
 
 addons:
   apt:
@@ -18,12 +20,17 @@ addons:
 matrix:
   include:
   - os: linux
+    name: PEP8 MyPy
     python: 3.7
-    install: pip install -e .[tests]
+    install: pip install -e .[tests,cov]
     script:
     - flake8
     - mypy .
+    - pytest -r a -v
+    after_success:
+    - pytest --cov
+    - coveralls
 
 install: pip install -e .[tests]
 
-script: pytest -v
+script: pytest -r a -v

asyncioffmpeg/ffplay.py

@@ -3,16 +3,18 @@
 This example uses a finite number of workers, rather than slamming the system with endless subprocesses.
 This is more effective than endless context switching for an overloaded CPU.
 """
+import queue
+import subprocess
 import asyncio
 from pathlib import Path
 import shutil
 import sys
 from typing import Iterable
 import os
 
-FFPLAY = shutil.which('ffplay')
+FFPLAY = shutil.which("ffplay")
 if not FFPLAY:
-    raise FileNotFoundError('FFPLAY not found')
+    raise ImportError("FFPLAY not found")
 
 
 async def ffplay(queue: asyncio.Queue):
@@ -25,14 +27,14 @@ async def ffplay(queue: asyncio.Queue):
     while True:
         filein = await queue.get()
 
-        cmd = [FFPLAY, '-loglevel', 'warning', '-autoexit', str(filein)]
+        cmd = [FFPLAY, "-loglevel", "warning", "-autoexit", str(filein)]
 
         proc = await asyncio.create_subprocess_exec(*cmd)
 
         ret = await proc.wait()
 
         if ret != 0:
-            print(filein, 'playback failure', file=sys.stderr)
+            print(filein, "playback failure", file=sys.stderr)
 
         queue.task_done()
 
@@ -42,18 +44,39 @@ async def main(flist: Iterable[Path]):
     Ntask = os.cpu_count()  # includes logical cores
     if not isinstance(Ntask, int):
         Ntask = 2
-# %% setup queue
+    # %% setup queue
     queue = asyncio.Queue()  # type: ignore
 
     for f in flist:
         await queue.put(f)
-# %% setup Tasks
-    tasks = [asyncio.create_task(ffplay(queue)) for i in range(Ntask)]
+    # %% setup Tasks
+    if sys.version_info >= (3, 7):
+        tasks = [asyncio.create_task(ffplay(queue)) for i in range(Ntask)]
+    else:
+        tasks = [asyncio.ensure_future(ffplay(queue)) for i in range(Ntask)]
 
     await queue.join()
 
-# %% program done, teardown Tasks
+    # %% program done, teardown Tasks
     for task in tasks:
         task.cancel()
 
     await asyncio.gather(*tasks, return_exceptions=True)
+
+
+def ffplay_sync(qin: queue.Queue):
+    """
+    Play media synchronously
+    """
+
+    while not qin.empty():
+        filein = qin.get(timeout=1.0)
+
+        cmd = [FFPLAY, "-v", "warning", "-autoexit", str(filein)]
+
+        ret = subprocess.run(cmd)
+
+        if ret.returncode != 0:
+            print(filein, "playback failure", cmd, file=sys.stderr)
+
+        qin.task_done()

asyncioffmpeg/ffplay_sync.py

@@ -6,50 +6,88 @@
 import asyncio
 import json
 import subprocess
-from typing import Dict, Iterable, AsyncGenerator
+import typing
 from pathlib import Path
 import shutil
 
-FFPROBE = shutil.which('ffprobe')
+FFPROBE = shutil.which("ffprobe")
 if not FFPROBE:
-    raise FileNotFoundError('FFPROBE not found')
+    raise ImportError("FFPROBE not found")
 
 
-async def get_meta(files: Iterable[Path]) -> AsyncGenerator[dict, None]:  # mypy bug Dict[str, str]
+def files2futures(path: Path, suffix: str) -> list:
+    path = Path(path).expanduser()
+    if not path.is_dir():
+        raise NotADirectoryError(path)
+    return [ffprobe(f) for f in path.iterdir() if f.is_file() and f.suffix in suffix]
 
-    futures = [ffprobe(file) for file in files]
-    for future in asyncio.as_completed(futures):
-        try:
-            meta = await future
-        except asyncio.TimeoutError:
-            continue
 
-        yield meta
+def print_meta(meta: typing.Dict[str, typing.Any]):
+    fn = Path(meta["format"]["filename"])
+    dur = float(meta["streams"][0]["duration"])
+    print("{:>40}  {:>5.1f}".format(fn.name, dur))
 
 
-async def ffprobe(file: Path) -> Dict[str, str]:
-    """ get media metadata """
+async def get_meta_gather(
+    path: Path, suffix: str
+) -> typing.List[typing.Dict[str, typing.Any]]:
+    """ for comparison with asyncio.as_completed"""
+    futures = files2futures(path, suffix)
+    metas = await asyncio.gather(*futures)
+    for meta in metas:
+        print_meta(meta)
 
-    proc = await asyncio.create_subprocess_exec(*[FFPROBE, '-loglevel', 'warning',
-                                                  '-print_format', 'json',
-                                                  '-show_streams',
-                                                  '-show_format',
-                                                  str(file)],
-                                                stdout=asyncio.subprocess.PIPE)
+    return metas
 
-    stdout, _ = await proc.communicate()
 
-    return json.loads(stdout.decode('utf8'))
+async def get_meta(
+    path: Path, suffix: str
+) -> typing.List[typing.Dict[str, typing.Any]]:
+    futures = files2futures(path, suffix)
+    metas = []
+    for file in asyncio.as_completed(futures):
+        meta = await file
+        print_meta(meta)
+        metas.append(meta)
 
+    return metas
 
-def ffprobe_sync(file: Path) -> dict:  # mypy bug Dict[str, str]
+
+async def ffprobe(file: Path) -> typing.Dict[str, typing.Any]:
     """ get media metadata """
+    proc = await asyncio.create_subprocess_exec(
+        *[
+            FFPROBE,
+            "-loglevel",
+            "warning",
+            "-print_format",
+            "json",
+            "-show_streams",
+            "-show_format",
+            str(file),
+        ],
+        stdout=asyncio.subprocess.PIPE
+    )
+
+    stdout, _ = await proc.communicate()
+
+    return json.loads(stdout.decode("utf8"))
 
-    meta = subprocess.check_output([FFPROBE, '-v', 'warning',
-                                    '-print_format', 'json',
-                                    '-show_streams',
-                                    '-show_format',
-                                    str(file)],
-                                   text=True)
+
+def ffprobe_sync(file: Path) -> typing.Dict[str, typing.Any]:
+    """ get media metadata """
+    meta = subprocess.check_output(
+        [
+            FFPROBE,
+            "-v",
+            "warning",
+            "-print_format",
+            "json",
+            "-show_streams",
+            "-show_format",
+            str(file),
+        ],
+        universal_newlines=True,
+    )
 
     return json.loads(meta)

asyncioffmpeg/ffprobe.py

@@ -0,0 +1,26 @@
+import os
+import sys
+import asyncio
+
+
+def runner(fun, *args):
+    """
+    Generic asyncio.run() equivalent for Python >= 3.5
+    """
+    if os.name == "nt" and (3, 7) <= sys.version_info < (3, 8):
+        asyncio.set_event_loop_policy(
+            asyncio.WindowsProactorEventLoopPolicy()  # type: ignore
+        )
+
+    if sys.version_info >= (3, 7):
+        result = asyncio.run(fun(*args))
+    else:
+        if os.name == "nt":
+            loop = asyncio.ProactorEventLoop()
+        else:
+            loop = asyncio.new_event_loop()
+            asyncio.get_child_watcher().attach_loop(loop)
+        result = loop.run_until_complete(fun(*args))
+        loop.close()
+
+    return result

asyncioffmpeg/runner.py

@@ -33,21 +33,26 @@
 import asyncio
 import sys
 from argparse import ArgumentParser
-
-if sys.version_info < (3, 6):
-    raise RuntimeError('Python >= 3.6 required')
+from asyncioffmpeg.runner import runner
 
 
 async def coro_worker(i: int, Niter: int, tic: float):
     """coroutine worker"""
     for _ in range(Niter):
         math.sin(3)
 
-    print(f'Coroutine worker {i} done at {time.time()-tic:.2f} sec.')
+    print("Coroutine worker {} done at {:.2f} sec.".format(i, time.monotonic() - tic))
 
 
 async def coro(Nworker: int, Niter: int, tic: float):
-    tasks = [asyncio.create_task(coro_worker(i, Niter, tic)) for i in range(Nworker)]
+    if sys.version_info >= (3, 7):
+        tasks = [
+            asyncio.create_task(coro_worker(i, Niter, tic)) for i in range(Nworker)
+        ]
+    else:
+        tasks = [
+            asyncio.ensure_future(coro_worker(i, Niter, tic)) for i in range(Nworker)
+        ]
     await asyncio.wait(tasks)
 
 
@@ -60,46 +65,59 @@ def __init__(self, i: int, Niter: int):
         self.i = i
 
     def run(self):
-        tic = time.time()
+        tic = time.monotonic()
         for _ in range(self.Niter):
             math.sin(3)
 
-        print(f'Thread worker {self.i} done at {time.time()-tic:.2f} sec.')
+        print(
+            "Thread worker {} done at {:.2f} sec.".format(
+                self.i, time.monotonic() - tic
+            )
+        )
 
 
 def mp_worker(i: int, Niter: int, tic: float):
     """ multiprocessing worker"""
     for _ in range(Niter):
         math.sin(3)
 
-    print(f'Process worker {i} done at {time.time()-tic:.2f} sec.')
-
-
-if __name__ == '__main__':
-    P = ArgumentParser(description='Demonstrate differences between coroutines, threads and proceses.')
-    P.add_argument('method', help='c: coroutine, t: threading, p: multiprocessing')
-    P.add_argument('-Nworker', help='number of workers', type=int, default=4)
-    P.add_argument('-Niter', help='number of loop iterations (arbitrary)', type=int, default=5000000)
+    print("Process worker {} done at {:.2f} sec.".format(i, time.monotonic() - tic))
+
+
+if __name__ == "__main__":
+    P = ArgumentParser(
+        description="Demonstrate differences between coroutines, threads and proceses."
+    )
+    P.add_argument("method", help="c: coroutine, t: threading, p: multiprocessing")
+    P.add_argument("-Nworker", help="number of workers", type=int, default=4)
+    P.add_argument(
+        "-Niter",
+        help="number of loop iterations (arbitrary)",
+        type=int,
+        default=5000000,
+    )
     A = P.parse_args()
 
-    if A.method not in ('c', 't', 'p'):
-        raise ValueError('Method must be one of: c t p')
+    if A.method not in ("c", "t", "p"):
+        raise ValueError("Method must be one of: c t p")
 
-    tic = time.time()
+    tic = time.monotonic()
     for i in range(A.Nworker):
-        if A.method == 't':
-            p = Thread_worker(i, A.Niter)
+        if A.method == "t":
+            p = Thread_worker(i, A.Niter)  # type: ignore
             p.start()
-        elif A.method == 'p':
-            p = multiprocessing.Process(target=mp_worker, args=(i, A.Niter, tic))  # type: ignore
+        elif A.method == "p":
+            p = multiprocessing.Process(  # type: ignore
+                target=mp_worker, args=(i, A.Niter, tic)  # type: ignore
+            )
             p.start()
-            print(f'started process workert {i}, PID: {p.pid}')  # type: ignore
+            print(
+                "started process workert {}, PID: {}".format(i, p.pid)  # type: ignore
+            )  # type: ignore
 
-    if A.method == 'c':
-        if sys.version_info < (3, 7):
-            raise RuntimeError('Python >= 3.7 required for this example')
-        asyncio.run(coro(A.Nworker, A.Niter, tic))
+    if A.method == "c":
+        runner(coro, A.Nworker, A.Niter, tic)
     else:
         p.join()
 
-    print(f'{time.time()-tic:.2f} sec.')
+    print("{:.2f} sec.".format(time.monotonic() - tic))