feat: add notch panel implementation

src/window/panel/notch/__init__.py

@@ -0,0 +1,3 @@
+from window.panel.notch.notch import Notch
+
+__all__ = ["Notch"]

src/window/panel/notch/notch.py

@@ -0,0 +1,197 @@
+from fabric.utils import Gdk, Gtk
+from fabric.widgets.box import Box
+from fabric.widgets.centerbox import CenterBox
+from fabric.widgets.shapes import Corner
+from fabric.widgets.stack import Stack
+from services.screencapture import screen_capture_service
+from window.panel.notch.recording import RecordingIndicator
+from window.panel.notch.player import NotchPlayer
+from window.controlcenter.player import get_shared_mpris_manager
+
+
+class Notch(Box):
+    def __init__(self, **kwargs):
+        super().__init__(
+            name="panel-notch-wrap",
+            orientation="h",
+            spacing=0,
+            v_align="center",
+            h_align="center",
+            h_expand=False,
+            **kwargs,
+        )
+
+        self.recording_indicator = RecordingIndicator()
+        self.recording_indicator.set_hexpand(True)
+        self.recording_indicator.set_halign(Gtk.Align.FILL)
+
+        self.player_widget = NotchPlayer()
+        self.player_widget.set_hexpand(True)
+        self.player_widget.set_halign(Gtk.Align.FILL)
+
+        self._mpris = get_shared_mpris_manager()
+        self._mpris.connect("new-player", self._on_new_player)
+        self._mpris.connect("player-vanish", self._on_player_vanish)
+
+        screen_capture_service.connect("recording-started", self._on_recording_started)
+        screen_capture_service.connect("recording-stopped", self._on_recording_stopped)
+
+        self._last_scroll_index = 0
+        self._playing_services: list[str] = []
+
+        self.notch_stack = Stack(
+            name="panel-notch-stack",
+            v_expand=True,
+            h_expand=True,
+            transition_type="crossfade",
+            transition_duration=200,
+            children=[self.recording_indicator, self.player_widget],
+        )
+
+        self.left_corner = Box(
+            name="panel-notch-corner-left",
+            orientation="v",
+            h_align="start",
+            children=[
+                Corner(
+                    name="panel-notch-corner",
+                    orientation="top-right",
+                    size=20,
+                )
+            ],
+        )
+
+        self.right_corner = Box(
+            name="panel-notch-corner-right",
+            orientation="v",
+            h_align="end",
+            children=[
+                Corner(
+                    name="panel-notch-corner",
+                    orientation="top-left",
+                    size=20,
+                )
+            ],
+        )
+
+        self.notch_box = CenterBox(
+            name="panel-notch",
+            orientation="h",
+            h_align="center",
+            v_align="center",
+            start_children=self.left_corner,
+            center_children=self.notch_stack,
+            end_children=self.right_corner,
+        )
+
+        self.add(self.notch_box)
+
+        self._init_state()
+
+        self.connect("scroll-event", self._on_scroll)
+
+    def _init_state(self):
+        """One-time init: check current state, then rely on signals."""
+        if screen_capture_service.is_recording:
+            self.recording_indicator.start_timer()
+
+        for name, svc in self._mpris.get_all_services().items():
+            if svc.playback_status.lower() == "playing":
+                self._playing_services.append(name)
+            svc.connect("play", self._on_any_play)
+            svc.connect("pause", self._on_any_pause)
+            svc.connect("artwork-change", self._on_any_artwork)
+            svc.connect("track-position", self._on_any_position)
+
+        self._apply_stack_state()
+
+    def _on_new_player(self, manager, name: str, service):
+        service.connect("play", self._on_any_play)
+        service.connect("pause", self._on_any_pause)
+        service.connect("artwork-change", self._on_any_artwork)
+        service.connect("track-position", self._on_any_position)
+
+        if service.playback_status.lower() == "playing":
+            self._playing_services.append(name)
+            self._apply_stack_state()
+
+    def _on_recording_started(self, service, path: str):
+        self.recording_indicator.start_timer()
+        self._apply_stack_state()
+
+    def _on_recording_stopped(self, service, path: str):
+        self.recording_indicator.stop_timer()
+        self._apply_stack_state()
+
+    def _on_any_artwork(self, service, local_path: str):
+        self.player_widget._on_artwork_change(service, local_path)
+
+    def _on_any_position(self, service, pos: float, dur: float):
+        self.player_widget._on_track_position(service, pos, dur)
+
+    def _on_player_vanish(self, manager, name: str):
+        if name in self._playing_services:
+            self._playing_services.remove(name)
+            self._apply_stack_state()
+
+    def _on_any_play(self, service):
+        if service.player_name not in self._playing_services:
+            self._playing_services.append(service.player_name)
+        self._apply_stack_state()
+
+    def _on_any_pause(self, service):
+        if service.player_name in self._playing_services:
+            self._playing_services.remove(service.player_name)
+        self._apply_stack_state()
+
+    def _on_scroll(self, widget, event):
+        if event.direction == Gdk.ScrollDirection.UP:
+            self._last_scroll_index -= 1
+        elif event.direction == Gdk.ScrollDirection.DOWN:
+            self._last_scroll_index += 1
+        else:
+            return
+
+        self._apply_stack_state()
+
+    def _apply_stack_state(self):
+        is_recording = screen_capture_service.is_recording
+        has_playing = len(self._playing_services) > 0
+
+        if has_playing:
+            name = self._playing_services[0]
+            svc = self._mpris.get_player_service(name)
+            self.player_widget.set_service(svc)
+        else:
+            self.player_widget.set_service(None)
+
+        active_pages = []
+        if is_recording:
+            active_pages.append(0)
+        if has_playing:
+            active_pages.append(1)
+
+        if not active_pages:
+            if self.get_visible():
+                self.set_visible(False)
+            return
+
+        if not self.get_visible():
+            self.set_visible(True)
+
+        if len(active_pages) == 1:
+            target = active_pages[0]
+        else:
+            idx = self._last_scroll_index % len(active_pages)
+            target = active_pages[idx]
+
+        current = self.notch_stack.get_visible_child()
+        desired = self.recording_indicator if target == 0 else self.player_widget
+
+        if current != desired:
+            self.notch_stack.set_visible_child(desired)
+
+    def destroy(self):
+        self.recording_indicator.destroy()
+        self.player_widget.destroy()
+        super().destroy()

