598 lines
20 KiB
Python
598 lines
20 KiB
Python
"""
|
||
EU-Utility - Notification System
|
||
|
||
Toast notification system for non-blocking user feedback.
|
||
Supports info, warning, error, and success notification types.
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from typing import List, Optional, Callable, Dict, Any
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from enum import Enum
|
||
from collections import deque
|
||
import threading
|
||
|
||
from PyQt6.QtWidgets import (
|
||
QWidget, QVBoxLayout, QHBoxLayout, QLabel,
|
||
QGraphicsDropShadowEffect, QApplication
|
||
)
|
||
from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QObject, QPoint
|
||
from PyQt6.QtGui import QColor, QIcon, QPixmap
|
||
|
||
from core.eu_styles import EU_COLORS, EU_RADIUS
|
||
|
||
|
||
class NotificationType(Enum):
|
||
"""Notification types with associated styling."""
|
||
INFO = "info"
|
||
WARNING = "warning"
|
||
ERROR = "error"
|
||
SUCCESS = "success"
|
||
|
||
|
||
NOTIFICATION_STYLES = {
|
||
NotificationType.INFO: {
|
||
'icon': 'ℹ',
|
||
'icon_color': '#4a9eff',
|
||
'border_color': 'rgba(74, 158, 255, 150)',
|
||
'bg_color': 'rgba(25, 35, 50, 240)',
|
||
},
|
||
NotificationType.WARNING: {
|
||
'icon': '⚠',
|
||
'icon_color': '#ffc107',
|
||
'border_color': 'rgba(255, 193, 7, 150)',
|
||
'bg_color': 'rgba(45, 40, 25, 240)',
|
||
},
|
||
NotificationType.ERROR: {
|
||
'icon': '✕',
|
||
'icon_color': '#f44336',
|
||
'border_color': 'rgba(244, 67, 54, 150)',
|
||
'bg_color': 'rgba(45, 25, 25, 240)',
|
||
},
|
||
NotificationType.SUCCESS: {
|
||
'icon': '✓',
|
||
'icon_color': '#4caf50',
|
||
'border_color': 'rgba(76, 175, 80, 150)',
|
||
'bg_color': 'rgba(25, 45, 30, 240)',
|
||
},
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class Notification:
|
||
"""A single notification entry."""
|
||
id: str
|
||
title: str
|
||
message: str
|
||
type: NotificationType
|
||
timestamp: datetime = field(default_factory=datetime.now)
|
||
sound_played: bool = False
|
||
duration: int = 5000 # ms
|
||
|
||
|
||
class ToastWidget(QWidget):
|
||
"""Individual toast notification widget."""
|
||
|
||
clicked = pyqtSignal(str) # Emits notification ID
|
||
closed = pyqtSignal(str) # Emits notification ID
|
||
expired = pyqtSignal(str) # Emits notification ID
|
||
|
||
def __init__(self, notification: Notification, parent=None):
|
||
super().__init__(parent)
|
||
self.notification = notification
|
||
self._opacity = 1.0
|
||
self._fade_timer = None
|
||
|
||
# Frameless, always on top
|
||
self.setWindowFlags(
|
||
Qt.WindowType.FramelessWindowHint |
|
||
Qt.WindowType.WindowStaysOnTopHint |
|
||
Qt.WindowType.Tool |
|
||
Qt.WindowType.WindowDoesNotFocus
|
||
)
|
||
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
|
||
self.setAttribute(Qt.WidgetAttribute.WA_ShowWithoutActivating)
|
||
|
||
self._setup_ui()
|
||
self._setup_auto_close()
|
||
|
||
def _setup_ui(self):
|
||
"""Setup toast UI with EU styling."""
|
||
style = NOTIFICATION_STYLES[self.notification.type]
|
||
|
||
# Main container styling
|
||
self.setStyleSheet(f"""
|
||
QWidget {{
|
||
background-color: {style['bg_color']};
|
||
border: 1px solid {style['border_color']};
|
||
border-radius: {EU_RADIUS['medium']};
|
||
}}
|
||
""")
|
||
|
||
# Shadow effect
|
||
shadow = QGraphicsDropShadowEffect()
|
||
shadow.setBlurRadius(20)
|
||
shadow.setColor(QColor(0, 0, 0, 80))
|
||
shadow.setOffset(0, 4)
|
||
self.setGraphicsEffect(shadow)
|
||
|
||
# Layout
|
||
layout = QHBoxLayout(self)
|
||
layout.setContentsMargins(12, 10, 12, 10)
|
||
layout.setSpacing(10)
|
||
|
||
# Icon
|
||
self.icon_label = QLabel(style['icon'])
|
||
self.icon_label.setStyleSheet(f"""
|
||
color: {style['icon_color']};
|
||
font-size: 16px;
|
||
font-weight: bold;
|
||
background: transparent;
|
||
""")
|
||
self.icon_label.setFixedSize(24, 24)
|
||
self.icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
layout.addWidget(self.icon_label)
|
||
|
||
# Content
|
||
content_layout = QVBoxLayout()
|
||
content_layout.setSpacing(4)
|
||
content_layout.setContentsMargins(0, 0, 0, 0)
|
||
|
||
# Title
|
||
if self.notification.title:
|
||
self.title_label = QLabel(self.notification.title)
|
||
self.title_label.setStyleSheet(f"""
|
||
color: {EU_COLORS['text_primary']};
|
||
font-size: 12px;
|
||
font-weight: bold;
|
||
background: transparent;
|
||
""")
|
||
self.title_label.setWordWrap(True)
|
||
content_layout.addWidget(self.title_label)
|
||
|
||
# Message
|
||
self.message_label = QLabel(self.notification.message)
|
||
self.message_label.setStyleSheet(f"""
|
||
color: {EU_COLORS['text_secondary']};
|
||
font-size: 11px;
|
||
background: transparent;
|
||
""")
|
||
self.message_label.setWordWrap(True)
|
||
self.message_label.setMaximumWidth(280)
|
||
content_layout.addWidget(self.message_label)
|
||
|
||
layout.addLayout(content_layout, 1)
|
||
|
||
# Close button
|
||
self.close_btn = QLabel("×")
|
||
self.close_btn.setStyleSheet(f"""
|
||
color: {EU_COLORS['text_muted']};
|
||
font-size: 16px;
|
||
font-weight: bold;
|
||
background: transparent;
|
||
""")
|
||
self.close_btn.setFixedSize(20, 20)
|
||
self.close_btn.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
self.close_btn.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
layout.addWidget(self.close_btn)
|
||
|
||
# Fixed width, auto height
|
||
self.setFixedWidth(320)
|
||
self.adjustSize()
|
||
|
||
def _setup_auto_close(self):
|
||
"""Setup auto-close timer."""
|
||
self._close_timer = QTimer(self)
|
||
self._close_timer.setSingleShot(True)
|
||
self._close_timer.timeout.connect(self._start_fade_out)
|
||
self._close_timer.start(self.notification.duration)
|
||
|
||
def _start_fade_out(self):
|
||
"""Start fade-out animation."""
|
||
self.expired.emit(self.notification.id)
|
||
self._fade_timer = QTimer(self)
|
||
self._fade_timer.timeout.connect(self._fade_step)
|
||
self._fade_timer.start(30) # 30ms per step
|
||
|
||
def _fade_step(self):
|
||
"""Fade out animation step."""
|
||
self._opacity -= 0.1
|
||
if self._opacity <= 0:
|
||
self._fade_timer.stop()
|
||
self.close()
|
||
self.closed.emit(self.notification.id)
|
||
else:
|
||
self.setWindowOpacity(self._opacity)
|
||
|
||
def mousePressEvent(self, event):
|
||
"""Handle click to dismiss."""
|
||
if event.button() == Qt.MouseButton.LeftButton:
|
||
# Check if close button was clicked
|
||
if self.close_btn.geometry().contains(event.pos()):
|
||
self._close_timer.stop()
|
||
self.close()
|
||
self.closed.emit(self.notification.id)
|
||
else:
|
||
self.clicked.emit(self.notification.id)
|
||
super().mousePressEvent(event)
|
||
|
||
def enterEvent(self, event):
|
||
"""Pause auto-close on hover."""
|
||
self._close_timer.stop()
|
||
super().enterEvent(event)
|
||
|
||
def leaveEvent(self, event):
|
||
"""Resume auto-close on leave."""
|
||
self._close_timer.start(1000) # Give 1 second after mouse leaves
|
||
super().leaveEvent(event)
|
||
|
||
def close_notification(self):
|
||
"""Close this notification immediately."""
|
||
self._close_timer.stop()
|
||
if self._fade_timer:
|
||
self._fade_timer.stop()
|
||
self.close()
|
||
self.closed.emit(self.notification.id)
|
||
|
||
|
||
class NotificationSignals(QObject):
|
||
"""Thread-safe signals for notifications."""
|
||
show_notification = pyqtSignal(Notification)
|
||
|
||
|
||
class NotificationManager:
|
||
"""Singleton manager for toast notifications.
|
||
|
||
Usage:
|
||
manager = NotificationManager.get_instance()
|
||
manager.notify("Title", "Message", type=NotificationType.INFO)
|
||
manager.notify_info("Info", "Something happened")
|
||
manager.notify_warning("Warning", "Be careful")
|
||
manager.notify_error("Error", "Something failed")
|
||
manager.notify_success("Success", "Operation completed")
|
||
"""
|
||
|
||
_instance = None
|
||
_lock = threading.Lock()
|
||
|
||
# Maximum notifications to show at once
|
||
MAX_VISIBLE = 5
|
||
|
||
# Spacing between toasts
|
||
TOAST_SPACING = 8
|
||
|
||
# Margin from screen edge
|
||
SCREEN_MARGIN = 20
|
||
|
||
def __new__(cls):
|
||
if cls._instance is None:
|
||
with cls._lock:
|
||
if cls._instance is None:
|
||
cls._instance = super().__new__(cls)
|
||
cls._instance._initialized = False
|
||
return cls._instance
|
||
|
||
def __init__(self):
|
||
if self._initialized:
|
||
return
|
||
|
||
self._initialized = True
|
||
self._app: Optional[QApplication] = None
|
||
self._active_toasts: Dict[str, ToastWidget] = {}
|
||
self._history: deque = deque(maxlen=100)
|
||
self._counter = 0
|
||
self._sound_enabled = True
|
||
self._sound_path: Optional[Path] = None
|
||
self._sound_cache: Dict[str, Any] = {}
|
||
self._signals = NotificationSignals()
|
||
self._signals.show_notification.connect(self._show_toast)
|
||
|
||
# Position: bottom-right corner
|
||
self._position = "bottom-right" # or "top-right", "bottom-left", "top-left"
|
||
|
||
@classmethod
|
||
def get_instance(cls) -> 'NotificationManager':
|
||
"""Get the singleton instance."""
|
||
return cls()
|
||
|
||
def initialize(self, app: QApplication = None, sound_path: Path = None):
|
||
"""Initialize the notification manager.
|
||
|
||
Args:
|
||
app: QApplication instance (optional, will try to get existing)
|
||
sound_path: Path to sound files directory
|
||
"""
|
||
if app:
|
||
self._app = app
|
||
else:
|
||
self._app = QApplication.instance()
|
||
|
||
if sound_path:
|
||
self._sound_path = Path(sound_path)
|
||
|
||
print(f"[Notifications] Initialized with {self._history.maxlen} history limit")
|
||
|
||
def _get_next_id(self) -> str:
|
||
"""Generate unique notification ID."""
|
||
self._counter += 1
|
||
return f"notif_{self._counter}_{datetime.now().strftime('%H%M%S')}"
|
||
|
||
def notify(self, title: str, message: str,
|
||
type: NotificationType = NotificationType.INFO,
|
||
sound: bool = False, duration: int = 5000) -> str:
|
||
"""Show a notification toast.
|
||
|
||
Args:
|
||
title: Notification title
|
||
message: Notification message
|
||
type: Notification type (info, warning, error, success)
|
||
sound: Play sound notification
|
||
duration: Display duration in milliseconds
|
||
|
||
Returns:
|
||
Notification ID
|
||
"""
|
||
notification = Notification(
|
||
id=self._get_next_id(),
|
||
title=title,
|
||
message=message,
|
||
type=type,
|
||
duration=duration
|
||
)
|
||
|
||
# Add to history
|
||
self._history.append(notification)
|
||
|
||
# Play sound if requested
|
||
if sound and self._sound_enabled:
|
||
self._play_sound(type)
|
||
notification.sound_played = True
|
||
|
||
# Emit signal to show toast (thread-safe)
|
||
self._signals.show_notification.emit(notification)
|
||
|
||
return notification.id
|
||
|
||
def notify_info(self, title: str, message: str,
|
||
sound: bool = False, duration: int = 5000) -> str:
|
||
"""Show an info notification."""
|
||
return self.notify(title, message, NotificationType.INFO, sound, duration)
|
||
|
||
def notify_warning(self, title: str, message: str,
|
||
sound: bool = False, duration: int = 5000) -> str:
|
||
"""Show a warning notification."""
|
||
return self.notify(title, message, NotificationType.WARNING, sound, duration)
|
||
|
||
def notify_error(self, title: str, message: str,
|
||
sound: bool = True, duration: int = 7000) -> str:
|
||
"""Show an error notification (sound on by default)."""
|
||
return self.notify(title, message, NotificationType.ERROR, sound, duration)
|
||
|
||
def notify_success(self, title: str, message: str,
|
||
sound: bool = False, duration: int = 5000) -> str:
|
||
"""Show a success notification."""
|
||
return self.notify(title, message, NotificationType.SUCCESS, sound, duration)
|
||
|
||
def _show_toast(self, notification: Notification):
|
||
"""Show toast widget (called via signal for thread safety)."""
|
||
if not self._app:
|
||
self._app = QApplication.instance()
|
||
if not self._app:
|
||
print("[Notifications] Error: No QApplication available")
|
||
return
|
||
|
||
# Create toast widget
|
||
toast = ToastWidget(notification)
|
||
toast.closed.connect(lambda: self._on_toast_closed(notification.id))
|
||
toast.expired.connect(lambda: self._on_toast_expired(notification.id))
|
||
|
||
# Store reference
|
||
self._active_toasts[notification.id] = toast
|
||
|
||
# Position the toast
|
||
self._position_toast(toast)
|
||
|
||
# Show
|
||
toast.show()
|
||
|
||
# Remove oldest if too many visible
|
||
if len(self._active_toasts) > self.MAX_VISIBLE:
|
||
oldest_id = next(iter(self._active_toasts))
|
||
self._active_toasts[oldest_id].close_notification()
|
||
|
||
def _position_toast(self, toast: ToastWidget):
|
||
"""Position toast on screen."""
|
||
if not self._app:
|
||
return
|
||
|
||
screen = self._app.primaryScreen().geometry()
|
||
|
||
# Calculate position
|
||
visible_count = len(self._active_toasts)
|
||
|
||
if self._position == "bottom-right":
|
||
x = screen.width() - toast.width() - self.SCREEN_MARGIN
|
||
y = screen.height() - (toast.height() + self.TOAST_SPACING) * (visible_count + 1) - self.SCREEN_MARGIN
|
||
elif self._position == "top-right":
|
||
x = screen.width() - toast.width() - self.SCREEN_MARGIN
|
||
y = self.SCREEN_MARGIN + (toast.height() + self.TOAST_SPACING) * visible_count
|
||
elif self._position == "bottom-left":
|
||
x = self.SCREEN_MARGIN
|
||
y = screen.height() - (toast.height() + self.TOAST_SPACING) * (visible_count + 1) - self.SCREEN_MARGIN
|
||
else: # top-left
|
||
x = self.SCREEN_MARGIN
|
||
y = self.SCREEN_MARGIN + (toast.height() + self.TOAST_SPACING) * visible_count
|
||
|
||
toast.move(x, max(y, self.SCREEN_MARGIN))
|
||
|
||
# Reposition all visible toasts to maintain stack
|
||
self._reposition_toasts()
|
||
|
||
def _reposition_toasts(self):
|
||
"""Reposition all visible toasts."""
|
||
if not self._app:
|
||
return
|
||
|
||
screen = self._app.primaryScreen().geometry()
|
||
toasts = list(self._active_toasts.values())
|
||
|
||
for i, toast in enumerate(toasts):
|
||
if self._position == "bottom-right":
|
||
x = screen.width() - toast.width() - self.SCREEN_MARGIN
|
||
y = screen.height() - (toast.height() + self.TOAST_SPACING) * (i + 1) - self.SCREEN_MARGIN
|
||
elif self._position == "top-right":
|
||
x = screen.width() - toast.width() - self.SCREEN_MARGIN
|
||
y = self.SCREEN_MARGIN + (toast.height() + self.TOAST_SPACING) * i
|
||
elif self._position == "bottom-left":
|
||
x = self.SCREEN_MARGIN
|
||
y = screen.height() - (toast.height() + self.TOAST_SPACING) * (i + 1) - self.SCREEN_MARGIN
|
||
else: # top-left
|
||
x = self.SCREEN_MARGIN
|
||
y = self.SCREEN_MARGIN + (toast.height() + self.TOAST_SPACING) * i
|
||
|
||
toast.move(x, max(y, self.SCREEN_MARGIN))
|
||
|
||
def _on_toast_closed(self, notification_id: str):
|
||
"""Handle toast closed."""
|
||
if notification_id in self._active_toasts:
|
||
del self._active_toasts[notification_id]
|
||
self._reposition_toasts()
|
||
|
||
def _on_toast_expired(self, notification_id: str):
|
||
"""Handle toast expired (started fading)."""
|
||
pass # Just for tracking if needed
|
||
|
||
def _play_sound(self, type: NotificationType):
|
||
"""Play notification sound."""
|
||
try:
|
||
# Try to use QSoundEffect for non-blocking playback
|
||
from PyQt6.QtMultimedia import QSoundEffect, QAudioDevice
|
||
|
||
sound_file = self._get_sound_file(type)
|
||
if not sound_file or not sound_file.exists():
|
||
return
|
||
|
||
# Use cached sound effect or create new
|
||
sound_key = str(sound_file)
|
||
if sound_key not in self._sound_cache:
|
||
effect = QSoundEffect()
|
||
effect.setSource(str(sound_file))
|
||
effect.setVolume(0.7)
|
||
self._sound_cache[sound_key] = effect
|
||
|
||
effect = self._sound_cache[sound_key]
|
||
effect.play()
|
||
|
||
except ImportError:
|
||
# Fallback: try simple audio playback
|
||
self._play_sound_fallback(type)
|
||
except Exception as e:
|
||
print(f"[Notifications] Sound error: {e}")
|
||
|
||
def _get_sound_file(self, type: NotificationType) -> Optional[Path]:
|
||
"""Get sound file path for notification type."""
|
||
if not self._sound_path:
|
||
return None
|
||
|
||
# Map types to sound files
|
||
sound_map = {
|
||
NotificationType.INFO: "info.wav",
|
||
NotificationType.WARNING: "warning.wav",
|
||
NotificationType.ERROR: "error.wav",
|
||
NotificationType.SUCCESS: "success.wav",
|
||
}
|
||
|
||
sound_file = self._sound_path / sound_map.get(type, "info.wav")
|
||
|
||
# Try alternative extensions
|
||
if not sound_file.exists():
|
||
for ext in ['.mp3', '.wav', '.ogg']:
|
||
alt_file = sound_file.with_suffix(ext)
|
||
if alt_file.exists():
|
||
return alt_file
|
||
|
||
return sound_file if sound_file.exists() else None
|
||
|
||
def _play_sound_fallback(self, type: NotificationType):
|
||
"""Fallback sound playback using system tools."""
|
||
import platform
|
||
import subprocess
|
||
|
||
sound_file = self._get_sound_file(type)
|
||
if not sound_file:
|
||
return
|
||
|
||
try:
|
||
system = platform.system()
|
||
if system == "Windows":
|
||
import winsound
|
||
winsound.MessageBeep()
|
||
elif system == "Darwin": # macOS
|
||
subprocess.run(["afplay", str(sound_file)], check=False)
|
||
else: # Linux
|
||
subprocess.run(["aplay", "-q", str(sound_file)], check=False)
|
||
except Exception:
|
||
pass # Silently fail on sound errors
|
||
|
||
# ========== Public API ==========
|
||
|
||
def close_all(self):
|
||
"""Close all visible notifications."""
|
||
for toast in list(self._active_toasts.values()):
|
||
toast.close_notification()
|
||
self._active_toasts.clear()
|
||
|
||
def get_history(self, limit: int = None, type: NotificationType = None) -> List[Notification]:
|
||
"""Get notification history.
|
||
|
||
Args:
|
||
limit: Maximum number of notifications to return
|
||
type: Filter by notification type
|
||
|
||
Returns:
|
||
List of notifications (newest first)
|
||
"""
|
||
history = list(self._history)
|
||
|
||
if type:
|
||
history = [n for n in history if n.type == type]
|
||
|
||
if limit:
|
||
history = history[-limit:]
|
||
|
||
return list(reversed(history))
|
||
|
||
def clear_history(self):
|
||
"""Clear notification history."""
|
||
self._history.clear()
|
||
|
||
def set_sound_enabled(self, enabled: bool):
|
||
"""Enable/disable notification sounds."""
|
||
self._sound_enabled = enabled
|
||
|
||
def set_sound_path(self, path: Path):
|
||
"""Set path to sound files."""
|
||
self._sound_path = Path(path)
|
||
|
||
def set_position(self, position: str):
|
||
"""Set notification position.
|
||
|
||
Args:
|
||
position: One of "bottom-right", "top-right", "bottom-left", "top-left"
|
||
"""
|
||
valid_positions = ["bottom-right", "top-right", "bottom-left", "top-left"]
|
||
if position in valid_positions:
|
||
self._position = position
|
||
|
||
def dismiss(self, notification_id: str):
|
||
"""Dismiss a specific notification by ID."""
|
||
if notification_id in self._active_toasts:
|
||
self._active_toasts[notification_id].close_notification()
|
||
|
||
|
||
# Convenience function to get manager
|
||
def get_notification_manager() -> NotificationManager:
|
||
"""Get the global NotificationManager instance."""
|
||
return NotificationManager.get_instance()
|