1373 lines
54 KiB
Python
1373 lines
54 KiB
Python
import os
|
|
import hashlib
|
|
import time
|
|
|
|
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 fabric.notifications import (
|
|
Notification,
|
|
NotificationAction,
|
|
NotificationCloseReason,
|
|
)
|
|
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.image import Image
|
|
from fabric.widgets.label import Label
|
|
from utils.roam import modus_service
|
|
from widgets.custom_image import CustomImage
|
|
from widgets.customrevealer import SlideRevealer
|
|
from widgets.wayland import WaylandWindow as Window
|
|
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)):
|
|
"""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)
|
|
|
|
# Check cache first for immediate return
|
|
cached_pixbuf = get_cached_pixbuf(
|
|
cache_key, size, NOTIFICATION_ICON_CACHE_DIR
|
|
)
|
|
if cached_pixbuf:
|
|
logger.debug(f"Cache hit for icon: {cache_key}")
|
|
return cached_pixbuf
|
|
|
|
# Load, cache, and return icon in one optimized flow
|
|
if source.startswith("file://"):
|
|
# Local file URL - process and cache immediately
|
|
file_path = source[7:]
|
|
pixbuf = load_and_cache_local_icon(file_path, cache_key, size)
|
|
elif os.path.exists(source):
|
|
# Direct file path - process and cache immediately
|
|
pixbuf = load_and_cache_local_icon(source, cache_key, size)
|
|
else:
|
|
# Icon name - resolve from theme and cache immediately
|
|
pixbuf = load_and_cache_theme_icon(source, cache_key, size)
|
|
|
|
return pixbuf
|
|
|
|
elif hasattr(source, "scale_simple"):
|
|
# Already a pixbuf - cache it directly with optimized flow
|
|
cache_key = get_cache_key(source, size)
|
|
|
|
# Check cache first
|
|
cached_pixbuf = get_cached_pixbuf(
|
|
cache_key, size, NOTIFICATION_ICON_CACHE_DIR
|
|
)
|
|
if cached_pixbuf:
|
|
logger.debug(f"Cache hit for pixbuf: {cache_key}")
|
|
return cached_pixbuf
|
|
|
|
# Scale once and cache immediately
|
|
scaled_pixbuf = source.scale_simple(
|
|
size[0], size[1], GdkPixbuf.InterpType.BILINEAR
|
|
)
|
|
save_pixbuf_to_cache(scaled_pixbuf, cache_key, NOTIFICATION_ICON_CACHE_DIR)
|
|
logger.debug(f"Cached new pixbuf icon: {cache_key}")
|
|
return scaled_pixbuf
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to cache notification icon: {e}")
|
|
|
|
# Return fallback with caching
|
|
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 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"""
|
|
try:
|
|
# Primary approach: Use notification ID and image hash for unique key
|
|
if 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
|
|
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}"
|
|
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())}"
|
|
except Exception:
|
|
# Absolute fallback
|
|
return f"notif_{notification_id}_{int(time.time())}"
|
|
|
|
|
|
def cache_notification_image(notification_id, image_pixbuf, size=(64, 64)):
|
|
"""Optimized notification image caching with robust error handling for persistent notifications"""
|
|
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")
|
|
|
|
# Check if already cached to avoid redundant work
|
|
if os.path.exists(cache_path):
|
|
logger.debug(f"Image cache hit: {cache_key}")
|
|
return cache_path, cache_key
|
|
|
|
# Try to scale and save the image with robust error handling
|
|
try:
|
|
scaled_pixbuf = image_pixbuf.scale_simple(
|
|
size[0], size[1], GdkPixbuf.InterpType.BILINEAR
|
|
)
|
|
scaled_pixbuf.savev(cache_path, "png", [], [])
|
|
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}")
|
|
|
|
return None, None
|
|
|
|
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, notification ID pattern, 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}")
|
|
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}")
|
|
|
|
|
|
def verify_cache_persistence():
|
|
"""Verify that cached icons persist and can be loaded after restart"""
|
|
try:
|
|
icon_cache_files = []
|
|
image_cache_files = []
|
|
|
|
if os.path.exists(NOTIFICATION_ICON_CACHE_DIR):
|
|
icon_cache_files = [f for f in os.listdir(NOTIFICATION_ICON_CACHE_DIR) if f.endswith('.png')]
|
|
|
|
if os.path.exists(NOTIFICATION_IMAGE_CACHE_DIR):
|
|
image_cache_files = [f for f in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR) if f.endswith('.png')]
|
|
|
|
logger.info(f"Cache persistence check: {len(icon_cache_files)} icons, {len(image_cache_files)} images cached")
|
|
|
|
# Test loading a few cached items to verify they work
|
|
for cache_file in icon_cache_files[:2]: # Test first 2 icon files
|
|
try:
|
|
cache_path = os.path.join(NOTIFICATION_ICON_CACHE_DIR, cache_file)
|
|
test_pixbuf = GdkPixbuf.Pixbuf.new_from_file(cache_path)
|
|
if test_pixbuf:
|
|
logger.debug(f"Successfully verified cached icon: {cache_file}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load cached icon {cache_file}: {e}")
|
|
|
|
for cache_file in image_cache_files[:2]: # Test first 2 image files
|
|
try:
|
|
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, cache_file)
|
|
test_pixbuf = GdkPixbuf.Pixbuf.new_from_file(cache_path)
|
|
if test_pixbuf:
|
|
logger.debug(f"Successfully verified cached image: {cache_file}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load cached image {cache_file}: {e}")
|
|
|
|
return len(icon_cache_files) + len(image_cache_files) > 0
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to verify cache persistence: {e}")
|
|
return False
|
|
|
|
|
|
def migrate_persistent_notifications():
|
|
"""Migrate persistent notifications to use cached assets when temp files are gone"""
|
|
try:
|
|
from services.modus import notification_service
|
|
|
|
migrated_count = 0
|
|
for cached_notification in notification_service.cached_notifications:
|
|
notification = cached_notification._notification
|
|
|
|
# Check if notification has image_pixbuf but temp file might be gone
|
|
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
|
|
try:
|
|
# Try to access pixel data to test if temp file still exists
|
|
notification.image_pixbuf.get_pixels()
|
|
except Exception:
|
|
# Temp file is gone, ensure app icon is cached as fallback
|
|
try:
|
|
if hasattr(notification, "app_icon") and notification.app_icon:
|
|
cache_notification_icon(notification.app_icon, (35, 35))
|
|
migrated_count += 1
|
|
logger.debug(f"Migrated notification for {notification.app_name} to use cached app icon")
|
|
except Exception as cache_error:
|
|
logger.debug(f"Failed to cache app icon for {notification.app_name}: {cache_error}")
|
|
|
|
if migrated_count > 0:
|
|
logger.info(f"Migrated {migrated_count} persistent notifications to use cached assets")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to migrate persistent notifications: {e}")
|
|
|
|
|
|
# Initialize cache and verify persistence on module load
|
|
ensure_notification_cache_dirs()
|
|
cleanup_old_cache_files()
|
|
verify_cache_persistence()
|
|
|
|
# Run migration for persistent notifications on startup
|
|
try:
|
|
migrate_persistent_notifications()
|
|
except Exception as e:
|
|
logger.debug(f"Migration skipped (service not ready): {e}")
|
|
|
|
|
|
def preload_notification_assets(notification):
|
|
"""Preload and cache notification assets with robust error handling for persistent notifications"""
|
|
try:
|
|
# Preload app icon at common sizes used in notifications
|
|
if hasattr(notification, "app_icon") and notification.app_icon:
|
|
try:
|
|
# Cache header size (24x24) and content size (35x35) simultaneously
|
|
cache_notification_icon(notification.app_icon, (24, 24))
|
|
cache_notification_icon(notification.app_icon, (35, 35))
|
|
logger.debug(f"Preloaded app icon for: {notification.app_name}")
|
|
except Exception as icon_error:
|
|
logger.debug(f"Failed to preload app icon for {notification.app_name}: {icon_error}")
|
|
|
|
# Preload notification image if present (with resilient handling)
|
|
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}")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to preload notification assets: {e}")
|
|
|
|
|
|
def smooth_revealer_animation(revealer: SlideRevealer, duration: int = 280):
|
|
"""Configure revealer for ultra-smooth animation"""
|
|
revealer.duration = duration
|
|
|
|
|
|
class ActionButton(Button):
|
|
def __init__(
|
|
self, action: NotificationAction, index: int, total: int, notification_box
|
|
):
|
|
super().__init__(
|
|
name="action-button",
|
|
h_expand=True,
|
|
on_clicked=self.on_clicked,
|
|
child=Label(name="button-label", label=action.label),
|
|
)
|
|
self.action = action
|
|
self.notification_box = notification_box
|
|
style_class = (
|
|
"start-action"
|
|
if index == 0
|
|
else "end-action" if index == total - 1 else "middle-action"
|
|
)
|
|
self.add_style_class(style_class)
|
|
self.connect(
|
|
"enter-notify-event", lambda *_: notification_box.hover_button(self)
|
|
)
|
|
self.connect(
|
|
"leave-notify-event", lambda *_: notification_box.unhover_button(self)
|
|
)
|
|
|
|
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")
|
|
|
|
|
|
class NotificationWidget(Box):
|
|
def __init__(
|
|
self,
|
|
notification: Notification,
|
|
timeout_ms=data.NOTIFICATION_TIMEOUT,
|
|
show_close_button=True,
|
|
name="notification",
|
|
**kwargs,
|
|
):
|
|
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),
|
|
name=name,
|
|
orientation="v",
|
|
h_align="fill",
|
|
h_expand=True,
|
|
children=[
|
|
self.create_content(notification),
|
|
self.create_action_buttons(notification),
|
|
],
|
|
)
|
|
|
|
self.notification = notification
|
|
self.timeout_ms = timeout_ms
|
|
self._timeout_id = None
|
|
|
|
# Add hover events to the main notification widget
|
|
self.connect("enter-notify-event", self._on_enter_notify)
|
|
self.connect("leave-notify-event", self._on_leave_notify)
|
|
|
|
self.start_timeout()
|
|
|
|
def create_header(self, notification):
|
|
"""Create notification header with optimized cached app icon"""
|
|
try:
|
|
# Use optimized caching for header app icon (24x24 for header display)
|
|
cached_app_icon_pixbuf = cache_notification_icon(notification.app_icon, (24, 24))
|
|
|
|
if cached_app_icon_pixbuf:
|
|
# Create image from cached pixbuf for consistent display
|
|
app_icon = CustomImage(pixbuf=cached_app_icon_pixbuf)
|
|
app_icon.set_name("notification-icon")
|
|
else:
|
|
# Fallback to theme icon if caching fails completely
|
|
app_icon = Image(
|
|
name="notification-icon",
|
|
icon_name="notifications",
|
|
icon_size=24,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load cached header icon: {e}")
|
|
# Ultimate fallback
|
|
app_icon = Image(
|
|
name="notification-icon",
|
|
icon_name="notifications",
|
|
icon_size=24,
|
|
)
|
|
|
|
return CenterBox(
|
|
name="notification-title",
|
|
start_children=[
|
|
Box(
|
|
spacing=4,
|
|
children=[
|
|
app_icon,
|
|
Label(
|
|
notification.app_name,
|
|
name="notification-app-name",
|
|
h_align="start",
|
|
),
|
|
],
|
|
)
|
|
],
|
|
end_children=[
|
|
self.create_close_button() if self.show_close_button else Box()
|
|
],
|
|
)
|
|
|
|
def create_content(self, notification):
|
|
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",
|
|
max_chars_width=40,
|
|
ellipsization="end",
|
|
),
|
|
# Label(
|
|
# name="notification-app-name",
|
|
# markup=" | " + notification.app_name,
|
|
# h_align="start",
|
|
# ellipsization="end",
|
|
# ),
|
|
],
|
|
),
|
|
(
|
|
Label(
|
|
markup=notification.body.replace("\n", " "),
|
|
h_align="start",
|
|
max_chars_width=45,
|
|
ellipsization="end",
|
|
)
|
|
if notification.body
|
|
else Label(
|
|
markup="",
|
|
h_align="start",
|
|
ellipsization="end",
|
|
)
|
|
),
|
|
],
|
|
),
|
|
Box(h_expand=True),
|
|
Box(
|
|
orientation="v",
|
|
children=[
|
|
Button(
|
|
name="notification-close-button",
|
|
image=CustomImage(icon_name="close-symbolic", icon_size=18),
|
|
visible=True, # Initially hidden
|
|
on_clicked=lambda *_: self._manual_close(),
|
|
),
|
|
Box(v_expand=True),
|
|
],
|
|
),
|
|
],
|
|
)
|
|
|
|
def _on_enter_notify(self, widget, event):
|
|
self._is_hovered = True
|
|
if self.close_button:
|
|
self.close_button.set_visible(True)
|
|
self.pause_timeout()
|
|
return False
|
|
|
|
def _on_leave_notify(self, widget, event):
|
|
self._is_hovered = False
|
|
if self.close_button:
|
|
self.close_button.set_visible(False)
|
|
self.resume_timeout()
|
|
return False
|
|
|
|
def get_pixbuf(self, icon_path, width, height):
|
|
"""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}")
|
|
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 get_fallback_notification_icon((width, height))
|
|
|
|
def _get_notification_pixbuf(self, notification):
|
|
"""Optimized notification pixbuf with cached-first approach for restart scenarios"""
|
|
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
|
|
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: {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:
|
|
# 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}")
|
|
# 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
|
|
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}")
|
|
return cached_app_icon
|
|
except Exception as e:
|
|
logger.debug(f"Failed to get cached app icon: {e}")
|
|
|
|
# Priority 4: Ultimate fallback with caching
|
|
logger.debug("Using fallback notification icon")
|
|
return get_fallback_notification_icon((35, 35))
|
|
|
|
def create_action_buttons(self, notification):
|
|
return Box(
|
|
name="notification-action-buttons",
|
|
spacing=4,
|
|
h_expand=True,
|
|
children=[
|
|
ActionButton(action, i, len(notification.actions), self)
|
|
for i, action in enumerate(notification.actions)
|
|
],
|
|
)
|
|
|
|
def start_timeout(self):
|
|
self.stop_timeout()
|
|
self._timeout_id = GLib.timeout_add(self.timeout_ms, self.close_notification)
|
|
|
|
def stop_timeout(self):
|
|
if self._timeout_id is not None:
|
|
GLib.source_remove(self._timeout_id)
|
|
self._timeout_id = None
|
|
|
|
def close_notification(self):
|
|
self.notification.close("expired")
|
|
self.stop_timeout()
|
|
return False
|
|
|
|
def pause_timeout(self):
|
|
self.stop_timeout()
|
|
|
|
def resume_timeout(self):
|
|
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
|
|
logger.debug(f"Manual close triggered for notification with cache_key: {getattr(self, 'notification_image_cache_key', 'None')} and app_icon: {getattr(self, 'app_icon_source', 'None')}")
|
|
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
|
|
def set_pointer_cursor(self, widget, cursor_name):
|
|
window = widget.get_window()
|
|
if window:
|
|
cursor = Gdk.Cursor.new_from_name(widget.get_display(), cursor_name)
|
|
window.set_cursor(cursor)
|
|
|
|
def hover_button(self, button):
|
|
self.pause_timeout()
|
|
self.set_pointer_cursor(button, "hand2")
|
|
|
|
def unhover_button(self, button):
|
|
# Don't resume timeout here since the notification itself might still be hovered
|
|
self.set_pointer_cursor(button, "arrow")
|
|
|
|
|
|
class NotificationRevealer(SlideRevealer):
|
|
def __init__(
|
|
self,
|
|
notification: Notification,
|
|
on_transition_end=None,
|
|
parent_window=None,
|
|
**kwargs,
|
|
):
|
|
self.notif_box = NotificationWidget(notification, show_close_button=False)
|
|
self.notification = notification
|
|
self.on_transition_end = on_transition_end
|
|
# Reference to NotificationCenter window for queue clearing
|
|
self.parent_window = parent_window
|
|
self._is_closing = False
|
|
|
|
# Enhanced swipe detection variables for Android-style animation
|
|
self._drag_start_y = 0
|
|
self._drag_start_x = 0
|
|
self._is_dragging = False
|
|
self._swipe_threshold = 80 # Distance to trigger auto-dismiss
|
|
self._swipe_velocity_threshold = (
|
|
150 # Velocity to trigger dismiss even on shorter swipes
|
|
)
|
|
self._swipe_in_progress = False
|
|
self._current_offset = 0
|
|
self._last_drag_time = 0
|
|
self._drag_velocity = 0
|
|
self._spring_back_duration = 200 # Duration for spring-back animation
|
|
self._dismiss_threshold = 0.3 # Dismiss if swiped 30% of width
|
|
|
|
# Animation state
|
|
self._animation_in_progress = False
|
|
self._spring_timer_id = None
|
|
self._css_provider = None
|
|
|
|
# Wrap notification in EventBox for swipe detection
|
|
self.event_box = EventBox(
|
|
events=[
|
|
"button-press-event",
|
|
"button-release-event",
|
|
"motion-notify-event",
|
|
],
|
|
child=self.notif_box,
|
|
)
|
|
|
|
super().__init__(
|
|
child=self.event_box,
|
|
direction="right",
|
|
duration=280, # Faster, smoother duration
|
|
)
|
|
|
|
smooth_revealer_animation(self)
|
|
|
|
# Connect our own handler that manages the slide animation
|
|
self.notification.connect("closed", self.on_resolved)
|
|
|
|
self._animation_in_progress = True
|
|
|
|
def _ease_out_cubic(self, t):
|
|
"""Smoother easing function for better animation quality"""
|
|
return 1 - pow(1 - t, 3)
|
|
|
|
def _ease_out_quart(self, t):
|
|
"""Even smoother easing for ultra-smooth animations"""
|
|
return 1 - pow(1 - t, 4)
|
|
|
|
def _apply_transform(self, offset_x, opacity, scale):
|
|
"""Apply smooth CSS transforms for animation"""
|
|
try:
|
|
# Create CSS transformation
|
|
transform_css = f"""
|
|
opacity: {opacity};
|
|
transform: translateX({offset_x}px) scale({scale});
|
|
transition: none;
|
|
"""
|
|
|
|
# Apply to the notification box
|
|
if hasattr(self.notif_box, "get_style_context"):
|
|
style_context = self.notif_box.get_style_context()
|
|
if style_context:
|
|
# Use CSS provider for smooth transforms
|
|
if not hasattr(self, "_css_provider") or not self._css_provider:
|
|
from gi.repository import Gtk
|
|
|
|
self._css_provider = Gtk.CssProvider()
|
|
style_context.add_provider(
|
|
self._css_provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION
|
|
)
|
|
|
|
css_data = f"* {{ {transform_css} }}"
|
|
self._css_provider.load_from_data(css_data.encode())
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Transform apply failed (non-critical): {e}")
|
|
|
|
def _animate_dismiss(self, start_offset):
|
|
"""Animate the notification sliding out with smooth 60fps animation"""
|
|
target_offset = NOTIFICATION_WIDTH + 50
|
|
duration = 200 # Slightly longer for smoother feel
|
|
|
|
if self._spring_timer_id:
|
|
GLib.source_remove(self._spring_timer_id)
|
|
|
|
start_time = GLib.get_monotonic_time() / 1000
|
|
offset_diff = target_offset - start_offset
|
|
|
|
def animate_step():
|
|
current_time = GLib.get_monotonic_time() / 1000
|
|
elapsed = current_time - start_time
|
|
progress = min(1.0, elapsed / duration)
|
|
|
|
# Use smoother easing for premium feel
|
|
eased_progress = self._ease_out_quart(progress)
|
|
current_offset = start_offset + (offset_diff * eased_progress)
|
|
|
|
# Smoother fade and scale transitions
|
|
opacity = max(0.0, 1.0 - (progress * 0.9)) # Gentler fade
|
|
scale = max(0.9, 1.0 - (progress * 0.1)) # Subtle scale
|
|
|
|
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
|
|
try:
|
|
self.notification.close("dismissed-by-user")
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
return True
|
|
|
|
# Use consistent 60fps timing
|
|
self._animation_in_progress = True
|
|
self._spring_timer_id = GLib.timeout_add(16, animate_step) # ~60 FPS
|
|
|
|
def _calculate_drag_velocity(self, current_x):
|
|
"""Calculate the velocity of the drag gesture"""
|
|
current_time = GLib.get_monotonic_time() / 1000
|
|
|
|
if self._last_drag_time > 0:
|
|
time_diff = current_time - self._last_drag_time
|
|
if time_diff > 0:
|
|
distance_diff = current_x - self._drag_start_x - self._current_offset
|
|
self._drag_velocity = abs(distance_diff / time_diff)
|
|
|
|
self._last_drag_time = current_time
|
|
|
|
def _on_animation_complete(self, is_hiding=False):
|
|
if is_hiding:
|
|
# Manually destroy the notification widget since we disconnected its handler
|
|
self.notif_box.destroy()
|
|
|
|
if self.on_transition_end:
|
|
self.on_transition_end()
|
|
self.destroy()
|
|
|
|
def on_resolved(
|
|
self,
|
|
_notification: Notification,
|
|
reason: NotificationCloseReason,
|
|
):
|
|
if self._is_closing:
|
|
return
|
|
|
|
self._is_closing = True
|
|
|
|
# Clean up any ongoing animations
|
|
if self._spring_timer_id:
|
|
GLib.source_remove(self._spring_timer_id)
|
|
|
|
# Use different slide directions based on dismiss reason
|
|
if reason == "expired":
|
|
# Gentle fade-out for auto-dismiss
|
|
self.set_slide_direction("left")
|
|
self.duration = 250 # Slightly slower for natural feel
|
|
elif self._swipe_in_progress:
|
|
# Quick slide for swipe dismissals
|
|
self.duration = 150
|
|
self.set_slide_direction("right")
|
|
else:
|
|
# Smooth slide for manual close
|
|
self.set_slide_direction("right")
|
|
self.duration = 200
|
|
|
|
self.hide()
|
|
# Consistent timing for smooth transitions
|
|
timeout_duration = self.duration + 50
|
|
GLib.timeout_add(timeout_duration, lambda: self._on_animation_complete(True))
|
|
|
|
def destroy(self):
|
|
# Clean up CSS provider and timers
|
|
if self._spring_timer_id:
|
|
GLib.source_remove(self._spring_timer_id)
|
|
super().destroy()
|
|
|
|
|
|
class NotificationState:
|
|
IDLE = 0
|
|
SHOWING = 1
|
|
HIDING = 2
|
|
TRANSITIONING = 3 # New state for smooth transitions
|
|
|
|
|
|
class ModusNoti(Window):
|
|
def __init__(self):
|
|
self._server = notification_service
|
|
|
|
self.notifications = Box(
|
|
v_expand=True,
|
|
h_expand=True,
|
|
style="margin: 1px 0px 1px 1px;",
|
|
orientation="v",
|
|
spacing=5,
|
|
)
|
|
|
|
# Enhanced queue system for ultra-smooth transitions
|
|
self.notification_queue = []
|
|
self.current_notification = None
|
|
self.notification_state = NotificationState.IDLE
|
|
self._transition_timer_id = None
|
|
self._debounce_timer_id = None
|
|
self._last_notification_time = 0
|
|
|
|
# Queue management settings for smooth behavior
|
|
self.MAX_QUEUE_SIZE = 3 # Limit queue to prevent overwhelming
|
|
self.TRANSITION_DELAY = 100 # Smoother transition timing
|
|
self.DEBOUNCE_DELAY = 50 # Prevent rapid fire notifications
|
|
|
|
# Initialize ignored apps list from config
|
|
self.ignored_apps = data.NOTIFICATION_IGNORED_APPS
|
|
|
|
self._server.connect("notification-added", self.on_new_notification)
|
|
super().__init__(
|
|
anchor="top right",
|
|
child=self.notifications,
|
|
layer="overlay",
|
|
title="modus-notifications", # More specific title for debugging
|
|
all_visible=True,
|
|
visible=False, # Start hidden, show only when we have content
|
|
exclusive=False,
|
|
)
|
|
|
|
def on_new_notification(self, fabric_notif, id):
|
|
notification: Notification = fabric_notif.get_notification_from_id(id)
|
|
|
|
# Check if the notification should be ignored completely (ignored apps)
|
|
if notification.app_name in self.ignored_apps:
|
|
return
|
|
|
|
if self._server.dont_disturb or modus_service.dont_disturb:
|
|
# Notification is already cached by the service, just don't show popup
|
|
return
|
|
|
|
# Preload assets immediately for optimal caching and display performance
|
|
preload_notification_assets(notification)
|
|
|
|
# Implement smart queue management for smooth transitions
|
|
current_time = GLib.get_monotonic_time() / 1000
|
|
|
|
# If queue is getting full, remove oldest notifications smoothly
|
|
if len(self.notification_queue) >= self.MAX_QUEUE_SIZE:
|
|
# Remove oldest notification from queue (not current showing one)
|
|
if self.notification_queue:
|
|
oldest = self.notification_queue.pop(0)
|
|
try:
|
|
oldest.close("dismissed-by-user")
|
|
except:
|
|
pass
|
|
|
|
# Add new notification to queue
|
|
self.notification_queue.append(notification)
|
|
|
|
# Debounce rapid notifications for smoother experience
|
|
if self._debounce_timer_id:
|
|
GLib.source_remove(self._debounce_timer_id)
|
|
|
|
self._debounce_timer_id = GLib.timeout_add(
|
|
self.DEBOUNCE_DELAY,
|
|
lambda: self._process_notification_queue_debounced() or False,
|
|
)
|
|
|
|
def _process_notification_queue_debounced(self):
|
|
"""Process queue after debounce delay for smooth transitions"""
|
|
self._debounce_timer_id = None
|
|
self._process_notification_queue()
|
|
return False
|
|
|
|
def _process_notification_queue(self):
|
|
# If we're currently showing a notification and there's a new one in queue
|
|
if (
|
|
self.notification_state == NotificationState.SHOWING
|
|
and self.current_notification
|
|
and self.notification_queue
|
|
):
|
|
# Smooth transition: start hiding current notification
|
|
self.notification_state = NotificationState.TRANSITIONING
|
|
self._start_smooth_transition()
|
|
|
|
elif (
|
|
self.notification_state == NotificationState.IDLE
|
|
and self.notification_queue
|
|
):
|
|
# If we're idle and have notifications in queue, show the next one
|
|
self._show_next_notification()
|
|
|
|
def _start_smooth_transition(self):
|
|
"""Start smooth transition between notifications"""
|
|
if self.current_notification and not self.current_notification._is_closing:
|
|
# Don't mark for cache cleanup during smooth transitions
|
|
# to maintain performance
|
|
|
|
# Use shorter timeout for smooth transitions
|
|
self.current_notification.notif_box.timeout_ms = 100
|
|
|
|
# Force close current notification with smooth animation
|
|
try:
|
|
self.current_notification.notification.close("expired")
|
|
except:
|
|
pass
|
|
|
|
def _show_next_notification(self):
|
|
if (
|
|
not self.notification_queue
|
|
or self.notification_state != NotificationState.IDLE
|
|
):
|
|
return
|
|
|
|
notification = self.notification_queue.pop(0)
|
|
self.notification_state = NotificationState.SHOWING
|
|
|
|
new_box = NotificationRevealer(
|
|
notification,
|
|
on_transition_end=lambda: self._on_notification_finished(new_box),
|
|
parent_window=self,
|
|
)
|
|
|
|
self.current_notification = new_box
|
|
|
|
# Clear any existing children
|
|
for child in list(self.notifications.children):
|
|
try:
|
|
self.notifications.remove(child)
|
|
except:
|
|
pass
|
|
|
|
self.notifications.children = [new_box]
|
|
|
|
# Show the window now that we have content to display
|
|
self.set_visible(True)
|
|
|
|
new_box.show_all()
|
|
self.notifications.queue_resize()
|
|
|
|
def start_animation():
|
|
if new_box.get_parent() and new_box.get_realized():
|
|
new_box.reveal()
|
|
return False
|
|
return True
|
|
|
|
GLib.idle_add(start_animation)
|
|
|
|
def _on_notification_finished(self, notification_box):
|
|
if notification_box != self.current_notification:
|
|
return
|
|
|
|
# Cancel any pending transition timer
|
|
if self._transition_timer_id:
|
|
GLib.source_remove(self._transition_timer_id)
|
|
self._transition_timer_id = None
|
|
|
|
# Safely remove notification box
|
|
try:
|
|
if notification_box in self.notifications.children:
|
|
self.notifications.remove(notification_box)
|
|
except:
|
|
pass
|
|
|
|
# Reset state
|
|
self.current_notification = None
|
|
self.notification_state = NotificationState.IDLE
|
|
|
|
# Process next notification with optimized delay for ultra-smooth transitions
|
|
if self.notification_queue:
|
|
self._transition_timer_id = GLib.timeout_add(
|
|
self.TRANSITION_DELAY, # Consistent smooth timing
|
|
lambda: self._show_next_notification() or False,
|
|
)
|
|
else:
|
|
# Hide window when no more notifications to show
|
|
self.set_visible(False)
|
|
|
|
def show_next_notification(self):
|
|
# Legacy method for compatibility - redirect to new implementation
|
|
self._show_next_notification()
|
|
|
|
def on_notification_finished(self, notification_box):
|
|
# Legacy method for compatibility - redirect to new implementation
|
|
self._on_notification_finished(notification_box)
|
|
|
|
def clear_notification_queue(self):
|
|
"""Clear queue with smooth cleanup"""
|
|
queue_length = len(self.notification_queue)
|
|
if queue_length > 0:
|
|
# Smooth dismissal of queued notifications
|
|
for notification in list(self.notification_queue):
|
|
try:
|
|
notification.close("dismissed-by-user")
|
|
except:
|
|
pass
|
|
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
|
|
|
|
# Clear animation timers
|
|
if self._transition_timer_id:
|
|
GLib.source_remove(self._transition_timer_id)
|
|
self._transition_timer_id = None
|
|
|
|
if self._debounce_timer_id:
|
|
GLib.source_remove(self._debounce_timer_id)
|
|
self._debounce_timer_id = None
|
|
|
|
def get_queue_length(self):
|
|
return len(self.notification_queue)
|