src/window/panel/notch/player.py

@@ -0,0 +1,90 @@
+from fabric.utils import os
+from fabric.widgets.box import Box
+from fabric.widgets.centerbox import CenterBox
+from fabric.widgets.label import Label
+
+from services.mpris import PlayerService
+from utils.utils import svg_file
+
+
+def _fmt_time(seconds: float) -> str:
+    secs = int(seconds)
+    if secs < 0:
+        return "0:00"
+    m = secs // 60
+    s = secs % 60
+    return f"{m}:{s:02d}"
+
+
+class NotchPlayer(Box):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        self._service: PlayerService | None = None
+
+        self.fallback_icon = svg_file("music.svg", size=16)
+
+        self.album_art = Box(
+            name="notch-player-art",
+            children=[self.fallback_icon],
+        )
+
+        self.time_label = Label(
+            name="notch-player-time",
+            markup="0:00",
+        )
+
+        self.player_box = CenterBox(
+            orientation="h",
+            h_expand=True,
+            start_children=self.album_art,
+            end_children=self.time_label,
+        )
+
+        self.add(self.player_box)
+
+    def set_service(self, service: PlayerService | None):
+        if self._service is service:
+            return
+
+        self._service = service
+
+        if service is None:
+            self.time_label.set_markup("0:00")
+            self._show_fallback()
+            return
+
+        self._update_from_service()
+
+    def _show_fallback(self):
+        self.album_art.set_style("")
+        self.fallback_icon.set_visible(True)
+
+    def _show_artwork(self, path: str):
+        self.fallback_icon.set_visible(False)
+        self.album_art.set_style(f"background-image:url('{path}')")
+
+    def _update_from_service(self):
+        if self._service is None:
+            return
+
+        cached = self._service.get_artwork()
+        if cached and os.path.exists(cached):
+            self._show_artwork(cached)
+        else:
+            self._show_fallback()
+
+        pos = self._service.position / 1_000_000
+        self.time_label.set_markup(_fmt_time(pos))
+
+    def _on_artwork_change(self, service, local_path: str):
+        if os.path.exists(local_path):
+            self._show_artwork(local_path)
+
+    def _on_track_position(self, service, pos: float, dur: float):
+        if self._service is service:
+            self.time_label.set_markup(_fmt_time(pos))
+
+    def destroy(self):
+        self._service = None
+        super().destroy()

