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

Caching Fixed

Browse Source
This commit is contained in:
tr1x_em committed 2025-08-14 16:05:29 +05:30
1 parent 662702ae0f
commit 3af502daa8
5 files changed
+322 -429
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

modules/controlcenter/main.py
+1 -1
View File
@@ -460,7 +460,7 @@ class ModusControlCenter(Window):
Box(
name="vol-box",
orientation="horizontal",
spacing=8,
spacing=12,
v_expand=True,
children=[
Box(
modules/notification/notification.py
+115 -286
View File
@@ -1,12 +1,21 @@
import os
import hashlib
import time
import uuid
from fabric.utils import get_relative_path
from gi.repository import Gdk, GdkPixbuf, GLib, Gtk # type: ignore
from loguru import logger
import config.data as data
from .unified_cache import (
get_unified_cache_key,
save_to_cache,
get_from_cache,
cleanup_cache,
get_fallback_icon,
ensure_cache_dir
)
from fabric.notifications import (
Notification,
NotificationAction,
@@ -30,62 +39,75 @@ 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")
# Unified notification cache directory (for both app icons and notification images)
UNIFIED_NOTIFICATION_CACHE_DIR = os.path.join(data.CACHE_DIR, "notifications")
# Backward compatibility constants
NOTIFICATION_ICON_CACHE_DIR = UNIFIED_NOTIFICATION_CACHE_DIR
NOTIFICATION_IMAGE_CACHE_DIR = UNIFIED_NOTIFICATION_CACHE_DIR
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)
"""Ensure unified notification cache directory exists"""
os.makedirs(UNIFIED_NOTIFICATION_CACHE_DIR, exist_ok=True)
def cleanup_old_cache_files():
"""Clean up old notification cache files (older than 7 days) from icon cache only"""
"""Clean up old notification cache files (older than 7 days)"""
try:
if not os.path.exists(NOTIFICATION_ICON_CACHE_DIR):
if not os.path.exists(UNIFIED_NOTIFICATION_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)
for filename in os.listdir(UNIFIED_NOTIFICATION_CACHE_DIR):
filepath = os.path.join(UNIFIED_NOTIFICATION_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}"
)
logger.debug(f"Cleaned up old notification 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()
def get_unified_cache_key(source_data, size=None, app_name=None):
"""Generate a unified cache key that works for both app icons and notification images"""
try:
if hasattr(source_data, "get_pixels"):
# For pixbuf data - use hash of pixel data for deterministic caching
try:
pixel_data = source_data.get_pixels()
image_hash = hashlib.md5(pixel_data).hexdigest()[:8]
return image_hash
except Exception:
# Fallback to timestamp if pixel data fails
return str(int(time.time()))[:8]
elif isinstance(source_data, str):
# For file paths - create hash-based name
if source_data.startswith("file://"):
source_data = source_data[7:]
# Create hash from file path and size
hash_input = source_data
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())}"
hash_input += f"_{size[0]}x{size[1]}"
return hashlib.md5(hash_input.encode()).hexdigest()[:8]
else:
# Fallback to timestamp
return str(int(time.time()))[:8]
except Exception:
# Ultimate fallback
return str(int(time.time()))[:8]
return hashlib.md5(cache_input.encode()).hexdigest()
# Backward compatibility
get_cache_key = get_unified_cache_key
def save_pixbuf_to_cache(pixbuf, cache_key, cache_dir):
@@ -123,18 +145,18 @@ def get_cached_pixbuf(cache_key, fallback_size=(48, 48), cache_dir=None):
return None
def cache_notification_icon(source, size=(48, 48)):
def cache_notification_icon(source, size=(48, 48), app_name=None):
"""Optimized notification icon caching with immediate pixbuf generation and caching"""
try:
ensure_notification_cache_dirs()
# Handle different source types with optimized caching
if isinstance(source, str):
cache_key = get_cache_key(source, size)
cache_key = get_unified_cache_key(source, size, app_name)
# Check cache first for immediate return
cached_pixbuf = get_cached_pixbuf(
cache_key, size, NOTIFICATION_ICON_CACHE_DIR
cache_key, fallback_size=size, cache_dir=NOTIFICATION_ICON_CACHE_DIR
)
if cached_pixbuf:
logger.debug(f"Cache hit for icon: {cache_key}")
@@ -156,11 +178,11 @@ def cache_notification_icon(source, size=(48, 48)):
elif hasattr(source, "scale_simple"):
# Already a pixbuf - cache it directly with optimized flow
cache_key = get_cache_key(source, size)
cache_key = get_unified_cache_key(source, size, app_name)
# Check cache first
cached_pixbuf = get_cached_pixbuf(
cache_key, size, NOTIFICATION_ICON_CACHE_DIR
cache_key, fallback_size=size, cache_dir=NOTIFICATION_ICON_CACHE_DIR
)
if cached_pixbuf:
logger.debug(f"Cache hit for pixbuf: {cache_key}")
@@ -198,23 +220,8 @@ def load_and_cache_local_icon(file_path, cache_key, 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}")
# For simplicity, just use fallback since theme icon loading is complex in GTK4
logger.debug(f"Using fallback for theme icon {icon_name}")
return get_fallback_notification_icon(size)
@@ -236,119 +243,47 @@ def get_fallback_notification_icon(size=(48, 48)):
return None
def find_existing_cached_image(notification_id):
"""Find existing cached image for a notification ID, regardless of hash - prioritizes restart scenarios"""
try:
if not os.path.exists(NOTIFICATION_IMAGE_CACHE_DIR):
logger.debug(
f"Notification image cache directory does not exist: {
NOTIFICATION_IMAGE_CACHE_DIR
}"
)
return None
# Look for any cached image that starts with this notification ID
matching_files = []
for filename in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR):
if filename.startswith(f"notif_{notification_id}_") and filename.endswith(
".png"
):
matching_files.append(filename)
if not matching_files:
logger.debug(
f"No cached images found for notification ID: {notification_id}"
)
return None
# Sort by modification time to get the most recent cache file
matching_files.sort(
key=lambda f: os.path.getmtime(
os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f)
),
reverse=True,
)
# Try to load the most recent cached image
for filename in matching_files:
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, filename)
try:
cached_pixbuf = GdkPixbuf.Pixbuf.new_from_file(cache_path)
if cached_pixbuf:
cache_key = filename[:-4] # Remove .png extension
logger.debug(
f"Found existing cached image for notification {
notification_id
}: {cache_key}"
)
return cached_pixbuf
except Exception as e:
logger.debug(f"Failed to load cached image {filename}: {e}")
continue
logger.debug(
f"Found {len(matching_files)} cache files for notification {
notification_id
} but none could be loaded"
)
return None
except Exception as e:
logger.debug(
f"Failed to find existing cached image for notification {notification_id}: {
e
}"
)
return None
def get_notification_image_cache_key(notification_id, image_pixbuf):
"""Generate a robust cache key for notification images, handling corrupted pixbufs"""
"""Generate a deterministic cache key based on image content to prevent duplicate caching"""
try:
# Primary approach: Use notification ID and image hash for unique key
if hasattr(image_pixbuf, "get_pixels"):
# Use image content hash for deterministic caching
if image_pixbuf and hasattr(image_pixbuf, "get_pixels"):
try:
pixel_data = image_pixbuf.get_pixels()
image_hash = hashlib.md5(pixel_data).hexdigest()[:8]
return f"notif_{notification_id}_{image_hash}"
except Exception as pixel_error:
logger.debug(
f"Failed to get pixel data for cache key (temp file likely gone): {
pixel_error
}"
)
# Fallback to notification ID + dimensions if pixel data fails
return image_hash
except Exception:
# If pixel data fails, use image dimensions + timestamp
try:
width = image_pixbuf.get_width()
height = image_pixbuf.get_height()
dimension_hash = hashlib.md5(
f"{width}x{height}".encode()
).hexdigest()[:8]
return f"notif_{notification_id}_{dimension_hash}"
dimension_hash = hashlib.md5(f"{width}x{height}".encode()).hexdigest()[:8]
return dimension_hash
except Exception:
# Ultimate fallback using just notification ID and timestamp
return f"notif_{notification_id}_{int(time.time())}"
else:
# Not a valid pixbuf, use notification ID + timestamp
return f"notif_{notification_id}_{int(time.time())}"
pass
# Fallback to timestamp for invalid pixbufs
return str(int(time.time()))[:8]
except Exception:
# Absolute fallback
return f"notif_{notification_id}_{int(time.time())}"
# Ultimate fallback
return str(int(time.time()))[:8]
def cache_notification_image(notification_id, image_pixbuf, size=(64, 64)):
"""Optimized notification image caching with robust error handling for persistent notifications"""
"""Smart notification image caching that avoids duplicate caching"""
try:
ensure_notification_cache_dirs()
# Generate deterministic cache key based on image content
cache_key = get_notification_image_cache_key(notification_id, image_pixbuf)
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f"{cache_key}.png")
# Check if already cached to avoid redundant work
if os.path.exists(cache_path):
logger.debug(f"Image cache hit: {cache_key}")
logger.debug(f"Image cache hit - already exists: {cache_key}")
return cache_path, cache_key
# Try to scale and save the image with robust error handling
# Try to scale and save the image
try:
scaled_pixbuf = image_pixbuf.scale_simple(
size[0], size[1], GdkPixbuf.InterpType.BILINEAR
@@ -357,29 +292,7 @@ def cache_notification_image(notification_id, image_pixbuf, size=(64, 64)):
logger.debug(f"Generated and cached notification image: {cache_key}")
return cache_path, cache_key
except Exception as scale_error:
# Handle cases where pixbuf is corrupted or references invalid files
logger.debug(
f"Failed to scale/save pixbuf (likely temp file gone): {scale_error}"
)
# Try to get pixel data directly to create a valid cache key even if file is gone
try:
# If we can't scale but pixbuf exists, create a fallback cache entry
fallback_cache_key = f"fallback_{notification_id}_{int(time.time())}"
fallback_cache_path = os.path.join(
NOTIFICATION_IMAGE_CACHE_DIR, f"{fallback_cache_key}.png"
)
# Create a fallback image based on notification ID
fallback_pixbuf = get_fallback_notification_icon(size)
if fallback_pixbuf:
fallback_pixbuf.savev(fallback_cache_path, "png", [], [])
logger.debug(f"Created fallback cached image: {fallback_cache_key}")
return fallback_cache_path, fallback_cache_key
except Exception as fallback_error:
logger.debug(f"Fallback cache creation failed: {fallback_error}")
logger.debug(f"Failed to cache image (temp file likely gone): {scale_error}")
return None, None
except Exception as e:
@@ -400,47 +313,16 @@ def get_cached_notification_image(cache_key):
def cleanup_notification_image_cache(cache_key=None):
"""Clean up notification image cache - specific key, notification ID pattern, or all"""
"""Clean up notification image cache - specific key or all"""
try:
ensure_notification_cache_dirs()
if cache_key:
# Handle notification ID-based cleanup (e.g., "notif_1_*")
if cache_key.endswith("_*"):
# Extract notification ID from pattern
notification_id_pattern = cache_key[:-2] # Remove "_*"
cleaned_count = 0
for filename in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR):
if filename.startswith(
notification_id_pattern
) and 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}"
)
cleaned_count += 1
except Exception as e:
logger.warning(
f"Failed to cleanup cache file {filename}: {e}"
)
if cleaned_count > 0:
logger.info(
f"Cleaned up {
cleaned_count
} cached images for notification ID pattern: {
notification_id_pattern
}"
)
else:
# 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}")
# 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):
@@ -448,9 +330,7 @@ def cleanup_notification_image_cache(cache_key=None):
filepath = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, filename)
try:
os.unlink(filepath)
logger.debug(
f"Cleaned up cached notification image: {filename}"
)
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:
@@ -468,7 +348,7 @@ def cleanup_notification_specific_caches(
# Clean up app icon cache for this specific source
if app_icon_source:
cache_key = get_cache_key(
cache_key = get_unified_cache_key(
app_icon_source, (24, 24)
) # Standard app icon size
cache_path = os.path.join(NOTIFICATION_ICON_CACHE_DIR, f"{cache_key}.png")
@@ -477,7 +357,7 @@ def cleanup_notification_specific_caches(
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_key_35 = get_unified_cache_key(app_icon_source, (35, 35))
cache_path_35 = os.path.join(
NOTIFICATION_ICON_CACHE_DIR, f"{cache_key_35}.png"
)
@@ -616,7 +496,7 @@ except Exception as e:
def preload_notification_assets(notification):
"""Preload and cache notification assets with robust error handling for persistent notifications"""
"""Preload and cache notification assets with robust error handling for persistent notifications - NO caching for image-pixmap"""
try:
# Preload app icon at common sizes used in notifications
if hasattr(notification, "app_icon") and notification.app_icon:
@@ -632,32 +512,9 @@ def preload_notification_assets(notification):
}"
)
# Preload notification image if present (with resilient handling)
# Cache notification image if available
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
try:
notification_id = getattr(notification, "id", int(time.time()))
# Try to cache the image, but don't fail if temp files are gone
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (35, 35)
)
if cache_path and cache_key:
logger.debug(
f"Preloaded notification image for: {notification.app_name}"
)
else:
logger.debug(
f"Skipped preloading notification image for {
notification.app_name
} (temp file likely gone)"
)
except Exception as image_error:
logger.debug(
f"Failed to preload notification image for {
notification.app_name
}: {image_error}"
)
cache_notification_image(notification.id, notification.image_pixbuf, (35, 35))
except Exception as e:
logger.warning(f"Failed to preload notification assets: {e}")
@@ -891,78 +748,38 @@ class NotificationWidget(Box):
try:
pixbuf = GdkPixbuf.Pixbuf.new_from_file(icon_path)
return pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR)
if pixbuf:
return pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR)
else:
return get_fallback_notification_icon((width, height))
except Exception as e:
logger.error(f"Failed to load or scale icon: {e}")
return get_fallback_notification_icon((width, height))
def _get_notification_pixbuf(self, notification):
"""Optimized notification pixbuf with cached-first approach for restart scenarios"""
"""Simplified notification pixbuf with NO CACHING for image-pixmap to prevent disk bloat"""
notification_id = getattr(notification, "id", int(time.time()))
# Priority 1: ALWAYS check for existing cached image first (especially important for restarts)
if notification_id:
existing_cached_image = find_existing_cached_image(notification_id)
if existing_cached_image:
# Store the cache key for cleanup when manually closed
self.notification_image_cache_key = (
f"notif_{notification_id}_*" # Wildcard for cleanup
)
logger.debug(
f"Using existing cached image for notification: {
notification_id
}, cache_key: {self.notification_image_cache_key}"
)
return existing_cached_image
try:
# Priority 2: Try to process current pixbuf only if we have no cached version
# Try to get cached notification image first
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
try:
cache_key = get_notification_image_cache_key(
notification_id, notification.image_pixbuf
)
cache_key = get_notification_image_cache_key(notification_id, notification.image_pixbuf)
cached_image = get_cached_notification_image(cache_key)
if cached_image:
logger.debug(f"Using cached notification image: {cache_key}")
logger.debug(f"Using cached notification image: {cache_key[:8]}")
return cached_image
# Try to cache the image now
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (35, 35)
)
if cache_path and cache_key:
# Store the cache key for cleanup when manually closed
self.notification_image_cache_key = cache_key
# Load and return the newly cached image
cached_pixbuf = get_cached_notification_image(cache_key)
if cached_pixbuf:
logger.debug(
f"Cached and using notification image: {cache_key}"
)
return cached_pixbuf
# Direct fallback if caching succeeded but loading failed
logger.debug("Using direct scaling for notification image")
return notification.image_pixbuf.scale_simple(
35, 35, GdkPixbuf.InterpType.BILINEAR
)
except Exception as image_error:
logger.debug(
f"Notification image processing failed (temp file likely gone): {image_error}"
)
logger.debug(f"Notification image processing failed: {image_error}")
# Continue to app icon fallback
except Exception as e:
logger.debug(f"Failed to process notification image: {e}")
# Priority 3: Use cached app icon with optimized cache-first approach
# Use cached app icon as fallback
try:
# Generate optimized cache key for app icon
app_icon_source = getattr(notification, "app_icon", None)
if app_icon_source:
# Cache icon immediately and return cached version
cached_app_icon = cache_notification_icon(app_icon_source, (35, 35))
if cached_app_icon:
logger.debug(f"Using cached app icon for: {app_icon_source}")
@@ -970,7 +787,7 @@ class NotificationWidget(Box):
except Exception as e:
logger.debug(f"Failed to get cached app icon: {e}")
# Priority 4: Ultimate fallback with caching
# Ultimate fallback
logger.debug("Using fallback notification icon")
return get_fallback_notification_icon((35, 35))
@@ -1289,6 +1106,10 @@ class ModusNoti(Window):
def on_new_notification(self, fabric_notif, id):
notification: Notification = fabric_notif.get_notification_from_id(id)
# Check if notification still exists (might have been removed already)
if not notification:
return
if self._server.dont_disturb or modus_service.dont_disturb:
# Notification is already cached by the service, just don't show popup
@@ -1369,6 +1190,14 @@ class ModusNoti(Window):
return
notification = self.notification_queue.pop(0)
# Check if notification is still valid (might have been removed)
if not notification or not hasattr(notification, 'app_icon'):
# Skip invalid notifications and try next one
if self.notification_queue:
self._show_next_notification()
return
self.notification_state = NotificationState.SHOWING
new_box = NotificationRevealer(
@@ -1391,7 +1220,7 @@ class ModusNoti(Window):
# Show the window now that we have content to display
self.set_visible(True)
new_box.show_all()
new_box.show()
self.notifications.queue_resize()
def start_animation():
modules/notification/notification_center.py
+58 -110
View File
@@ -8,7 +8,7 @@ from fabric.widgets.eventbox import EventBox
from fabric.widgets.label import Label
from fabric.widgets.revealer import Revealer
from fabric.widgets.scrolledwindow import ScrolledWindow
from gi.repository import GLib
from gi.repository import GLib, GdkPixbuf
from loguru import logger
from modules.notification.notification import (
@@ -17,7 +17,6 @@ from modules.notification.notification import (
cache_notification_image,
get_cached_notification_image,
get_notification_image_cache_key,
find_existing_cached_image,
preload_notification_assets,
cleanup_all_notification_caches,
cleanup_notification_specific_caches,
@@ -309,84 +308,57 @@ class ExpandableNotificationGroup(Box):
self.expanded_container.set_visible(False)
def _get_notification_pixbuf(self, notification):
"""Use cached-first approach prioritizing existing cache over pixbuf processing"""
"""Simplified notification pixbuf for notification center"""
notification_id = getattr(notification, "id", None)
# Priority 1: ALWAYS check for existing cached image first (critical for restart scenarios)
if notification_id:
existing_cached_image = find_existing_cached_image(notification_id)
if existing_cached_image:
logger.debug(
f"Using existing cached image in center for notification: {
notification_id
}"
)
return existing_cached_image
try:
# Priority 2: Try to process current pixbuf only if we have no cached version
if (
hasattr(notification, "image_pixbuf")
and notification.image_pixbuf
and notification_id
):
# Try to process current pixbuf if available
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
try:
cache_key = get_notification_image_cache_key(
notification_id, notification.image_pixbuf
)
cached_image = get_cached_notification_image(cache_key)
if cached_image:
logger.debug(
f"Using cached notification image in center: {cache_key}"
)
logger.debug(f"Using cached notification image: {cache_key}")
return cached_image
# Try to cache the image now
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (35, 35)
)
if cache_path and cache_key:
cached_image = get_cached_notification_image(cache_key)
if cached_image:
logger.debug(
f"Cached and using notification image in center: {cache_key}"
)
return cached_image
cached_pixbuf = get_cached_notification_image(cache_key)
if cached_pixbuf:
logger.debug(f"Cached and using notification image: {cache_key}")
return cached_pixbuf
# Direct scaling fallback if caching works but loading fails
logger.debug(
"Using direct scaling for notification image in center"
# Direct fallback if caching succeeded but loading failed
logger.debug("Using direct scaling for notification image")
return notification.image_pixbuf.scale_simple(
35, 35, GdkPixbuf.InterpType.BILINEAR
)
# GdkPixbuf.InterpType.BILINEAR = 2
return notification.image_pixbuf.scale_simple(35, 35, 2)
except Exception as image_error:
logger.debug(
f"Notification image processing failed in center (temp file likely gone): {
image_error
}"
)
logger.debug(f"Notification image processing failed: {image_error}")
# Continue to app icon fallback
except Exception as e:
logger.debug(f"Failed to process notification image in center: {e}")
logger.debug(f"Failed to process notification image: {e}")
# Priority 3: Use cached app icon with optimized cache-first approach
# Use cached app icon as fallback
try:
# Use optimized caching for app icon
app_icon_source = getattr(notification, "app_icon", None)
if app_icon_source:
cached_app_icon = cache_notification_icon(app_icon_source, (35, 35))
if cached_app_icon:
logger.debug(
f"Using cached app icon in center for: {app_icon_source}"
)
logger.debug(f"Using cached app icon for: {app_icon_source}")
return cached_app_icon
except Exception as e:
logger.debug(f"Failed to get cached app icon in center: {e}")
logger.debug(f"Failed to get cached app icon: {e}")
# Priority 4: Ultimate fallback with caching
logger.debug("Using fallback notification icon in center")
# Ultimate fallback
logger.debug("Using fallback notification icon")
return get_fallback_notification_icon((35, 35))
def on_clicked(self, widget, event):
@@ -446,49 +418,41 @@ class ExpandableNotificationGroup(Box):
# Close all notifications in this group
for notification in self.notifications:
try:
# Get notification ID for cache cleanup
notification_id = getattr(notification._notification, "id", None)
app_icon_source = getattr(notification._notification, "app_icon", None)
# Clean up caches using notification ID pattern
if notification_id:
cleanup_notification_specific_caches(
app_icon_source=app_icon_source,
notification_image_cache_key=f"notif_{notification_id}_*",
)
logger.debug(
f"Cleaned up caches for notification ID: {notification_id}"
)
# Get notification cache metadata for cleanup
cache_metadata = getattr(notification, 'cache_metadata', {})
# Clean up caches using stored metadata
from modules.notification.notification import cleanup_notification_specific_caches
cleanup_notification_specific_caches(
app_icon_source=notification._notification.app_icon,
notification_image_cache_key=cache_metadata.get('notification_image_cache_key')
)
logger.debug(f"Cleaned up caches for notification ID: {notification._notification.id}")
notification_service.remove_cached_notification(notification.cache_id)
except Exception as e:
logger.error(
f"Error removing notification {notification.cache_id}: {e}"
)
logger.error(f"Error removing notification {notification.cache_id}: {e}")
def _close_single_notification(self, notification):
"""Close a single notification from this group with proper cache cleanup"""
try:
# Get notification ID for cache cleanup
notification_id = getattr(notification._notification, "id", None)
app_icon_source = getattr(notification._notification, "app_icon", None)
# Clean up caches using notification ID pattern
if notification_id:
cleanup_notification_specific_caches(
app_icon_source=app_icon_source,
notification_image_cache_key=f"notif_{notification_id}_*",
)
logger.debug(
f"Cleaned up caches for notification ID: {notification_id}"
)
# Get notification cache metadata for cleanup
cache_metadata = getattr(notification, 'cache_metadata', {})
# Clean up caches using stored metadata
from modules.notification.notification import cleanup_notification_specific_caches
cleanup_notification_specific_caches(
app_icon_source=notification._notification.app_icon,
notification_image_cache_key=cache_metadata.get('notification_image_cache_key')
)
logger.debug(f"Cleaned up caches for notification ID: {notification._notification.id}")
notification_service.remove_cached_notification(notification.cache_id)
logger.debug(f"Closed single notification: {notification.cache_id}")
except Exception as e:
logger.error(
f"Error removing single notification {notification.cache_id}: {e}"
)
logger.error(f"Error removing single notification {notification.cache_id}: {e}")
def _close_single_notification_and_stop_propagation(self, notification):
"""Close notification and prevent click from expanding the group"""
@@ -608,19 +572,19 @@ class NotificationCenterWidget(NotificationWidget):
def _on_close_clicked(self, *args):
try:
# Get notification ID for cache cleanup
notification_id = getattr(self.notification, "id", None)
app_icon_source = getattr(self.notification, "app_icon", None)
# Clean up caches using notification ID pattern
if notification_id:
cleanup_notification_specific_caches(
app_icon_source=app_icon_source,
notification_image_cache_key=f"notif_{notification_id}_*",
)
logger.debug(
f"Cleaned up caches for notification center ID: {notification_id}"
)
# Get notification cache metadata for cleanup
cache_metadata = getattr(self, 'cache_metadata', {})
if hasattr(self.notification, '_cache_metadata'):
cache_metadata = self.notification._cache_metadata
# Clean up caches using stored metadata
from modules.notification.notification import cleanup_notification_specific_caches
cleanup_notification_specific_caches(
app_icon_source=self.notification.app_icon,
notification_image_cache_key=cache_metadata.get('notification_image_cache_key')
)
logger.debug(f"Cleaned up caches for notification center ID: {self.notification.id}")
notification_service.remove_cached_notification(self.notification_id)
except Exception as e:
@@ -734,22 +698,6 @@ class NotificationCenter(Window):
# Preload assets for each cached notification to ensure display consistency
preload_notification_assets(cached_notification._notification)
# Check if we can find cached images for this notification
if notification_id:
existing_image = find_existing_cached_image(notification_id)
if existing_image:
logger.debug(
f"✓ Found cached image for notification {
notification_id
} during rebuild"
)
else:
logger.debug(
f"✗ No cached image found for notification {
notification_id
} during rebuild"
)
self.notification_groups[app_name].append(cached_notification)
rebuild_count += 1
modules/todo/todo_widget.py
+2 -2
View File
@@ -111,7 +111,7 @@ class TodoItem(Box):
# Edit button - using SVG icon
self.edit_icon = Svg(
name="todo-edit-icon",
size=14,
size=12,
svg_file=get_relative_path("../../config/assets/icons/todo/edit.svg"),
)
self.edit_button = Button(
@@ -123,7 +123,7 @@ class TodoItem(Box):
# Delete button - using SVG icon
self.delete_icon = Svg(
name="todo-delete-icon",
size=14,
size=12,
svg_file=get_relative_path(
"../../config/assets/icons/todo/delete-symbolic.svg"
),
services/custom_notification.py
+146 -30
View File
@@ -1,10 +1,15 @@
# Standard library imports
import json
import os
import time
from typing import List
# Fabric imports
import gi
gi.require_version("Gtk", "3.0")
gi.require_version("GdkPixbuf", "2.0")
from gi.repository import GdkPixbuf
import config.data as data
@@ -17,31 +22,34 @@ from fabric.notifications import (
)
gi.require_version("Gtk", "3.0")
gi.require_version("GdkPixbuf", "2.0")
NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notification"
NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notification_history.json"
class CachedNotification(Service):
@classmethod
def create_from_dict(cls, data, **kwargs):
"""Create CachedNotification from enhanced JSON data"""
data["timeout"] = 0
self = cls.__new__(cls)
Service.__init__(self, **kwargs)
self._notification = Notification.deserialize(data)
self.cache_id = data["cached-id"]
self._cache_id = data["cached-id"] # Set directly to private var
# Store cache metadata for cleanup
self.cache_metadata = data.get("cache_metadata", {})
self.timestamp = data.get("timestamp", int(time.time()))
return self
@Signal
def removed_from_cache(self) -> None: ...
@Property(int, "read-write")
@Property(int, "readable")
def cache_id(self) -> int:
return self._cache_id
@cache_id.setter
def cache_id(self, cache_id: int):
self._cache_id = cache_id
@Property(str, "readable")
def app_name(self) -> str:
return self._notification.app_name
@@ -82,8 +90,8 @@ class CachedNotification(Service):
def image_file(self) -> str:
return self._notification.image_file # type: ignore
@Property(GdkPixbuf.Pixbuf, "readable")
def image_pixbuf(self) -> GdkPixbuf.Pixbuf:
@Property(object, "readable")
def image_pixbuf(self) -> GdkPixbuf.Pixbuf | None:
if self.image_pixmap:
return self.image_pixmap.as_pixbuf()
if self.image_file and os.path.exists(self.image_file):
@@ -92,10 +100,36 @@ class CachedNotification(Service):
except Exception:
# If file can't be loaded, return None
pass
return None # type: ignore
return None
@Property(dict, "readable")
def serialized(self) -> dict:
"""Enhanced serialization with cache metadata"""
from modules.notification.notification import (
get_cache_key,
get_notification_image_cache_key,
)
# Get better cache keys for icons
app_icon_cache_key = None
notification_image_cache_key = None
if self.app_icon:
app_icon_cache_key = get_cache_key(self.app_icon, (35, 35), self.app_name)
if self.id and hasattr(self._notification, 'image_pixbuf'):
# Only try to access image_pixbuf if we can safely do so
try:
# Check if image_pixbuf exists and can be accessed
image_pixbuf = getattr(self._notification, 'image_pixbuf', None)
if image_pixbuf:
notification_image_cache_key = get_notification_image_cache_key(
self.id, image_pixbuf
)
except (AttributeError, OSError, Exception):
# If temp file is gone or any other error, just mark as None
pass
return {
"cached-id": self.cache_id,
"id": self.id,
@@ -110,12 +144,23 @@ class CachedNotification(Service):
"image-pixmap": (
self.image_pixmap.serialize() if self.image_pixmap else None
),
"timestamp": int(time.time()),
"group": self.app_name, # Group notifications by app name
# Enhanced cache metadata
"cache_metadata": {
"app_icon_cache_key": app_icon_cache_key,
"notification_image_cache_key": notification_image_cache_key,
"has_cached_image": notification_image_cache_key is not None,
"cache_timestamp": int(time.time())
}
}
def __init__(self, notification: Notification, cache_id: int, **kwargs):
super().__init__()
self._notification: Notification = notification
self._cache_id = cache_id
self.cache_metadata = {}
self.timestamp = int(time.time())
def remove_from_cache(self):
self.removed_from_cache.emit()
@@ -153,9 +198,8 @@ class CachedNotifications(Notifications):
def dont_disturb(self) -> bool:
"""Return the pause status."""
return self._dont_disturb
@dont_disturb.setter
def dont_disturb(self, value: bool):
def set_dont_disturb(self, value: bool):
"""Set the pause status."""
self._dont_disturb = value
self.notify("dont-disturb")
@@ -166,8 +210,14 @@ class CachedNotifications(Notifications):
self._signal_handlers = {} # Store signal handlers by notification_id
self._dont_disturb = False
self._count = 0
self._next_cache_id = 1 # Track next available cache ID
self.load_cached_notifications()
# Connect to the notification_added signal to cache new notifications
# Note: self here refers to the CachedNotifications service, which inherits from Notifications
# So we connect to our own notification_added signal
super().notification_added.connect(self.on_notification_added)
def load_cached_notifications(self) -> dict[int, CachedNotification]:
"""Load cached notifications from a JSON file (deserialization)."""
@@ -178,21 +228,26 @@ class CachedNotifications(Notifications):
# If file doesn't exist or is corrupted, start with empty list
data = []
max_cache_id = 0
for notification in data:
cached_notification = CachedNotification.create_from_dict(notification)
cache_id = cached_notification.cache_id
max_cache_id = max(max_cache_id, cache_id)
handler_id = cached_notification.connect(
"removed-from-cache",
lambda *args: self.remove_cached_notification(
notification_id=cached_notification.cache_id
notification_id=cache_id
),
)
self._signal_handlers[cached_notification.cache_id] = handler_id
self._cached_notifications[cached_notification.cache_id] = (
cached_notification
)
self._signal_handlers[cache_id] = handler_id
self._cached_notifications[cache_id] = cached_notification
self._count += 1
# Set next cache ID to be higher than any existing ID
self._next_cache_id = max_cache_id + 1
self.notify("count")
return self._cached_notifications
def cache_notifications(self) -> None:
"""Save cached notifications to a JSON file."""
@@ -206,26 +261,41 @@ class CachedNotifications(Notifications):
json.dump(serialized_data, file, indent=4)
def clear_all_cached_notifications(self):
"""Empty the notifications."""
"""Empty the notifications with enhanced cache cleanup"""
# Clean up all cached files before clearing notifications
from modules.notification.notification import cleanup_all_notification_caches
for cached_notification in self._cached_notifications.values():
handler_id = self._signal_handlers.pop(cached_notification.cache_id, None)
if handler_id:
cached_notification.disconnect(handler_id)
# Clear all notification caches (icons and images)
cleanup_all_notification_caches()
self._cached_notifications = {}
self.cache_notifications()
self._count = 0
self._next_cache_id = 1 # Reset cache ID counter
self.notify("count")
self.clear_all.emit()
def notification_added(self, notification_id: int) -> None:
"""Handle notification added and cache it."""
super().notification_added(notification_id)
def on_notification_added(self, service, notification_id: int) -> None:
"""Handle notification added and cache it with enhanced metadata"""
# Don't call super() - we're handling this ourselves
notification = self.get_notification_from_id(notification_id)
if notification:
# Import here to avoid circular imports
from config import data
from modules.notification.notification import (
preload_notification_assets,
cache_notification_icon,
cache_notification_image,
get_cache_key,
get_notification_image_cache_key
)
# Check if this app should be ignored for history (don't cache)
if notification.app_name in data.NOTIFICATION_IGNORED_APPS_HISTORY:
@@ -234,12 +304,48 @@ class CachedNotifications(Notifications):
# Cache notifications (except for ignored apps)
# DND only affects popup display, not caching
cache_id = self._next_cache_id
self._next_cache_id += 1
self._count += 1
cache_id = self._count
cached_notification = CachedNotification(
notification=notification, cache_id=cache_id
)
# Set cache_id directly since it's read-only property
cached_notification._cache_id = cache_id
# Preload assets and store cache metadata
preload_notification_assets(notification)
# Store enhanced cache metadata
app_icon_cache_key = None
notification_image_cache_key = None
if notification.app_icon:
app_icon_cache_key = get_cache_key(notification.app_icon, (35, 35), notification.app_name)
cache_notification_icon(notification.app_icon, (35, 35), notification.app_name)
if hasattr(notification, 'image_pixbuf'):
try:
# Safely try to access image_pixbuf
image_pixbuf = getattr(notification, 'image_pixbuf', None)
if image_pixbuf:
notification_image_cache_key = get_notification_image_cache_key(
notification.id, image_pixbuf
)
cache_notification_image(notification.id, image_pixbuf, (35, 35))
except (AttributeError, OSError, Exception):
# If temp file is gone or any other error, skip caching
pass
# Store metadata in cached notification
cached_notification.cache_metadata = {
"app_icon_cache_key": app_icon_cache_key,
"notification_image_cache_key": notification_image_cache_key,
"has_cached_image": notification_image_cache_key is not None,
"cache_timestamp": int(time.time())
}
handler_id = cached_notification.connect(
"removed-from-cache",
lambda *args: self.remove_cached_notification(notification_id=cache_id),
@@ -253,22 +359,32 @@ class CachedNotifications(Notifications):
self.emit("cached-notification-added", cached_notification)
def remove_cached_notification(self, notification_id: int):
"""Remove the notification of given id."""
"""Remove the notification of given id with enhanced cache cleanup"""
if notification_id in self._cached_notifications:
cached_notification = self._cached_notifications.pop(
notification_id
) # Remove from cache
cached_notification = self._cached_notifications.pop(notification_id)
# Enhanced cache cleanup using stored metadata
if hasattr(cached_notification, 'cache_metadata'):
cache_metadata = cached_notification.cache_metadata
# Clean up specific cached files using stored keys
from modules.notification.notification import cleanup_notification_specific_caches
cleanup_notification_specific_caches(
app_icon_source=cached_notification.app_icon,
notification_image_cache_key=cache_metadata.get('notification_image_cache_key')
)
self.cache_notifications() # Update JSON
self._count -= 1
self.notify("count")
# Get the stored signal handler ID and disconnect it
handler_id = self._signal_handlers.pop(notification_id, None)
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
self.set_dont_disturb(not self.dont_disturb)