asyncgenerator

Author: scivision

README.md

@@ -9,15 +9,38 @@ In general to show the latest techniques, Python ≥ 3.7 is required.
 
 ## FFprobe
 
-* `probe_sync`: Use FFprobe to retrieve file metadata synchronously
-* `probe_coroutine`: Use FFprobe to retrieve file metadata asynchronously using Python `asyncio` coroutine event loop.
+Both synchronous (traditional for loop) and asynchronous pipeline are demonstrated.
+They call FFprobe executable to return JSON formatted metadata.
+
+### probe_sync.py
+
+retrieve file metadata synchronously.
+
+### probe_coroutine.py
+
+retrieve file metadata in an asynchronous pipeline (asyncio generator) using Python `asyncio` coroutine event loop.
 
 ## FFplay
 
 I like to test asynchronous techniques with video playback, as it makes some effects obvious.
+The FFplay asyncio example is more advanced than the FFprobe example.
+In the FFprobe example, the lazy asyncio generator produces metadata concurrently as fast as it's requested.
+There is no resource throttling in the FFprobe example, so the CPU could become overwhelmed with context switching.
+
+The FFplay example in contrast is an example of a task using resource throttling via asyncio.Queue.
+The queueing could also be implemented for FFprobe style task if desired.
+However, the rationale employed is that the FFprobe task is overall lightweight, and thus other parts of the pipeline inherently limit resource utilization.
+If the FFprobe task was in an asyncio.gather() algorithm, resource utilization could get too high.
+Thus we have a "win-win" by using asyncio generator for FFprobe--the throttling comes implicitly from other parts of the pipeline.
+
+
+### play_threadpool.py
+
+Even though coroutines are more efficient in many applications, the syntax of `concurrent.futures.ThreadPoolExecutor` is perhaps the simplest possible way to spawn independent processes in a controlled fashion
+
+### play_coroutine.py
 
-* `play_threadpool`: Even though coroutines are more efficient, the syntax of `concurrent.futures.ThreadPoolExecutor` is perhaps the simplest possible way to spawn independent processes in a controlled fashion
-* `play_coroutine`: Use Python `asyncio` coroutine event loop to spawn processes.
+Use Python `asyncio` coroutine event loop to spawn processes.
 
 ### Fortran
 

asyncioffmpeg/ffplay.py

@@ -7,13 +7,9 @@
 from pathlib import Path
 import shutil
 import sys
-from typing import List, Generator, Union
+from typing import Iterable
 import os
 
-if sys.version_info < (3, 7):
-    raise RuntimeError('Python >= 3.7 required')
-
-
 FFPLAY = shutil.which('ffplay')
 if not FFPLAY:
     raise FileNotFoundError('FFPLAY not found')
@@ -28,9 +24,8 @@ async def ffplay(queue: asyncio.Queue):
 
     while True:
         filein = await queue.get()
-        assert isinstance(filein, Path)
 
-        cmd = [FFPLAY, '-v', 'warning', '-autoexit', str(filein)]
+        cmd = [FFPLAY, '-loglevel', 'warning', '-autoexit', str(filein)]
 
         proc = await asyncio.create_subprocess_exec(*cmd)
 
@@ -42,7 +37,7 @@ async def ffplay(queue: asyncio.Queue):
         queue.task_done()
 
 
-async def main(flist: Union[List[Path], Generator[Path, None, None]]):
+async def main(flist: Iterable[Path]):
 
     Ntask = os.cpu_count()  # includes logical cores
     if not isinstance(Ntask, int):

asyncioffmpeg/ffprobe.py

@@ -1,46 +1,55 @@
 #!/usr/bin/env python
 """
-This file contains connections from Python to FFprobe, useful for extracting
+use Python with FFprobe to extract
 JSON metadata from any kind of media file that FFprobe can read.
-
-Moreover, this file demonstrates asynchronous and synchronous access to
-thread-safe subprocesses
 """
 import asyncio
 import json
