From 829693af00b04f78cabcfaf7b7e0a726ec4b4ab2 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Sun, 5 Jul 2026 00:06:07 +0545 Subject: [PATCH] refactor(src/window/panel/notch): simplify recording logic and add idle widget handling --- src/window/panel/notch/notch.py | 120 +++++++++++++++------------- src/window/panel/notch/player.py | 96 +++++++++++++++------- src/window/panel/notch/recording.py | 65 ++++----------- 3 files changed, 148 insertions(+), 133 deletions(-) diff --git a/src/window/panel/notch/notch.py b/src/window/panel/notch/notch.py index 3cafc3a..45fb9e3 100644 --- a/src/window/panel/notch/notch.py +++ b/src/window/panel/notch/notch.py @@ -1,4 +1,4 @@ -from fabric.utils import Gdk, Gtk +from fabric.utils import Gdk, Gtk, GLib from fabric.widgets.box import Box from fabric.widgets.centerbox import CenterBox from fabric.widgets.shapes import Corner @@ -29,6 +29,11 @@ class Notch(Box): self.player_widget.set_hexpand(True) self.player_widget.set_halign(Gtk.Align.FILL) + self.idle_widget = Box( + name="notch-idle", + h_expand=True, + ) + self._mpris = get_shared_mpris_manager() self._mpris.connect("new-player", self._on_new_player) self._mpris.connect("player-vanish", self._on_player_vanish) @@ -37,16 +42,19 @@ class Notch(Box): screen_capture_service.connect("recording-stopped", self._on_recording_stopped) self._last_scroll_index = 0 - self._playing_services: list[str] = [] + self._last_scroll_time = 0 + self._last_recording_check = 0 + self._is_active_recording = False + self._music_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], + transition_type="none", + children=[self.idle_widget, self.recording_indicator, self.player_widget], ) + self._last_stack_page = 0 self.left_corner = Box( name="panel-notch-corner-left", @@ -88,78 +96,83 @@ class Notch(Box): self._init_state() - self.connect("scroll-event", self._on_scroll) + self.notch_stack.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._is_active_recording = screen_capture_service.is_recording + if self._is_active_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) + self._music_services.append(name) 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) + if name not in self._music_services: + self._music_services.append(name) self._apply_stack_state() def _on_recording_started(self, service, path: str): + self._is_active_recording = True self.recording_indicator.start_timer() self._apply_stack_state() def _on_recording_stopped(self, service, path: str): + self._is_active_recording = False 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) + if name in self._music_services: + self._music_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 + now = GLib.get_monotonic_time() + if now - self._last_scroll_time < 100_000: + return True + self._last_scroll_time = now + dy = 0 + if event.direction == Gdk.ScrollDirection.UP: + dy = -1 + elif event.direction == Gdk.ScrollDirection.DOWN: + dy = 1 + elif event.direction == Gdk.ScrollDirection.SMOOTH: + if event.delta_y < -0.1: + dy = -1 + elif event.delta_y > 0.1: + dy = 1 + else: + return True + else: + return True + + self._last_scroll_index += dy self._apply_stack_state() + return True def _apply_stack_state(self): - is_recording = screen_capture_service.is_recording - has_playing = len(self._playing_services) > 0 + is_recording = self._is_active_recording - if has_playing: - name = self._playing_services[0] + now = GLib.get_monotonic_time() + if now - self._last_recording_check > 3_000_000: + self._last_recording_check = now + if self._is_active_recording and not screen_capture_service.is_recording: + self._is_active_recording = False + is_recording = False + self.recording_indicator.stop_timer() + + has_music = len(self._music_services) > 0 + + if has_music: + name = self._music_services[0] svc = self._mpris.get_player_service(name) self.player_widget.set_service(svc) else: @@ -167,28 +180,25 @@ class Notch(Box): active_pages = [] if is_recording: - active_pages.append(0) - if has_playing: active_pages.append(1) + if has_music: + active_pages.append(2) if not active_pages: - if self.get_visible(): - self.set_visible(False) + if self.notch_stack.get_visible_child() != self.idle_widget: + self.notch_stack.set_visible_child(self.idle_widget) return - if not self.get_visible(): - self.set_visible(True) + pages = {1: self.recording_indicator, 2: self.player_widget} if len(active_pages) == 1: target = active_pages[0] else: - idx = self._last_scroll_index % len(active_pages) - target = active_pages[idx] + target = active_pages[self._last_scroll_index % len(active_pages)] - current = self.notch_stack.get_visible_child() - desired = self.recording_indicator if target == 0 else self.player_widget + desired = pages[target] - if current != desired: + if self.notch_stack.get_visible_child() != desired: self.notch_stack.set_visible_child(desired) def destroy(self): diff --git a/src/window/panel/notch/player.py b/src/window/panel/notch/player.py index 6b401d4..cf9995c 100644 --- a/src/window/panel/notch/player.py +++ b/src/window/panel/notch/player.py @@ -1,19 +1,43 @@ -from fabric.utils import os +import random + +from fabric.utils import Gtk, GLib, 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 Visualizer(Box): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._heights = [4, 4, 4] + self.set_valign(Gtk.Align.CENTER) + self.set_vexpand(False) + self.set_size_request(-1, 20) + self.set_style("margin-right: 8px; margin-bottom: 2px; padding: 0 4px;") + self.connect("draw", self._on_draw) + + def set_heights(self, heights): + self._heights = heights + self.queue_draw() + + def _on_draw(self, widget, cr): + w = self.get_allocated_width() + h = self.get_allocated_height() + n = len(self._heights) + if n == 0 or w < 3 or h < 4: + return + + spacing = 3 + bar_w = max(3, (w - spacing * (n - 1)) // n) + cr.set_source_rgb(1, 1, 1) + + for i, bh in enumerate(self._heights): + x = i * (bar_w + spacing) + y = h - bh + cr.rectangle(x, y, bar_w, bh) + cr.fill() class NotchPlayer(Box): @@ -21,24 +45,22 @@ class NotchPlayer(Box): super().__init__(**kwargs) self._service: PlayerService | None = None + self._vis_timeout = None - self.fallback_icon = svg_file("music.svg", size=16) + self.fallback_icon = svg_file("music.svg", size=18) self.album_art = Box( name="notch-player-art", children=[self.fallback_icon], ) - self.time_label = Label( - name="notch-player-time", - markup="0:00", - ) + self.visualizer = Visualizer() self.player_box = CenterBox( orientation="h", h_expand=True, start_children=self.album_art, - end_children=self.time_label, + end_children=self.visualizer, ) self.add(self.player_box) @@ -49,19 +71,45 @@ class NotchPlayer(Box): self._service = service - if service is None: - self.time_label.set_markup("0:00") - self._show_fallback() - return + self._stop_visualizer() + self._show_fallback() - self._update_from_service() + if service is not None: + self._update_from_service() + self._start_visualizer() + + def _start_visualizer(self): + if self._vis_timeout is not None: + return + self._vis_timeout = GLib.timeout_add(200, self._animate_bars) + + def _stop_visualizer(self): + if self._vis_timeout is not None: + GLib.source_remove(self._vis_timeout) + self._vis_timeout = None + + def _animate_bars(self): + if self._service is None: + self._stop_visualizer() + return False + + playing = self._service.playback_status + if playing != "Playing": + self.visualizer.set_heights([4, 4, 4]) + return True + + heights = [random.randint(4, 20) for _ in range(3)] + self.visualizer.set_heights(heights) + return True def _show_fallback(self): + self.album_art.set_size_request(-1, -1) 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_size_request(18, 18) self.album_art.set_style(f"background-image:url('{path}')") def _update_from_service(self): @@ -74,17 +122,11 @@ class NotchPlayer(Box): 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._stop_visualizer() self._service = None super().destroy() diff --git a/src/window/panel/notch/recording.py b/src/window/panel/notch/recording.py index b1f3c5f..a78d844 100644 --- a/src/window/panel/notch/recording.py +++ b/src/window/panel/notch/recording.py @@ -1,4 +1,4 @@ -from fabric.utils import GLib, os, time, logger +from fabric.utils import GLib, time from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.label import Label @@ -22,10 +22,7 @@ class RecordingIndicator(Box): 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}") + setup_cursor_hover(self.recording_button, "pointer") self.time_label = Label( name="notch-recording-time", @@ -47,9 +44,9 @@ class RecordingIndicator(Box): 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) + self.recording_start_time = time.time() + self._update_display() + self.timer_timeout_id = GLib.timeout_add(1000, self._update_display) def stop_timer(self): self.recording_start_time = None @@ -58,53 +55,19 @@ class RecordingIndicator(Box): GLib.source_remove(self.timer_timeout_id) self.timer_timeout_id = None - def update_timer_display(self): + def _update_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}" + elapsed_seconds = int(time.time() - self.recording_start_time) + minutes = elapsed_seconds // 60 + seconds = elapsed_seconds % 60 - 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() + self.time_label.set_markup(f"{minutes:02d}:{seconds:02d}") + self.set_tooltip_text( + f"Recording in progress ({minutes:02d}:{seconds:02d}) - Click to stop" + ) + return True def on_stop_recording(self, *args): try: