sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

🔧 refactor(config, notification_utils): remove music_player visibility management

Browse Source
This commit is contained in:
sankalp committed 2025-06-29 12:23:14 +05:45
1 parent 02c148a12a
commit 3ec4c6229d
14 files changed
+450 -1533
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

config/cavalcade/cava.ini
-36
View File
@@ -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
config/data.py
-2
View File
@@ -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,
}
config/hypr/colors.conf
+17 -17
View File
@@ -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
config/settings/constants.py
-1
View File
@@ -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,
config/settings/gui.py
+16 -8
View File
@@ -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))
install.sh
-1
View File
@@ -22,7 +22,6 @@ INSTALL_DIR="$HOME/.config/Modus"
PACKAGES=(
fabric-cli-git
cava
cliphist
gnome-bluetooth-3.0
gobject-introspection
memory_monitor.py
-1
View File
@@ -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'],
modules/dock/components/__init__.py
+3 -21
View File
@@ -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):
modules/dock/components/music_player.py
-577
View File
@@ -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()
modules/dock/components/notifications.py
+21 -112
View File
@@ -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),
modules/notification_popup.py
+11 -235
View File
@@ -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
styles/components.css
-180
View File
@@ -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;
}
utils/audio_visualizer.py
-342
View File
@@ -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
utils/notification_utils.py
+382
View File
@@ -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