+import subprocess
+from typing import Dict, Iterable, AsyncGenerator
 from pathlib import Path
 import shutil
-import sys
 
 FFPROBE = shutil.which('ffprobe')
 if not FFPROBE:
     raise FileNotFoundError('FFPROBE not found')
 
-TIMEOUT = 5.0  # 2.0 is too short for Windows
+
+async def get_meta(files: Iterable[Path]) -> AsyncGenerator[dict, None]:  # mypy bug Dict[str, str]
+
+    futures = [ffprobe(file) for file in files]
+    for future in asyncio.as_completed(futures):
+        try:
+            meta = await future
+        except asyncio.TimeoutError:
+            continue
+
+        yield meta
 
 
-# %% Asynchronous FFprobe
-async def ffprobe(filein: Path) -> dict:
+async def ffprobe(file: Path) -> Dict[str, str]:
     """ get media metadata """
-    assert isinstance(FFPROBE, str)
 
-    proc = await asyncio.create_subprocess_exec(*[FFPROBE, '-v', 'warning',
+    proc = await asyncio.create_subprocess_exec(*[FFPROBE, '-loglevel', 'warning',
                                                   '-print_format', 'json',
                                                   '-show_streams',
                                                   '-show_format',
-                                                  str(filein)],
-                                                stdout=asyncio.subprocess.PIPE,
-                                                stderr=asyncio.subprocess.PIPE)
+                                                  str(file)],
+                                                stdout=asyncio.subprocess.PIPE)
 
-    stdout, stderr = await proc.communicate()
+    stdout, _ = await proc.communicate()
 
     return json.loads(stdout.decode('utf8'))
 
 
-async def get_meta(filein: Path) -> dict:
-    try:
-        meta = await asyncio.wait_for(ffprobe(filein), timeout=TIMEOUT)
-        return meta
-    except asyncio.TimeoutError:
-        print('timeout ', filein, file=sys.stderr)
-        return {}
+def ffprobe_sync(file: Path) -> dict:  # mypy bug Dict[str, str]
+    """ get media metadata """
+
+    meta = subprocess.check_output([FFPROBE, '-v', 'warning',
+                                    '-print_format', 'json',
+                                    '-show_streams',
+                                    '-show_format',
+                                    str(file)],
+                                   text=True)
+
+    return json.loads(meta)

asyncioffmpeg/ffprobe_sync.py

@@ -1,4 +1,6 @@
 [mypy]
 ignore_missing_imports = True
 strict_optional = False
-allow_redefinition = True
+allow_redefinition = True
+show_error_context = False
+show_column_numbers = True

mypy.ini

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 import asyncio
 import os
+import sys
 from pathlib import Path
 from argparse import ArgumentParser
 
@@ -12,15 +13,16 @@
         description="Plays media files asynchronously with FFplay")
     p.add_argument('path', help='directory where media files are kept')
     p.add_argument('-suffix', help='file suffixes of desired media file types',
-                   nargs='+', default=['.mp4', '.avi', '.ogv', '.wmv', '.flv', '.mov'])
+                   nargs='+', default=['.mp4', '.avi', '.ogv', '.wmv', '.flv', '.mov', '.ogg'])
     P = p.parse_args()
 
     path = Path(P.path).expanduser()
+    if not path.is_dir():
+        raise NotADirectoryError(path)
 
     flist = (f for f in path.iterdir() if f.is_file() and f.suffix in P.suffix)
 
-    if os.name == 'nt':
-        loop = asyncio.ProactorEventLoop()  # type: ignore
-        loop.run_until_complete(play.main(flist))
-    else:
-        asyncio.run(play.main(flist))
+    if os.name == 'nt' and sys.version_info < (3, 8):
+        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())  # type: ignore
+
+    asyncio.run(play.main(flist))

play_coroutine.py

@@ -1,24 +1,23 @@
 #!/usr/bin/env python
 from argparse import ArgumentParser
 from pathlib import Path
