Addded Caching System

This commit is contained in:
tr1x_em committed 2025-08-07 09:38:10 +05:30
1 parent 4852fa01e2
commit e7fac41852
3 files changed
+466 -25

No files matched your search

+371 -9
View File
@@ -1,4 +1,6 @@
import os
import hashlib
import time
from fabric.utils import get_relative_path
from gi.repository import Gdk, GdkPixbuf, GLib, Gtk # type: ignore
@@ -25,6 +27,306 @@ from services.modus import notification_service
NOTIFICATION_WIDTH = 360
NOTIFICATION_IMAGE_SIZE = 48
NOTIFICATION_WIDTH = 360
NOTIFICATION_IMAGE_SIZE = 48
# Notification icon cache directory (for small notification and app icons)
NOTIFICATION_ICON_CACHE_DIR = os.path.join(data.CACHE_DIR, "notification_icons")
# Notification image cache directory (for large notification images)
NOTIFICATION_IMAGE_CACHE_DIR = os.path.join(data.CACHE_DIR, "notification_images")
def ensure_notification_cache_dirs():
"""Ensure notification cache directories exist"""
os.makedirs(NOTIFICATION_ICON_CACHE_DIR, exist_ok=True)
os.makedirs(NOTIFICATION_IMAGE_CACHE_DIR, exist_ok=True)
def cleanup_old_cache_files():
"""Clean up old notification cache files (older than 7 days) from icon cache only"""
try:
if not os.path.exists(NOTIFICATION_ICON_CACHE_DIR):
return
current_time = time.time()
week_ago = current_time - (7 * 24 * 60 * 60) # 7 days
for filename in os.listdir(NOTIFICATION_ICON_CACHE_DIR):
filepath = os.path.join(NOTIFICATION_ICON_CACHE_DIR, filename)
try:
if os.path.isfile(filepath):
file_mtime = os.path.getmtime(filepath)
if file_mtime < week_ago:
os.unlink(filepath)
logger.debug(f"Cleaned up old notification icon cache: {filename}")
except Exception as e:
logger.warning(f"Failed to cleanup cache file {filename}: {e}")
except Exception as e:
logger.warning(f"Failed to cleanup notification cache: {e}")
def get_cache_key(source_data, size=None):
"""Generate a cache key from source data and optional size"""
if isinstance(source_data, str):
# For file paths
cache_input = source_data
if size:
cache_input += f"_{size[0]}x{size[1]}"
else:
# For pixbuf data - use hash of pixel data
try:
pixel_data = source_data.get_pixels()
cache_input = hashlib.md5(pixel_data).hexdigest()
if size:
cache_input += f"_{size[0]}x{size[1]}"
except Exception:
# Fallback to timestamp if pixel data fails
cache_input = f"pixbuf_{int(time.time())}"
return hashlib.md5(cache_input.encode()).hexdigest()
def save_pixbuf_to_cache(pixbuf, cache_key, cache_dir):
"""Save a pixbuf to the specified cache directory"""
try:
ensure_notification_cache_dirs()
cache_path = os.path.join(cache_dir, f"{cache_key}.png")
# Don't overwrite existing cache
if os.path.exists(cache_path):
return cache_path
pixbuf.savev(cache_path, "png", [], [])
logger.debug(f"Cached notification icon: {cache_key}")
return cache_path
except Exception as e:
logger.warning(f"Failed to cache notification icon: {e}")
return None
def get_cached_pixbuf(cache_key, fallback_size=(48, 48), cache_dir=None):
"""Get a cached pixbuf or return None if not found"""
if cache_dir is None:
cache_dir = NOTIFICATION_ICON_CACHE_DIR
try:
cache_path = os.path.join(cache_dir, f"{cache_key}.png")
if os.path.exists(cache_path):
logger.debug(f"Using cached notification icon: {cache_key}")
return GdkPixbuf.Pixbuf.new_from_file_at_scale(
cache_path, fallback_size[0], fallback_size[1], True
)
except Exception as e:
logger.warning(f"Failed to load cached notification icon: {e}")
return None
def cache_notification_icon(source, size=(48, 48)):
"""Cache a notification icon from local sources (file, pixbuf) - PERMANENT cache"""
try:
ensure_notification_cache_dirs()
# Handle different source types
if isinstance(source, str):
cache_key = get_cache_key(source, size)
# Check if already cached
cached_pixbuf = get_cached_pixbuf(cache_key, size, NOTIFICATION_ICON_CACHE_DIR)
if cached_pixbuf:
return cached_pixbuf
# Load and cache the icon (local only)
if source.startswith('file://'):
# Local file URL
file_path = source[7:]
pixbuf = load_and_cache_local_icon(file_path, cache_key, size)
elif os.path.exists(source):
# Direct file path
pixbuf = load_and_cache_local_icon(source, cache_key, size)
else:
# Icon name - try to resolve from theme
pixbuf = load_and_cache_theme_icon(source, cache_key, size)
return pixbuf
elif hasattr(source, 'scale_simple'):
# Already a pixbuf
cache_key = get_cache_key(source, size)
# Check if already cached
cached_pixbuf = get_cached_pixbuf(cache_key, size, NOTIFICATION_ICON_CACHE_DIR)
if cached_pixbuf:
return cached_pixbuf
# Scale and cache
scaled_pixbuf = source.scale_simple(size[0], size[1], GdkPixbuf.InterpType.BILINEAR)
save_pixbuf_to_cache(scaled_pixbuf, cache_key, NOTIFICATION_ICON_CACHE_DIR)
return scaled_pixbuf
except Exception as e:
logger.warning(f"Failed to cache notification icon: {e}")
# Return fallback
return get_fallback_notification_icon(size)
def load_and_cache_local_icon(file_path, cache_key, size):
"""Load a local icon file and cache it"""
try:
if os.path.exists(file_path):
pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale(
file_path, size[0], size[1], True
)
save_pixbuf_to_cache(pixbuf, cache_key, NOTIFICATION_ICON_CACHE_DIR)
return pixbuf
except Exception as e:
logger.warning(f"Failed to load local notification icon {file_path}: {e}")
return get_fallback_notification_icon(size)
def load_and_cache_theme_icon(icon_name, cache_key, size):
"""Load an icon from the current theme and cache it"""
try:
icon_theme = Gtk.IconTheme.get_default()
icon_info = icon_theme.lookup_icon(icon_name, min(size), 0)
if icon_info:
pixbuf = icon_info.load_icon()
if pixbuf:
scaled_pixbuf = pixbuf.scale_simple(size[0], size[1], GdkPixbuf.InterpType.BILINEAR)
save_pixbuf_to_cache(scaled_pixbuf, cache_key, NOTIFICATION_ICON_CACHE_DIR)
return scaled_pixbuf
except Exception as e:
logger.warning(f"Failed to load theme notification icon {icon_name}: {e}")
return get_fallback_notification_icon(size)
def get_fallback_notification_icon(size=(48, 48)):
"""Get the fallback notification icon"""
try:
fallback_path = get_relative_path("../../config/assets/icons/notification.png")
return GdkPixbuf.Pixbuf.new_from_file_at_scale(
fallback_path, size[0], size[1], True
)
except Exception as e:
logger.warning(f"Failed to load fallback notification icon: {e}")
# Create a simple colored rectangle as ultimate fallback
try:
return GdkPixbuf.Pixbuf.new(GdkPixbuf.Colorspace.RGB, True, 8, size[0], size[1])
except:
return None
def get_notification_image_cache_key(notification_id, image_pixbuf):
"""Generate a cache key for notification images"""
try:
# Use notification ID and image hash for unique key
if hasattr(image_pixbuf, 'get_pixels'):
pixel_data = image_pixbuf.get_pixels()
image_hash = hashlib.md5(pixel_data).hexdigest()[:8]
else:
image_hash = str(int(time.time()))
return f"notif_{notification_id}_{image_hash}"
except Exception:
return f"notif_{notification_id}_{int(time.time())}"
def cache_notification_image(notification_id, image_pixbuf, size=(64, 64)):
"""Cache a notification image pixbuf - PERMANENT cache until manual clear"""
try:
ensure_notification_cache_dirs()
cache_key = get_notification_image_cache_key(notification_id, image_pixbuf)
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f"{cache_key}.png")
# Don't overwrite existing cache
if os.path.exists(cache_path):
return cache_path, cache_key
# Scale and save the image
scaled_pixbuf = image_pixbuf.scale_simple(size[0], size[1], GdkPixbuf.InterpType.BILINEAR)
scaled_pixbuf.savev(cache_path, "png", [], [])
logger.debug(f"Cached notification image: {cache_key}")
return cache_path, cache_key
except Exception as e:
logger.warning(f"Failed to cache notification image: {e}")
return None, None
def get_cached_notification_image(cache_key):
"""Get a cached notification image or return None if not found"""
try:
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f"{cache_key}.png")
if os.path.exists(cache_path):
logger.debug(f"Using cached notification image: {cache_key}")
return GdkPixbuf.Pixbuf.new_from_file(cache_path)
except Exception as e:
logger.warning(f"Failed to load cached notification image: {e}")
return None
def cleanup_notification_image_cache(cache_key=None):
"""Clean up notification image cache - specific key or all"""
try:
ensure_notification_cache_dirs()
if cache_key:
# Remove specific cached image
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f"{cache_key}.png")
if os.path.exists(cache_path):
os.unlink(cache_path)
logger.debug(f"Cleaned up cached notification image: {cache_key}")
else:
# Remove all cached images
for filename in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR):
if filename.endswith('.png'):
filepath = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, filename)
try:
os.unlink(filepath)
logger.debug(f"Cleaned up cached notification image: {filename}")
except Exception as e:
logger.warning(f"Failed to cleanup cache file {filename}: {e}")
except Exception as e:
logger.warning(f"Failed to cleanup notification image cache: {e}")
def cleanup_notification_specific_caches(app_icon_source=None, notification_image_cache_key=None):
"""Clean up caches specific to a notification (both app icon and notification image)"""
try:
# Clean up notification image cache
if notification_image_cache_key:
cleanup_notification_image_cache(notification_image_cache_key)
# Clean up app icon cache for this specific source
if app_icon_source:
cache_key = get_cache_key(app_icon_source, (24, 24)) # Standard app icon size
cache_path = os.path.join(NOTIFICATION_ICON_CACHE_DIR, f"{cache_key}.png")
if os.path.exists(cache_path):
os.unlink(cache_path)
logger.debug(f"Cleaned up cached app icon: {cache_key}")
# Also clean 35x35 version used in notifications
cache_key_35 = get_cache_key(app_icon_source, (35, 35))
cache_path_35 = os.path.join(NOTIFICATION_ICON_CACHE_DIR, f"{cache_key_35}.png")
if os.path.exists(cache_path_35):
os.unlink(cache_path_35)
logger.debug(f"Cleaned up cached app icon (35x35): {cache_key_35}")
except Exception as e:
logger.warning(f"Failed to cleanup notification specific caches: {e}")
def cleanup_all_notification_caches():
"""Clean up ALL notification caches (icons and images)"""
try:
# Clean icon cache
if os.path.exists(NOTIFICATION_ICON_CACHE_DIR):
for filename in os.listdir(NOTIFICATION_ICON_CACHE_DIR):
if filename.endswith('.png'):
filepath = os.path.join(NOTIFICATION_ICON_CACHE_DIR, filename)
try:
os.unlink(filepath)
logger.debug(f"Cleaned up cached notification icon: {filename}")
except Exception as e:
logger.warning(f"Failed to cleanup icon cache file {filename}: {e}")
# Clean image cache
cleanup_notification_image_cache()
logger.info("Cleaned up all notification caches")
except Exception as e:
logger.warning(f"Failed to cleanup all notification caches: {e}")
# Initialize cache on module load
ensure_notification_cache_dirs()
cleanup_old_cache_files()
def smooth_revealer_animation(revealer: SlideRevealer, duration: int = 350):
"""Configure revealer for macOS-like smooth animation"""
@@ -57,6 +359,8 @@ class ActionButton(Button):
)
def on_clicked(self, *_):
# Mark for cache cleanup when action button is clicked
self.notification_box._should_cleanup_cache = True
self.action.invoke()
self.action.parent.close("dismissed-by-user")
@@ -72,6 +376,9 @@ class NotificationWidget(Box):
self.show_close_button = show_close_button
self.close_button = None
self._is_hovered = False
self.notification_image_cache_key = None # Track cached image for cleanup
self.app_icon_source = notification.app_icon # Track app icon source for cleanup
self._should_cleanup_cache = False # Only cleanup cache on manual dismissal
super().__init__(
size=(NOTIFICATION_WIDTH, -1),
@@ -187,9 +494,7 @@ class NotificationWidget(Box):
name="notification-close-button",
image=CustomImage(icon_name="close-symbolic", icon_size=18),
visible=True, # Initially hidden
on_clicked=lambda *_: self.notification.close(
"dismissed-by-user"
),
on_clicked=lambda *_: self._manual_close(),
),
Box(v_expand=True),
],
@@ -212,32 +517,65 @@ class NotificationWidget(Box):
return False
def get_pixbuf(self, icon_path, width, height):
if icon_path.startswith("file://") and os.path.exists(icon_path):
"""Get pixbuf with caching support"""
try:
# Use the icon caching system
cached_pixbuf = cache_notification_icon(icon_path, (width, height))
if cached_pixbuf:
return cached_pixbuf
except Exception as e:
logger.warning(f"Failed to get cached pixbuf for {icon_path}: {e}")
# Fallback to original method if caching fails
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}")
icon_path = get_relative_path("../../config/assets/icons/notification.png")
return get_fallback_notification_icon((width, height))
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
return get_fallback_notification_icon((width, height))
def _get_notification_pixbuf(self, notification):
"""Safely get notification pixbuf, fallback to app icon if needed"""
"""Safely get notification pixbuf with caching for image notifications"""
try:
if hasattr(notification, 'image_pixbuf') and notification.image_pixbuf:
# Try to get/cache the notification image
notification_id = getattr(notification, 'id', int(time.time()))
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (64, 64)
)
if cache_path and cache_key:
# Store cache key for reference (but don't auto-delete)
self.notification_image_cache_key = cache_key
# Load the cached image
cached_pixbuf = get_cached_notification_image(cache_key)
if cached_pixbuf:
return cached_pixbuf.scale_simple(35, 35, GdkPixbuf.InterpType.BILINEAR)
# Fallback to direct scaling if caching fails
return notification.image_pixbuf.scale_simple(
35, 35, GdkPixbuf.InterpType.BILINEAR
)
except Exception as e:
logger.warning(f"Failed to get notification image: {e}")
# Fallback to app icon
return self.get_pixbuf(notification.app_icon, 35, 35)
# Fallback to cached app icon
try:
cached_app_icon = cache_notification_icon(notification.app_icon, (35, 35))
if cached_app_icon:
return cached_app_icon
except Exception as e:
logger.warning(f"Failed to get cached app icon: {e}")
# Ultimate fallback
return get_fallback_notification_icon((35, 35))
def create_action_buttons(self, notification):
return Box(
@@ -271,8 +609,22 @@ class NotificationWidget(Box):
if not self._is_hovered: # Only resume if not hovered
self.start_timeout()
def _manual_close(self):
"""Handle manual close button click - mark for cache cleanup"""
self._should_cleanup_cache = True
self.notification.close("dismissed-by-user")
def destroy(self):
self.stop_timeout()
# Only clean up caches if this was a manual dismissal
if self._should_cleanup_cache:
cleanup_notification_specific_caches(
app_icon_source=getattr(self, 'app_icon_source', None),
notification_image_cache_key=getattr(self, 'notification_image_cache_key', None)
)
logger.debug(f"Cleaned up caches for manually dismissed notification")
else:
logger.debug(f"Preserved caches for timeout/auto-dismissed notification")
super().destroy()
# @staticmethod
@@ -384,6 +736,8 @@ class NotificationRevealer(SlideRevealer):
self._apply_transform(current_offset, opacity, scale)
if progress >= 1.0:
# Mark notification for cache cleanup on swipe dismissal
self.notif_box._should_cleanup_cache = True
# Trigger the actual notification dismissal immediately
try:
self.notification.close("dismissed-by-user")
@@ -546,6 +900,9 @@ class ModusNoti(Window):
):
self.notification_state = NotificationState.HIDING
# Mark for cache cleanup when forcibly hiding for new notification
self.current_notification.notif_box._should_cleanup_cache = True
# Force close the current notification to trigger hiding animation
try:
self.current_notification.notification.close("dismissed-by-user")
@@ -634,6 +991,11 @@ class ModusNoti(Window):
pass # Ignore errors if notification is already closed
self.notification_queue.clear()
# Also clean current notification if showing
if self.current_notification:
# Mark for cache cleanup when clearing queue
self.current_notification.notif_box._should_cleanup_cache = True
# Cancel any pending transition timer
if self._transition_timer_id:
GLib.source_remove(self._transition_timer_id)
+92 -16
View File
@@ -4,7 +4,7 @@ from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from fabric.widgets.scrolledwindow import ScrolledWindow
from modules.notification.notification import NotificationWidget
from modules.notification.notification import NotificationWidget, cleanup_notification_image_cache, cleanup_all_notification_caches, cleanup_notification_specific_caches
from services.modus import notification_service
from widgets.wayland import WaylandWindow as Window
from fabric.widgets.image import Image
@@ -18,21 +18,79 @@ class NotificationCenterWidget(NotificationWidget):
notification._notification, timeout_ms=0, show_close_button=True, **kwargs
)
def create_close_button(self):
close_button = Button(
def create_content(self, notification):
# Create our custom close button for notification center
from widgets.custom_image import CustomImage
self.close_button = Button(
name="notif-close-button",
child=Image(
icon_name="dialog-cancel-symbolic", name="notif-close-label", size=5
),
image=CustomImage(icon_name="close-symbolic", icon_size=18),
visible=True, # Always visible in notification center
on_clicked=self._on_close_clicked,
)
close_button.connect(
"enter-notify-event", lambda *_: self.hover_button(close_button)
self.close_button.connect(
"enter-notify-event", lambda *_: self.hover_button(self.close_button)
)
close_button.connect(
"leave-notify-event", lambda *_: self.unhover_button(close_button)
self.close_button.connect(
"leave-notify-event", lambda *_: self.unhover_button(self.close_button)
)
# Create the content box manually with our custom close button
from fabric.widgets.box import Box
from fabric.widgets.label import Label
return Box(
name="notification-content",
spacing=8,
children=[
Box(
name="notification-image",
children=CustomImage(
pixbuf=self._get_notification_pixbuf(notification)
),
),
Box(
name="notification-text",
orientation="v",
v_align="center",
children=[
Box(
name="notification-summary-box",
orientation="h",
children=[
Label(
name="notification-summary",
markup=notification.summary.replace("\n", " "),
h_align="start",
ellipsization="end",
),
],
),
(
Label(
markup=notification.body.replace("\n", " "),
h_align="start",
ellipsization="end",
)
if notification.body
else Label(
markup="",
h_align="start",
ellipsization="end",
)
),
],
),
Box(h_expand=True),
Box(
orientation="v",
children=[
self.close_button, # Use our custom close button
Box(v_expand=True),
],
),
],
)
return close_button
# Override to disable the action buttons
def create_action_buttons(self, notification):
@@ -40,11 +98,16 @@ class NotificationCenterWidget(NotificationWidget):
def _on_close_clicked(self, *args):
try:
# Clean up this notification's cached icons AND images
cleanup_notification_specific_caches(
app_icon_source=getattr(self, 'app_icon_source', None),
notification_image_cache_key=getattr(self, 'notification_image_cache_key', None)
)
logger.debug(f"Cleaned up all caches for notification center: {self.notification_id}")
notification_service.remove_cached_notification(self.notification_id)
except Exception as e:
logger.error(f"Error removing notification {self.notification_id}: {e}")
finally:
self.destroy()
# Override to disable timeout functionality
def start_timeout(self):
@@ -162,25 +225,35 @@ class NotificationCenter(Window):
def on_notification_removed(self, service, cached_notification):
try:
# Find and remove the corresponding widget
# Find and remove the corresponding widget with cache cleanup
for child in self.notifications_box.get_children():
if (
hasattr(child, "notification_id")
and child.notification_id == cached_notification.cache_id
):
# Clean up the widget's cached icons AND images before destroying
cleanup_notification_specific_caches(
app_icon_source=getattr(child, 'app_icon_source', None),
notification_image_cache_key=getattr(child, 'notification_image_cache_key', None)
)
logger.debug(f"Cleaned up all caches for removed notification: {child.notification_id}")
child.destroy()
break
logger.debug(
f"Removed notification widget for {cached_notification.app_name}"
f"Removed notification widget and all caches for {cached_notification.app_name}"
)
except Exception as e:
logger.error(f"Error removing notification widget: {e}")
def on_clear_all(self, service):
try:
# Clear all remaining cached notification images AND icons
# Individual notifications clean up their own images, this handles any remaining
cleanup_all_notification_caches()
for child in self.notifications_box.get_children():
child.destroy()
logger.debug("Cleared all notification widgets")
logger.debug("Cleared all notification widgets and remaining cached images")
except Exception as e:
logger.error(f"Error clearing notification widgets: {e}")
@@ -191,6 +264,9 @@ class NotificationCenter(Window):
self.scrolled.set_visible(current_count > 0)
def clear_all_notifications(self, *_):
# Clear all remaining cached notification images AND icons when clear all is clicked
# Individual notifications clean up their own images, this handles any remaining
cleanup_all_notification_caches() # Clear ALL caches (icons + images)
notification_service.clear_all_cached_notifications()
if hasattr(self, "mousecapture"):
self.mousecapture.hide_child_window()
+3
View File
@@ -258,6 +258,9 @@ class CachedNotifications(Notifications):
if handler_id:
# Disconnect the signal handler
cached_notification.disconnect(handler_id)
# Emit signal to notify UI that notification was removed
self.emit("cached-notification-removed", cached_notification)
def toggle_dnd(self):
self.dont_disturb = not self.dont_disturb