audio/replaygain: Add a ReplayGain filter

With options for album mode, track mode, and completely disabled.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

emmental/audio/replaygain.py

@@ -0,0 +1,62 @@
+# Copyright 2022 (c) Anna Schumaker.
+"""A custom Gst.Bin for selecting ReplayGain mode."""
+import collections
+from gi.repository import GObject
+from gi.repository import Gst
+
+RequestPads = collections.namedtuple("RequestPads", ["src", "sink"])
+
+
+class Filter(Gst.Bin):
+    """The ReplayGain filter element."""
+
+    mode = GObject.Property(type=str, default="disabled")
+
+    def __init__(self):
+        """Initialize the ReplayGain element."""
+        super().__init__()
+        self._src = Gst.ElementFactory.make("output-selector")
+        self._sink = Gst.ElementFactory.make("input-selector")
+        self._rgalbum = Gst.ElementFactory.make("rgvolume")
+        self._rgtrack = Gst.ElementFactory.make("rgvolume")
+        self._rglimit = Gst.ElementFactory.make("rglimiter")
+
+        for elm in [self._src, self._rgalbum, self._rgtrack,
+                    self._rglimit, self._sink]:
+            self.add(elm)
+
+        self._disabled = self.__request_pads(self._rglimit)
+        self._album_mode = self.__request_pads(self._rgalbum)
+        self._track_mode = self.__request_pads(self._rgtrack)
+
+        self._rgalbum.set_property("pre-amp", 6.0)
+        self._rgtrack.set_property("pre-amp", 6.0)
+        self._rgtrack.set_property("album-mode", False)
+
+        self._src.set_property("pad-negotiation-mode", 2)
+        self._src.set_property("active-pad", self._disabled.src)
+        self._sink.set_property("active-pad", self._disabled.sink)
+
+        self.__add_ghost_pad("sink", self._src)
+        self.__add_ghost_pad("src", self._sink)
+
+        self.connect("notify::mode", self.__notify_mode)
+
+    def __add_ghost_pad(self, pad: str, elm: Gst.Element) -> None:
+        self.add_pad(Gst.GhostPad.new(pad, elm.get_static_pad(pad)))
+
+    def __request_pads(self, elm: Gst.Element) -> RequestPads:
+        pads = RequestPads(src=self._src.request_pad_simple("src_%u"),
+                           sink=self._sink.request_pad_simple("sink_%u"))
+        pads.src.link(elm.get_static_pad("sink"))
+        elm.get_static_pad("src").link(pads.sink)
+        return pads
+
+    def __notify_mode(self, filter: Gst.Bin, param) -> None:
+        match self.mode:
+            case "album": pads = self._album_mode
+            case "track": pads = self._track_mode
+            case _: pads = self._disabled
+        print(f"audio: setting ReplayGain mode to '{self.mode}'")
+        self._src.set_property("active-pad", pads.src)
+        self._sink.set_property("active-pad", pads.sink)

tests/audio/test_replaygain.py

