move more ffmpeg-specific functions to ffmpeg.py

.flake8

@@ -1,3 +1,5 @@
 [flake8]
 max-line-length = 132
 exclude = .git,__pycache__,.eggs/,doc/,docs/,build/,dist/,archive/
+per-file-ignores =
+    __init__.py:F401

pylivestream/__init__.py

@@ -5,6 +5,7 @@
 #
 from . import stream
 from . import utils
+from . import ffmpeg
 
 __all__ = ['FileIn', 'Microphone', 'SaveDisk', 'Screenshare', 'Webcam']
 

pylivestream/ffmpeg.py

@@ -4,22 +4,22 @@
 import os
 from pathlib import Path
 import shutil
+import json
 
 
-class Ffmpeg():
-
+class Ffmpeg:
     def __init__(self):
 
-        self.ERROR = ['-loglevel', 'error']
-        self.WARNING = ['-loglevel', 'warning']
-        self.INFO = ['-loglevel', 'info']
+        self.ERROR = ["-loglevel", "error"]
+        self.WARNING = ["-loglevel", "warning"]
+        self.INFO = ["-loglevel", "info"]
 
-        self.YES = ['-y']
+        self.YES = ["-y"]
 
         # default 8, increasing can help avoid warnings
-        self.QUEUE = ['-thread_queue_size', '8']
+        self.QUEUE = ["-thread_queue_size", "8"]
 
-        self.THROTTLE = '-re'
+        self.THROTTLE = "-re"
 
     def timelimit(self, t: Union[str, int, float]) -> List[str]:
         if t is None:
@@ -30,7 +30,7 @@ def timelimit(self, t: Union[str, int, float]) -> List[str]:
         t = str(t)
 
         if len(t) > 0:
-            return ['-t', str(t)]
+            return ["-t", str(t)]
         else:
             return []
 
@@ -39,16 +39,15 @@ def drawtext(self, text: str = None) -> List[str]:
         if not text:  # None or '' or [] etc.
             return []
 
-        fontcolor = 'fontcolor=white'
-        fontsize = 'fontsize=24'
-        box = 'box=1'
-        boxcolor = '[email protected]'
-        border = 'boxborderw=5'
-        x = 'x=(w-text_w)/2'
-        y = 'y=(h-text_h)*3/4'
+        fontcolor = "fontcolor=white"
+        fontsize = "fontsize=24"
+        box = "box=1"
+        boxcolor = "[email protected]"
+        border = "boxborderw=5"
+        x = "x=(w-text_w)/2"
+        y = "y=(h-text_h)*3/4"
 
