"""
Notification Center for Qtile Expanded.
This module provides a centralized notification management system that:
- Tracks all notifications
- Manages notification history
- Provides popup display functionality
- Integrates with DBus notifications
This allows widgets like NotificationBell to display notifications without
having to manage the popup logic themselves.
"""
import time
from datetime import datetime
from typing import Any, Callable, Optional
from libqtile.log_utils import logger
[docs]
class NotificationCenter:
"""
Centralized notification management for Qtile.
This class maintains a list of notifications and provides methods to:
- Add notifications (from DBus or manually)
- Remove notifications
- Show notifications in a popup
- Persist notifications across reloads
Usage:
from qtile_expanded.extensions.notification_center import NotificationCenter
# Create a notification center
center = NotificationCenter()
# Add a notification
center.add_notification(
app_name="Firefox",
summary="New message",
body="Hello from Alice!"
)
# Show popup
center.show_popup(qtile)
# Clear all
center.clear()
Attributes:
notifications: List of notification dicts
max_notifications: Maximum notifications to keep
"""
_instance: Optional["NotificationCenter"] = None
[docs]
def __init__(
self,
max_notifications: int = 50,
storage_namespace: str = "notification_center",
persist_state: bool = True,
):
"""
Initialize NotificationCenter.
Args:
max_notifications: Maximum number of notifications to keep in history
storage_namespace: Namespace for state storage
persist_state: Whether to persist notifications across reloads
"""
self.max_notifications = max_notifications
self.storage_namespace = storage_namespace
self.persist_state = persist_state
self.notifications: list[dict[str, Any]] = []
self._popup = None
self._qtile = None
# Try to load persisted state
if self.persist_state:
self._load_state()
# Set up DBus listener
self._setup_dbus()
# Register as singleton
NotificationCenter._instance = self
[docs]
@classmethod
def get_instance(cls) -> "NotificationCenter":
"""Get or create the singleton instance."""
if cls._instance is None:
cls._instance = cls()
return cls._instance
def _load_state(self) -> None:
"""Load notifications from storage."""
try:
from qtile_expanded.storage import StateStorage
storage = StateStorage(self.storage_namespace)
self.notifications = storage.get("notifications", [])
logger.info(
f"NotificationCenter: Loaded {len(self.notifications)} notifications from storage"
)
except ImportError:
logger.warning(
"NotificationCenter: qtile_expanded.storage not available, "
"state persistence disabled"
)
except Exception as e:
logger.warning(f"NotificationCenter: Could not load state: {e}")
def _save_state(self) -> None:
"""Save notifications to storage."""
if not self.persist_state:
return
try:
from qtile_expanded.storage import StateStorage
storage = StateStorage(self.storage_namespace)
storage.set("notifications", self.notifications)
except Exception as e:
logger.warning(f"NotificationCenter: Could not save state: {e}")
def _setup_dbus(self) -> None:
"""Set up DBus notification listener."""
try:
import dbus
from dbus.mainloop.glib import DBusGMainLoop
DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
bus.add_signal_receiver(
self._on_dbus_notification,
signal_name="Notify",
dbus_interface="org.freedesktop.Notifications",
)
logger.info("NotificationCenter: DBus notification listener started")
except ImportError:
logger.warning(
"NotificationCenter: dbus-python not installed. "
"DBus notification listening disabled."
)
except Exception as e:
logger.warning(f"NotificationCenter: Could not set up DBus listener: {e}")
def _on_dbus_notification(self, *args, **kwargs) -> None:
"""Handle incoming DBus notification."""
# args: app_name, replaces_id, app_icon, summary, body, actions, hints, expire_timeout
app_name = args[0] if len(args) > 0 else "Unknown"
replaces_id = args[1] if len(args) > 1 else 0
app_icon = args[2] if len(args) > 2 else None
summary = args[3] if len(args) > 3 else ""
body = args[4] if len(args) > 4 else ""
actions = args[5] if len(args) > 5 else []
hints = args[6] if len(args) > 6 else {}
self.add_notification(
app_name=app_name,
summary=summary,
body=body,
icon=app_icon,
replaces_id=replaces_id,
hints=hints,
)
[docs]
def add_notification(
self,
summary: str = "",
body: str = "",
app_name: str = "",
icon: Optional[str] = None,
urgency: int = 1,
replaces_id: int = 0,
hints: Optional[dict] = None,
**kwargs,
) -> dict[str, Any]:
"""
Add a notification.
Args:
summary: Notification summary/title
body: Notification body text
app_name: Application name
icon: Icon path
urgency: Urgency level (0=low, 1=normal, 2=critical)
replaces_id: ID of notification this replaces
hints: Additional hints from DBus
**kwargs: Additional metadata
Returns:
The created notification dict
"""
# Check if this replaces an existing notification
if replaces_id:
for i, n in enumerate(self.notifications):
if n.get("id") == replaces_id or n.get("replaces_id") == replaces_id:
self.notifications[i] = {
"app_name": app_name,
"summary": summary,
"body": body,
"icon": icon,
"urgency": urgency,
"timestamp": datetime.now().isoformat(),
"id": replaces_id,
"replaces_id": replaces_id,
}
self._save_state()
if self._popup:
self._popup.update_notifications(self.notifications)
return self.notifications[i]
# Create new notification
notification = {
"app_name": app_name,
"summary": summary,
"body": body,
"icon": icon,
"urgency": urgency,
"timestamp": datetime.now().isoformat(),
"id": int(time.time() * 1000) + len(self.notifications),
"replaces_id": replaces_id,
}
# Add any additional kwargs
notification.update(kwargs)
# Add to list
self.notifications.append(notification)
# Trim if exceeds max
if len(self.notifications) > self.max_notifications:
self.notifications = self.notifications[-self.max_notifications:]
self._save_state()
# Update popup if open
if self._popup:
self._popup.update_notifications(self.notifications)
return notification
[docs]
def dismiss_notification(self, index: int) -> bool:
"""
Dismiss a notification by index.
Args:
index: Index of notification to dismiss
Returns:
True if notification was dismissed, False otherwise
"""
if 0 <= index < len(self.notifications):
del self.notifications[index]
self._save_state()
if self._popup:
self._popup.update_notifications(self.notifications)
return True
return False
[docs]
def clear(self) -> None:
"""Clear all notifications."""
self.notifications = []
self._save_state()
if self._popup:
self._popup.hide()
def _on_popup_close(self) -> None:
"""Handle popup close."""
pass
[docs]
def get_count(self) -> int:
"""Get the number of notifications."""
return len(self.notifications)
[docs]
def get_unread_count(self) -> int:
"""Get the number of unread notifications."""
# For now, all are considered unread
# Could be extended with read/unread tracking
return len(self.notifications)
[docs]
def mark_all_as_read(self) -> None:
"""Mark all notifications as read."""
# Could be extended with read/unread tracking
pass
[docs]
def __len__(self) -> int:
"""Get the number of notifications."""
return len(self.notifications)
[docs]
def __getitem__(self, index: int) -> dict[str, Any]:
"""Get a notification by index."""
return self.notifications[index]
[docs]
def __iter__(self):
"""Iterate over notifications."""
return iter(self.notifications)