feat(analyzer): analyze replaygain using ffmpeg

- remove pycairo pip install
- fix py36 compatibility
- reraise when executable was not found

BREAKING CHANGE: The analyzer requires 'ffmpeg'. The 'rgain3' python package and it's system dependencies can be removed.

analyzer/libretime_analyzer/ffmpeg.py

@@ -0,0 +1,56 @@
+import re
+from pathlib import Path
+from typing import Optional
+
+from .utils import run_
+
+
+def _ffmpeg(*args, **kwargs):
+    return run_(
+        "ffmpeg",
+        *args,
+        "-f",
+        "null",
+        "/dev/null",
+        "-hide_banner",
+        "-nostats",
+        **kwargs,
+    )
+
+
+def _ffprobe(*args, **kwargs):
+    return run_("ffprobe", *args, **kwargs)
+
+
+_PROBE_REPLAYGAIN_RE = re.compile(
+    r".*REPLAYGAIN_TRACK_GAIN: ([-+]?[0-9]+\.[0-9]+) dB.*",
+)
+
+
+def probe_replaygain(filepath: Path) -> Optional[float]:
+    """
+    Probe replaygain will probe the given audio file and return the replaygain if available.
+    """
+    cmd = _ffprobe("-i", filepath)
+
+    track_gain_match = _PROBE_REPLAYGAIN_RE.search(cmd.stderr)
+
+    if track_gain_match:
+        return float(track_gain_match.group(1))
+
+
+_COMPUTE_REPLAYGAIN_RE = re.compile(
+    r".* track_gain = ([-+]?[0-9]+\.[0-9]+) dB.*",
+)
+
+
+def compute_replaygain(filepath: Path) -> Optional[float]:
+    """
+    Compute replaygain will analyse the given audio file and return the replaygain if available.
+    """
+    cmd = _ffmpeg("-i", filepath, "-vn", "-filter", "replaygain")
+
+    track_gain_match = _COMPUTE_REPLAYGAIN_RE.search(cmd.stderr)
+
+    if track_gain_match:
+        return float(track_gain_match.group(1))

analyzer/libretime_analyzer/steps/analyze_replaygain.py

@@ -1,42 +1,27 @@
-import re
-import subprocess
+from subprocess import CalledProcessError
 from typing import Any, Dict
 
-from loguru import logger
-
-REPLAYGAIN_EXECUTABLE = "replaygain"  # From the rgain3 python package
+from ..ffmpeg import compute_replaygain, probe_replaygain
 
 
-def analyze_replaygain(filename: str, metadata: Dict[str, Any]):
-    """Extracts the Replaygain loudness normalization factor of a track.
-    :param filename: The full path to the file to analyzer
-    :param metadata: A metadata dictionary where the results will be put
-    :return: The metadata dictionary
+def analyze_replaygain(filepath: str, metadata: Dict[str, Any]):
     """
-    """ The -d flag means do a dry-run, ie. don't modify the file directly.
+    Extracts the Replaygain loudness normalization factor of a track using ffmpeg.
     """
-    command = [REPLAYGAIN_EXECUTABLE, "-d", filename]
     try:
-        results = subprocess.check_output(
-            command,
-            stderr=subprocess.STDOUT,
-            close_fds=True,
-            universal_newlines=True,
-        )
-        gain_match = (
-            r"Calculating Replay Gain information \.\.\.(?:\n|.)*?:([\d.-]*) dB"
-        )
-        replaygain = re.search(gain_match, results).group(1)
-        metadata["replay_gain"] = float(replaygain)
+        # First probe for existing replaygain metadata.
+        track_gain = probe_replaygain(filepath)
+        if track_gain is not None:
+            metadata["replay_gain"] = track_gain
+            return metadata
+    except (CalledProcessError, OSError):
+        pass
 
-    except OSError as e:  # replaygain was not found
-        logger.warning(
-            "Failed to run: %s - %s. %s"
-            % (command[0], e.strerror, "Do you have python-rgain installed?")
-        )
-    except subprocess.CalledProcessError as e:  # replaygain returned an error code
-        logger.warning("%s %s %s", e.cmd, e.output, e.returncode)
-    except Exception as e:
-        logger.warning(e)
+    try:
+        track_gain = compute_replaygain(filepath)
+        if track_gain is not None:
+            metadata["replay_gain"] = track_gain
+    except (CalledProcessError, OSError):
+        pass
 
     return metadata

analyzer/libretime_analyzer/utils.py

@@ -0,0 +1,24 @@
+from subprocess import PIPE, CalledProcessError, CompletedProcess, run
+
+from loguru import logger
+
+
+def run_(*args, **kwargs) -> CompletedProcess:
+    try:
+        return run(
+            args,
+            check=True,
+            stdout=PIPE,
+            stderr=PIPE,
+            universal_newlines=True,
+            **kwargs,
+        )
+
+    except OSError as exception:  # executable was not found
+        cmd = args[0]
+        logger.warning(f"Failed to run: {cmd} - {exception}. Is {cmd} installed?")
+        raise exception
+
+    except CalledProcessError as exception:  # returned an error code
+        logger.error(exception)
+        raise exception