Files
Modus/modules/dock/components/notifications.py
T

852 lines
29 KiB
Python

import json
import os
from datetime import datetime
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.centerbox import CenterBox
from fabric.widgets.label import Label
from fabric.widgets.revealer import Revealer
from fabric.widgets.scrolledwindow import ScrolledWindow
from gi.repository import Gdk, GLib, Gtk
import config.data as data
import utils.icons as icons
from modules.notification_popup import NotificationBox, get_limited_apps_history, get_history_ignored_apps
from utils.custom_image import CustomImage
from utils.notification_utils import (
CONFIG_FILE,
PERSISTENT_DIR,
PERSISTENT_HISTORY_FILE,
cleanup_orphan_cached_images,
compute_time_label,
create_historical_notification_from_data,
get_date_header,
get_shared_notification_history,
load_scaled_pixbuf,
save_persistent_history,
schedule_midnight_update,
)
from utils.wayland import WaylandWindow as Window
class NotificationHistoryPopup(Window):
"""Popup window that shows notification history with click-based reveal"""
def __init__(self, dock_component=None, **kwargs):
# Get dock position for proper anchoring
dock_position = data.DOCK_POSITION
popup_anchor = self._get_popup_anchor(dock_position)
margin = self._get_popup_margin(dock_position)
super().__init__(
name="notification-history-popup",
layer="top",
anchor=popup_anchor,
margin=margin,
exclusive=False,
keyboard_mode="on-demand",
visible=False,
all_visible=False,
**kwargs,
)
self.dock_component = dock_component
# Get transition type based on dock position
transition_type = self._get_revealer_transition(dock_position)
# Create notification history content
self.notification_history = NotificationHistory()
# Create scrolled window for the history
self.scrolled = ScrolledWindow(
name="notifications-scrolled",
child=self.notification_history,
h_scrollbar_policy="never",
v_scrollbar_policy="automatic",
min_content_size=(400, 450),
max_content_size=(450, 500),
h_expand=True,
v_expand=True,
propagate_width=True,
propagate_height=True,
)
# Create main content box
self.content_box = Box(
name="notification-history-popup-content",
orientation="v",
spacing=8,
h_expand=True,
v_expand=True,
children=[self.scrolled],
style_classes=["notification-history-popup"],
)
# Create revealer for slide transition
self.revealer = Revealer(
transition_type=transition_type,
transition_duration=300,
child=self.content_box,
reveal_child=False,
)
self.add(self.revealer)
# Connect to escape key to close
self.connect("key-press-event", self.on_key_press)
self.connect("button-press-event", self.on_button_press)
self.connect("enter-notify-event", self.on_enter_notify)
self.connect("leave-notify-event", self.on_leave_notify)
self.set_can_focus(True)
def _get_popup_anchor(self, dock_position):
"""Get popup anchor based on dock position"""
anchor_map = {
"Top": "top",
"Bottom": "bottom right",
"Left": "left",
"Right": "right",
}
return anchor_map.get(dock_position, "bottom")
def _get_popup_margin(self, dock_position):
"""Get popup margin based on dock position"""
margin_map = {
"Top": "60px 10px 10px 10px",
"Bottom": "10px 650px 60px 10px",
"Left": "10px 10px 10px 60px",
"Right": "10px 60px 10px 10px",
}
return margin_map.get(dock_position, "10px 10px 60px 10px")
def _get_revealer_transition(self, dock_position):
"""Get revealer transition type based on dock position"""
transition_map = {
"Top": "slide-down",
"Bottom": "slide-up",
"Left": "slide-right",
"Right": "slide-left",
}
return transition_map.get(dock_position, "slide-up")
def on_key_press(self, widget, event):
"""Handle key press events"""
if event.keyval == Gdk.KEY_Escape:
self.hide_popup()
return True
elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_d:
current_dnd = self.notification_history.do_not_disturb_enabled
self.notification_history.header_switch.set_active(not current_dnd)
return True
elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_a:
self.notification_history.clear_history()
return True
return False
def on_button_press(self, widget, event):
"""Handle button press events - close popup when clicking outside"""
# Close popup when clicking outside
if self.dock_component:
self.dock_component.popup_visible = False
self.hide_popup()
return False
def on_enter_notify(self, widget, event):
"""Handle mouse entering popup"""
# No action needed for click-based popup
return False
def on_leave_notify(self, widget, event):
"""Handle mouse leaving popup"""
# No action needed for click-based popup
return False
def show_popup(self):
"""Show the notification history popup"""
# Ensure notification history is loaded before showing
self.notification_history._ensure_history_loaded()
self.set_visible(True)
self.show_all()
# Use GLib.idle_add to ensure the window is shown before revealing
GLib.idle_add(self._reveal_popup)
def _reveal_popup(self):
"""Reveal the popup after window is shown"""
self.revealer.set_reveal_child(True)
self.grab_focus()
return False # Don't repeat
def hide_popup(self):
"""Hide the notification history popup"""
self.revealer.set_reveal_child(False)
# Hide window after transition completes
GLib.timeout_add(350, lambda: self.set_visible(False))
class NotificationIndicator(Box):
def __init__(self, **kwargs):
self.notification_icon = Label(
name="notification-icon", markup=icons.notifications
)
# Create button for the icon
self.icon_button = Button(
name="button-bar-notifications",
child=self.notification_icon,
on_clicked=self.on_button_clicked,
)
super().__init__(
name="notification-indicator",
children=[self.icon_button],
**kwargs
)
# Initialize popup as None, will be created in background
self.popup = None
self.popup_visible = False
self.popup_creating = False
self._connect_to_dnd_state()
# Connect to destroy signal for cleanup
self.connect("destroy", self.on_destroy)
# Start background creation of popup after a short delay
GLib.timeout_add(100, self._create_popup_in_background)
def on_destroy(self, widget):
"""Hide popup when component is destroyed"""
try:
# Hide popup instead of destroying it (only if it was created)
if self.popup is not None:
try:
self.popup.hide_popup()
self.popup_visible = False
except Exception:
pass
except Exception:
pass
def _create_popup_in_background(self):
"""Create popup in background for instant access"""
if self.popup is None and not self.popup_creating:
print("Creating notification popup in background...") # Debug print
self.popup_creating = True
# Use idle_add to create popup without blocking UI
GLib.idle_add(self._do_background_creation)
return False # Don't repeat timeout
def _do_background_creation(self):
"""Actually create the popup in background"""
try:
self.popup = NotificationHistoryPopup(dock_component=self)
print("Notification popup created in background!") # Debug print
except Exception as e:
print(f"Error creating popup in background: {e}")
finally:
self.popup_creating = False
return False # Don't repeat
def on_button_clicked(self, button):
"""Handle button click to toggle notification popup"""
print("Notification button clicked!") # Debug print
# If popup is still being created, create it synchronously
if self.popup is None:
if self.popup_creating:
print("Popup still creating, waiting...") # Debug print
# Force synchronous creation if user clicked while background creation is happening
self.popup = NotificationHistoryPopup(dock_component=self)
self.popup_creating = False
else:
print("Creating notification popup immediately...") # Debug print
self.popup = NotificationHistoryPopup(dock_component=self)
if self.popup_visible:
# Hide popup if currently visible
self.popup.hide_popup()
self.popup_visible = False
else:
# Show popup if currently hidden
self.popup.show_popup()
self.popup_visible = True
def _connect_to_dnd_state(self):
try:
self.notification_history = get_shared_notification_history()
self.notification_history.connect(
"dnd-state-changed", self._on_dnd_state_changed
)
self._update_icon_for_dnd_state(
self.notification_history.do_not_disturb_enabled
)
except Exception as e:
print(f"Could not connect to DND state: {e}")
def _on_dnd_state_changed(self, notification_history, dnd_enabled):
self._update_icon_for_dnd_state(dnd_enabled)
def _update_icon_for_dnd_state(self, dnd_enabled):
if dnd_enabled:
self.notification_icon.set_markup(icons.notifications_off)
else:
self.notification_icon.set_markup(icons.notifications)
class NotificationHistory(Box):
def __init__(self, **kwargs):
super().__init__(name="notification-history", orientation="v", **kwargs)
self.containers = []
self.header_label = Label(
name="nhh",
label="Notifications",
h_align="start",
h_expand=True,
)
self.header_switch = Gtk.Switch(name="dnd-switch")
self.header_switch.set_vexpand(False)
self.header_switch.set_valign(Gtk.Align.CENTER)
self.header_switch.set_active(False)
self.header_clean = Button(
name="nhh-button",
child=Label(name="nhh-button-label", markup=icons.trash),
on_clicked=self.clear_history,
)
self.do_not_disturb_enabled = False
self.header_switch.connect("notify::active", self.on_do_not_disturb_changed)
self.dnd_label = Label(name="dnd-label", markup=icons.notifications_off)
self.history_header = CenterBox(
name="notification-history-header",
spacing=8,
start_children=[self.header_switch, self.dnd_label],
center_children=[self.header_label],
end_children=[self.header_clean],
)
self.notifications_list = Box(
name="notifications-list",
orientation="v",
spacing=4,
h_expand=True,
v_expand=True,
h_align="fill",
v_align="fill",
)
self.no_notifications_label = Label(
name="no-notif",
markup=icons.notifications_clear,
v_align="fill",
h_align="fill",
v_expand=True,
h_expand=True,
justification="center",
)
self.no_notifications_box = Box(
name="no-notifications-box",
v_align="fill",
h_align="fill",
v_expand=True,
h_expand=True,
children=[self.no_notifications_label],
)
self.scrolled_window = ScrolledWindow(
name="notification-history-scrolled-window",
orientation="v",
h_expand=True,
v_expand=True,
h_align="fill",
v_align="fill",
propagate_width=False,
propagate_height=False,
)
self.scrolled_window_viewport_box = Box(
orientation="v",
children=[self.notifications_list, self.no_notifications_box],
)
self.scrolled_window.add_with_viewport(self.scrolled_window_viewport_box)
self.persistent_notifications = []
self.add(self.history_header)
self.add(self.scrolled_window)
# Initialize history loading state
self.history_loaded = False
self.history_loading = False
self.schedule_midnight_update()
self._connect_to_shared_history()
self._load_and_sync_dnd_state()
# Start background loading of history after a short delay
GLib.timeout_add(200, self._load_history_in_background)
def _connect_to_shared_history(self):
try:
shared_history = get_shared_notification_history()
shared_history.connect(
"notification-added", self._on_shared_notification_added
)
except Exception as e:
print(f"Could not connect to shared notification history: {e}")
def _on_shared_notification_added(self, shared_history):
GLib.idle_add(self._refresh_history)
def _refresh_history(self):
try:
for child in self.notifications_list.get_children()[:]:
self.notifications_list.remove(child)
if hasattr(child, "destroy"):
child.destroy()
self.containers.clear()
if os.path.exists(PERSISTENT_HISTORY_FILE):
with open(PERSISTENT_HISTORY_FILE, "r") as f:
self.persistent_notifications = json.load(f)
for note in reversed(self.persistent_notifications[-50:]):
self._add_historical_notification(note)
self.update_no_notifications_label_visibility()
except Exception as e:
print(f"Error refreshing notification history: {e}")
def _load_and_sync_dnd_state(self):
try:
dnd_enabled = False
if os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE, "r") as f:
config_data = json.load(f)
dnd_enabled = config_data.get("dnd_enabled", False)
shared_history = get_shared_notification_history()
shared_history.do_not_disturb_enabled = dnd_enabled
self.header_switch.set_active(dnd_enabled)
self.do_not_disturb_enabled = dnd_enabled
except Exception as e:
print(f"Could not sync DND state: {e}")
def _load_history_in_background(self):
"""Start loading notification history in background"""
if not self.history_loaded and not self.history_loading:
print("Starting background loading of notification history...") # Debug print
self.history_loading = True
# Use idle_add to load history without blocking UI
GLib.idle_add(self._do_background_history_loading)
return False # Don't repeat timeout
def _do_background_history_loading(self):
"""Actually load the history in background"""
try:
print("Loading notification history in background...") # Debug print
self._load_persistent_history()
cleanup_orphan_cached_images(self.persistent_notifications)
self.history_loaded = True
print("Notification history loaded in background!") # Debug print
except Exception as e:
print(f"Error loading history in background: {e}")
finally:
self.history_loading = False
return False # Don't repeat
def _ensure_history_loaded(self):
"""Ensure notification history is loaded (force if needed)"""
if not self.history_loaded:
if self.history_loading:
print("History still loading in background, forcing synchronous load...") # Debug print
# Force synchronous loading if user opened popup while background loading
self._load_persistent_history()
cleanup_orphan_cached_images(self.persistent_notifications)
self.history_loaded = True
self.history_loading = False
else:
print("Loading notification history synchronously...") # Debug print
self._load_persistent_history()
cleanup_orphan_cached_images(self.persistent_notifications)
self.history_loaded = True
def schedule_midnight_update(self):
schedule_midnight_update(self.on_midnight)
def on_midnight(self):
self.rebuild_with_separators()
self.schedule_midnight_update()
return GLib.SOURCE_REMOVE
def create_date_separator(self, date_header):
return Box(
name="notif-date-sep",
children=[
Label(
name="notif-date-sep-label",
label=date_header,
h_align="center",
h_expand=True,
)
],
)
def rebuild_with_separators(self):
GLib.idle_add(self._do_rebuild_with_separators)
def _do_rebuild_with_separators(self):
children = list(self.notifications_list.get_children())
for child in children:
self.notifications_list.remove(child)
current_date_header = None
last_date_header = None
for container in sorted(
self.containers, key=lambda x: x.arrival_time, reverse=True
):
arrival_time = container.arrival_time
date_header = get_date_header(arrival_time)
if date_header != current_date_header:
sep = self.create_date_separator(date_header)
self.notifications_list.add(sep)
current_date_header = date_header
last_date_header = date_header
self.notifications_list.add(container)
if not self.containers and last_date_header:
for child in list(self.notifications_list.get_children()):
if child.get_name() == "notif-date-sep":
self.notifications_list.remove(child)
self.notifications_list.show_all()
self.update_no_notifications_label_visibility()
def on_do_not_disturb_changed(self, switch, pspec):
self.do_not_disturb_enabled = switch.get_active()
try:
shared_history = get_shared_notification_history()
shared_history.set_do_not_disturb_enabled(self.do_not_disturb_enabled)
except Exception as e:
print(f"Could not update shared DND state: {e}")
def clear_history(self, *args):
for child in self.notifications_list.get_children()[:]:
container = child
notif_box = (
container.notification_box
if hasattr(container, "notification_box")
else None
)
if notif_box:
notif_box.destroy(from_history_delete=True)
self.notifications_list.remove(child)
child.destroy()
if os.path.exists(PERSISTENT_HISTORY_FILE):
try:
os.remove(PERSISTENT_HISTORY_FILE)
except Exception as e:
print(f"Error deleting persistent history file: {e}")
self.persistent_notifications = []
self.containers = []
self.rebuild_with_separators()
def _load_persistent_history(self):
if not os.path.exists(PERSISTENT_DIR):
os.makedirs(PERSISTENT_DIR, exist_ok=True)
if os.path.exists(PERSISTENT_HISTORY_FILE):
try:
with open(PERSISTENT_HISTORY_FILE, "r") as f:
self.persistent_notifications = json.load(f)
for note in reversed(self.persistent_notifications[-50:]):
self._add_historical_notification(note)
except Exception as e:
print(f"Error loading persistent history: {e}")
self.persistent_notifications = []
else:
self.persistent_notifications = []
GLib.idle_add(self.update_no_notifications_label_visibility)
def _save_persistent_history(self):
save_persistent_history(self.persistent_notifications)
def update_no_notifications_label_visibility(self):
has_notifications = bool(self.containers)
self.no_notifications_box.set_visible(not has_notifications)
self.notifications_list.set_visible(has_notifications)
def _add_historical_notification(self, note):
hist_notif = create_historical_notification_from_data(note)
hist_box = NotificationBox(hist_notif, timeout_ms=0)
hist_box.uuid = hist_notif.id
hist_box.cached_image_path = hist_notif.cached_image_path
hist_box.set_is_history(True)
for child in hist_box.get_children():
if child.get_name() == "notification-action-buttons":
hist_box.remove(child)
container = Box(
name="notification-container",
orientation="v",
h_align="fill",
h_expand=True,
)
container.notification_box = hist_box
try:
arrival = datetime.fromisoformat(hist_notif.timestamp)
except Exception:
arrival = datetime.now()
container.arrival_time = arrival
self.hist_time_label = Label(
name="notification-timestamp",
markup=compute_time_label(container.arrival_time),
h_align="start",
ellipsization="end",
)
self.hist_notif_image_box = Box(
name="notification-image",
orientation="v",
children=[
CustomImage(pixbuf=load_scaled_pixbuf(hist_box, 48, 48)),
Box(v_expand=True),
],
)
self.hist_notif_summary_label = Label(
name="notification-summary",
markup=hist_notif.summary,
h_align="start",
ellipsization="end",
)
self.hist_notif_app_name_label = Label(
name="notification-app-name",
markup=f"{hist_notif.app_name}",
h_align="start",
ellipsization="end",
)
self.hist_notif_body_label = (
Label(
name="notification-body",
markup=hist_notif.body,
h_align="start",
ellipsization="end",
line_wrap="word-char",
)
if hist_notif.body
else Box()
)
self.hist_notif_body_label.set_single_line_mode(
True
) if hist_notif.body else None
self.hist_notif_summary_box = Box(
name="notification-summary-box",
orientation="h",
children=[
self.hist_notif_summary_label,
Box(
name="notif-sep",
h_expand=False,
v_expand=False,
h_align="center",
v_align="center",
),
self.hist_notif_app_name_label,
Box(
name="notif-sep",
h_expand=False,
v_expand=False,
h_align="center",
v_align="center",
),
self.hist_time_label,
],
)
self.hist_notif_text_box = Box(
name="notification-text",
orientation="v",
v_align="center",
h_expand=True,
children=[
self.hist_notif_summary_box,
self.hist_notif_body_label,
],
)
self.hist_notif_close_button = Button(
name="notif-close-button",
child=Label(name="notif-close-label", markup=icons.cancel),
on_clicked=lambda *_: self.delete_historical_notification(
hist_notif.id, container
),
)
self.hist_notif_close_button_box = Box(
orientation="v",
children=[
self.hist_notif_close_button,
Box(v_expand=True),
],
)
content_box = Box(
name="notification-box-hist",
spacing=8,
children=[
self.hist_notif_image_box,
self.hist_notif_text_box,
self.hist_notif_close_button_box,
],
)
container.add(content_box)
self.containers.insert(0, container)
self.rebuild_with_separators()
self.update_no_notifications_label_visibility()
def delete_historical_notification(self, note_id, container):
if hasattr(container, "notification_box"):
notif_box = container.notification_box
notif_box.destroy(from_history_delete=True)
target_note_id_str = str(note_id)
new_persistent_notifications = []
removed_from_list = False
for note_in_list in self.persistent_notifications:
current_note_id_str = str(note_in_list.get("id"))
if current_note_id_str == target_note_id_str:
removed_from_list = True
continue
new_persistent_notifications.append(note_in_list)
if removed_from_list:
self.persistent_notifications = new_persistent_notifications
self._save_persistent_history()
container.destroy()
self.containers = [c for c in self.containers if c != container]
self.rebuild_with_separators()
class NotificationHistoryWindow(Window):
def __init__(self):
self.notification_history = NotificationHistory()
scrolled = ScrolledWindow(
name="notifications-scrolled",
child=self.notification_history,
h_scrollbar_policy="never",
v_scrollbar_policy="automatic",
min_content_size=(400, 450),
max_content_size=(450, 500),
h_expand=True,
v_expand=True,
propagate_width=False,
propagate_height=False,
)
main_box = Box(
name="notifications-popup-main",
orientation="v",
spacing=8,
h_expand=True,
v_expand=True,
children=[scrolled],
)
dock_position = data.DOCK_POSITION
popup_anchor = self._get_popup_anchor(dock_position)
margin = self._get_popup_margin(dock_position)
super().__init__(
anchor=popup_anchor,
margin=margin,
child=main_box,
layer="top",
exclusive=False,
keyboard_mode="on-demand",
visible=True,
all_visible=True,
)
self.connect("button-press-event", self.on_button_press)
self.connect("key-press-event", self.on_key_press)
self.set_can_focus(True)
self.grab_focus()
self.set_tooltip_text(
"Keyboard shortcuts:\n• Escape: Close popup\n• Ctrl+D: Toggle DND\n• Ctrl+A: Clear all notifications"
)
def _get_popup_anchor(self, dock_position):
anchor_map = {
"Top": "top", # Dock at top -> popup at top right
"Bottom": "bottom", # Dock at bottom -> popup at bottom right
"Left": "left", # Dock at left -> popup at top left
"Right": "right", # Dock at right -> popup at top right
}
return anchor_map.get(dock_position, "bottom")
def _get_popup_margin(self, dock_position):
margin_map = {
"Top": "60px 10px 10px 10px",
"Bottom": "10px 10px 60px 10px",
"Left": "10px 10px 10px 60px",
"Right": "10px 60px 10px 10px",
}
return margin_map.get(dock_position, "10px 10px 10px 10px")
def refresh_popup(self):
pass
def on_key_press(self, _widget, event):
if event.keyval == Gdk.KEY_Escape:
self.set_visible(False)
return True
elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_d:
current_dnd = self.notification_history.do_not_disturb_enabled
self.notification_history.header_switch.set_active(not current_dnd)
return True
elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_a:
self.notification_history.clear_history()
return True
return False
def on_button_press(self, _widget, _event):
return False
class Notifications(Box):
def __init__(self, **kwargs):
super().__init__(
orientation="h" if not data.VERTICAL else "v",
spacing=4,
children=[NotificationIndicator()],
**kwargs,
)
self.show_all()