import hashlib import os import time from fabric.notifications import ( Notification, NotificationAction, NotificationCloseReason, ) from fabric.utils import get_relative_path from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.eventbox import EventBox from fabric.widgets.image import Image from fabric.widgets.label import Label from gi.repository import Gdk, GdkPixbuf, GLib # type: ignore from loguru import logger import config.data as data from services.modus import notification_service from utils.functions import escape_markup_text 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 .unified_cache import ( get_unified_cache_key, ) NOTIFICATION_WIDTH = 360 NOTIFICATION_IMAGE_SIZE = 48 NOTIFICATION_WIDTH = 360 NOTIFICATION_IMAGE_SIZE = 48 # 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 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)""" try: 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(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 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_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: 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] # Backward compatibility get_cache_key = get_unified_cache_key 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}") 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), 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_unified_cache_key(source, size, app_name) # Check cache first for immediate return cached_pixbuf = get_cached_pixbuf( cache_key, fallback_size=size, cache_dir=NOTIFICATION_ICON_CACHE_DIR ) if cached_pixbuf: 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_unified_cache_key(source, size, app_name) # Check cache first cached_pixbuf = get_cached_pixbuf( cache_key, fallback_size=size, cache_dir=NOTIFICATION_ICON_CACHE_DIR ) if cached_pixbuf: 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) 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""" # 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) 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 deterministic cache key based on image content to prevent duplicate caching""" try: # 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 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 dimension_hash except Exception: pass # Fallback to timestamp for invalid pixbufs return str(int(time.time()))[:8] except Exception: # Ultimate fallback return str(int(time.time()))[:8] def cache_notification_image(notification_id, image_pixbuf, size=(64, 64)): """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 - already exists: {cache_key}") return cache_path, cache_key # Try to scale and save the image 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: logger.debug( f"Failed to cache image (temp file likely gone): {scale_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): 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) - SINGLE ICON SIZE""" 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 (only 35x35 version) if app_icon_source: # Only clean 35x35 version since we only cache this size now 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" ) 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 - SINGLE ICON SIZE""" try: # Cache app icon only at content size (35x35) - scale down for headers at runtime if hasattr(notification, "app_icon") and notification.app_icon: try: # Only cache at 35x35 to reduce disk usage - headers will scale this down cache_notification_icon(notification.app_icon, (35, 35)) except Exception as icon_error: logger.debug( f"Failed to preload app icon for {notification.app_name}: { icon_error }" ) # Cache notification image if available if hasattr(notification, "image_pixbuf") and notification.image_pixbuf: cache_notification_image( notification.id, notification.image_pixbuf, (35, 35) ) 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 - SINGLE CACHE SIZE""" try: # Get 35x35 cached icon and scale down to 24x24 for header cached_app_icon_pixbuf = cache_notification_icon( notification.app_icon, (35, 35) ) if cached_app_icon_pixbuf: # Scale down the 35x35 cached icon to 24x24 for header display header_icon_pixbuf = cached_app_icon_pixbuf.scale_simple( 24, 24, GdkPixbuf.InterpType.BILINEAR ) app_icon = CustomImage(pixbuf=header_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=escape_markup_text( 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=escape_markup_text( 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) 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): """Simplified notification pixbuf with NO CACHING for image-pixmap to prevent disk bloat""" notification_id = getattr(notification, "id", int(time.time())) try: # 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 ) cached_image = get_cached_notification_image(cache_key) if cached_image: return cached_image except Exception as 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}") # Use cached app icon as fallback try: 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: return cached_app_icon except Exception as e: logger.debug(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( 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 - just close notification""" 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("Cleaned up caches for manually dismissed notification") else: logger.debug("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 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 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 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) # 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( 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() 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)