♻️ refactor(network): use NetworkClient and update connection handling
This commit is contained in:
4 files changed
+22
-989
No files matched your search
@@ -1,7 +1,7 @@
|
||||
from fabric.widgets.label import Label
|
||||
|
||||
from modules.dashboard.tile import Tile
|
||||
from services.nett import NetworkService
|
||||
from services.network import NetworkClient
|
||||
|
||||
import utils.icons as icons
|
||||
|
||||
@@ -22,13 +22,21 @@ class Network(Tile):
|
||||
menu=True,
|
||||
**kwargs,
|
||||
)
|
||||
self.nm = NetworkService()
|
||||
self.nm.connect("connection-change", self.handle_connection_change)
|
||||
self.nm.init_props()
|
||||
self.nm = NetworkClient()
|
||||
self.nm.connect("changed", self.handle_connection_change)
|
||||
self.nm.connect("ready", self.handle_connection_change)
|
||||
|
||||
def handle_connection_change(self, *args):
|
||||
# Get current network state from NetworkClient properties
|
||||
connected = False
|
||||
con_label = "Disconnected"
|
||||
|
||||
if self.nm.wifi_device and self.nm.wifi_device.active_access_point:
|
||||
connected = True
|
||||
con_label = self.nm.wifi_device.active_access_point.ssid
|
||||
elif not self.nm.wireless_enabled:
|
||||
con_label = "Off"
|
||||
|
||||
def handle_connection_change(
|
||||
self, source, con_label: str, connected: bool, status: str
|
||||
):
|
||||
if self.state != connected:
|
||||
self.state = connected
|
||||
if self.state:
|
||||
@@ -39,14 +47,4 @@ class Network(Tile):
|
||||
self.add_style_class("off")
|
||||
print("State change:", connected)
|
||||
|
||||
if not connected:
|
||||
if status == "Wi-Fi Off":
|
||||
self.label.set_label("Off")
|
||||
elif status == "Wi-Fi On (No Connection)":
|
||||
self.label.set_label("Disconnected")
|
||||
elif status == "Connecting…":
|
||||
self.label.set_label("Connecting…")
|
||||
else:
|
||||
self.label.set_label("Unknown")
|
||||
else:
|
||||
self.label.set_label(con_label or "Connected")
|
||||
self.label.set_label(con_label)
|
||||
@@ -12,7 +12,6 @@ from gi.repository import GLib
|
||||
|
||||
import config.data as data
|
||||
import utils.icons as icons
|
||||
from .notifications import Notifications
|
||||
from services.network import NetworkClient
|
||||
|
||||
|
||||
@@ -188,16 +187,14 @@ class RecordingIndicator(Button):
|
||||
if data.VERTICAL:
|
||||
# Vertical dock: show only icon
|
||||
self.recording_box = Box(
|
||||
orientation="v",
|
||||
spacing=2,
|
||||
children=[self.recording_label]
|
||||
orientation="v", spacing=2, children=[self.recording_label]
|
||||
)
|
||||
else:
|
||||
# Horizontal dock: place icon and time side by side
|
||||
self.recording_box = Box(
|
||||
orientation="h",
|
||||
spacing=4,
|
||||
children=[self.recording_label, self.time_label]
|
||||
children=[self.recording_label, self.time_label],
|
||||
)
|
||||
self.add(self.recording_box)
|
||||
|
||||
@@ -232,7 +229,10 @@ class RecordingIndicator(Button):
|
||||
if not data.VERTICAL:
|
||||
time_text = f"{minutes:02d}:{seconds:02d}"
|
||||
self.time_label.set_markup(time_text)
|
||||
self.set_tooltip_text(f"Recording in progress ({minutes:02d}:{seconds:02d}) - Click to stop")
|
||||
self.set_tooltip_text(
|
||||
f"Recording in progress ({minutes:02d}:{
|
||||
seconds:02d}) - Click to stop"
|
||||
)
|
||||
else:
|
||||
self.set_tooltip_text("Recording in progress - Click to stop")
|
||||
else:
|
||||
@@ -279,7 +279,6 @@ class RecordingIndicator(Button):
|
||||
class Indicators(Box):
|
||||
def __init__(self, **kwargs):
|
||||
self.recording_indicator = RecordingIndicator()
|
||||
self.notifications = Notifications()
|
||||
|
||||
super().__init__(
|
||||
name="indicator",
|
||||
@@ -288,7 +287,6 @@ class Indicators(Box):
|
||||
children=[
|
||||
WifiIndicator(),
|
||||
BluetoothIndicator(),
|
||||
self.notifications,
|
||||
self.recording_indicator,
|
||||
],
|
||||
**kwargs,
|
||||
|
||||
@@ -1,851 +0,0 @@
|
||||
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()
|
||||
@@ -1,112 +0,0 @@
|
||||
import gi
|
||||
from fabric.core.service import Service, Signal
|
||||
from loguru import logger
|
||||
|
||||
try:
|
||||
gi.require_version("NM", "1.0")
|
||||
from gi.repository import NM
|
||||
except ValueError:
|
||||
logger.error("Failed to start network manager")
|
||||
|
||||
|
||||
class NetworkService(Service):
|
||||
|
||||
@Signal
|
||||
def connection_change(
|
||||
self, ssid: str, connected: bool, status: str = ""
|
||||
) -> None: ...
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.client = NM.Client.new(None)
|
||||
self.client.connect("device-added", self.on_device_added)
|
||||
self.client.connect("device-removed", self.on_device_removed)
|
||||
# self.client.connect("notify::state", self.handle_client_state_change)
|
||||
|
||||
# active_connections = self.client.get_active_connections()
|
||||
# for ac in active_connections:
|
||||
# connection = ac.get_connection()
|
||||
# type = ac.get_connection_type()
|
||||
# print("Active connection:", connection.get_id(), type)
|
||||
|
||||
devices = self.client.get_devices()
|
||||
for dev in devices:
|
||||
if dev.get_device_type() == NM.DeviceType.WIFI:
|
||||
self.wifi_dev = dev
|
||||
active_ap = self.wifi_dev.get_active_access_point()
|
||||
if active_ap:
|
||||
ssid = active_ap.get_ssid()
|
||||
print("Connected SSID:", ssid.get_data().decode("utf-8"))
|
||||
|
||||
if self.wifi_dev:
|
||||
self.wifi_dev.connect("state-changed", self.handle_device_state_change)
|
||||
self.wifi_dev.connect(
|
||||
"notify::active-access-point", self.handle_prop_change
|
||||
)
|
||||
|
||||
def init_props(self):
|
||||
self.handle_prop_change(self.wifi_dev, None)
|
||||
|
||||
def handle_prop_change(self, source, pspec):
|
||||
if source is None:
|
||||
self.connection_change("", False, "No Device")
|
||||
return
|
||||
|
||||
device_state = source.get_state()
|
||||
if device_state == NM.DeviceState.UNAVAILABLE:
|
||||
self.connection_change("", False, "Wi-Fi Off")
|
||||
return
|
||||
|
||||
active_ap = source.get_active_access_point()
|
||||
if active_ap:
|
||||
ssid = active_ap.get_ssid().get_data().decode("utf-8")
|
||||
print("Connected SSID:", ssid)
|
||||
self.connection_change(ssid, True, "Connected")
|
||||
else:
|
||||
self.connection_change("", False, "Wi-Fi On (No Connection)")
|
||||
|
||||
def handle_device_state_change(self, device, new_state, old_state, reason):
|
||||
print("Device state changed:", old_state, "→", new_state)
|
||||
match new_state:
|
||||
case NM.DeviceState.UNAVAILABLE:
|
||||
self.connection_change("", False, "Wi-Fi Off")
|
||||
case NM.DeviceState.DISCONNECTED:
|
||||
self.connection_change("", False, "Wi-Fi On (No Connection)")
|
||||
case (
|
||||
NM.DeviceState.PREPARE
|
||||
| NM.DeviceState.CONFIG
|
||||
| NM.DeviceState.NEED_AUTH
|
||||
| NM.DeviceState.IP_CONFIG
|
||||
| NM.DeviceState.IP_CHECK
|
||||
| NM.DeviceState.SECONDARIES
|
||||
):
|
||||
self.connection_change("", False, "Connecting…")
|
||||
case NM.DeviceState.ACTIVATED:
|
||||
active_ap = device.get_active_access_point()
|
||||
if active_ap:
|
||||
ssid = active_ap.get_ssid().get_data().decode("utf-8")
|
||||
self.connection_change(ssid, True, "Connected")
|
||||
else:
|
||||
self.connection_change("", True, "Connected (No SSID)")
|
||||
case _:
|
||||
self.connection_change("", False, "Unknown")
|
||||
|
||||
def on_device_added(self, source, device):
|
||||
print("Added device")
|
||||
if device.get_device_type() == NM.DeviceType.WIFI:
|
||||
self.wifi_dev = device
|
||||
self.wifi_dev.connect("state-changed", self.handle_device_state_change)
|
||||
self.wifi_dev.connect(
|
||||
"notify::active-access-point", self.handle_prop_change
|
||||
)
|
||||
|
||||
def on_device_removed(self, client, device):
|
||||
print("Removed device")
|
||||
if self.wifi_dev and device.get_path() == self.wifi_dev.get_path():
|
||||
self.wifi_dev = None
|
||||
self.handle_prop_change(self.wifi_dev, None)
|
||||
|
||||
def print_state(self):
|
||||
print(
|
||||
"wifi: ", self.wifi_dev.get_state(), " client: ", self.client.get_state()
|
||||
)
|
||||
Reference in new issue
Block a user