-        return ['-vf',
-                f"drawtext=text='{text}':{fontcolor}:{fontsize}:{box}:{boxcolor}:{border}:{x}:{y}"]
+        return ["-vf", f"drawtext=text='{text}':{fontcolor}:{fontsize}:{box}:{boxcolor}:{border}:{x}:{y}"]
 
     def listener(self):
         """
@@ -63,19 +62,19 @@ def listener(self):
 
         TIMEOUT = 0.5
 
-        FFPLAY = shutil.which('ffplay')
+        FFPLAY = shutil.which("ffplay")
         if not FFPLAY:
-            raise FileNotFoundError('FFplay not found, cannot start listener')
+            raise FileNotFoundError("FFplay not found, cannot start listener")
 
-        cmd = [FFPLAY, '-loglevel', 'error', '-timeout', '5', '-autoexit', 'rtmp://localhost']
+        cmd = [FFPLAY, "-loglevel", "error", "-timeout", "5", "-autoexit", "rtmp://localhost"]
 
-        print('starting Localhost RTMP listener. \n\n', ' '.join(cmd), '\n\n Press   q   in this terminal to end stream.')
+        print("starting Localhost RTMP listener. \n\n", " ".join(cmd), "\n\n Press   q   in this terminal to end stream.")
 
         proc = subprocess.Popen(cmd)
 
-#        proc = subprocess.Popen(['ffmpeg', '-v', 'fatal', '-timeout', '5',
-#                                 '-i', 'rtmp://localhost', '-f', 'null', '-'],
-#                                stdout=subprocess.DEVNULL)
+        #        proc = subprocess.Popen(['ffmpeg', '-v', 'fatal', '-timeout', '5',
+        #                                 '-i', 'rtmp://localhost', '-f', 'null', '-'],
+        #                                stdout=subprocess.DEVNULL)
 
         sleep(TIMEOUT)
 
@@ -86,7 +85,44 @@ def movingBG(self, bgfn: Path = None) -> List[str]:
             return []
 
         bg = str(bgfn)
-        if os.name == 'nt':
-            bg = bg.replace('\\', '/')   # for PureWindowsPath
+        if os.name == "nt":
+            bg = bg.replace("\\", "/")  # for PureWindowsPath
+
+        return ["-filter_complex", f"movie={bg}:loop=0,setpts=N/FRAME_RATE/TB"]
+
+
+def get_exe(exein: str) -> str:
+    """checks that host streaming program is installed"""
+
+    exe = str(Path(exein).expanduser())
+    # %% verify
+    if not shutil.which(exe):
+        raise FileNotFoundError(
+            f"""
+
+*** Must have FFmpeg + FFprobe installed to use PyLivestream.
+https://www.ffmpeg.org/download.html
+
+could not find {exein}
+"""
+        )
+
+    return exe
+
+
+def get_meta(fn: Path, exein: str = None) -> Union[None, dict]:
+    if not fn:  # audio-only
+        return None
+
+    fn = Path(fn).expanduser()
+
+    if not fn.is_file():
+        raise FileNotFoundError(fn)
+
+    exe = get_exe("ffprobe") if exein is None else exein
+
+    cmd = [str(exe), "-loglevel", "error", "-print_format", "json", "-show_streams", "-show_format", str(fn)]
 
-        return ['-filter_complex', f'movie={bg}:loop=0,setpts=N/FRAME_RATE/TB']
+    ret = subprocess.check_output(cmd, universal_newlines=True)
+    # %% decode JSON from FFprobe
+    return json.loads(ret)

pylivestream/pylivestream.ini

@@ -9,6 +9,7 @@ keyframe_sec: 2
 audio_bps: 128k
 preset: veryfast
 exe: ffmpeg
+ffprobe_exe: ffprobe
 timelimit:
 
 # indexed by sys.platform

pylivestream/stream.py

@@ -7,7 +7,7 @@
 from typing import List
 #
 from . import utils
-from .ffmpeg import Ffmpeg
+from .ffmpeg import Ffmpeg, get_exe
 
 # %%  Col0: vertical pixels (height). Col1: video kbps. Interpolates.
 # NOTE: Python >= 3.6 has guaranteed dict() order.
@@ -81,16 +81,16 @@ def osparam(self, key: str):
 
         C.read_string(Path(self.inifn).expanduser().read_text(), source=str(self.inifn))
 
+        self.exe = get_exe(C.get(sys.platform, 'exe', fallback='ffmpeg'))
+        self.probeexe = get_exe(C.get(sys.platform, 'ffprobe_exe', fallback='ffprobe'))
+
         if self.site not in C:
             raise ValueError(f'streaming site {self.site} not found in configuration file {self.inifn}')
 
         if 'XDG_SESSION_TYPE' in os.environ:
             if os.environ['XDG_SESSION_TYPE'] == 'wayland':
                 logging.error('Wayland may only give black output. Try X11')
 
-        self.exe, self.probeexe = utils.getexe(C.get(sys.platform, 'exe',
-                                                     fallback='ffmpeg'))
-
         if self.vidsource == 'camera':
             self.res: List[str] = C.get(self.site, 'webcam_res').split('x')
             self.fps: float = C.getint(self.site, 'webcam_fps')

pylivestream/utils.py

@@ -1,16 +1,13 @@
-import json
 import logging
 import subprocess
 from pathlib import Path
 import sys
-import shutil
-from typing import Tuple, Union, List
+import typing
 import pkg_resources
+from .ffmpeg import get_meta
 
-DEVNULL = subprocess.DEVNULL
 
-
-def run(cmd: List[str]):
+def run(cmd: typing.Sequence[str]):
     """
     FIXME: shell=True for Windows seems necessary to specify devices enclosed by "" quotes
     """
@@ -38,7 +35,7 @@ def run(cmd: List[str]):
 """
 
 
