From 3ec4c6229da1a3374efda92b1143514f14586de3 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Sun, 29 Jun 2025 12:23:14 +0545 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=A7=20refactor(config,=20notification?= =?UTF-8?q?=5Futils):=20remove=20music=5Fplayer=20visibility=20management?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/cavalcade/cava.ini | 36 -- config/data.py | 2 - config/hypr/colors.conf | 34 +- config/settings/constants.py | 1 - config/settings/gui.py | 24 +- install.sh | 1 - memory_monitor.py | 1 - modules/dock/components/__init__.py | 24 +- modules/dock/components/music_player.py | 577 ----------------------- modules/dock/components/notifications.py | 133 +----- modules/notification_popup.py | 246 +--------- styles/components.css | 180 ------- utils/audio_visualizer.py | 342 -------------- utils/notification_utils.py | 382 +++++++++++++++ 14 files changed, 450 insertions(+), 1533 deletions(-) delete mode 100644 config/cavalcade/cava.ini delete mode 100644 modules/dock/components/music_player.py delete mode 100644 utils/audio_visualizer.py create mode 100644 utils/notification_utils.py diff --git a/config/cavalcade/cava.ini b/config/cavalcade/cava.ini deleted file mode 100644 index 63c0d55..0000000 --- a/config/cavalcade/cava.ini +++ /dev/null @@ -1,36 +0,0 @@ -[general] -bars = 24 - -sensitivity = 100 -framerate = 60 -lower_cutoff_freq = 50 -higher_cutoff_freq = 8000 -autosens = 1 - -# these parameters should not be changed -# bar width and spacing goes wrong with raw output -bar_width = 1 -bar_spacing = 0 - -[output] -method = raw -raw_target = /tmp/cava.fifo -bit_format = 16bit -channels = stereo - -[smoothing] -gravity = 200 -integral = 70 -ignore = 0 -monstercat = 1 - -[eq] -1 = 1.00 -2 = 1.00 -3 = 1.00 -4 = 1.00 -5 = 1.00 -6 = 1.00 -7 = 1.00 -8 = 1.00 -9 = 1.00 diff --git a/config/data.py b/config/data.py index f1aa9b9..d4c5e8e 100644 --- a/config/data.py +++ b/config/data.py @@ -83,7 +83,6 @@ if os.path.exists(CONFIG_FILE): "notifications": config.get("dock_notifications_visible", True), "systray": config.get("dock_tray_visible", True), "applications": config.get("dock_applications_visible", True), - "music_player": config.get("dock_music_player_visible", True), "language": config.get("dock_language_visible", True), } @@ -116,7 +115,6 @@ else: "notifications": True, "systray": True, "applications": True, - "music_player": True, "language": True, } diff --git a/config/hypr/colors.conf b/config/hypr/colors.conf index 6269b38..ff92bcb 100644 --- a/config/hypr/colors.conf +++ b/config/hypr/colors.conf @@ -1,27 +1,27 @@ -$wallpaper = /home/sankalp/Pictures/wallpapers/97.jpg -$background = 101417 -$foreground = dfe3e7 +$wallpaper = /home/sankalp/Pictures/wallpapers/-13.webp +$background = 191114 +$foreground = eedfe3 -$primary = 93cdf6 -$secondary = b6c9d8 -$tertiary = cdc0e9 -$surface = 101417 -$surface_bright = 353a3d -$outline = 8b9198 +$primary = fdb0d4 +$secondary = e0bdcb +$tertiary = f2bb9a +$surface = 191114 +$surface_bright = 40373a +$outline = 9d8d92 $error = ff5945 $shadow = 000000 $red = ffb2b9 -$green = 95d5a7 -$yellow = b8cf84 -$blue = afc6ff -$magenta = e4b7f3 +$green = b7d085 +$yellow = dec56e +$blue = cebdfe +$magenta = f7b1de $cyan = 82d3e2 $white = 82d3e0 $red_dim = ff4c5c -$green_dim = 4cb86a -$yellow_dim = 8baa43 -$blue_dim = 497dff -$magenta_dim = c360e4 +$green_dim = 89ad42 +$yellow_dim = bb9b2b +$blue_dim = 8359fc +$magenta_dim = ee54b7 $cyan_dim = 30b5ce diff --git a/config/settings/constants.py b/config/settings/constants.py index 52f57af..1e2214b 100644 --- a/config/settings/constants.py +++ b/config/settings/constants.py @@ -61,7 +61,6 @@ DEFAULTS = { "workspace_nums": False, "workspace_use_chinese_numerals": False, "workspace_dots": False, - "dock_music_player": True, "window_switcher_items_per_row": 13, "metrics_visible": { "cpu": True, diff --git a/config/settings/gui.py b/config/settings/gui.py index ddfaea7..1660fe2 100644 --- a/config/settings/gui.py +++ b/config/settings/gui.py @@ -40,14 +40,14 @@ class HyprConfGUI(Window): root_box = Box(orientation="v", spacing=10, style="margin: 10px;") self.add(root_box) - main_content_box = Box(orientation="h", spacing=6, v_expand=True, h_expand=True) + main_content_box = Box(orientation="h", spacing=6, v_expand=False, h_expand=False) root_box.add(main_content_box) self.tab_stack = Stack( transition_type="slide-up-down", transition_duration=250, - v_expand=True, - h_expand=True, + v_expand=False, + h_expand=False, ) self.key_bindings_tab_content = self.create_key_bindings_tab() @@ -79,15 +79,24 @@ class HyprConfGUI(Window): button_box.add(accept_btn) root_box.add(button_box) + # Force window size after all content is added + self.connect("realize", self._force_window_size) + + def _force_window_size(self, widget): + """Force the window to respect the declared size""" + self.resize(640, 640) + self.set_size_request(640, 640) + def create_key_bindings_tab(self): scrolled_window = ScrolledWindow( h_scrollbar_policy="never", v_scrollbar_policy="automatic", - h_expand=True, - v_expand=True, - propagate_width=False, - propagate_height=False, + h_expand=False, + v_expand=False, + propagate_width=True, + propagate_height=True, ) + scrolled_window.set_size_request(620, 580) # Set explicit size for content area main_vbox = Box(orientation="v", spacing=10, style="margin: 15px;") scrolled_window.add(main_vbox) @@ -516,7 +525,6 @@ class HyprConfGUI(Window): "indicators": "Indicators", "applications": "Taskbar", "language": "Language Indicator", - "music_player": "Music Player", } self.corners_switch = Gtk.Switch(active=bind_vars.get("corners_visible", True)) diff --git a/install.sh b/install.sh index f350eaa..062dc56 100755 --- a/install.sh +++ b/install.sh @@ -22,7 +22,6 @@ INSTALL_DIR="$HOME/.config/Modus" PACKAGES=( fabric-cli-git - cava cliphist gnome-bluetooth-3.0 gobject-introspection diff --git a/memory_monitor.py b/memory_monitor.py index 0354d2b..9c5c8f6 100755 --- a/memory_monitor.py +++ b/memory_monitor.py @@ -48,7 +48,6 @@ class ModusMemoryMonitor: 'dock_controls': ['controls'], 'dock_indicators': ['indicators'], 'dock_metrics': ['metrics'], - 'dock_music_player': ['music_player', 'music'], 'dock_notifications': ['dock_notifications'], 'dock_workspaces': ['workspaces'], diff --git a/modules/dock/components/__init__.py b/modules/dock/components/__init__.py index b8a94f3..20c9a1a 100644 --- a/modules/dock/components/__init__.py +++ b/modules/dock/components/__init__.py @@ -18,7 +18,6 @@ from .battery import Battery from .controls import Controls from .indicators import Indicators from .metrics import Metrics -from .music_player import MusicPlayer from .workspaces import workspace import gi @@ -111,7 +110,6 @@ class DockComponents(Box): ) self.controls = Controls() self.indicators = Indicators() - self.music_player = MusicPlayer() self.systray = SystemTray( icon_size=18, spacing=4, @@ -127,7 +125,6 @@ class DockComponents(Box): self.applications, self.indicators, self.battery, - self.music_player, self.date_time, self.language, self.systray, @@ -158,7 +155,6 @@ class DockComponents(Box): "date_time": self.date_time, "controls": self.controls, "indicators": self.indicators, - "music_player": self.music_player, "systray": self.systray, "applications": self.applications, "language": self.language, @@ -174,11 +170,6 @@ class DockComponents(Box): widget._update_visibility_based_on_content() else: widget.set_visible(visibility) - # Special handling for music player - let it manage its own visibility based on media state - elif component_name == "music_player": - # Music player manages its own visibility based on config and media state - # Don't override its visibility here, let it handle it internally - pass # Special handling for battery - let it manage its own visibility based on config and battery state elif component_name == "battery": # Battery component manages its own visibility based on config and battery presence @@ -195,7 +186,6 @@ class DockComponents(Box): "date_time": self.date_time, "controls": self.controls, "indicators": self.indicators, - "music_player": self.music_player, "systray": self.systray, "applications": self.applications, "language": self.language, @@ -206,17 +196,9 @@ class DockComponents(Box): component_name ] - # Special handling for music player - update the global config and let it handle its own visibility - if component_name == "music_player": - # Update the global config data - import config.data as data - data.DOCK_COMPONENTS_VISIBILITY[component_name] = self.component_visibility[component_name] - # Let the music player update its own visibility based on the new config - components[component_name]._update_display() - else: - components[component_name].set_visible( - self.component_visibility[component_name] - ) + components[component_name].set_visible( + self.component_visibility[component_name] + ) config_file = get_relative_path("../../../config/assets/config.json") if os.path.exists(config_file): diff --git a/modules/dock/components/music_player.py b/modules/dock/components/music_player.py deleted file mode 100644 index 9f631db..0000000 --- a/modules/dock/components/music_player.py +++ /dev/null @@ -1,577 +0,0 @@ -import os -import tempfile -import urllib.parse - -import requests -from fabric.widgets.box import Box -from fabric.widgets.button import Button -from fabric.widgets.centerbox import CenterBox -from fabric.widgets.eventbox import EventBox -from fabric.widgets.label import Label -from fabric.widgets.stack import Stack -from gi.repository import Gdk, GdkPixbuf, GLib - -import config.data as data -import utils.icons as icons -from services.mpris import MediaManager -from utils.audio_visualizer import SpectrumRender -from utils.circle_image import CircleImage - - -class MusicPlayer(EventBox): - def __init__(self, **kwargs): - super().__init__(name="music-player", events=["button-press"], **kwargs) - - orientation = "v" if data.VERTICAL else "h" - - self.main_container = Box( - name="music-player-container", - spacing=4, - orientation=orientation, - ) - - self._media_manager = MediaManager() - self._media_manager.connect("player-appeared", self._on_player_changed) - self._media_manager.connect("player-vanished", self._on_player_changed) - - self._current_player = None - self._manually_selected_player = None - self._update_timeout_id = None - self._current_player_signals = [] - - self._marquee_timeout_id = None - self._marquee_position = 0 - self._original_title = "" - self._title_needs_marquee = False - - self._spin_timeout_id = None - self._spin_angle = 0 - - self._show_visualizer = False - self._create_music_player_view() - self.add(self.main_container) - self.connect("button-press-event", self._on_music_player_clicked) - - # Start hidden - visibility will be managed by _update_display() based on media availability - self.set_visible(False) - - # Ensure visualizer is disabled in vertical mode - if data.VERTICAL: - self._show_visualizer = False - - GLib.timeout_add(100, self._delayed_initial_update) - - def _create_music_player_view(self): - orientation = "v" if data.VERTICAL else "h" - - self.album_thumbnail = CircleImage( - name="music-album-thumbnail", - size=30, - v_align="center", - visible=True, - ) - - self.album_thumbnail.add_events(Gdk.EventMask.BUTTON_PRESS_MASK) - self.album_thumbnail.connect("button-press-event", self._on_thumbnail_clicked) - - # Create track label (hidden in vertical mode) - self.track_label = Label( - name="music-track", - label="No Media", - ellipsize=0, - h_align="center", - ) - - if data.VERTICAL: - self.track_label.set_visible(False) # Hide label in vertical mode - self.track_label.set_size_request(-1, -1) - else: - self.track_label.set_visible(True) - self.track_label.set_size_request(100, -1) - - self.track_label.set_hexpand(False) - self.track_label.set_vexpand(False) - - self.cavalcade = None - self.cavalcade_box = None - - self.center_stack = Stack( - name="music-player-stack", - transition_type="crossfade", - transition_duration=100, - v_align="center", - v_expand=False, - children=[self.track_label], - ) - self.center_stack.set_visible_child(self.track_label) - - self._create_controls() - - self.music_player_box = CenterBox( - name="music-player-centerbox", - orientation=orientation, - h_expand=True, - h_align="fill", - v_align="center", - v_expand=False, - start_children=self.album_thumbnail, - center_children=self.center_stack, - end_children=self.controls_box, - ) - - # Add vertical class for CSS styling - if data.VERTICAL: - self.music_player_box.add_style_class("vertical") - self.main_container.add_style_class("vertical") - - self.main_container.add(self.music_player_box) - - def _load_album_art(self, art_url): - if hasattr(self, "_last_art_url") and self._last_art_url == art_url: - return - - if not art_url: - self._load_fallback_image() - self._last_art_url = None - return - - try: - if art_url.startswith("file://"): - file_path = urllib.parse.unquote(art_url[7:]) - if os.path.exists(file_path): - self.album_thumbnail.set_image_from_file(file_path) - self._show_album_art() - self._last_art_url = art_url - else: - self._load_fallback_image() - self._last_art_url = None - elif art_url.startswith(("http://", "https://")): - response = requests.get(art_url) - if response.status_code == 200: - with tempfile.NamedTemporaryFile( - delete=False, suffix=".jpg" - ) as temp_file: - temp_file.write(response.content) - temp_path = temp_file.name - - pixbuf = GdkPixbuf.Pixbuf.new_from_file(temp_path) - self.album_thumbnail.set_image_from_pixbuf(pixbuf) - self._show_album_art() - self._last_art_url = art_url - - os.unlink(temp_path) - else: - self._load_fallback_image() - self._last_art_url = None - else: - self._load_fallback_image() - self._last_art_url = None - except Exception: - self._load_fallback_image() - self._last_art_url = None - - def _load_fallback_image(self): - fallback = os.path.expanduser("~/.current.wall") - - try: - if os.path.exists(fallback): - self.album_thumbnail.set_image_from_file(fallback) - self._show_album_art() - else: - self._hide_album_art() - except Exception: - self._hide_album_art() - - def _show_album_art(self): - self.album_thumbnail.set_visible(True) - self._start_spin_animation() - - def _hide_album_art(self): - self.album_thumbnail.set_visible(False) - - def _start_spin_animation(self): - if not self._spin_timeout_id: - self._spin_timeout_id = GLib.timeout_add( - 100, self._animate_spin - ) # 100ms to save CPU - - def _stop_spin_animation(self): - if self._spin_timeout_id: - GLib.source_remove(self._spin_timeout_id) - self._spin_timeout_id = None - self._spin_angle = 0 - self.album_thumbnail.angle = 0 - - def _animate_spin(self): - if not self._current_player or not self.album_thumbnail.get_visible(): - return False - - if self._current_player.status == "playing": - self._spin_angle = (self._spin_angle + 5) % 360 - self.album_thumbnail.angle = self._spin_angle - - return True - - def _update_player_indicators(self): - available_players = self._media_manager.players - - if len(available_players) <= 1: - for child in self.player_indicators_box.get_children(): - self.player_indicators_box.remove(child) - return - - for child in self.player_indicators_box.get_children(): - self.player_indicators_box.remove(child) - - current_player_name = ( - self._current_player.player_name if self._current_player else "None" - ) - - for player in available_players: - player_name = player.player_name or "Unknown" - player_icon_markup = self._get_player_icon(player_name) - - indicator_button = Button( - name="music-player-indicator-button", - child=Label( - name="music-player-indicator", - markup=player_icon_markup, - ), - can_focus=False, - ) - - is_current_player = player.player_name == current_player_name - tooltip_text = f"Player: {player_name}" - if is_current_player: - tooltip_text += " (Active)" - else: - tooltip_text += " - Click to switch" - indicator_button.set_tooltip_text(tooltip_text) - - if is_current_player: - indicator_button.get_child().set_name("music-player-indicator-active") - else: - indicator_button.get_child().set_name("music-player-indicator") - - indicator_button.connect( - "clicked", self._on_player_indicator_clicked, player - ) - - self.player_indicators_box.add(indicator_button) - - self.player_indicators_box.show_all() - - def _create_controls(self): - self.controls_box = Box( - name="music-controls", - orientation="h" if not data.VERTICAL else "v", - spacing=2, - ) - - self.prev_button = Button( - name="music-prev", - child=Label(markup=icons.prev), - can_focus=False, - ) - self.prev_button.connect("clicked", self._on_previous_clicked) - self.prev_button.set_tooltip_text("Previous Track") - - self.play_pause_button = Button( - name="music-play-pause", - child=Label(name="play-pause-icon", markup=icons.play), - can_focus=False, - ) - self.play_pause_button.connect("clicked", self._on_play_pause_clicked) - self.play_pause_button.set_tooltip_text("Play/Pause") - - self.next_button = Button( - name="music-next", - child=Label(markup=icons.next), - can_focus=False, - ) - self.next_button.connect("clicked", self._on_next_clicked) - self.next_button.set_tooltip_text("Next Track") - - self.player_indicators_box = Box( - name="music-player-indicators", - orientation="h" if not data.VERTICAL else "v", - spacing=2, - ) - - self.controls_box.add(self.prev_button) - self.controls_box.add(self.play_pause_button) - self.controls_box.add(self.next_button) - self.controls_box.add(self.player_indicators_box) - - def _delayed_initial_update(self): - self._update_display() - return False - - def _connect_player_signals(self, player): - if player: - metadata_signal = player.connect( - "metadata_changed", self._on_metadata_changed - ) - status_signal = player.connect( - "playback_status_changed", self._on_playback_status_changed - ) - self._current_player_signals = [metadata_signal, status_signal] - - def _disconnect_player_signals(self): - if self._current_player and self._current_player_signals: - for signal_id in self._current_player_signals: - try: - self._current_player.disconnect(signal_id) - except: - pass # Signal might already be disconnected - self._current_player_signals = [] - - def _on_metadata_changed(self, *_args): - GLib.idle_add(self._update_display) - - def _on_playback_status_changed(self, *_args): - GLib.idle_add(self._update_display) - - def _on_player_changed(self, *_args): - GLib.idle_add(self._update_display) - - def _update_display(self): - previous_player = self._current_player - - if self._manually_selected_player: - manually_selected_available = next( - ( - p - for p in self._media_manager.players - if p.player_name == self._manually_selected_player.player_name - ), - None, - ) - if manually_selected_available: - self._current_player = manually_selected_available - self._manually_selected_player = manually_selected_available - else: - self._manually_selected_player = None - self._current_player = self._media_manager.current_player - else: - self._current_player = self._media_manager.current_player - - if self._current_player != previous_player: - self._disconnect_player_signals() - if self._current_player: - self._connect_player_signals(self._current_player) - - if not self._current_player: - # Hide the component when there's no media player available - self.set_visible(False) - - if self._update_timeout_id: - GLib.source_remove(self._update_timeout_id) - self._update_timeout_id = None - if self._marquee_timeout_id: - GLib.source_remove(self._marquee_timeout_id) - self._marquee_timeout_id = None - self._stop_spin_animation() - return False - - # Only show if enabled in configuration - if data.DOCK_COMPONENTS_VISIBILITY.get("music_player", True): - self.set_visible(True) - else: - self.set_visible(False) - return False - - album_art_url = self._current_player.album_image_url - if album_art_url: - self._load_album_art(album_art_url) - else: - self._load_fallback_image() - - self._update_stack_display() - self._update_player_indicators() - - play_pause_icon = self.play_pause_button.get_child() - if self._current_player.status == "playing": - play_pause_icon.set_markup(icons.pause) - else: - play_pause_icon.set_markup(icons.play) - - # Skip text processing in vertical mode since label is hidden - if not data.VERTICAL: - track_title = self._current_player.track_title or "Unknown Track" - max_display_chars = 12 - - self._original_title = track_title - self._title_needs_marquee = len(track_title) > max_display_chars - - if self._original_title != getattr(self, "_last_original_title", ""): - self._last_original_title = self._original_title - self._marquee_position = 0 - self._start_marquee_animation() - - self._update_marquee_text() - self._update_tooltips() - - self.prev_button.set_sensitive(self._current_player.can_go_previous) - self.next_button.set_sensitive(self._current_player.can_go_next) - self.play_pause_button.set_sensitive(self._current_player.can_pause) - - if self._update_timeout_id: - GLib.source_remove(self._update_timeout_id) - self._update_timeout_id = None - - return True - - def _start_marquee_animation(self): - if self._marquee_timeout_id: - GLib.source_remove(self._marquee_timeout_id) - self._marquee_timeout_id = None - - if self._title_needs_marquee: - self._marquee_timeout_id = GLib.timeout_add(250, self._animate_marquee) - - def _animate_marquee(self): - if not self._current_player: - return False - - self._marquee_position += 1 - title_length = len(self._original_title) - loop_length = title_length + 3 - - if self._marquee_position >= loop_length: - self._marquee_position = 0 - - self._update_marquee_text() - return True - - def _update_marquee_text(self): - if not self._current_player: - return - - # Skip updating label text in vertical mode since it's hidden - if data.VERTICAL: - return - - # Use the same character limit as in _update_display - max_display_chars = 12 - - if self._title_needs_marquee: - extended_text = self._original_title + " " - start_pos = self._marquee_position - end_pos = start_pos + max_display_chars - - if end_pos <= len(extended_text): - display_title = extended_text[start_pos:end_pos] - else: - part1 = extended_text[start_pos:] - part2 = self._original_title[: end_pos - len(extended_text)] - display_title = (part1 + part2)[:max_display_chars] - else: - display_title = self._original_title[:max_display_chars] - - self.track_label.set_label(display_title) - - def _get_player_icon(self, player_name): - if not player_name: - return icons.disc - - player_name_lower = player_name.lower() - - if "firefox" in player_name_lower: - return icons.firefox - elif "spotify" in player_name_lower: - return icons.spotify - elif any( - browser in player_name_lower - for browser in ["chromium", "chrome", "edge", "brave", "opera", "vivaldi"] - ): - return icons.chromium - else: - return icons.disc - - def _update_tooltips(self): - if not self._current_player: - return - - full_track_title = self._current_player.track_title or "Unknown Track" - track_artist = self._current_player.track_artist or "Unknown Artist" - player_name = self._current_player.player_name or "Media Player" - - tooltip_text = f"{full_track_title}\nby {track_artist}\n\nPlayer: {player_name}" - - if data.VERTICAL: - # In vertical mode, show tooltip on the main container and album thumbnail - self.set_tooltip_text(tooltip_text) - self.album_thumbnail.set_tooltip_text(tooltip_text) - else: - # In horizontal mode, show tooltip on the track label - self.track_label.set_tooltip_text(tooltip_text) - - def _on_play_pause_clicked(self, _button): - if self._current_player: - self._current_player.play_pause() - - def _on_previous_clicked(self, _button): - if self._current_player: - self._current_player.previous() - - def _on_next_clicked(self, _button): - if self._current_player: - self._current_player.next() - - def _on_player_indicator_clicked(self, _button, player): - if player: - self._manually_selected_player = player - self._update_display() - - def _on_thumbnail_clicked(self, _widget, _event): - if self._current_player and not data.VERTICAL: - # Only toggle visualizer in horizontal mode - self._show_visualizer = not self._show_visualizer - self._update_stack_display() - return True - - def _on_music_player_clicked(self, _widget, _event): - if self._current_player and not data.VERTICAL: - # Only toggle visualizer in horizontal mode - self._show_visualizer = not self._show_visualizer - self._update_stack_display() - return True - - def _create_visualizer_if_needed(self): - if not self.cavalcade: - self.cavalcade = SpectrumRender() - self.cavalcade_box = self.cavalcade.get_spectrum_box() - self.center_stack.add(self.cavalcade_box) - - def _destroy_visualizer(self): - if self.cavalcade_box: - self.center_stack.remove(self.cavalcade_box) - self.cavalcade_box = None - if self.cavalcade and hasattr(self.cavalcade, "cava"): - self.cavalcade.cava.close() - self.cavalcade = None - - def _update_stack_display(self): - # Never show visualizer in vertical mode - if self._show_visualizer and self._current_player and not data.VERTICAL: - self._create_visualizer_if_needed() - if self.cavalcade_box: - self.cavalcade_box.set_visible(True) - self.cavalcade_box.show_all() - self.center_stack.set_visible_child(self.cavalcade_box) - else: - self.center_stack.set_visible_child(self.track_label) - self._destroy_visualizer() - - def __del__(self): - self._disconnect_player_signals() - - if self._update_timeout_id: - GLib.source_remove(self._update_timeout_id) - if self._marquee_timeout_id: - GLib.source_remove(self._marquee_timeout_id) - if self._spin_timeout_id: - GLib.source_remove(self._spin_timeout_id) - - self._destroy_visualizer() diff --git a/modules/dock/components/notifications.py b/modules/dock/components/notifications.py index e82b57a..39e41ac 100644 --- a/modules/dock/components/notifications.py +++ b/modules/dock/components/notifications.py @@ -1,9 +1,7 @@ import json -import locale import os -from datetime import datetime, timedelta +from datetime import datetime -from fabric.utils.helpers import get_relative_path from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox @@ -13,13 +11,22 @@ from gi.repository import GLib, Gtk, Gdk import config.data as data import utils.icons as icons -from modules.notification_popup import NotificationBox, load_scaled_pixbuf, get_shared_notification_history +from modules.notification_popup import NotificationBox from utils.custom_image import CustomImage from utils.wayland import WaylandWindow as Window - -PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" -PERSISTENT_HISTORY_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") -CONFIG_FILE = get_relative_path("../../../config/assets/config.json") +from utils.notification_utils import ( + PERSISTENT_DIR, + PERSISTENT_HISTORY_FILE, + CONFIG_FILE, + load_scaled_pixbuf, + get_shared_notification_history, + cleanup_orphan_cached_images, + get_date_header, + schedule_midnight_update, + compute_time_label, + create_historical_notification_from_data, + save_persistent_history +) class NotificationIndicator(Button): @@ -80,20 +87,7 @@ class NotificationIndicator(Button): self.set_tooltip_text("Notifications") -class HistoricalNotification(object): - def __init__( - self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None - ): - self.id = id - self.app_icon = app_icon - self.summary = summary - self.body = body - self.app_name = app_name - self.timestamp = timestamp - self.cached_image_path = cached_image_path - self.image_pixbuf = None - self.actions = [] - self.cached_scaled_pixbuf = None + class NotificationHistory(Box): @@ -172,7 +166,7 @@ class NotificationHistory(Box): self.add(self.history_header) self.add(self.scrolled_window) self._load_persistent_history() - self._cleanup_orphan_cached_images() + cleanup_orphan_cached_images(self.persistent_notifications) self.schedule_midnight_update() @@ -236,45 +230,8 @@ class NotificationHistory(Box): except Exception as e: print(f"Could not sync DND state: {e}") - def get_ordinal(self, n): - if 11 <= (n % 100) <= 13: - return "th" - else: - return {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th") - - def get_date_header(self, dt): - now = datetime.now() - today = now.date() - date = dt.date() - if date == today: - return "Today" - elif date == today - timedelta(days=1): - return "Yesterday" - else: - original_locale = locale.getlocale(locale.LC_TIME) - try: - locale.setlocale(locale.LC_TIME, ("en_US", "UTF-8")) - except locale.Error: - locale.setlocale(locale.LC_TIME, "C") - try: - day = dt.day - ordinal = self.get_ordinal(day) - month = dt.strftime("%B") - if dt.year == now.year: - result = f"{month} {day}{ordinal}" - else: - result = f"{month} {day}{ordinal}, {dt.year}" - finally: - locale.setlocale(locale.LC_TIME, original_locale) - return result - def schedule_midnight_update(self): - now = datetime.now() - next_midnight = datetime.combine( - now.date() + timedelta(days=1), datetime.min.time() - ) - delta_seconds = (next_midnight - now).total_seconds() - GLib.timeout_add_seconds(int(delta_seconds), self.on_midnight) + schedule_midnight_update(self.on_midnight) def on_midnight(self): self.rebuild_with_separators() @@ -308,7 +265,7 @@ class NotificationHistory(Box): self.containers, key=lambda x: x.arrival_time, reverse=True ): arrival_time = container.arrival_time - date_header = self.get_date_header(arrival_time) + date_header = get_date_header(arrival_time) if date_header != current_date_header: sep = self.create_date_separator(date_header) self.notifications_list.add(sep) @@ -385,62 +342,17 @@ class NotificationHistory(Box): GLib.idle_add(self.update_no_notifications_label_visibility) def _save_persistent_history(self): - try: - with open(PERSISTENT_HISTORY_FILE, "w") as f: - json.dump(self.persistent_notifications, f) - except Exception as e: - print(f"Error saving persistent history: {e}") + save_persistent_history(self.persistent_notifications) def update_no_notifications_label_visibility(self): has_notifications = bool(self.containers) self.no_notifications_box.set_visible(not has_notifications) self.notifications_list.set_visible(has_notifications) - def _cleanup_orphan_cached_images(self): - - if not os.path.exists(PERSISTENT_DIR): - - return - - cached_files = [ - f - for f in os.listdir(PERSISTENT_DIR) - if f.startswith("notification_") and f.endswith(".png") - ] - if not cached_files: - - return - - history_uuids = { - note.get("id") for note in self.persistent_notifications if note.get("id") - } - deleted_count = 0 - for cached_file in cached_files: - try: - uuid_from_filename = cached_file[len("notification_") : -len(".png")] - if uuid_from_filename not in history_uuids: - cache_file_path = os.path.join(PERSISTENT_DIR, cached_file) - os.remove(cache_file_path) - - deleted_count += 1 - - except Exception as e: - print( - f"Error processing cached file {cached_file} during cleanup: {e}" - ) - def _add_historical_notification(self, note): - hist_notif = HistoricalNotification( - id=note.get("id"), - app_icon=note.get("app_icon"), - summary=note.get("summary"), - body=note.get("body"), - app_name=note.get("app_name"), - timestamp=note.get("timestamp"), - cached_image_path=note.get("cached_image_path"), - ) + hist_notif = create_historical_notification_from_data(note) hist_box = NotificationBox(hist_notif, timeout_ms=0) hist_box.uuid = hist_notif.id @@ -462,9 +374,6 @@ class NotificationHistory(Box): arrival = datetime.now() container.arrival_time = arrival - def compute_time_label(arrival_time): - return arrival_time.strftime("%H:%M") - self.hist_time_label = Label( name="notification-timestamp", markup=compute_time_label(container.arrival_time), diff --git a/modules/notification_popup.py b/modules/notification_popup.py index b1913ca..6f34670 100644 --- a/modules/notification_popup.py +++ b/modules/notification_popup.py @@ -1,121 +1,27 @@ -import json -import os import uuid -from datetime import datetime from fabric.notifications.service import Notification, NotificationAction, Notifications -from fabric.utils.helpers import get_relative_path from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.image import Image from fabric.widgets.label import Label from fabric.widgets.revealer import Revealer -from gi.repository import GdkPixbuf, GLib, GObject, Gtk +from gi.repository import GLib, Gtk from loguru import logger import config.data as data import utils.icons as icons from utils.custom_image import CustomImage from utils.wayland import WaylandWindow as Window - -PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" -PERSISTENT_HISTORY_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") -CONFIG_FILE = get_relative_path("../config/assets/config.json") - -# Global shared notification history instance for DND state -_shared_notification_history = None - - -def get_shared_notification_history(): - global _shared_notification_history - if _shared_notification_history is None: - _shared_notification_history = NotificationHistory() - return _shared_notification_history - - -def cache_notification_pixbuf(notification_box): - notification = notification_box.notification - if notification.image_pixbuf: - os.makedirs(PERSISTENT_DIR, exist_ok=True) - cache_file = os.path.join( - PERSISTENT_DIR, f"notification_{notification_box.uuid}.png" - ) - - try: - scaled = notification.image_pixbuf.scale_simple( - 48, 48, GdkPixbuf.InterpType.BILINEAR - ) - scaled.savev(cache_file, "png", [], []) - - return cache_file - except Exception as e: - logger.error(f"Error caching image for notification {notification.id}: {e}") - return None - else: - logger.debug(f"Notification {notification.id} has no image_pixbuf to cache.") - return None - - -def load_scaled_pixbuf(notification_box, width, height): - notification = notification_box.notification - if not hasattr(notification_box, "notification") or notification is None: - return None - - pixbuf = None - if ( - hasattr(notification_box, "cached_image_path") - and notification_box.cached_image_path - and os.path.exists(notification_box.cached_image_path) - ): - try: - logger.debug( - f"Attempting to load cached image from: { - notification_box.cached_image_path - } for notification {notification.id}" - ) - pixbuf = GdkPixbuf.Pixbuf.new_from_file(notification_box.cached_image_path) - if pixbuf: - pixbuf = pixbuf.scale_simple( - width, height, GdkPixbuf.InterpType.BILINEAR - ) - - return pixbuf - except Exception as e: - logger.error( - f"Error loading cached image from { - notification_box.cached_image_path - } for notification {notification.id}: {e}" - ) - logger.warning( - f"Falling back to notification.image_pixbuf for notification { - notification.id - }" - ) - - if notification.image_pixbuf: - pixbuf = notification.image_pixbuf.scale_simple( - width, height, GdkPixbuf.InterpType.BILINEAR - ) - return pixbuf - - return get_app_icon_pixbuf(notification.app_icon, width, height) - - -def get_app_icon_pixbuf(icon_path, width, height): - if not icon_path: - return None - if icon_path.startswith("file://"): - icon_path = icon_path[7:] - if not os.path.exists(icon_path): - logger.warning(f"Icon path does not exist: {icon_path}") - return None - try: - pixbuf = GdkPixbuf.Pixbuf.new_from_file(icon_path) - return pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) - except Exception as e: - logger.error(f"Failed to load or scale icon: {e}") - return None +from utils.notification_utils import ( + LIMITED_APPS, + cache_notification_pixbuf, + load_scaled_pixbuf, + get_shared_notification_history, + NotificationHistory, + delete_cached_image +) class ActionButton(Button): @@ -394,16 +300,9 @@ class NotificationBox(Box): if ( hasattr(self, "cached_image_path") and self.cached_image_path - and os.path.exists(self.cached_image_path) and (not self._is_history or from_history_delete) ): - try: - os.remove(self.cached_image_path) - logger.info(f"Deleted cached image: {self.cached_image_path}") - except Exception as e: - logger.error( - f"Error deleting cached image {self.cached_image_path}: {e}" - ) + delete_cached_image(self.cached_image_path) self._destroyed = True self.stop_timeout() super().destroy() @@ -417,130 +316,7 @@ class NotificationBox(Box): self._container.resume_all_timeouts() -class NotificationHistory(Box): - __gsignals__ = { - "dnd-state-changed": (GObject.SignalFlags.RUN_FIRST, None, (bool,)), - "notification-added": (GObject.SignalFlags.RUN_FIRST, None, ()), - } - - def __init__(self, **kwargs): - super().__init__(name="notification-history", orientation="v", **kwargs) - self.do_not_disturb_enabled = False - self.persistent_notifications = [] - self._load_persistent_history() - self._load_dnd_state() - - def set_do_not_disturb_enabled(self, enabled): - if self.do_not_disturb_enabled != enabled: - self.do_not_disturb_enabled = enabled - self._save_dnd_state() - self.emit("dnd-state-changed", enabled) - - def _load_persistent_history(self): - if not os.path.exists(PERSISTENT_DIR): - os.makedirs(PERSISTENT_DIR, exist_ok=True) - - if os.path.exists(PERSISTENT_HISTORY_FILE): - try: - with open(PERSISTENT_HISTORY_FILE, "r") as f: - self.persistent_notifications = json.load(f) - logger.info( - f"Loaded { - len(self.persistent_notifications) - } notifications from persistent history" - ) - except Exception as e: - logger.error(f"Error loading persistent history: {e}") - self.persistent_notifications = [] - else: - self.persistent_notifications = [] - - def _load_dnd_state(self): - try: - if os.path.exists(CONFIG_FILE): - with open(CONFIG_FILE, "r") as f: - config_data = json.load(f) - self.do_not_disturb_enabled = config_data.get("dnd_enabled", False) - logger.info(f"Loaded DND state: {self.do_not_disturb_enabled}") - except Exception as e: - logger.error(f"Error loading DND state: {e}") - self.do_not_disturb_enabled = False - - def _save_dnd_state(self): - try: - config_data = {} - if os.path.exists(CONFIG_FILE): - with open(CONFIG_FILE, "r") as f: - config_data = json.load(f) - - config_data["dnd_enabled"] = self.do_not_disturb_enabled - - with open(CONFIG_FILE, "w") as f: - json.dump(config_data, f, indent=4) - logger.debug(f"Saved DND state: {self.do_not_disturb_enabled}") - except Exception as e: - logger.error(f"Error saving DND state: {e}") - - def _save_persistent_history(self): - try: - with open(PERSISTENT_HISTORY_FILE, "w") as f: - json.dump(self.persistent_notifications, f, indent=2) - logger.info( - f"Saved { - len(self.persistent_notifications) - } notifications to persistent history" - ) - except Exception as e: - logger.error(f"Error saving persistent history: {e}") - - def add_notification(self, notification_box): - try: - notification = notification_box.notification - - hist_data = { - "id": notification_box.uuid, - "app_icon": getattr( - notification, "app_icon", "dialog-information-symbolic" - ), - "summary": getattr(notification, "summary", "No summary"), - "body": getattr(notification, "body", ""), - "app_name": getattr(notification, "app_name", "Unknown"), - "timestamp": datetime.now().isoformat(), - "cached_image_path": getattr( - notification_box, "cached_image_path", None - ), - } - self.persistent_notifications.append(hist_data) - if len(self.persistent_notifications) > 100: - old_notif = self.persistent_notifications.pop(0) - - self._save_persistent_history() - - self.emit("notification-added") - - except Exception as e: - logger.error(f"Error adding notification to history: {e}") - - def clear_history_for_app(self, app_name): - try: - original_count = len(self.persistent_notifications) - self.persistent_notifications = [ - notif - for notif in self.persistent_notifications - if notif.get("app_name") != app_name - ] - removed_count = original_count - len(self.persistent_notifications) - if removed_count > 0: - self._save_persistent_history() - logger.info( - f"Cleared {removed_count} notifications for app: {app_name}" - ) - except Exception as e: - logger.error(f"Error clearing history for app {app_name}: {e}") - - class NotificationContainer(Box): - LIMITED_APPS = ["Spotify"] def __init__( self, @@ -641,7 +417,7 @@ class NotificationContainer(Box): notification.connect("closed", self.on_notification_closed) app_name = notification.app_name - if app_name in self.LIMITED_APPS: + if app_name in LIMITED_APPS: notification_history_instance.clear_history_for_app(app_name) existing_notification_index = -1 diff --git a/styles/components.css b/styles/components.css index c4c2fed..f408bd6 100644 --- a/styles/components.css +++ b/styles/components.css @@ -177,184 +177,4 @@ menu > menuitem:hover > label { color: var(--shadow); } -/* - * ── music player ────────────────────────────────────────────────────────── -*/ -#music-player { - background-color: var(--surface); - padding: 4px; -} - -#music-player.invert { - border-radius: 12px; -} - -/* Track info styles */ -#music-track { - color: var(--primary); - font-weight: bold; - font-size: 12px; -} - -/* Control buttons base styles */ -#music-prev, -#music-next { - background: none; - border: none; - padding: 4px; -} - -#music-prev:hover, -#music-next:hover { - background-color: var(--surface-bright); -} - -#music-prev:active, -#music-next:active { - background-color: var(--primary); -} - -#music-prev label, -#music-next label { - color: var(--primary); - font-size: 14px; -} - -#music-prev:hover label, -#music-next:hover label { - color: var(--primary); -} - -#music-prev:active label, -#music-next:active label { - color: var(--shadow); -} - -/* Play/Pause button styling */ -#music-play-pause { - background-color: var(--surface); - border-radius: 12px; - padding: 6px; - border: none; -} - -#music-play-pause label { - color: var(--primary); - font-size: 18px; -} - -#music-play-pause:hover { - background-color: var(--surface-bright); -} - -#music-play-pause:hover label { - color: var(--primary); -} - -#music-play-pause:active { - background-color: var(--primary); -} - -#music-play-pause:active label { - color: var(--shadow); -} - -/* Player indicators container */ -#music-player-indicators { - padding: 4px; -} - -/* Player indicator buttons */ -#music-player-indicator-button { - background: none; - border: none; - padding: 2px; - border-radius: 6px; -} - -#music-player-indicator-button:hover { - background-color: var(--surface-bright); -} - -#music-player-indicator-button:active { - background-color: var(--primary); -} - -/* Player indicator labels */ -#music-player-indicator { - color: var(--surface-bright); - font-size: 12px; -} - -#music-player-indicator-active { - color: var(--primary); - font-size: 12px; - font-weight: bold; -} - -#music-player-indicator-button:hover #music-player-indicator { - color: var(--primary); -} - -#music-player-indicator-button:active #music-player-indicator, -#music-player-indicator-button:active #music-player-indicator-active { - color: var(--shadow); -} - -/* Vertical mode specific styles */ -#music-player-container.vertical { - padding: 8px 4px; -} - -#music-player-centerbox.vertical { - padding: 4px 0; -} - -#music-player-centerbox.vertical #music-track { - font-size: 10px; - margin: 4px 0; -} - -#music-player.vertical { - padding: 8px 4px; -} - -/* Adjust controls layout for vertical mode */ -#music-controls.vertical { - padding: 2px; -} - -#music-controls.vertical #music-prev, -#music-controls.vertical #music-next, -#music-controls.vertical #music-play-pause { - padding: 2px; - margin: 1px 0; -} - -#music-controls.vertical #music-play-pause { - padding: 4px; -} - -#music-controls.vertical label { - font-size: 12px; -} - -#music-controls.vertical #music-play-pause label { - font-size: 14px; -} - -/* Player indicators in vertical mode */ -#music-player-indicators.vertical { - padding: 2px; -} - -#music-player-indicators.vertical #music-player-indicator-button { - padding: 1px; - margin: 1px 0; -} - -#music-player-indicators.vertical #music-player-indicator, -#music-player-indicators.vertical #music-player-indicator-active { - font-size: 10px; -} diff --git a/utils/audio_visualizer.py b/utils/audio_visualizer.py deleted file mode 100644 index e473cb4..0000000 --- a/utils/audio_visualizer.py +++ /dev/null @@ -1,342 +0,0 @@ -import configparser -import ctypes -import os -import re -import signal -import struct -import subprocess -from math import pi - -from fabric.utils.helpers import get_relative_path -from fabric.widgets.overlay import Overlay -from gi.repository import Gdk, GLib, Gtk -from loguru import logger - -import config.data as data - - -def get_bars(file_path): - config = configparser.ConfigParser() - config.read(file_path) - return int(config["general"]["bars"]) - - -CAVA_CONFIG = get_relative_path("../config/cavalcade/cava.ini") - -bars = get_bars(CAVA_CONFIG) - - -def set_death_signal(): - """ - Set the death signal of the child process to SIGTERM so that if the parent - process is killed, the child (cava) is automatically terminated. - """ - libc = ctypes.CDLL("libc.so.6") - PR_SET_PDEATHSIG = 1 - libc.prctl(PR_SET_PDEATHSIG, signal.SIGTERM) - - -class Cava: - """ - CAVA wrapper. - Launch cava process with certain settings and read output. - """ - - NONE = 0 - RUNNING = 1 - RESTARTING = 2 - CLOSING = 3 - - def __init__(self, mainapp): - self.bars = bars - self.path = "/tmp/cava.fifo" - - self.cava_config_file = CAVA_CONFIG - self.data_handler = mainapp.draw.update - self.command = ["cava", "-p", self.cava_config_file] - self.state = self.NONE - - self.env = dict(os.environ) - self.env["LC_ALL"] = "en_US.UTF-8" # not sure if it's necessary - - is_16bit = True - self.byte_type, self.byte_size, self.byte_norm = ( - ("H", 2, 65535) if is_16bit else ("B", 1, 255) - ) - - if not os.path.exists(self.path): - os.mkfifo(self.path) - - self.fifo_fd = None - self.fifo_dummy_fd = None - self.io_watch_id = None - - def _run_process(self): - logger.debug("Launching cava process...") - try: - self.process = subprocess.Popen( - self.command, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=self.env, - # Ensure cava gets killed when the parent dies. - preexec_fn=set_death_signal, - ) - logger.debug("cava successfully launched!") - self.state = self.RUNNING - except Exception: - logger.exception("Fail to launch cava") - - def _start_io_reader(self): - logger.debug("Activating GLib IO watch for cava stream handler") - # Open FIFO in non-blocking mode for reading - self.fifo_fd = os.open(self.path, os.O_RDONLY | os.O_NONBLOCK) - # Open dummy write end to prevent getting an EOF on our FIFO - self.fifo_dummy_fd = os.open(self.path, os.O_WRONLY | os.O_NONBLOCK) - self.io_watch_id = GLib.io_add_watch( - self.fifo_fd, GLib.IO_IN, self._io_callback - ) - - def _io_callback(self, source, condition): - chunk = self.byte_size * self.bars # number of bytes for given format - try: - data = os.read(self.fifo_fd, chunk) - except OSError: - # logger.error("Error reading FIFO: {}".format(e)) - return False - - # When no data is read, do not remove the IO watch immediately. - if len(data) < chunk: - # Instead of closing the FIFO, we log a warning and continue. - # logger.warning("Incomplete data packet received (expected {} bytes, got {}). Waiting for more data...".format(chunk, len(data))) - # Returning True keeps the IO watch active. A real EOF will only occur when the writer closes. - return True - - fmt = self.byte_type * self.bars # format string for struct.unpack - sample = [i / self.byte_norm for i in struct.unpack(fmt, data)] - GLib.idle_add(self.data_handler, sample) - return True - - def _on_stop(self): - logger.debug("Cava stream handler deactivated") - if self.state == self.RESTARTING: - self.start() - elif self.state == self.RUNNING: - self.state = self.NONE - logger.error("Cava process was unexpectedly terminated.") - # self.restart() # May cause infinity loop, need more check - - def start(self): - """Launch cava""" - self._start_io_reader() - self._run_process() - - def restart(self): - """Restart cava process""" - if self.state == self.RUNNING: - logger.debug("Restarting cava process (normal mode) ...") - self.state = self.RESTARTING - if self.process.poll() is None: - self.process.kill() - elif self.state == self.NONE: - logger.warning("Restarting cava process (after crash) ...") - self.start() - - def close(self): - """Stop cava process""" - self.state = self.CLOSING - if self.process.poll() is None: - self.process.kill() - if self.io_watch_id: - GLib.source_remove(self.io_watch_id) - if self.fifo_fd: - os.close(self.fifo_fd) - if self.fifo_dummy_fd: - os.close(self.fifo_dummy_fd) - if os.path.exists(self.path): - os.remove(self.path) - - -class AttributeDict(dict): - """Dictionary with keys as attributes. Does nothing but easy reading""" - - def __getattr__(self, attr): - return self.get(attr, 3) - - def __setattr__(self, attr, value): - self[attr] = value - - -class Spectrum: - """Spectrum drawing""" - - def __init__(self): - self.silence_value = 0 - self.audio_sample = [] - self.color = None - - self.area = Gtk.DrawingArea() - self.area.connect("draw", self.redraw) - self.area.add_events(Gdk.EventMask.BUTTON_PRESS_MASK) - - self.sizes = AttributeDict() - self.sizes.area = AttributeDict() - self.sizes.bar = AttributeDict() - - self.silence = 10 - self.max_height = 12 - - self.area.connect("configure-event", self.size_update) - self.color_update() - - def is_silence(self, value): - """Check if volume level critically low during last iterations""" - self.silence_value = 0 if value > 0 else self.silence_value + 1 - return self.silence_value > self.silence - - def update(self, data): - """Audio data processing""" - self.color_update() - self.audio_sample = data - if not self.is_silence(self.audio_sample[0]): - self.area.queue_draw() - elif self.silence_value == (self.silence + 1): - self.audio_sample = [0] * self.sizes.number - self.area.queue_draw() - - def redraw(self, widget, cr): - """Draw spectrum graph""" - cr.set_source_rgba(*self.color) - - if data.VERTICAL: - # Vertical mode: draw bars horizontally (rotated 90 degrees) - dy = 3 - center_x = self.sizes.area.width / 2 # center horizontal of the drawing area - - for i, value in enumerate(self.audio_sample): - height = self.sizes.area.height / self.sizes.number - self.sizes.padding - radius = height / 2 - width = max(self.sizes.bar.width * min(value, 1), self.sizes.zero) / 2 - if width == self.sizes.zero / 2 + 1: - width *= 0.5 - - width = min(width, self.max_height) - - # Draw rectangle and arcs for rounded ends (horizontal bars) - cr.rectangle(center_x - width, dy, width * 2, height) - cr.arc(center_x - width, dy + radius, radius, 0, 2 * pi) - cr.arc(center_x + width, dy + radius, radius, 0, 2 * pi) - - cr.close_path() - dy += height + self.sizes.padding - else: - # Horizontal mode: draw bars vertically (original behavior) - dx = 3 - center_y = self.sizes.area.height / 2 # center vertical of the drawing area - - for i, value in enumerate(self.audio_sample): - width = self.sizes.area.width / self.sizes.number - self.sizes.padding - radius = width / 2 - height = max(self.sizes.bar.height * min(value, 1), self.sizes.zero) / 2 - if height == self.sizes.zero / 2 + 1: - height *= 0.5 - - height = min(height, self.max_height) - - # Draw rectangle and arcs for rounded ends (vertical bars) - cr.rectangle(dx, center_y - height, width, height * 2) - cr.arc(dx + radius, center_y - height, radius, 0, 2 * pi) - cr.arc(dx + radius, center_y + height, radius, 0, 2 * pi) - - cr.close_path() - dx += width + self.sizes.padding - cr.fill() - - def size_update(self, *args): - """Update drawing geometry""" - self.sizes.number = bars - self.sizes.padding = 100 / bars - self.sizes.zero = 0 - - self.sizes.area.width = self.area.get_allocated_width() - self.sizes.area.height = self.area.get_allocated_height() - 2 - - tw = self.sizes.area.width - self.sizes.padding * (self.sizes.number - 1) - self.sizes.bar.width = max(int(tw / self.sizes.number), 1) - self.sizes.bar.height = self.sizes.area.height - - def color_update(self): - """Set drawing color according to current settings by reading primary color from CSS""" - color = "#a5c8ff" # default value - try: - with open(get_relative_path("../styles/colors.css"), "r") as f: - content = f.read() - m = re.search(r"--primary:\s*(#[0-9a-fA-F]{6})", content) - if m: - color = m.group(1) - except Exception as e: - logger.error("Failed to read primary color: {}".format(e)) - red = int(color[1:3], 16) / 255 - green = int(color[3:5], 16) / 255 - blue = int(color[5:7], 16) / 255 - self.color = Gdk.RGBA(red=red, green=green, blue=blue, alpha=1.0) - - -class SpectrumRender: - def __init__(self, mode=None, **kwargs): - self.mode = mode - - self.draw = Spectrum() - self.cava = Cava(self) - self.cava.start() - - def get_spectrum_box(self): - # Get the spectrum box - box = Overlay(name="cavalcade", h_align="center", v_align="center") - box.set_size_request(90, 34) # Better size for music player - box.add_overlay(self.draw.area) - box.show_all() # Ensure everything is visible - return box - - -class AudioVisualizer(Overlay): - """ - Simple AudioVisualizer wrapper around existing spectrum components - for compatibility with the music player component. - """ - - def __init__(self, name=None, width=150, height=24, bars=20, **kwargs): - super().__init__(name=name, **kwargs) - - self.width = width - self.height = height - self.bars = bars - self.is_playing = False - - # Create spectrum renderer - self.spectrum_render = SpectrumRender() - self.spectrum_box = self.spectrum_render.get_spectrum_box() - - # Set size - self.spectrum_box.set_size_request(width, height) - - # Add to overlay - self.add_overlay(self.spectrum_box) - - # Set alignment - self.set_halign(Gtk.Align.CENTER) - self.set_valign(Gtk.Align.CENTER) - - def set_visible(self, visible): - """Override set_visible to control spectrum rendering""" - super().set_visible(visible) - if visible and self.is_playing: - # Ensure spectrum is active when visible and playing - if ( - hasattr(self.spectrum_render, "cava") - and self.spectrum_render.cava.state == 0 - ): # NONE - self.spectrum_render.cava.start() - elif not visible: - # Optionally pause spectrum when not visible to save resources - pass diff --git a/utils/notification_utils.py b/utils/notification_utils.py new file mode 100644 index 0000000..30cbd8c --- /dev/null +++ b/utils/notification_utils.py @@ -0,0 +1,382 @@ +""" +Common notification utilities and shared functionality +""" +import json +import locale +import os +import uuid +from datetime import datetime, timedelta + +from fabric.utils.helpers import get_relative_path +from fabric.widgets.box import Box +from gi.repository import GdkPixbuf, GLib, GObject +from loguru import logger + +import config.data as data + +# Constants +PERSISTENT_DIR = f"/tmp/{data.APP_NAME}/notifications" +PERSISTENT_HISTORY_FILE = os.path.join(PERSISTENT_DIR, "notification_history.json") +CONFIG_FILE = get_relative_path("../config/assets/config.json") +NOTIFICATION_IMAGE_PREFIX = "notification_" +NOTIFICATION_IMAGE_SUFFIX = ".png" +DEFAULT_NOTIFICATION_IMAGE_SIZE = 48 +MAX_PERSISTENT_NOTIFICATIONS = 100 +LIMITED_APPS = ["Spotify"] + +# Global shared notification history instance +_shared_notification_history = None + + +class HistoricalNotification: + """Data structure for historical notification data""" + + def __init__(self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None): + self.id = id + self.app_icon = app_icon + self.summary = summary + self.body = body + self.app_name = app_name + self.timestamp = timestamp + self.cached_image_path = cached_image_path + self.image_pixbuf = None + self.actions = [] + self.cached_scaled_pixbuf = None + + +# Image handling functions +def cache_notification_pixbuf(notification_box): + """Cache notification image to disk and return cache file path""" + notification = notification_box.notification + if notification.image_pixbuf: + os.makedirs(PERSISTENT_DIR, exist_ok=True) + cache_file = os.path.join( + PERSISTENT_DIR, f"{NOTIFICATION_IMAGE_PREFIX}{notification_box.uuid}{NOTIFICATION_IMAGE_SUFFIX}" + ) + + try: + scaled = notification.image_pixbuf.scale_simple( + DEFAULT_NOTIFICATION_IMAGE_SIZE, DEFAULT_NOTIFICATION_IMAGE_SIZE, GdkPixbuf.InterpType.BILINEAR + ) + scaled.savev(cache_file, "png", [], []) + return cache_file + except Exception as e: + logger.error(f"Error caching image for notification {notification.id}: {e}") + return None + else: + logger.debug(f"Notification {notification.id} has no image_pixbuf to cache.") + return None + + +def load_scaled_pixbuf(notification_box, width, height): + """Load and scale pixbuf from notification box""" + notification = notification_box.notification + if not hasattr(notification_box, "notification") or notification is None: + return None + + pixbuf = None + + # Try to load from cached image first + if ( + hasattr(notification_box, "cached_image_path") + and notification_box.cached_image_path + and os.path.exists(notification_box.cached_image_path) + ): + try: + logger.debug( + f"Attempting to load cached image from: {notification_box.cached_image_path} " + f"for notification {notification.id}" + ) + pixbuf = GdkPixbuf.Pixbuf.new_from_file(notification_box.cached_image_path) + if pixbuf: + pixbuf = pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + return pixbuf + except Exception as e: + logger.error( + f"Error loading cached image from {notification_box.cached_image_path} " + f"for notification {notification.id}: {e}" + ) + logger.warning(f"Falling back to notification.image_pixbuf for notification {notification.id}") + + # Fall back to notification image_pixbuf + if notification.image_pixbuf: + pixbuf = notification.image_pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + return pixbuf + + # Fall back to app icon + return get_app_icon_pixbuf(notification.app_icon, width, height) + + +def get_app_icon_pixbuf(icon_path, width, height): + """Load and scale app icon pixbuf""" + if not icon_path: + return None + if icon_path.startswith("file://"): + icon_path = icon_path[7:] + if not os.path.exists(icon_path): + logger.warning(f"Icon path does not exist: {icon_path}") + return None + try: + pixbuf = GdkPixbuf.Pixbuf.new_from_file(icon_path) + return pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + except Exception as e: + logger.error(f"Failed to load or scale icon: {e}") + return None + + +def cleanup_orphan_cached_images(persistent_notifications): + """Clean up cached images that no longer have corresponding notifications""" + if not os.path.exists(PERSISTENT_DIR): + return + + cached_files = [ + f for f in os.listdir(PERSISTENT_DIR) + if f.startswith(NOTIFICATION_IMAGE_PREFIX) and f.endswith(NOTIFICATION_IMAGE_SUFFIX) + ] + if not cached_files: + return + + history_uuids = { + note.get("id") for note in persistent_notifications if note.get("id") + } + for cached_file in cached_files: + try: + uuid_from_filename = cached_file[len(NOTIFICATION_IMAGE_PREFIX):-len(NOTIFICATION_IMAGE_SUFFIX)] + if uuid_from_filename not in history_uuids: + cache_file_path = os.path.join(PERSISTENT_DIR, cached_file) + os.remove(cache_file_path) + except Exception as e: + print(f"Error processing cached file {cached_file} during cleanup: {e}") + + +def delete_cached_image(cached_image_path): + """Delete a specific cached image file""" + if cached_image_path and os.path.exists(cached_image_path): + try: + os.remove(cached_image_path) + logger.info(f"Deleted cached image: {cached_image_path}") + except Exception as e: + logger.error(f"Error deleting cached image {cached_image_path}: {e}") + + +# Time and date utilities +def get_ordinal(n): + """Get ordinal suffix for a number (1st, 2nd, 3rd, etc.)""" + if 11 <= (n % 100) <= 13: + return "th" + else: + return {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th") + + +def get_date_header(dt): + """Get human-readable date header for a datetime""" + now = datetime.now() + today = now.date() + date = dt.date() + + if date == today: + return "Today" + elif date == today - timedelta(days=1): + return "Yesterday" + else: + original_locale = locale.getlocale(locale.LC_TIME) + try: + locale.setlocale(locale.LC_TIME, ("en_US", "UTF-8")) + except locale.Error: + locale.setlocale(locale.LC_TIME, "C") + try: + day = dt.day + ordinal = get_ordinal(day) + month = dt.strftime("%B") + if dt.year == now.year: + result = f"{month} {day}{ordinal}" + else: + result = f"{month} {day}{ordinal}, {dt.year}" + finally: + locale.setlocale(locale.LC_TIME, original_locale) + return result + + +def schedule_midnight_update(callback): + """Schedule a callback to run at midnight""" + now = datetime.now() + next_midnight = datetime.combine( + now.date() + timedelta(days=1), datetime.min.time() + ) + delta_seconds = (next_midnight - now).total_seconds() + return GLib.timeout_add_seconds(int(delta_seconds), callback) + + +def compute_time_label(arrival_time): + """Compute time label for notification timestamp""" + return arrival_time.strftime("%H:%M") + + +# Data creation utilities +def create_notification_data(notification_box): + """Create notification data dictionary from notification box""" + notification = notification_box.notification + + return { + "id": notification_box.uuid, + "app_icon": getattr(notification, "app_icon", "dialog-information-symbolic"), + "summary": getattr(notification, "summary", "No summary"), + "body": getattr(notification, "body", ""), + "app_name": getattr(notification, "app_name", "Unknown"), + "timestamp": datetime.now().isoformat(), + "cached_image_path": getattr(notification_box, "cached_image_path", None), + } + + +def create_historical_notification_from_data(note_data): + """Create HistoricalNotification object from data dictionary""" + return HistoricalNotification( + id=note_data.get("id"), + app_icon=note_data.get("app_icon"), + summary=note_data.get("summary"), + body=note_data.get("body"), + app_name=note_data.get("app_name"), + timestamp=note_data.get("timestamp"), + cached_image_path=note_data.get("cached_image_path"), + ) + + +# Storage utilities +def ensure_persistent_dir(): + """Ensure the persistent directory exists""" + if not os.path.exists(PERSISTENT_DIR): + os.makedirs(PERSISTENT_DIR, exist_ok=True) + + +def load_persistent_history(): + """Load notification history from persistent storage""" + ensure_persistent_dir() + + if os.path.exists(PERSISTENT_HISTORY_FILE): + try: + with open(PERSISTENT_HISTORY_FILE, "r") as f: + notifications = json.load(f) + logger.info(f"Loaded {len(notifications)} notifications from persistent history") + return notifications + except Exception as e: + logger.error(f"Error loading persistent history: {e}") + return [] + else: + return [] + + +def save_persistent_history(notifications): + """Save notification history to persistent storage""" + try: + ensure_persistent_dir() + with open(PERSISTENT_HISTORY_FILE, "w") as f: + json.dump(notifications, f, indent=2) + logger.info(f"Saved {len(notifications)} notifications to persistent history") + except Exception as e: + logger.error(f"Error saving persistent history: {e}") + + +def load_dnd_state(): + """Load Do Not Disturb state from config""" + try: + if os.path.exists(CONFIG_FILE): + with open(CONFIG_FILE, "r") as f: + config_data = json.load(f) + dnd_enabled = config_data.get("dnd_enabled", False) + logger.info(f"Loaded DND state: {dnd_enabled}") + return dnd_enabled + except Exception as e: + logger.error(f"Error loading DND state: {e}") + + return False + + +def save_dnd_state(dnd_enabled): + """Save Do Not Disturb state to config""" + try: + config_data = {} + if os.path.exists(CONFIG_FILE): + with open(CONFIG_FILE, "r") as f: + config_data = json.load(f) + + config_data["dnd_enabled"] = dnd_enabled + + with open(CONFIG_FILE, "w") as f: + json.dump(config_data, f, indent=4) + logger.debug(f"Saved DND state: {dnd_enabled}") + except Exception as e: + logger.error(f"Error saving DND state: {e}") + + +# Shared notification history class +class NotificationHistory(Box): + """Shared notification history with DND state management""" + + __gsignals__ = { + "dnd-state-changed": (GObject.SignalFlags.RUN_FIRST, None, (bool,)), + "notification-added": (GObject.SignalFlags.RUN_FIRST, None, ()), + } + + def __init__(self, **kwargs): + super().__init__(name="notification-history", orientation="v", **kwargs) + self.do_not_disturb_enabled = False + self.persistent_notifications = [] + self._load_persistent_history() + self._load_dnd_state() + + def set_do_not_disturb_enabled(self, enabled): + """Set DND state and emit signal if changed""" + if self.do_not_disturb_enabled != enabled: + self.do_not_disturb_enabled = enabled + save_dnd_state(enabled) + self.emit("dnd-state-changed", enabled) + + def _load_persistent_history(self): + """Load persistent notification history""" + self.persistent_notifications = load_persistent_history() + + def _load_dnd_state(self): + """Load DND state from config""" + self.do_not_disturb_enabled = load_dnd_state() + + def _save_persistent_history(self): + """Save persistent notification history""" + save_persistent_history(self.persistent_notifications) + + def add_notification(self, notification_box): + """Add notification to history""" + try: + hist_data = create_notification_data(notification_box) + self.persistent_notifications.append(hist_data) + + # Remove old notifications if we exceed the limit + if len(self.persistent_notifications) > MAX_PERSISTENT_NOTIFICATIONS: + self.persistent_notifications.pop(0) + + self._save_persistent_history() + self.emit("notification-added") + except Exception as e: + print(f"Error adding notification to history: {e}") + + def clear_history_for_app(self, app_name): + """Clear history for specific app""" + try: + original_count = len(self.persistent_notifications) + self.persistent_notifications = [ + notif for notif in self.persistent_notifications + if notif.get("app_name") != app_name + ] + removed_count = original_count - len(self.persistent_notifications) + if removed_count > 0: + self._save_persistent_history() + logger.info(f"Cleared {removed_count} notifications for app: {app_name}") + except Exception as e: + print(f"Error clearing history for app {app_name}: {e}") + + +def get_shared_notification_history(): + """Get or create the shared notification history instance""" + global _shared_notification_history + if _shared_notification_history is None: + _shared_notification_history = NotificationHistory() + return _shared_notification_history