Factor out the video player

lazy_player/__init__.py

@@ -8,6 +8,7 @@ from typing import Any, cast
 import gi
 
 from .file_model import FileItem, FileListModel, FileType
+from .video_player import VideoPlayer
 
 gi.require_version("Gdk", "4.0")
 gi.require_version("Gtk", "4.0")
@@ -23,9 +24,7 @@ class MainWindow(Gtk.ApplicationWindow):
     list_model: FileListModel
     selection_model: Gtk.SingleSelection
     video_widget: Gtk.Picture
-    pipeline: Gst.Pipeline
+    tick_callback_id: int
-    playbin: Gst.Element
-    overlay_tick_callback_id: int
     overlay_label: Gtk.Label
     overlay_hide_time: float
     last_position_save: float
@@ -47,13 +46,7 @@ class MainWindow(Gtk.ApplicationWindow):
         self.overlay_label.set_wrap(True)
         self.overlay_label.set_wrap_mode(Pango.WrapMode.WORD_CHAR)
 
-        def update_overlay(widget: Gtk.Widget, frame_clock: Gdk.FrameClock, data: Any) -> bool:
+        self.tick_callback_id = self.add_tick_callback(self._on_tick, None)
-            current_time = frame_clock.get_frame_time() / 1_000_000
-            if current_time >= self.overlay_hide_time:
-                self.overlay_label.set_visible(False)
-            return True
-
-        self.overlay_tick_callback_id = self.add_tick_callback(update_overlay, None)
 
         # Make window fullscreen and borderless
         self.set_decorated(False)
@@ -89,25 +82,8 @@ class MainWindow(Gtk.ApplicationWindow):
 
         overlay_box.append(overlay)
 
-        # Setup GStreamer pipeline
+        # Setup video player
-        self.pipeline = Gst.Pipeline.new("video-player")
+        self.video_player = VideoPlayer(self.video_widget)
-
-        playbin = Gst.ElementFactory.make("playbin", "playbin")
-        if not playbin:
-            raise RuntimeError("Failed to create playbin element")
-
-        self.playbin = playbin
-
-        video_sink = Gst.ElementFactory.make("gtk4paintablesink", "gtk4paintablesink")
-        if not video_sink:
-            raise RuntimeError("Failed to create gtk4paintablesink element")
-
-        self.playbin.set_property("video-sink", video_sink)
-        self.pipeline.add(self.playbin)
-
-        # Link video widget to sink
-        paintable = video_sink.get_property("paintable")
-        self.video_widget.set_paintable(paintable)
 
         # Add both main menu and overlay to stack
         self.stack.add_named(main_box, "menu")
@@ -161,35 +137,8 @@ class MainWindow(Gtk.ApplicationWindow):
                     position = 0
 
                 # Start playing the video
-                full_path = os.path.abspath(file_item.full_path)
-                self.playbin.set_property("uri", f"file://{full_path}")
-
                 track = file_item.load_attribute("subtitle_track", -2)
-
+                self.video_player.play(file_item.full_path, position, track)
-                if track >= 0:
-                    flags = self.playbin.get_property("flags")
-                    flags |= 0x00000004  # TEXT flag
-                    self.playbin.set_property("flags", flags)
-                    self.playbin.set_property("current-text", track)
-                elif track == -1:
-                    flags = self.playbin.get_property("flags")
-                    flags &= ~0x00000004  # TEXT flag
-                    self.playbin.set_property("flags", flags)
-
-                if position:
-                    # Pause and wait for it to complete.
-                    self.pipeline.set_state(Gst.State.PAUSED)
-                    self.pipeline.get_state(Gst.CLOCK_TIME_NONE)
-
-                    # Seek to saved position.
-                    self.pipeline.seek_simple(
-                        Gst.Format.TIME,
-                        Gst.SeekFlags.FLUSH | Gst.SeekFlags.ACCURATE,
-                        position,
-                    )
-
-                # Now start playing
-                self.pipeline.set_state(Gst.State.PLAYING)
 
                 self.stack.set_visible_child_name("overlay")
                 self.show_overlay_text(f"Playing: {file_item.name}")
@@ -212,6 +161,14 @@ class MainWindow(Gtk.ApplicationWindow):
 
         self.set_child(self.stack)
 