-def check_device(cmd: List[str]) -> bool:
+def check_device(cmd: typing.Sequence[str]) -> bool:
     try:
         run(cmd)
         ok = True
@@ -55,7 +52,7 @@ def check_display(fn: str = None) -> bool:
     if not fn:
         fn = pkg_resources.resource_filename(__name__, 'logo.png')
 
-    cmd = ['ffplay', '-v', 'error', '-t', '1.0', '-autoexit', fn]
+    cmd = ['ffplay', '-loglevel', 'error', '-t', '1.0', '-autoexit', fn]
 
     ret = subprocess.run(cmd, timeout=10).returncode
 
@@ -73,49 +70,6 @@ def get_inifile(fn: str) -> Path:
     return inifn
 
 
-def getexe(exein: str = None) -> Tuple[str, str]:
-    """checks that host streaming program is installed"""
-
-    if not exein:
-        exe = 'ffmpeg'
-        probeexe = 'ffprobe'
-    else:
-        exe = str(Path(exein).expanduser())
-        probeexe = str(Path(exein).expanduser().parent / 'ffprobe')
-# %% verify
-    if not shutil.which(exe):
-        print('\n\n *** Must have FFmpeg installed to use PyLivestream.', file=sys.stderr)
-        print('https://www.ffmpeg.org/download.html \n\n', file=sys.stderr)
-        raise FileNotFoundError(exe)
-
-    if not shutil.which(probeexe):
-        print('\n\n *** You must have FFmpeg + FFprobe installed to use PyLivestream.', file=sys.stderr)
-        print('https://www.ffmpeg.org/download.html \n\n', file=sys.stderr)
-        raise FileNotFoundError(probeexe)
-
-    return exe, probeexe
-
-
-def get_meta(fn: Path, exein: str = None) -> Union[None, dict]:
-    if not fn:  # audio-only
-        return None
-
-    fn = Path(fn).expanduser()
-
-    if not fn.is_file():
-        raise FileNotFoundError(fn)
-
-    exe = getexe()[1] if exein is None else exein
-
-    cmd = [str(exe), '-v', 'error', '-print_format', 'json',
-           '-show_streams',
-           '-show_format', str(fn)]
-
-    ret = subprocess.check_output(cmd, universal_newlines=True)
-# %% decode JSON from FFprobe
-    return json.loads(ret)
-
-
 def meta_caption(meta) -> str:
     """makes text from metadata for captioning video"""
     caption = ''
@@ -133,7 +87,7 @@ def meta_caption(meta) -> str:
     return caption
 
 
-def get_resolution(fn: Path, exe: str = None) -> List[str]:
+def get_resolution(fn: Path, exe: str = None) -> typing.List[str]:
     """
     get resolution (widthxheight) of video file
     http://trac.ffmpeg.org/wiki/FFprobeTips#WidthxHeight

tests/test_class.py

@@ -33,15 +33,10 @@ def test_bad_key(key, excp):
         assert pls.utils.getstreamkey(key) is None
 
 
-@pytest.mark.parametrize('rex', (None, '', 'ffmpeg'))
+@pytest.mark.parametrize('rex', ('ffmpeg', 'ffprobe'))
 def test_exe(rex):
-    exe, pexe = pls.utils.getexe()
-    assert 'ffmpeg' in exe
-    assert 'ffprobe' in pexe
-
-    exe, pexe = pls.utils.getexe(rex)
-    assert 'ffmpeg' in exe
-    assert 'ffprobe' in pexe
+    exe = pls.ffmpeg.get_exe(rex)
+    assert rex in exe
 
 
 @pytest.mark.parametrize('inp', (None, ''))