refactor(src/window/panel/notch): simplify recording logic and add idle widget handling

This commit is contained in:
sankalp committed 2026-07-05 00:06:07 +05:45
1 parent 4d1f2ce7d1
commit 829693af00
3 files changed
+148 -133

No files matched your search

+65 -55
View File
@@ -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):
+69 -27
View File
@@ -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()
+14 -51
View File
@@ -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: