audio: Load a track requested by the user
Either through the command line, mpris2, or the open button in the header. Implements: #7 (Add MPRIS2 Support) Implements: #47 (Signal that the track has changed when it actually changes) Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
parent
51096104ce
commit
35d0d815ca
|
@ -1,5 +1,6 @@
|
|||
# Copyright 2022 (c) Anna Schumaker.
|
||||
"""Set up our Application."""
|
||||
import pathlib
|
||||
from . import gsetup
|
||||
from . import audio
|
||||
from . import db
|
||||
|
@ -10,6 +11,8 @@ from . import options
|
|||
from . import window
|
||||
from gi.repository import GObject
|
||||
from gi.repository import GLib
|
||||
from gi.repository import Gio
|
||||
from gi.repository import Gst
|
||||
from gi.repository import Adw
|
||||
|
||||
MAJOR_VERSION = 3
|
||||
|
@ -28,9 +31,18 @@ class Application(Adw.Application):
|
|||
def __init__(self):
|
||||
"""Initialize an Application."""
|
||||
super().__init__(application_id=gsetup.APPLICATION_ID,
|
||||
resource_base_path=gsetup.RESOURCE_PATH)
|
||||
resource_base_path=gsetup.RESOURCE_PATH,
|
||||
flags=Gio.ApplicationFlags.HANDLES_OPEN)
|
||||
self.add_main_option_entries([options.Version])
|
||||
|
||||
def __load_file(self, file: pathlib.Path) -> None:
|
||||
self.player.set_state_sync(Gst.State.READY)
|
||||
self.player.file = file
|
||||
self.player.set_state_sync(Gst.State.PLAYING)
|
||||
|
||||
def __load_path(self, src: GObject.GObject, path: pathlib.Path) -> None:
|
||||
self.__load_file(path)
|
||||
|
||||
def __set_replaygain(self, *args) -> None:
|
||||
enabled = self.db.settings["audio.replaygain.enabled"]
|
||||
mode = self.db.settings["audio.replaygain.mode"]
|
||||
|
@ -48,6 +60,7 @@ class Application(Adw.Application):
|
|||
|
||||
hdr.connect("notify::rg-enabled", self.__set_replaygain)
|
||||
hdr.connect("notify::rg-mode", self.__set_replaygain)
|
||||
hdr.connect("track-requested", self.__load_path)
|
||||
self.__set_replaygain()
|
||||
return hdr
|
||||
|
||||
|
@ -75,6 +88,7 @@ class Application(Adw.Application):
|
|||
self.mpris.app.connect("Quit", self.win.close)
|
||||
|
||||
self.mpris.player.link_property("Volume", self.win.header, "volume")
|
||||
self.mpris.player.connect("OpenPath", self.__load_path)
|
||||
|
||||
def do_handle_local_options(self, opts: GLib.VariantDict) -> int:
|
||||
"""Handle any command line options."""
|
||||
|
@ -103,6 +117,12 @@ class Application(Adw.Application):
|
|||
Adw.Application.do_activate(self)
|
||||
self.win.present()
|
||||
|
||||
def do_open(self, files: list, n_files: int, hint: str) -> None:
|
||||
"""Play an audio file passed from the command line."""
|
||||
if n_files > 0:
|
||||
self.__load_file(pathlib.Path(files[0].get_path()))
|
||||
self.activate()
|
||||
|
||||
def do_shutdown(self) -> None:
|
||||
"""Handle the Adw.Application::shutdown signal."""
|
||||
Adw.Application.do_shutdown(self)
|
||||
|
|
|
@ -1,8 +1,10 @@
|
|||
# Copyright 2022 (c) Anna Schumaker.
|
||||
"""A custom GObject managing a GStreamer playbin."""
|
||||
import pathlib
|
||||
from gi.repository import GObject
|
||||
from gi.repository import Gst
|
||||
from . import replaygain
|
||||
from .. import path
|
||||
|
||||
|
||||
class Player(GObject.GObject):
|
||||
|
@ -10,6 +12,9 @@ class Player(GObject.GObject):
|
|||
|
||||
volume = GObject.Property(type=float, default=1.0)
|
||||
|
||||
file = GObject.Property(type=GObject.TYPE_PYOBJECT)
|
||||
have_track = GObject.Property(type=bool, default=False)
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the audio Player."""
|
||||
super().__init__()
|
||||
|
@ -21,8 +26,28 @@ class Player(GObject.GObject):
|
|||
Gst.ElementFactory.make("fakesink"))
|
||||
self._playbin.set_state(Gst.State.READY)
|
||||
|
||||
bus = self._playbin.get_bus()
|
||||
bus.add_signal_watch()
|
||||
bus.connect("message::eos", self.__msg_eos)
|
||||
bus.connect("message::stream-start", self.__msg_stream_start)
|
||||
|
||||
self.bind_property("volume", self._playbin, "volume")
|
||||
|
||||
self.connect("notify::file", self.__notify_file)
|
||||
|
||||
def __msg_eos(self, bus: Gst.Bus, message: Gst.Message) -> None:
|
||||
self.emit("eos")
|
||||
|
||||
def __msg_stream_start(self, bus: Gst.Bus, message: Gst.Message) -> None:
|
||||
self.emit("file-loaded",
|
||||
path.from_uri(self._playbin.get_property("current-uri")))
|
||||
|
||||
def __notify_file(self, player: GObject.GObject, param) -> None:
|
||||
if self.file:
|
||||
uri = self.file.as_uri()
|
||||
print(f"audio: loading {uri}")
|
||||
self._playbin.set_property("uri", uri)
|
||||
|
||||
def get_replaygain(self) -> tuple[bool, str | None]:
|
||||
"""Get the current ReplayGain mode."""
|
||||
mode = self._replaygain.mode
|
||||
|
@ -36,6 +61,25 @@ class Player(GObject.GObject):
|
|||
"""Set the ReplayGain mode."""
|
||||
self._replaygain.mode = mode if enabled else "disabled"
|
||||
|
||||
def set_state_sync(self, state: Gst.State) -> None:
|
||||
"""Set the state of the playbin, and wait for it to change."""
|
||||
if self._playbin.set_state(state) == Gst.StateChangeReturn.ASYNC:
|
||||
self.get_state()
|
||||
|
||||
def shutdown(self) -> None:
|
||||
"""Shut down the player."""
|
||||
self._playbin.set_state(Gst.State.NULL)
|
||||
|
||||
@GObject.Signal(arg_types=(GObject.TYPE_PYOBJECT,))
|
||||
def file_loaded(self, file: pathlib.Path) -> None:
|
||||
"""Signal that a new URI has started."""
|
||||
print("audio: file loaded")
|
||||
self.have_track = True
|
||||
|
||||
@GObject.Signal
|
||||
def eos(self) -> None:
|
||||
"""Signal that the current track has ended."""
|
||||
print("audio: end of stream")
|
||||
self.set_state_sync(Gst.State.READY)
|
||||
self.have_track = False
|
||||
self.file = None
|
||||
|
|
|
@ -2,6 +2,7 @@
|
|||
"""Our Mpris2 Player dbus Object."""
|
||||
import pathlib
|
||||
from gi.repository import GObject
|
||||
from .. import path
|
||||
from . import dbus
|
||||
|
||||
PLAYER_XML = pathlib.Path(__file__).parent / "Player.xml"
|
||||
|
@ -69,3 +70,8 @@ class Player(dbus.Object):
|
|||
@GObject.Signal(arg_types=(str,))
|
||||
def OpenUri(self, uri: str) -> None:
|
||||
"""Open the given uri."""
|
||||
self.emit("OpenPath", path.from_uri(uri))
|
||||
|
||||
@GObject.Signal(arg_types=(GObject.TYPE_PYOBJECT,))
|
||||
def OpenPath(self, filepath: pathlib.Path) -> None:
|
||||
"""Open the given path."""
|
||||
|
|
|
@ -0,0 +1,11 @@
|
|||
# Copyright 2022 (c) Anna Schumaker.
|
||||
"""Extra path handling for URIs."""
|
||||
import pathlib
|
||||
import urllib
|
||||
|
||||
|
||||
def from_uri(uri: str) -> pathlib.Path:
|
||||
"""Make a path from a uri."""
|
||||
if parsed := urllib.parse.urlparse(uri):
|
||||
return pathlib.Path(urllib.parse.unquote(parsed.path))
|
||||
return pathlib.Path(uri)
|
|
@ -4,7 +4,9 @@ import io
|
|||
import unittest
|
||||
import unittest.mock
|
||||
import emmental.audio
|
||||
import tests.util
|
||||
from gi.repository import GObject
|
||||
from gi.repository import GLib
|
||||
from gi.repository import Gst
|
||||
|
||||
|
||||
|
@ -15,11 +17,23 @@ class TestAudio(unittest.TestCase):
|
|||
"""Set up common variables."""
|
||||
self.player = emmental.audio.Player()
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up the playbin."""
|
||||
self.player.shutdown()
|
||||
|
||||
def main_loop(self) -> None:
|
||||
"""Run a GLib main loop."""
|
||||
while GLib.main_context_default().iteration():
|
||||
pass
|
||||
|
||||
def test_player(self):
|
||||
"""Test that the audio player was set up correctly."""
|
||||
self.assertIsInstance(self.player, GObject.GObject)
|
||||
self.assertEqual(self.player.get_state(), Gst.State.READY)
|
||||
|
||||
self.assertIsNone(self.player.file)
|
||||
self.assertFalse(self.player.have_track)
|
||||
|
||||
def test_playbin(self):
|
||||
"""Test that the playbin was configured correctly."""
|
||||
self.assertIsInstance(self.player._playbin, Gst.Element)
|
||||
|
@ -30,6 +44,38 @@ class TestAudio(unittest.TestCase):
|
|||
self.assertRegex(self.player._playbin.get_property("video-sink").name,
|
||||
r"fakesink\d+")
|
||||
|
||||
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_eos(self, mock_stdout: io.StringIO):
|
||||
"""Test handling an EOS message."""
|
||||
self.player.file = tests.util.TRACK_OGG
|
||||
|
||||
eos = Gst.Message.new_eos(self.player._playbin)
|
||||
self.player._playbin.get_bus().post(eos)
|
||||
self.main_loop()
|
||||
|
||||
self.assertRegex(mock_stdout.getvalue(), "audio: end of stream")
|
||||
self.assertEqual(self.player.get_state(), Gst.State.READY)
|
||||
self.assertFalse(self.player.have_track)
|
||||
self.assertIsNone(self.player.file)
|
||||
|
||||
@unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
|
||||
def test_file(self, mock_stdout: io.StringIO):
|
||||
"""Test that the file property works as expected."""
|
||||
self.player.file = tests.util.TRACK_OGG
|
||||
self.assertEqual(self.player._playbin.get_property("uri"),
|
||||
tests.util.TRACK_OGG.as_uri())
|
||||
|
||||
started = unittest.mock.Mock()
|
||||
self.player.connect("file-loaded", started)
|
||||
self.player.set_state_sync(Gst.State.PAUSED)
|
||||
self.main_loop()
|
||||
started.assert_called_with(self.player, tests.util.TRACK_OGG)
|
||||
self.assertTrue(self.player.have_track)
|
||||
|
||||
self.assertEqual(mock_stdout.getvalue(),
|
||||
f"audio: loading {tests.util.TRACK_OGG.as_uri()}\n"
|
||||
"audio: file loaded\n")
|
||||
|
||||
def test_volume(self):
|
||||
"""Test that the volume property works as expected."""
|
||||
self.assertEqual(self.player.volume, 1.0)
|
||||
|
|
|
@ -2,6 +2,7 @@
|
|||
"""Tests our Mpris Player object."""
|
||||
import pathlib
|
||||
import unittest
|
||||
import unittest.mock
|
||||
import emmental.mpris2.player
|
||||
from gi.repository import Gio
|
||||
|
||||
|
@ -55,4 +56,8 @@ class TestPlayer(unittest.TestCase):
|
|||
self.player.emit("Seek", 12345)
|
||||
self.player.emit("SetPosition",
|
||||
"/com/nowheycreamery/emmental/123", 12345)
|
||||
|
||||
open_path = unittest.mock.Mock()
|
||||
self.player.connect("OpenPath", open_path)
|
||||
self.player.emit("OpenUri", "/a/b/c.ogg")
|
||||
open_path.assert_called_with(self.player, pathlib.Path("/a/b/c.ogg"))
|
||||
|
|
|
@ -0,0 +1,21 @@
|
|||
# Copyright 2022 (c) Anna Schumaker.
|
||||
"""Tests our custom Path class."""
|
||||
import pathlib
|
||||
import unittest
|
||||
import emmental.path
|
||||
|
||||
|
||||
class TestPath(unittest.TestCase):
|
||||
"""Test our path module."""
|
||||
|
||||
def test_from_uri(self):
|
||||
"""Test getting a path from a uri."""
|
||||
p = emmental.path.from_uri("file:///a/b/c.txt")
|
||||
self.assertIsInstance(p, pathlib.Path)
|
||||
self.assertEqual(str(p), "/a/b/c.txt")
|
||||
|
||||
p = emmental.path.from_uri("file:///a/b%20c/d.txt")
|
||||
self.assertEqual(str(p), "/a/b c/d.txt")
|
||||
|
||||
p = emmental.path.from_uri("/a/b/c.txt")
|
||||
self.assertEqual(str(p), "/a/b/c.txt")
|
|
@ -1,10 +1,14 @@
|
|||
# Copyright 2022 (c) Anna Schumaker
|
||||
"""Helper utilities for testing."""
|
||||
import pathlib
|
||||
import unittest
|
||||
import emmental.db
|
||||
from gi.repository import GObject
|
||||
|
||||
|
||||
TRACK_OGG = pathlib.Path(__file__).parent / "track.ogg"
|
||||
|
||||
|
||||
class TestCase(unittest.TestCase):
|
||||
"""A TestCase that handles database setup and cleanup."""
|
||||
|
||||
|
|
Binary file not shown.
Loading…
Reference in New Issue