-from typing import List
+from typing import Sequence
 import asyncio
+import sys
 import os
 
 import asyncioffmpeg.ffprobe as probe
 
 
-async def main(path: Path, suffix: List[str]):
+async def main(path: Path, suffix: Sequence[str]):
 
-    path = Path(P.path).expanduser()
+    path = Path(path).expanduser()
+    if not path.is_dir():
+        raise NotADirectoryError(path)
 
-    flist = (f for f in path.iterdir() if f.is_file() and f.suffix in suffix)
+    files = (f for f in path.iterdir() if f.is_file() and f.suffix in suffix)
 
-    futures = [probe.get_meta(f) for f in flist]
-
-    metas = await asyncio.gather(*futures)
-
-    for meta in metas:
+    async for meta in probe.get_meta(files):
         if not meta:
             continue
         fn = meta['format']['filename']
@@ -31,11 +30,10 @@ async def main(path: Path, suffix: List[str]):
         description="Get media metadata asynchronously with FFprobe")
     p.add_argument('path', help='directory where media files are kept')
     p.add_argument('-suffix', help='file suffixes of desired media file types',
-                   nargs='+', default=['.mp4', '.avi', '.ogv'])
+                   nargs='+', default=['.mp4', '.avi', '.ogv', '.ogg'])
     P = p.parse_args()
 
-    if os.name == 'nt':
-        loop = asyncio.ProactorEventLoop()  # type: ignore
-        loop.run_until_complete(main(P.path, P.suffix))
-    else:
-        asyncio.run(main(P.path, P.suffix))
+    if os.name == 'nt' and sys.version_info < (3, 8):
+        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())  # type: ignore
+
+    asyncio.run(main(P.path, P.suffix))

probe_coroutine.py

@@ -2,7 +2,7 @@
 from pathlib import Path
 import shutil
 from argparse import ArgumentParser
-import asyncioffmpeg.ffprobe_sync as probe
+import asyncioffmpeg.ffprobe as probe
 
 FFPROBE = shutil.which('ffprobe')
 if not FFPROBE:
@@ -14,7 +14,7 @@
         description="Get media metadata synchronously with FFprobe")
     p.add_argument('path', help='directory where media files are kept')
     p.add_argument('-suffix', help='file suffixes of desired media file types',
-                   nargs='+', default=['.mp4', '.avi', '.ogv'])
+                   nargs='+', default=['.mp4', '.avi', '.ogv', '.ogg'])
     P = p.parse_args()
 
     path = Path(P.path).expanduser()

probe_sync.py

@@ -20,7 +20,8 @@ classifiers =
   Topic :: Multimedia :: Video :: Conversion
   Topic :: System :: Networking
   Topic :: Utilities
-license_file = LICENSE
+license_files =
+  LICENSE.txt
 long_description = file: README.md
 long_description_content_type = text/markdown
 
@@ -29,7 +30,6 @@ python_requires = >= 3.7
 setup_requires =
   setuptools >= 38.6
   pip >= 10
-  twine >= 1.11
 packages = find:
 scripts =
   demo.py

setup.cfg

@@ -2,7 +2,6 @@
 import os
 import pytest
 import asyncioffmpeg.ffprobe as probe
-import asyncioffmpeg.ffprobe_sync as probe_sync
 
 
 def get_duration(meta: dict) -> float:
@@ -14,14 +13,14 @@ def get_duration(meta: dict) -> float:
 async def test_ffprobe(genpat):
     vid = genpat
 
-    meta = await probe.get_meta(vid)
-    assert get_duration(meta) == pytest.approx(5.)
+    async for meta in probe.get_meta([vid]):
+        assert get_duration(meta) == pytest.approx(5.)
 
 
 def test_ffprobe_sync(genpat):
     vid = genpat
 
-    meta = probe_sync.ffprobe_sync(vid)
+    meta = probe.ffprobe_sync(vid)
     assert get_duration(meta) == pytest.approx(5.)