@@ -0,0 +1,100 @@
+# Copyright 2022 (c) Anna Schumaker.
+"""Test the ReplayGain filter element."""
+import io
+import unittest
+import unittest.mock
+import emmental.audio.replaygain
+from gi.repository import Gst
+
+
+class TestReplayGain(unittest.TestCase):
+    """Test the ReplayGain filter."""
+
+    def setUp(self):
+        """Set up common variables."""
+        self.filter = emmental.audio.replaygain.Filter()
+
+    def test_request_pads(self):
+        """Test that we can create a RequestPads named tuple."""
+        req = emmental.audio.replaygain.RequestPads(1, 2)
+        self.assertIsInstance(req, tuple)
+        self.assertEqual(req.src, 1)
+        self.assertEqual(req.sink, 2)
+
+    def test_init(self):
+        """Test that the filter is set up properly."""
+        self.assertIsInstance(self.filter, Gst.Bin)
+        self.assertIsInstance(self.filter._src, Gst.Element)
+        self.assertIsInstance(self.filter._sink, Gst.Element)
+        self.assertEqual(self.filter.mode, "disabled")
+
+        self.assertRegex(self.filter._src.name, r"outputselector\d+")
+        self.assertEqual(self.filter._src.get_property("pad-negotiation-mode"),
+                         2)
+        self.assertEqual(self.filter._src.get_property("active-pad"),
+                         self.filter._disabled.src)
+
+        self.assertRegex(self.filter._sink.name, r"inputselector\d+")
+        self.assertEqual(self.filter._sink.get_property("active-pad"),
+                         self.filter._disabled.sink)
+
+        self.assertIsInstance(self.filter.get_static_pad("src"), Gst.GhostPad)
+        self.assertIsInstance(self.filter.get_static_pad("sink"), Gst.GhostPad)
+
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_disabled(self, mock_stdout: io.StringIO):
+        """Test the ReplayGain filter with ReplayGain disabled."""
+        self.assertIsInstance(self.filter._rglimit, Gst.Element)
+        self.assertIsInstance(self.filter._disabled,
+                              emmental.audio.replaygain.RequestPads)
+        self.assertIsInstance(self.filter._disabled.src, Gst.Pad)
+        self.assertIsInstance(self.filter._disabled.sink, Gst.Pad)
+
+        self.assertRegex(self.filter._rglimit.name, r"rglimiter\d+")
+
+        self.filter.mode = "track"
+        self.filter.mode = "disabled"
+        self.assertRegex(mock_stdout.getvalue(),
+                         r"audio: setting ReplayGain mode to 'disabled'")
+        self.assertEqual(self.filter._src.get_property("active-pad"),
+                         self.filter._disabled.src)
+        self.assertEqual(self.filter._sink.get_property("active-pad"),
+                         self.filter._disabled.sink)
+
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_album_mode(self, mock_stdout: io.StringIO):
+        """Test the ReplayGain filter with ReplayGain set to album mode."""
+        self.assertIsInstance(self.filter._rgalbum, Gst.Element)
+        self.assertIsInstance(self.filter._album_mode,
+                              emmental.audio.replaygain.RequestPads)
+        self.assertIsInstance(self.filter._album_mode.src, Gst.Pad)
+        self.assertIsInstance(self.filter._album_mode.sink, Gst.Pad)
+
+        self.assertRegex(self.filter._rgalbum.name, r"rgvolume\d+")
+        self.assertEqual(self.filter._rgalbum.get_property("pre-amp"), 6.0)
+        self.assertTrue(self.filter._rgalbum.get_property("album-mode"))
+
+        self.filter.mode = "album"
+        self.assertRegex(mock_stdout.getvalue(),
+                         r"audio: setting ReplayGain mode to 'album'")
+        self.assertEqual(self.filter._src.get_property("active-pad"),
+                         self.filter._album_mode.src)
+        self.assertEqual(self.filter._sink.get_property("active-pad"),
+                         self.filter._album_mode.sink)
+
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_track_mode(self, mock_stdout: io.StringIO):
+        """Test the ReplayGain filter with ReplayGain set to track mode."""
+        self.assertIsInstance(self.filter._rgtrack, Gst.Element)
+
+        self.assertRegex(self.filter._rgtrack.name, r"rgvolume\d+")
+        self.assertEqual(self.filter._rgtrack.get_property("pre-amp"), 6.0)
+        self.assertFalse(self.filter._rgtrack.get_property("album-mode"))
+
+        self.filter.mode = "track"
+        self.assertRegex(mock_stdout.getvalue(),
+                         r"audio: setting ReplayGain mode to 'track'")
+        self.assertEqual(self.filter._src.get_property("active-pad"),
+                         self.filter._track_mode.src)
+        self.assertEqual(self.filter._sink.get_property("active-pad"),
+                         self.filter._track_mode.sink)