audio: Notify when playback is almost done

The application can use this to pre-load the next track for gapless
playback.

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

emmental/audio/__init__.py

@@ -31,6 +31,7 @@ class Player(GObject.GObject):
     playing = GObject.Property(type=bool, default=False)
     status = GObject.Property(type=str, default="Stopped")
     have_track = GObject.Property(type=bool, default=False)
+    almost_done = GObject.Property(type=bool, default=False)
     playtime = GObject.Property(type=float)
     savedtime = GObject.Property(type=float)
 
@@ -58,6 +59,11 @@ class Player(GObject.GObject):
 
         self.connect("notify::file", self.__notify_file)
 
+    def __check_last_second(self) -> None:
+        if self.duration - self.position <= 2 * (Gst.SECOND / Gst.USECOND):
+            if not self.almost_done:
+                self.emit("about-to-finish")
+
     def __get_current_playtime(self) -> float:
         if not self._playbin.clock:
             return 0.0
@@ -132,6 +138,8 @@ class Player(GObject.GObject):
         for tag in ["album-disc-number", "track-number",
                     "position", "playtime", "savedtime"]:
             self.set_property(tag, 0)
+
+        self.almost_done = False
         self.artwork = artwork
         self.duration = duration
 
@@ -139,6 +147,7 @@ class Player(GObject.GObject):
         (res, pos) = self._playbin.query_position(Gst.Format.TIME)
         self.position = pos / Gst.USECOND if res else 0
         self.playtime = self.__get_current_playtime() + self.savedtime
+        self.__check_last_second()
         return GLib.SOURCE_CONTINUE
 
     def __update_timeout(self) -> None:
@@ -194,6 +203,12 @@ class Player(GObject.GObject):
         """Stop playback."""
         self.set_state_sync(Gst.State.READY)
 
+    @GObject.Signal
+    def about_to_finish(self) -> None:
+        """Signal that playback is almost done."""
+        print("audio: about to finish")
+        self.almost_done = True
+
     @GObject.Signal(arg_types=(GObject.TYPE_PYOBJECT,))
     def file_loaded(self, file: pathlib.Path) -> None:
         """Signal that a new URI has started."""

tests/audio/test_audio.py

@@ -61,6 +61,7 @@ class TestAudio(unittest.TestCase):
         self.player.position = 8
         self.player.playtime = 6
         self.player.savedtime = 4
+        self.player.almost_done = True
         self.player.artwork = pathlib.Path("/a/b/c.jpg")
 
         eos = Gst.Message.new_eos(self.player._playbin)
@@ -79,8 +80,34 @@ class TestAudio(unittest.TestCase):
         self.assertEqual(self.player.get_state(), Gst.State.READY)
         self.assertEqual(self.player.status, "Stopped")
         self.assertFalse(self.player.have_track)
+        self.assertFalse(self.player.almost_done)
         self.assertIsNone(self.player.file)
 
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_almost_done(self, mock_stdout: io.StringIO):
+        """Test notifying that the current track is almost done."""
+        self.assertFalse(self.player.almost_done)
+
+        self.player.almost_done = True
+        self.player.file = tests.util.TRACK_OGG
+        self.player.pause()
+        self.main_loop()
+        self.assertFalse(self.player.almost_done)
+
+        self.player.duration = (5 * Gst.SECOND) / Gst.USECOND
+        with unittest.mock.patch.object(self.player._playbin,
+                                        "query_position") as query_position:
+            query_position.return_value = (True, 2 * Gst.SECOND)
+            self.player._Player__update_position()
+            self.assertFalse(self.player.almost_done)
+            self.assertNotRegex(mock_stdout.getvalue(),
+                                "audio: about to finish")
+
+            query_position.return_value = (True, 3 * Gst.SECOND)
+            self.player._Player__update_position()
+            self.assertTrue(self.player.almost_done)
+            self.assertRegex(mock_stdout.getvalue(), "audio: about to finish")
+
     @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."""