src/window/panel/notch/recording.py

@@ -0,0 +1,118 @@
+from fabric.utils import GLib, os, time, logger
+from fabric.widgets.button import Button
+from fabric.widgets.centerbox import CenterBox
+from fabric.widgets.label import Label
+from fabric.widgets.box import Box
+from services.screencapture import screen_capture_service
+
+from utils.utils import setup_cursor_hover, svg_file
+
+
+class RecordingIndicator(Box):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        self.recording_start_time = None
+        self.timer_timeout_id = None
+
+        self.recording_icon = svg_file("misc/media-record.svg", size=24)
+
+        self.recording_button = Button(
+            name="panel-button",
+            child=self.recording_icon,
+        )
+        self.recording_button.connect("clicked", self.on_stop_recording)
+        try:
+            setup_cursor_hover(self.recording_button, "pointer")
+        except Exception as e:
+            logger.error(f"An error occurred: {e}")
+
+        self.time_label = Label(
+            name="notch-recording-time",
+            markup="00:00",
+            max_width_chars=5,
+            ellipsize="none",
+        )
+
+        self.recording_box = CenterBox(
+            orientation="h",
+            h_expand=True,
+            start_children=self.recording_button,
+            end_children=self.time_label,
+        )
+
+        self.add(self.recording_box)
+
+    def start_timer(self):
+        if self.timer_timeout_id is not None:
+            return
+
+        self.recording_start_time = self._get_recording_start_time()
+        self.update_timer_display()
+        self.timer_timeout_id = GLib.timeout_add(1000, self.update_timer_display)
+
+    def stop_timer(self):
+        self.recording_start_time = None
+
+        if self.timer_timeout_id is not None:
+            GLib.source_remove(self.timer_timeout_id)
+            self.timer_timeout_id = None
+
+    def update_timer_display(self):
+        if self.recording_start_time is None:
+            return False
+
+        try:
+            elapsed_seconds = int(time.time() - self.recording_start_time)
+            minutes = elapsed_seconds // 60
+            seconds = elapsed_seconds % 60
+            time_text = f"{minutes:02d}:{seconds:02d}"
+
+            self.time_label.set_markup(time_text)
+            self.set_tooltip_text(
+                f"Recording in progress ({time_text}) - Click to stop"
+            )
+
+            return True
+        except Exception as e:
+            logger.error(f"[DEBUG] Error updating timer display: {e}")
+            return False
+
+    def _get_recording_start_time(self):
+        wf_file = "/tmp/recording_start_time.txt"
+        gpu_file = "/tmp/gpu_recording_start_time.txt"
+
+        def read_timestamp(path):
+            try:
+                with open(path, "r") as f:
+                    content = f.read().strip()
+                    if content:
+                        t = float(content)
+                        if abs(t - time.time()) <= 3600:
+                            return t
+                    return os.path.getmtime(path)
+            except (OSError, ValueError):
+                return None
+
+        if os.path.exists(wf_file):
+            t = read_timestamp(wf_file)
+            if t:
+                return t
+
+        if os.path.exists(gpu_file):
+            t = read_timestamp(gpu_file)
+            if t:
+                return t
+
+        return time.time()
+
+    def on_stop_recording(self, *args):
+        try:
+            screen_capture_service.stop_recording()
+            self.stop_timer()
+        except Exception:
+            pass
+
+    def destroy(self):
+        self.stop_timer()
+        super().destroy()