diff --git a/src/window/panel/notch/__init__.py b/src/window/panel/notch/__init__.py new file mode 100644 index 0000000..faaa1a2 --- /dev/null +++ b/src/window/panel/notch/__init__.py @@ -0,0 +1,3 @@ +from window.panel.notch.notch import Notch + +__all__ = ["Notch"] diff --git a/src/window/panel/notch/notch.py b/src/window/panel/notch/notch.py new file mode 100644 index 0000000..3cafc3a --- /dev/null +++ b/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() diff --git a/src/window/panel/notch/player.py b/src/window/panel/notch/player.py new file mode 100644 index 0000000..6b401d4 --- /dev/null +++ b/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() diff --git a/src/window/panel/notch/recording.py b/src/window/panel/notch/recording.py new file mode 100644 index 0000000..b1f3c5f --- /dev/null +++ b/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()