+    @property
+    def now(self) -> float:
+        frame_clock = self.get_frame_clock()
+        if frame_clock is None:
+            return 0
+
+        return frame_clock.get_frame_time() / 1_000_000
+
     @property
     def selection(self) -> FileItem:
         selected_item = self.selection_model.get_selected_item()
@@ -312,11 +269,7 @@ class MainWindow(Gtk.ApplicationWindow):
 
     def _toggle_play_pause(self) -> None:
         """Toggle between play and pause states"""
-        _, state, _ = self.pipeline.get_state(0)
+        self.video_player.toggle_play_pause()
-        if state == Gst.State.PLAYING:
-            self.pipeline.set_state(Gst.State.PAUSED)
-        else:
-            self.pipeline.set_state(Gst.State.PLAYING)
 
     def _on_video_key_pressed(
         self,
@@ -330,50 +283,47 @@ class MainWindow(Gtk.ApplicationWindow):
 
         elif keyval == Gdk.keyval_from_name("Escape"):
             self._save_position()
-            self.pipeline.set_state(Gst.State.NULL)
+            self.video_player.stop()
             self.stack.set_visible_child_name("menu")
             self.list_view.grab_focus()
             return True
 
         elif keyval == Gdk.keyval_from_name("Left"):
-            self._seek_relative(-10)
+            self.video_player.seek_relative(-10)
             return True
 
         elif keyval == Gdk.keyval_from_name("Right"):
-            self._seek_relative(10)
+            self.video_player.seek_relative(10)
-            return True
-
-        elif keyval == Gdk.keyval_from_name("Up"):
-            self._seek_relative(60)
             return True
 
         elif keyval == Gdk.keyval_from_name("Down"):
-            self._seek_relative(-60)
+            self.video_player.seek_relative(-60)
+            return True
+
+        elif keyval == Gdk.keyval_from_name("Up"):
+            self.video_player.seek_relative(60)
             return True
 
         elif keyval == Gdk.keyval_from_name("Home"):
-            self.pipeline.seek_simple(
+            self.video_player.seek_start()
-                Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT, 0
-            )
             return True
 
         elif keyval == Gdk.keyval_from_name("End"):
-            success, duration = self.pipeline.query_duration(Gst.Format.TIME)
+            self.video_player.seek_end()
-            if success:
-                self.pipeline.seek_simple(
-                    Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT, duration
-                )
             return True
 
         elif keyval == Gdk.keyval_from_name("j"):
-            self._cycle_subtitles()
+            msg = self.video_player.cycle_subtitles()
+            self.show_overlay_text(msg)
             return True
 
         return False
 
     def _toggle_watched_status(self) -> None:
         """Toggle watched status for the selected file"""
+
         file_item = self.selection
+
         if file_item.file_type == FileType.DIRECTORY:
             return
 
@@ -423,110 +373,47 @@ class MainWindow(Gtk.ApplicationWindow):
         else:
             return self._on_menu_key_pressed(keyval, keycode, state)
 
-    def _seek_relative(self, offset: int) -> None:
-        """Seek relative to current position by offset seconds"""
-
-        # Query current position
-        success, current = self.pipeline.query_position(Gst.Format.TIME)
-        if not success:
-            return
-
-        # Convert offset to nanoseconds and add to current
-        new_pos = current + (offset * Gst.SECOND)
-
-        # Ensure we don't seek before start
-        if new_pos < 0:
-            new_pos = 0
-
-        # Perform seek
-        self.pipeline.seek_simple(
-            Gst.Format.TIME, Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT, new_pos
-        )
-
-    def _get_subtitle_info(self, track_index: int) -> str:
-        """Get subtitle track info including language if available"""
-
-        # Query the subtitle track's tags
-        caps: Gst.TagList | None = self.playbin.emit("get-text-tags", track_index)
-        if not caps:
-            return str(track_index)
-
-        found, lang = caps.get_string("language-code")
-        return f"{track_index} ({lang})" if found else str(track_index)
-
     def _save_position(self) -> None:
         """Save current playback position as xattr"""
+
+        if not self.video_player.is_playing:
+            return
+
         file_item = self.selection
-        assert file_item.file_type != FileType.DIRECTORY
 
-        success, position = self.pipeline.query_position(Gst.Format.TIME)
+        if file_item.file_type == FileType.DIRECTORY:
-        success2, duration = self.pipeline.query_duration(Gst.Format.TIME)
+            return
 
-        if success and success2:
+        position = self.video_player.get_position()
+        if position is not None:
             file_item.save_attribute("position", position)
+
+        duration = self.video_player.get_duration()
+        if duration is not None:
             file_item.save_attribute("duration", duration)
 
-    def _cycle_subtitles(self) -> None:
-        """Cycle through available subtitle tracks, including off state"""
-
-        # Get current flags and subtitle track
-        flags = self.playbin.get_property("flags")
-        current = self.playbin.get_property("current-text")
-        n_text = self.playbin.get_property("n-text")
-
-        if n_text == 0:
-            self.show_overlay_text("No subtitles available")
-            return
-
-        # If subtitles are disabled, enable them and set to first track
-        if not (flags & 0x00000004):  # TEXT flag
-            flags |= 0x00000004
-            self.playbin.set_property("flags", flags)
-            self.playbin.set_property("current-text", 0)
-            track_info = self._get_subtitle_info(0)
-            self.show_overlay_text(f"Subtitle track: {track_info}")
-            file_item = self.selection
-            assert file_item.file_type != FileType.DIRECTORY
-            file_item.save_attribute("subtitle_track", 0)
-            return
-
-        # If we're on the last track, disable subtitles
-        if current >= n_text - 1:
-            flags &= ~0x00000004  # TEXT flag
-            self.playbin.set_property("flags", flags)
-            self.show_overlay_text("Subtitles: Off")
-            file_item = self.selection
-            assert file_item.file_type != FileType.DIRECTORY
-            file_item.save_attribute("subtitle_track", -1)
-            return
-
-        # Otherwise cycle to next track
-        next_track = current + 1
-        self.playbin.set_property("current-text", next_track)
-        track_info = self._get_subtitle_info(next_track)
-        self.show_overlay_text(f"Subtitle track: {track_info}")
-        file_item = self.selection
-        assert file_item.file_type != FileType.DIRECTORY
-        file_item.save_attribute("subtitle_track", next_track)
-
     def show_overlay_text(self, text: str, timeout_seconds: float = 1.0) -> None:
         """Show text in a centered overlay that disappears after timeout"""
         self.overlay_label.set_text(text)
         self.overlay_label.set_visible(True)
 
         # Set absolute time when overlay should hide
-        frame_clock = self.get_frame_clock()
+        self.overlay_hide_time = self.now + timeout_seconds
-        if frame_clock is None:
-            return
 
-        frame_time = frame_clock.get_frame_time() / 1_000_000  # Convert to seconds
+    def _on_tick(self, widget: Gtk.Widget, frame_clock: Gdk.FrameClock, data: Any) -> bool:
-        self.overlay_hide_time = frame_time + timeout_seconds
+        current_time = frame_clock.get_frame_time() / 1_000_000
+
+        if current_time >= self.overlay_hide_time:
+            self.overlay_label.set_visible(False)
 
         # Save position every 60 seconds
+        frame_time = frame_clock.get_frame_time() / 1_000_000
         if frame_time - self.last_position_save >= 60.0:
             self._save_position()
             self.last_position_save = frame_time
 
+        return True
+
     def _populate_file_list(self) -> None:
         items: list[FileItem] = []
 

lazy_player/video_player.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import gi
+
+gi.require_version("Gtk", "4.0")
+gi.require_version("Gst", "1.0")
+from gi.repository import Gst, Gtk  # NOQA: E402
+
+
+class VideoPlayer:
+    pipeline: Gst.Pipeline
+    playbin: Gst.Element
+    is_playing: bool
+
+    def __init__(self, picture: Gtk.Picture):
+        self.is_playing = False
+
+        self.pipeline = Gst.Pipeline.new("video-player")
+
+        playbin = Gst.ElementFactory.make("playbin", "playbin")
+        if not playbin:
+            raise RuntimeError("Failed to create playbin element")
+
+        self.playbin = playbin
+
+        video_sink = Gst.ElementFactory.make("gtk4paintablesink", "gtk4paintablesink")
+        if not video_sink:
+            raise RuntimeError("Failed to create gtk4paintablesink element")
+
+        self.playbin.set_property("video-sink", video_sink)
+        self.pipeline.add(self.playbin)
+
+        # Link picture to sink
+        paintable = video_sink.get_property("paintable")
+        picture.set_paintable(paintable)
+
+    def play(self, file_path: Path | str, position: int = 0, subtitle_track: int = -2) -> None:
+        """Start playing a video file"""
+
+        if isinstance(file_path, Path):
+            file_path = os.path.abspath(file_path)
+
+        self.playbin.set_property("uri", f"file://{file_path}")
+
+        if subtitle_track >= 0:
+            flags = self.playbin.get_property("flags")
+            flags |= 0x00000004  # TEXT flag
+            self.playbin.set_property("flags", flags)
+            self.playbin.set_property("current-text", subtitle_track)
+        elif subtitle_track == -1:
+            flags = self.playbin.get_property("flags")
+            flags &= ~0x00000004  # TEXT flag
+            self.playbin.set_property("flags", flags)
+
+        if position:
+            # Pause and wait for it to complete
+            self.pipeline.set_state(Gst.State.PAUSED)
+            self.pipeline.get_state(Gst.CLOCK_TIME_NONE)
+
+            # Seek to saved position
+            self.pipeline.seek_simple(
+                Gst.Format.TIME,
+                Gst.SeekFlags.FLUSH | Gst.SeekFlags.ACCURATE,
+                position,
+            )
+
+        # Start playing
+        self.pipeline.set_state(Gst.State.PLAYING)
+        self.is_playing = True
+
+    def stop(self) -> None:
+        """Stop playback and release resources"""
+        self.pipeline.set_state(Gst.State.NULL)
+        self.is_playing = False
+
+    def toggle_play_pause(self) -> None:
+        """Toggle between play and pause states"""
+        _, state, _ = self.pipeline.get_state(0)
+        if state == Gst.State.PLAYING:
+            self.pipeline.set_state(Gst.State.PAUSED)
+        else:
+            self.pipeline.set_state(Gst.State.PLAYING)
+
+    def seek_relative(self, offset: float) -> None:
+        """Seek relative to current position by offset seconds"""
+        success, current = self.pipeline.query_position(Gst.Format.TIME)
+        if not success:
+            return
+
+        new_pos = current + int(offset * Gst.SECOND)
+        if new_pos < 0:
+            new_pos = 0
+
+        self.pipeline.seek_simple(
+            Gst.Format.TIME,
+            Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT,
+            new_pos,
+        )
+
+    def seek_start(self):
+        """Seek to the start of the video."""
+        self.pipeline.seek_simple(
+            Gst.Format.TIME,
+            Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT,
+            0,
+        )
+
+    def seek_end(self):
+        """Seek to the end of the video."""
+        duration = self.get_duration()
+        if duration:
+            self.pipeline.seek_simple(
+                Gst.Format.TIME,
+                Gst.SeekFlags.FLUSH | Gst.SeekFlags.KEY_UNIT,
+                duration,
+            )
+
+    def get_position(self) -> int | None:
+        """Get current playback position in nanoseconds"""
+        success, position = self.pipeline.query_position(Gst.Format.TIME)
+        return position if success else None
+
+    def get_duration(self) -> int | None:
+        """Get total duration in nanoseconds"""
+        success, duration = self.pipeline.query_duration(Gst.Format.TIME)
+        return duration if success else None
+
+    def cycle_subtitles(self) -> str:
+        """Cycle through available subtitle tracks, including off state"""
+
+        flags = self.playbin.get_property("flags")
+        current = self.playbin.get_property("current-text")
+        n_text = self.playbin.get_property("n-text")
+
+        if n_text == 0:
+            return "No subtitles available"
+
+        if not (flags & 0x00000004):  # TEXT flag
+            flags |= 0x00000004
+            self.playbin.set_property("flags", flags)
+            self.playbin.set_property("current-text", 0)
+            return f"Subtitle track: {self._get_subtitle_info(0)}"
+
+        if current >= n_text - 1:
+            flags &= ~0x00000004  # TEXT flag
+            self.playbin.set_property("flags", flags)
+            return "Subtitles: Off"
+
+        next_track = current + 1
+        self.playbin.set_property("current-text", next_track)
+        return f"Subtitle track: {self._get_subtitle_info(next_track)}"
+
+    def _get_subtitle_info(self, track_index: int) -> str:
+        """Get subtitle track info including language if available"""
+        caps: Gst.TagList | None = self.playbin.emit("get-text-tags", track_index)
+        if not caps:
+            return str(track_index)
+
+        found, lang = caps.get_string("language-code")
+        return f"{track_index} ({lang})" if found else str(track_index)