feat(root): implement dynamic todo list refresh and UI enhancements

This change adds a mechanism to refresh the todo list dynamically when the underlying service updates, improving UI responsiveness and user experience. It also updates documentation and removes deprecated files.

- New Features:
  - Added `_refresh_todos` method to rebuild the todo list when `get_todo_service().todos` changes.
  - Introduced `TodoListCapture` class for mouse capture handling of the todo list window.
- Enhancements:
  - Updated README with corrected image paths, installation tips, and manual installation commands.
  - Introduced `start.py` to launch the application or lock mode.
  - Removed deprecated `config.json` and `debug_memory.py`.
- Bug Fixes:
  - Fixed lock screen screenshot path in README.
This commit is contained in:
sankalp committed 2026-04-10 17:11:32 +05:45
1 parent 235f53b6a2
commit e4813f35a0
267 files changed
+24582 -394

No files matched your search

Whitespace-only changes.
+615
View File
@@ -0,0 +1,615 @@
import subprocess
from fabric.bluetooth import BluetoothClient, BluetoothDevice
from fabric.utils import Gdk, GLib, Gtk, exec_shell_command, logger
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.centerbox import CenterBox
from fabric.widgets.image import Image
from fabric.widgets.label import Label
from fabric.widgets.revealer import Revealer
from fabric.widgets.scrolledwindow import ScrolledWindow
from fabric.widgets.separator import Separator
from utils.utils import svg_file
def get_battery_icon_file(
percentage: int, is_charging: bool, icon_path: str = ""
) -> str:
"""
Get the battery icon file path based on percentage and charging status.
Args:
percentage: Battery percentage (0-100)
is_charging: Whether the device is charging
icon_path: Base path for icons (unused, kept for compatibility)
Returns:
Relative path to the battery icon file
"""
clamped = max(0, min(100, percentage))
step = (clamped // 10) * 10
filename = f"battery-{step:03d}{'-charging' if is_charging else ''}.svg"
return f"battery/{filename}"
def set_bluetooth_enabled_with_fallback(client, enabled: bool):
try:
# Try fabric bluetooth first
client.set_enabled(enabled)
except Exception as e:
logger.warning(f"Fabric bluetooth set_enabled({enabled}) failed: {e}")
# Fallback to rfkill to unblock/block bluetooth
if enabled:
command = "rfkill unblock bluetooth"
else:
command = "rfkill block bluetooth"
result = exec_shell_command(command)
if result is False:
logger.error(f"rfkill fallback failed: command '{command}' returned False")
elif isinstance(result, str) and result.strip():
# If result is a non-empty string, it might be an error message
logger.warning(f"rfkill command output: {result.strip()}")
else:
logger.info(f"rfkill fallback succeeded: {command}")
class BluetoothDeviceSlot(CenterBox):
def __init__(self, device: BluetoothDevice, **kwargs):
super().__init__(h_expand=True, name="device-button", **kwargs)
self.device = device
self.device.connect("changed", self.on_changed)
self.device.connect(
"notify::closed", lambda *_: self.device.closed and self.destroy()
)
self.styles = [
"connected" if self.device.connected else "",
"paired" if self.device.paired else "",
]
self.dimage = Image(
icon_name=device.icon_name + "-symbolic", # type: ignore
size=5,
name="device-icon",
style_classes=" ".join(self.styles),
)
self.device_button = Button(
on_clicked=lambda *_: self.toggle_connecting(),
child=Box(
orientation="h",
h_expand=True,
children=[self.dimage, Label(label=device.name)],
), # type: ignore
)
self.start_children = [self.device_button]
# # Add battery info if available
if hasattr(device, "battery_percentage") and device.battery_percentage > 0:
battery_box = Box(orientation="h", spacing=4)
# Create battery icon
battery_icon = svg_file(
get_battery_icon_file(
device.battery_percentage,
False, # Not charging for bluetooth devices
)
)
# Create battery percentage label
battery_label = Label(
label=f"{device.battery_percentage:.0f}%", name="battery-label"
)
battery_box.children = [battery_icon, battery_label]
self.end_children = [battery_box]
self.device_button.connect("enter-notify-event", self.on_button_enter)
self.device_button.connect("leave-notify-event", self.on_button_leave)
self.device.emit("changed") # to update display status
def on_button_enter(self, widget, event):
self.add_style_class("button-hovered")
def on_button_leave(self, widget, event):
self.remove_style_class("button-hovered")
def toggle_connecting(self):
self.device.emit("changed")
self.device.set_connecting(not self.device.connected)
def on_changed(self, *_):
try:
# Update connection and pairing status
new_styles = [
"connected" if self.device.connected else "",
"paired" if self.device.paired else "",
]
self.styles = new_styles
self.dimage.set_property("style-classes", " ".join(self.styles))
except Exception:
return
# Update battery info if available
if (
hasattr(self.device, "battery_percentage")
and self.device.battery_percentage > 0
):
if not self.end_children: # Add battery display if not already present
battery_box = Box(orientation="h", spacing=4)
# Create battery icon
battery_icon = svg_file(
get_battery_icon_file(
self.device.battery_percentage,
False, # Not charging for bluetooth devices
)
)
# Create battery percentage label
battery_label = Label(
label=f"{self.device.battery_percentage:.0f}%", name="battery-label"
)
battery_box.children = [battery_icon, battery_label]
self.end_children = [battery_box]
else: # Update existing battery display
battery_box = self.end_children[0]
if hasattr(battery_box, "children") and len(battery_box.children) >= 2:
battery_icon = battery_box.children[0]
battery_label = battery_box.children[1]
# Update battery icon
battery_icon.dynamic_file(
get_battery_icon_file(
self.device.battery_percentage,
False, # Not charging for bluetooth devices
)
)
# Update battery percentage
battery_label.set_label(f"{self.device.battery_percentage:.0f}%")
elif self.end_children: # Remove battery display if no longer available
self.end_children = []
return
class BluetoothConnections(Box):
def __init__(
self, parent, show_hidden_devices: bool = False, show_back_button=True, **kwargs
):
super().__init__(
spacing=8,
orientation="vertical",
name="bluetooth-connections",
**kwargs,
)
self.parent = parent
self.show_hidden_devices = show_hidden_devices
self.is_scanning = False # Track scanning state
self.refresh_timer = None # Timer for periodic device refresh
self._update_in_progress = False # Prevent concurrent updates
self._destroyed = False # Track if widget is destroyed
self.client = BluetoothClient(on_device_added=self.on_device_added)
# Create pull-to-refresh indicator
self.refresh_indicator = Label(
name="bluetooth-refresh-indicator",
label="↓ Pull to scan for devices",
h_align="center",
visible=False,
style="color: #fff; font-size: 12px; padding: 5px;",
)
# Create title with optional back button
title_children = []
if show_back_button:
title_children.append(
Button(
image=Image(icon_name="back", size=10),
on_clicked=lambda *_: self.parent.close_bluetooth(),
)
)
title_children.append(Label("Bluetooth", name="bluetooth-title"))
self.title = Box(
orientation="h",
children=title_children,
)
self.toggle_button = Gtk.Switch(visible=True, name="toggle-button")
# Safely set initial state
self.toggle_button.set_active(self.client.enabled)
self.toggle_button.connect(
"notify::active",
lambda *_: set_bluetooth_enabled_with_fallback(
self.client, self.toggle_button.get_active()
),
)
# Connect client signals
self.client.connect(
"notify::enabled",
lambda *_: self.toggle_button.set_active(self.client.enabled),
)
self.client.connect("notify::scanning", lambda *_: self.update_scan_label())
# Connect to device changes
self.client.connect("device-added", self.update_devices)
self.client.connect("device-removed", self.update_devices)
# Connect to additional signals for better real-time monitoring
self.client.connect("changed", self.on_client_changed)
# Create Devices section
self.paired_devices_label = Label(
label="Devices", h_align="start", name="networks-title"
)
self.paired_devices = Box(
spacing=4, orientation="vertical", name="known-networks"
)
# Create "No devices available" message
self.no_devices_label = Label(
label="No devices available",
h_align="center",
name="no-networks-label",
visible=False,
)
# Create Other Devices section with clickable title
self.other_devices_button = Button(
child=Label("Other Devices", h_align="start"),
name="wifi-other-button",
on_clicked=self.toggle_other_devices,
)
self.other_devices = Box(spacing=4, orientation="vertical")
# Create scrolled window for other devices
self.other_devices_scrolled = ScrolledWindow(
min_content_size=(303, 150),
child=self.other_devices,
overlay_scroll=True,
)
# Add pull-to-refresh functionality to scrolled window
self.setup_pull_to_refresh()
# Create revealer for Other Devices section
self.other_devices_revealer = Revealer(
child=self.other_devices_scrolled,
transition_type="slide-down",
transition_duration=100,
child_revealed=False,
)
# Create More Settings button (same style as Other Devices button)
self.more_settings_button = Button(
child=Label("More Settings", h_align="start"),
name="wifi-other-button",
on_clicked=self.open_bluetooth_settings,
)
self.children = [
CenterBox(
start_children=self.title,
end_children=self.toggle_button,
name="bluetooth-widget-top",
),
self.refresh_indicator,
Separator(orientation="h", name="separator"),
self.paired_devices_label,
self.paired_devices,
self.no_devices_label,
Separator(orientation="h", name="separator"),
self.other_devices_button,
self.other_devices_revealer,
Separator(orientation="h", name="separator"),
self.more_settings_button,
]
# Connect cleanup on destroy
self.connect("destroy", self.on_destroy)
self.client.notify("scanning")
self.client.notify("enabled")
# Initial device update
self.update_devices()
# Start periodic device monitoring for real-time updates
self.start_device_monitoring()
def toggle_other_devices(self, *_):
"""Toggle the visibility of other devices section"""
current_state = self.other_devices_revealer.child_revealed
self.other_devices_revealer.child_revealed = not current_state
# Handle scanning based on section visibility
if self.client:
if self.other_devices_revealer.child_revealed:
# Start scanning when revealing other devices section
if not self.client.scanning:
self.client.toggle_scan()
# Also force an immediate device refresh to catch any missed connections
self.force_device_refresh()
else:
# Stop scanning when hiding other devices section
if self.client.scanning:
self.client.toggle_scan()
def open_bluetooth_settings(self, *_):
"""Open Blueman bluetooth manager"""
try:
subprocess.Popen(["blueman-manager"], start_new_session=True)
if self.parent and hasattr(self.parent, "hide_controlcenter"):
self.parent.hide_controlcenter()
except FileNotFoundError:
pass
except Exception:
pass
def update_scan_label(self):
"""Update scanning state appearance"""
if self.client.scanning:
# Show scanning feedback in refresh indicator
self.refresh_indicator.set_label("Scanning for devices...")
self.refresh_indicator.set_visible(True)
self.refresh_indicator.add_style_class("scanning")
else:
# Hide scanning feedback
self.refresh_indicator.set_visible(False)
self.refresh_indicator.remove_style_class("scanning")
def update_devices(self, *_):
"""Update the list of available devices"""
# Prevent concurrent updates and check if destroyed
if self._update_in_progress or self._destroyed or not self.client:
return
self._update_in_progress = True
try:
# Store current device addresses to detect changes
current_paired_addresses = {
child.device.address
for child in self.paired_devices.get_children()
if hasattr(child, "device")
}
current_other_addresses = {
child.device.address
for child in self.other_devices.get_children()
if hasattr(child, "device")
}
# Get current devices safely
devices = self.client.devices
paired_devices = []
other_devices = []
new_paired_addresses = set()
new_other_addresses = set()
for device in devices:
try:
if device.name and device.name != "Unknown":
# Categorize devices: paired devices go to "Devices (Paired)"
# All others go to "Other Devices"
if device.paired:
paired_devices.append(device)
new_paired_addresses.add(device.address)
else:
other_devices.append(device)
new_other_addresses.add(device.address)
except Exception:
continue
# Check if we need to update (devices added/removed)
paired_changed = current_paired_addresses != new_paired_addresses
other_changed = current_other_addresses != new_other_addresses
# Only rebuild if something actually changed
if paired_changed or other_changed:
# Clear existing devices safely
for child in list(self.paired_devices.get_children()):
if not self._destroyed:
child.destroy()
for child in list(self.other_devices.get_children()):
if not self._destroyed:
child.destroy()
# Add paired devices
for device in paired_devices:
if not self._destroyed:
device_slot = BluetoothDeviceSlot(device)
self.paired_devices.add(device_slot)
# Add other devices
for device in other_devices:
if not self._destroyed:
device_slot = BluetoothDeviceSlot(device)
self.other_devices.add(device_slot)
# Show/hide sections based on available devices
if not self._destroyed:
has_paired_devices = len(paired_devices) > 0
has_other_devices = len(other_devices) > 0
has_any_devices = has_paired_devices or has_other_devices
# Show paired devices section only if there are paired devices
self.paired_devices_label.set_visible(has_paired_devices)
self.paired_devices.set_visible(has_paired_devices)
# Show "No devices available" message if no devices at all
self.no_devices_label.set_visible(not has_any_devices)
# Always show the other devices button, regardless of available devices
self.other_devices_button.set_visible(True) # Always visible
except Exception:
pass
finally:
self._update_in_progress = False
def start_device_monitoring(self):
"""Start periodic monitoring for device changes"""
# Monitor for device changes every 5 seconds (less aggressive)
# This helps catch devices that connect from external sources
self.refresh_timer = GLib.timeout_add_seconds(5, self.periodic_device_refresh)
def stop_device_monitoring(self):
"""Stop periodic monitoring"""
if self.refresh_timer:
GLib.source_remove(self.refresh_timer)
self.refresh_timer = None
def periodic_device_refresh(self):
"""Periodically refresh device list to catch external connections"""
# Skip if update in progress, destroyed, or client not available
if (
self._update_in_progress
or self._destroyed
or not self.client
or not self.client.enabled
):
return True # Continue monitoring
try:
# Simple check - just trigger update_devices which has its own safety checks
# Don't force signal emissions as that can cause race conditions
self.update_devices()
except Exception:
pass
return True # Continue monitoring
def force_device_refresh(self):
"""Force an immediate refresh of the device list"""
if self._update_in_progress or self._destroyed:
return
try:
# Simply trigger update_devices which has its own safety checks
# Avoid forcing signal emissions to prevent race conditions
self.update_devices()
except Exception:
pass
def on_client_changed(self, *_):
"""Handle when the bluetooth client state changes"""
# Update devices when client state changes
self.update_devices()
def on_destroy(self, widget):
"""Cleanup when widget is destroyed"""
# Mark as destroyed to prevent further updates
self._destroyed = True
# Stop monitoring
self.stop_device_monitoring()
# Make sure other devices revealer is collapsed when closing
try:
self.other_devices_revealer.child_revealed = False
except Exception:
pass # Widget might already be destroyed
def close_bluetooth(self):
"""Called when Bluetooth panel is being closed"""
# Collapse the other devices section when closing
self.other_devices_revealer.child_revealed = False
def setup_pull_to_refresh(self):
"""Setup pull-to-refresh gesture for the scrolled window"""
# Get the scrolled window's vertical adjustment
self.vadjustment = self.other_devices_scrolled.get_vadjustment()
# Track gesture state
self.pull_start_y = 0
self.is_pulling = False
self.pull_threshold = 50 # pixels to trigger refresh
# Connect to scroll events
self.other_devices_scrolled.connect("scroll-event", self.on_scroll_event)
self.other_devices_scrolled.connect("button-press-event", self.on_button_press)
self.other_devices_scrolled.connect(
"button-release-event", self.on_button_release
)
self.other_devices_scrolled.connect(
"motion-notify-event", self.on_motion_notify
)
# Enable events
self.other_devices_scrolled.set_events(
Gdk.EventMask.SCROLL_MASK
| Gdk.EventMask.BUTTON_PRESS_MASK
| Gdk.EventMask.BUTTON_RELEASE_MASK
| Gdk.EventMask.POINTER_MOTION_MASK
)
def on_scroll_event(self, widget, event):
"""Handle scroll events for pull-to-refresh"""
# Only handle pull-to-refresh when at the top
if self.vadjustment.get_value() <= 0:
if event.direction == Gdk.ScrollDirection.UP:
# Scrolling up at the top - toggle scan and force refresh
self.client.toggle_scan()
self.force_device_refresh()
return True # Consume the event
return False # Let normal scrolling continue
def on_button_press(self, widget, event):
"""Handle button press for touch/drag gestures"""
if self.vadjustment.get_value() <= 0:
self.pull_start_y = event.y
self.is_pulling = True
return False
def on_button_release(self, widget, event):
"""Handle button release for touch/drag gestures"""
if self.is_pulling:
pull_distance = event.y - self.pull_start_y
if pull_distance > self.pull_threshold:
# Toggle scan and force refresh
self.client.toggle_scan()
self.force_device_refresh()
# Hide refresh indicator
self.refresh_indicator.set_visible(False)
self.refresh_indicator.remove_style_class("ready-to-refresh")
self.is_pulling = False
return False
def on_motion_notify(self, widget, event):
"""Handle motion events for visual feedback during pull"""
if self.is_pulling and self.vadjustment.get_value() <= 0:
pull_distance = event.y - self.pull_start_y
if pull_distance > 0:
# Show refresh indicator when pulling down
self.refresh_indicator.set_visible(True)
if pull_distance >= self.pull_threshold:
if self.client.scanning:
self.refresh_indicator.set_label("↑ Release to stop scanning")
else:
self.refresh_indicator.set_label("↑ Release to scan")
self.refresh_indicator.add_style_class("ready-to-refresh")
else:
if self.client.scanning:
self.refresh_indicator.set_label("↓ Pull to stop scanning")
else:
self.refresh_indicator.set_label("↓ Pull to scan for devices")
self.refresh_indicator.remove_style_class("ready-to-refresh")
else:
self.refresh_indicator.set_visible(False)
return False
def on_device_added(self, client: BluetoothClient, address: str):
"""Handle when a new device is added"""
# Update the device list when devices are added
self.update_devices()
+615
View File
@@ -0,0 +1,615 @@
import subprocess
from fabric.bluetooth import BluetoothClient, BluetoothDevice
from fabric.utils import Gdk, GLib, Gtk, exec_shell_command, logger
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.centerbox import CenterBox
from fabric.widgets.image import Image
from fabric.widgets.label import Label
from fabric.widgets.revealer import Revealer
from fabric.widgets.scrolledwindow import ScrolledWindow
from fabric.widgets.separator import Separator
from utils.utils import svg_file
def get_battery_icon_file(
percentage: int, is_charging: bool, icon_path: str = ""
) -> str:
"""
Get the battery icon file path based on percentage and charging status.
Args:
percentage: Battery percentage (0-100)
is_charging: Whether the device is charging
icon_path: Base path for icons (unused, kept for compatibility)
Returns:
Relative path to the battery icon file
"""
clamped = max(0, min(100, percentage))
step = int((clamped // 10) * 10)
filename = f"battery-{step:03d}{'-charging' if is_charging else ''}.svg"
return f"battery/{filename}"
def set_bluetooth_enabled_with_fallback(client, enabled: bool):
try:
# Try fabric bluetooth first
client.set_enabled(enabled)
except Exception as e:
logger.warning(f"Fabric bluetooth set_enabled({enabled}) failed: {e}")
# Fallback to rfkill to unblock/block bluetooth
if enabled:
command = "rfkill unblock bluetooth"
else:
command = "rfkill block bluetooth"
result = exec_shell_command(command)
if result is False:
logger.error(f"rfkill fallback failed: command '{command}' returned False")
elif isinstance(result, str) and result.strip():
# If result is a non-empty string, it might be an error message
logger.warning(f"rfkill command output: {result.strip()}")
else:
logger.info(f"rfkill fallback succeeded: {command}")
class BluetoothDeviceSlot(CenterBox):
def __init__(self, device: BluetoothDevice, **kwargs):
super().__init__(h_expand=True, name="device-button", **kwargs)
self.device = device
self.device.connect("changed", self.on_changed)
self.device.connect(
"notify::closed", lambda *_: self.device.closed and self.destroy()
)
self.styles = [
"connected" if self.device.connected else "",
"paired" if self.device.paired else "",
]
self.dimage = Image(
icon_name=device.icon_name + "-symbolic", # type: ignore
size=5,
name="device-icon",
style_classes=" ".join(self.styles),
)
self.device_button = Button(
on_clicked=lambda *_: self.toggle_connecting(),
child=Box(
orientation="h",
h_expand=True,
children=[self.dimage, Label(label=device.name)],
), # type: ignore
)
self.start_children = [self.device_button]
# # Add battery info if available
if hasattr(device, "battery_percentage") and device.battery_percentage > 0:
battery_box = Box(orientation="h", spacing=4)
# Create battery icon
battery_icon = svg_file(
get_battery_icon_file(
device.battery_percentage,
False, # Not charging for bluetooth devices
)
)
# Create battery percentage label
battery_label = Label(
label=f"{device.battery_percentage:.0f}%", name="battery-label"
)
battery_box.children = [battery_icon, battery_label]
self.end_children = [battery_box]
self.device_button.connect("enter-notify-event", self.on_button_enter)
self.device_button.connect("leave-notify-event", self.on_button_leave)
self.device.emit("changed") # to update display status
def on_button_enter(self, widget, event):
self.add_style_class("button-hovered")
def on_button_leave(self, widget, event):
self.remove_style_class("button-hovered")
def toggle_connecting(self):
self.device.emit("changed")
self.device.set_connecting(not self.device.connected)
def on_changed(self, *_):
try:
# Update connection and pairing status
new_styles = [
"connected" if getattr(self.device, "connected", False) else "",
"paired" if getattr(self.device, "paired", False) else "",
]
self.styles = new_styles
self.dimage.set_property("style-classes", " ".join(self.styles))
except Exception:
return
# Update battery info if available
if (
hasattr(self.device, "battery_percentage")
and self.device.battery_percentage > 0
):
if not self.end_children: # Add battery display if not already present
battery_box = Box(orientation="h", spacing=4)
# Create battery icon
battery_icon = svg_file(
get_battery_icon_file(
self.device.battery_percentage,
False, # Not charging for bluetooth devices
)
)
# Create battery percentage label
battery_label = Label(
label=f"{self.device.battery_percentage:.0f}%", name="battery-label"
)
battery_box.children = [battery_icon, battery_label]
self.end_children = [battery_box]
else: # Update existing battery display
battery_box = self.end_children[0]
if hasattr(battery_box, "children") and len(battery_box.children) >= 2:
battery_icon = battery_box.children[0]
battery_label = battery_box.children[1]
# Update battery icon
battery_icon.dynamic_file(
get_battery_icon_file(
self.device.battery_percentage,
False, # Not charging for bluetooth devices
)
)
# Update battery percentage
battery_label.set_label(f"{self.device.battery_percentage:.0f}%")
elif self.end_children: # Remove battery display if no longer available
self.end_children = []
return
class BluetoothConnections(Box):
def __init__(
self, parent, show_hidden_devices: bool = False, show_back_button=True, **kwargs
):
super().__init__(
spacing=8,
orientation="vertical",
name="bluetooth-connections",
**kwargs,
)
self.parent = parent
self.show_hidden_devices = show_hidden_devices
self.is_scanning = False # Track scanning state
self.refresh_timer = None # Timer for periodic device refresh
self._update_in_progress = False # Prevent concurrent updates
self._destroyed = False # Track if widget is destroyed
self.client = BluetoothClient(on_device_added=self.on_device_added)
# Create pull-to-refresh indicator
self.refresh_indicator = Label(
name="bluetooth-refresh-indicator",
label="↓ Pull to scan for devices",
h_align="center",
visible=False,
style="color: #fff; font-size: 12px; padding: 5px;",
)
# Create title with optional back button
title_children = []
if show_back_button:
title_children.append(
Button(
image=Image(icon_name="back", size=10),
on_clicked=lambda *_: self.parent.close_bluetooth(),
)
)
title_children.append(Label("Bluetooth", name="bluetooth-title"))
self.title = Box(
orientation="h",
children=title_children,
)
self.toggle_button = Gtk.Switch(visible=True, name="toggle-button")
# Safely set initial state
self.toggle_button.set_active(self.client.enabled)
self.toggle_button.connect(
"notify::active",
lambda *_: set_bluetooth_enabled_with_fallback(
self.client, self.toggle_button.get_active()
),
)
# Connect client signals
self.client.connect(
"notify::enabled",
lambda *_: self.toggle_button.set_active(self.client.enabled),
)
self.client.connect("notify::scanning", lambda *_: self.update_scan_label())
# Connect to device changes
self.client.connect("device-added", self.update_devices)
self.client.connect("device-removed", self.update_devices)
# Connect to additional signals for better real-time monitoring
self.client.connect("changed", self.on_client_changed)
# Create Devices section
self.paired_devices_label = Label(
label="Devices", h_align="start", name="networks-title"
)
self.paired_devices = Box(
spacing=4, orientation="vertical", name="known-networks"
)
# Create "No devices available" message
self.no_devices_label = Label(
label="No devices available",
h_align="center",
name="no-networks-label",
visible=False,
)
# Create Other Devices section with clickable title
self.other_devices_button = Button(
child=Label("Other Devices", h_align="start"),
name="wifi-other-button",
on_clicked=self.toggle_other_devices,
)
self.other_devices = Box(spacing=4, orientation="vertical")
# Create scrolled window for other devices
self.other_devices_scrolled = ScrolledWindow(
min_content_size=(303, 150),
child=self.other_devices,
overlay_scroll=True,
)
# Add pull-to-refresh functionality to scrolled window
self.setup_pull_to_refresh()
# Create revealer for Other Devices section
self.other_devices_revealer = Revealer(
child=self.other_devices_scrolled,
transition_type="slide-down",
transition_duration=100,
child_revealed=False,
)
# Create More Settings button (same style as Other Devices button)
self.more_settings_button = Button(
child=Label("More Settings", h_align="start"),
name="wifi-other-button",
on_clicked=self.open_bluetooth_settings,
)
self.children = [
CenterBox(
start_children=self.title,
end_children=self.toggle_button,
name="bluetooth-widget-top",
),
self.refresh_indicator,
Separator(orientation="h", name="separator"),
self.paired_devices_label,
self.paired_devices,
self.no_devices_label,
Separator(orientation="h", name="separator"),
self.other_devices_button,
self.other_devices_revealer,
Separator(orientation="h", name="separator"),
self.more_settings_button,
]
# Connect cleanup on destroy
self.connect("destroy", self.on_destroy)
self.client.notify("scanning")
self.client.notify("enabled")
# Initial device update
self.update_devices()
# Start periodic device monitoring for real-time updates
self.start_device_monitoring()
def toggle_other_devices(self, *_):
"""Toggle the visibility of other devices section"""
current_state = self.other_devices_revealer.child_revealed
self.other_devices_revealer.child_revealed = not current_state
# Handle scanning based on section visibility
if self.client:
if self.other_devices_revealer.child_revealed:
# Start scanning when revealing other devices section
if not self.client.scanning:
self.client.toggle_scan()
# Also force an immediate device refresh to catch any missed connections
self.force_device_refresh()
else:
# Stop scanning when hiding other devices section
if self.client.scanning:
self.client.toggle_scan()
def open_bluetooth_settings(self, *_):
"""Open Blueman bluetooth manager"""
try:
subprocess.Popen(["blueman-manager"], start_new_session=True)
if self.parent and hasattr(self.parent, "hide_controlcenter"):
self.parent.hide_controlcenter()
except FileNotFoundError:
pass
except Exception:
pass
def update_scan_label(self):
"""Update scanning state appearance"""
if self.client.scanning:
# Show scanning feedback in refresh indicator
self.refresh_indicator.set_label("Scanning for devices...")
self.refresh_indicator.set_visible(True)
self.refresh_indicator.add_style_class("scanning")
else:
# Hide scanning feedback
self.refresh_indicator.set_visible(False)
self.refresh_indicator.remove_style_class("scanning")
def update_devices(self, *_):
"""Update the list of available devices"""
# Prevent concurrent updates and check if destroyed
if self._update_in_progress or self._destroyed or not self.client:
return
self._update_in_progress = True
try:
# Store current device addresses to detect changes
current_paired_addresses = {
child.device.address
for child in self.paired_devices.get_children()
if hasattr(child, "device")
}
current_other_addresses = {
child.device.address
for child in self.other_devices.get_children()
if hasattr(child, "device")
}
# Get current devices safely
devices = self.client.devices
paired_devices = []
other_devices = []
new_paired_addresses = set()
new_other_addresses = set()
for device in devices:
try:
if device.name and device.name != "Unknown":
# Categorize devices: paired devices go to "Devices (Paired)"
# All others go to "Other Devices"
if device.paired:
paired_devices.append(device)
new_paired_addresses.add(device.address)
else:
other_devices.append(device)
new_other_addresses.add(device.address)
except Exception:
continue
# Check if we need to update (devices added/removed)
paired_changed = current_paired_addresses != new_paired_addresses
other_changed = current_other_addresses != new_other_addresses
# Only rebuild if something actually changed
if paired_changed or other_changed:
# Clear existing devices safely
for child in list(self.paired_devices.get_children()):
if not self._destroyed:
child.destroy()
for child in list(self.other_devices.get_children()):
if not self._destroyed:
child.destroy()
# Add paired devices
for device in paired_devices:
if not self._destroyed:
device_slot = BluetoothDeviceSlot(device)
self.paired_devices.add(device_slot)
# Add other devices
for device in other_devices:
if not self._destroyed:
device_slot = BluetoothDeviceSlot(device)
self.other_devices.add(device_slot)
# Show/hide sections based on available devices
if not self._destroyed:
has_paired_devices = len(paired_devices) > 0
has_other_devices = len(other_devices) > 0
has_any_devices = has_paired_devices or has_other_devices
# Show paired devices section only if there are paired devices
self.paired_devices_label.set_visible(has_paired_devices)
self.paired_devices.set_visible(has_paired_devices)
# Show "No devices available" message if no devices at all
self.no_devices_label.set_visible(not has_any_devices)
# Always show the other devices button, regardless of available devices
self.other_devices_button.set_visible(True) # Always visible
except Exception:
pass
finally:
self._update_in_progress = False
def start_device_monitoring(self):
"""Start periodic monitoring for device changes"""
# Monitor for device changes every 5 seconds (less aggressive)
# This helps catch devices that connect from external sources
self.refresh_timer = GLib.timeout_add_seconds(5, self.periodic_device_refresh)
def stop_device_monitoring(self):
"""Stop periodic monitoring"""
if self.refresh_timer:
GLib.source_remove(self.refresh_timer)
self.refresh_timer = None
def periodic_device_refresh(self):
"""Periodically refresh device list to catch external connections"""
# Skip if update in progress, destroyed, or client not available
if (
self._update_in_progress
or self._destroyed
or not self.client
or not self.client.enabled
):
return True # Continue monitoring
try:
# Simple check - just trigger update_devices which has its own safety checks
# Don't force signal emissions as that can cause race conditions
self.update_devices()
except Exception:
pass
return True # Continue monitoring
def force_device_refresh(self):
"""Force an immediate refresh of the device list"""
if self._update_in_progress or self._destroyed:
return
try:
# Simply trigger update_devices which has its own safety checks
# Avoid forcing signal emissions to prevent race conditions
self.update_devices()
except Exception:
pass
def on_client_changed(self, *_):
"""Handle when the bluetooth client state changes"""
# Update devices when client state changes
self.update_devices()
def on_destroy(self, widget):
"""Cleanup when widget is destroyed"""
# Mark as destroyed to prevent further updates
self._destroyed = True
# Stop monitoring
self.stop_device_monitoring()
# Make sure other devices revealer is collapsed when closing
try:
self.other_devices_revealer.child_revealed = False
except Exception:
pass # Widget might already be destroyed
def close_bluetooth(self):
"""Called when Bluetooth panel is being closed"""
# Collapse the other devices section when closing
self.other_devices_revealer.child_revealed = False
def setup_pull_to_refresh(self):
"""Setup pull-to-refresh gesture for the scrolled window"""
# Get the scrolled window's vertical adjustment
self.vadjustment = self.other_devices_scrolled.get_vadjustment()
# Track gesture state
self.pull_start_y = 0
self.is_pulling = False
self.pull_threshold = 50 # pixels to trigger refresh
# Connect to scroll events
self.other_devices_scrolled.connect("scroll-event", self.on_scroll_event)
self.other_devices_scrolled.connect("button-press-event", self.on_button_press)
self.other_devices_scrolled.connect(
"button-release-event", self.on_button_release
)
self.other_devices_scrolled.connect(
"motion-notify-event", self.on_motion_notify
)
# Enable events
self.other_devices_scrolled.set_events(
Gdk.EventMask.SCROLL_MASK
| Gdk.EventMask.BUTTON_PRESS_MASK
| Gdk.EventMask.BUTTON_RELEASE_MASK
| Gdk.EventMask.POINTER_MOTION_MASK
)
def on_scroll_event(self, widget, event):
"""Handle scroll events for pull-to-refresh"""
# Only handle pull-to-refresh when at the top
if self.vadjustment.get_value() <= 0:
if event.direction == Gdk.ScrollDirection.UP:
# Scrolling up at the top - toggle scan and force refresh
self.client.toggle_scan()
self.force_device_refresh()
return True # Consume the event
return False # Let normal scrolling continue
def on_button_press(self, widget, event):
"""Handle button press for touch/drag gestures"""
if self.vadjustment.get_value() <= 0:
self.pull_start_y = event.y
self.is_pulling = True
return False
def on_button_release(self, widget, event):
"""Handle button release for touch/drag gestures"""
if self.is_pulling:
pull_distance = event.y - self.pull_start_y
if pull_distance > self.pull_threshold:
# Toggle scan and force refresh
self.client.toggle_scan()
self.force_device_refresh()
# Hide refresh indicator
self.refresh_indicator.set_visible(False)
self.refresh_indicator.remove_style_class("ready-to-refresh")
self.is_pulling = False
return False
def on_motion_notify(self, widget, event):
"""Handle motion events for visual feedback during pull"""
if self.is_pulling and self.vadjustment.get_value() <= 0:
pull_distance = event.y - self.pull_start_y
if pull_distance > 0:
# Show refresh indicator when pulling down
self.refresh_indicator.set_visible(True)
if pull_distance >= self.pull_threshold:
if self.client.scanning:
self.refresh_indicator.set_label("↑ Release to stop scanning")
else:
self.refresh_indicator.set_label("↑ Release to scan")
self.refresh_indicator.add_style_class("ready-to-refresh")
else:
if self.client.scanning:
self.refresh_indicator.set_label("↓ Pull to stop scanning")
else:
self.refresh_indicator.set_label("↓ Pull to scan for devices")
self.refresh_indicator.remove_style_class("ready-to-refresh")
else:
self.refresh_indicator.set_visible(False)
return False
def on_device_added(self, client: BluetoothClient, address: str):
"""Handle when a new device is added"""
# Update the device list when devices are added
self.update_devices()
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+138
View File
@@ -0,0 +1,138 @@
import subprocess
from fabric.utils import logger
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from utils.utils import svg_file
class NightLightControl:
"""Control night light using hyprsunset"""
def __init__(self):
self.is_active = False
self._check_initial_state()
def _check_initial_state(self):
"""Check if hyprsunset is currently running"""
try:
result = subprocess.run(
["pgrep", "-f", "hyprsunset"], capture_output=True, text=True
)
self.is_active = bool(result.stdout.strip())
except Exception as e:
logger.warning(f"Failed to check hyprsunset status: {e}")
self.is_active = False
def toggle(self):
"""Toggle night light on/off"""
try:
if self.is_active:
# Turn off night light by killing hyprsunset
subprocess.run(["pkill", "hyprsunset"], check=False)
self.is_active = False
logger.debug("Night light turned off")
else:
# Turn on night light with default temperature (3000K)
subprocess.Popen(["hyprsunset", "-t", "4500"], start_new_session=True)
self.is_active = True
logger.debug("Night light turned on")
return True
except Exception as e:
logger.warning(f"Failed to toggle night light: {e}")
return False
def set_temperature(self, temperature: int):
"""Set night light temperature (1000-6500K)"""
try:
# Kill existing process
subprocess.run(["pkill", "hyprsunset"], check=False)
# Start with new temperature
subprocess.Popen(
["hyprsunset", "-t", str(temperature)], start_new_session=True
)
self.is_active = True
logger.debug(f"Night light temperature set to {temperature}K")
return True
except Exception as e:
logger.warning(f"Failed to set night light temperature: {e}")
return False
def create_night_light_widget(control_center):
"""Create night light widget for control center"""
# Initialize night light control
night_light = NightLightControl()
# Create icon
night_light_icon = svg_file(
(
"applets/redshift-status-on.svg"
if night_light.is_active
else "applets/redshift-status-off.svg"
),
size=42,
)
# Create status label
night_light_status_label = Label(
label="On" if night_light.is_active else "Off",
name="nightlight-widget-label",
style_classes="status-label",
max_chars_width=15,
ellipsization="end",
h_align="start",
)
def toggle_night_light(*_):
"""Toggle night light and update UI"""
if night_light.toggle():
# Update icon
night_light_icon.dynamic_file(
"applets/redshift-status-on.svg"
if night_light.is_active
else "applets/redshift-status-off.svg"
)
# Update status label
night_light_status_label.set_label("On" if night_light.is_active else "Off")
# Create widget box (similar to bluetooth_widget structure)
night_light_widget = Box(
name="nightlight-widget",
orientation="h",
h_expand=True,
children=[
Button(
name="nightlight-icon-button",
child=night_light_icon,
on_clicked=toggle_night_light,
),
Button(
name="nightlight-info-button",
child=Box(
name="nightlight-widget-info",
h_expand=True,
v_expand=True,
v_align="center",
h_align="start",
orientation="vertical",
children=[
Label(
name="nightlight-widget-name",
label="Night Light",
style_classes="ct",
h_align="start",
),
night_light_status_label,
],
),
on_clicked=toggle_night_light,
),
],
)
return night_light_widget
+332
View File
@@ -0,0 +1,332 @@
from fabric.utils import GLib
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.image import Image
from fabric.widgets.label import Label
from fabric.widgets.scale import Scale
from fabric.widgets.scrolledwindow import ScrolledWindow
from fabric.widgets.separator import Separator
# Local imports
from utils.roam import audio_service
class PerAppVolumeControl(Box):
"""Per-application volume control widget"""
def __init__(self, control_center, **kwargs):
super().__init__(
orientation="vertical",
name="per-app-volume-control",
spacing=5,
**kwargs,
)
self.control_center = control_center
self._updating_volumes = set()
self._app_widgets = {}
self._signal_connections = []
# Header with back button
self.header = Box(
orientation="horizontal",
name="per-app-volume-header",
style_classes="menu-header",
children=[
Button(
image=Image(icon_name="back", size=10),
on_clicked=lambda *_: self.control_center.close_per_app_volume(),
),
Label("App Volume", name="app-volume-header"),
],
)
# Scrollable container for app volume controls
self.apps_container = Box(
orientation="vertical",
name="apps-scrolled-container",
spacing=2,
)
self.scrolled_window = ScrolledWindow(
name="apps-scrolled-window",
child=self.apps_container,
size=(300, 500),
)
# Add escape key binding for navigation back
try:
if hasattr(self.control_center, "add_keybinding"):
self.control_center.add_keybinding("Escape", self._go_back)
except Exception:
pass # Ignore if keybinding fails
self.children = [self.header, self.scrolled_window]
# Connect to audio service changes
if audio_service:
self._signal_connections.append(
audio_service.connect("stream-added", self._on_stream_changed)
)
self._signal_connections.append(
audio_service.connect("stream-removed", self._on_stream_changed)
)
# Initial population
self._populate_apps()
# Set up auto-refresh timer for audio streams
self._refresh_timer = GLib.timeout_add_seconds(2, self._auto_refresh)
def _auto_refresh(self):
"""Auto-refresh the application list every 2 seconds"""
self._populate_apps()
return True # Continue the timer
def _go_back(self, *_):
"""Return to main control center view"""
self.control_center.close_per_app_volume()
def _get_app_icon(self, app):
"""Get icon for application"""
# Don't use fabric's generic audio icon, use description/name for better detection
fabric_icon = getattr(app, "icon_name", "")
if fabric_icon and fabric_icon != "audio":
return fabric_icon
# Use description first (more accurate), then name
app_description = getattr(app, "description", "").lower()
app_name = app.name.lower()
# Check description first as it's more reliable
search_text = app_description if app_description else app_name
icon_mapping = {
"spotify": "spotify",
"firefox": "firefox",
"chromium": "chromium-browser",
"chrome": "google-chrome",
"vlc": "vlc",
"discord": "discord",
"steam": "steam",
"zen": "zen-browser",
"code": "vscode",
"visual studio code": "vscode",
"telegram": "telegram-desktop",
"pulse": "audio-card",
"pipewire": "audio-card",
"alsa": "audio-card",
"sink": "audio-speakers",
"source": "audio-input-microphone",
"youtube": "youtube",
"music": "rhythmbox",
"media": "multimedia-player",
}
# Check if any key in mapping is contained in search text
for key, icon in icon_mapping.items():
if key in search_text:
return icon
# Also check app name as fallback
if search_text != app_name:
for key, icon in icon_mapping.items():
if key in app_name:
return icon
# Default audio icon for unknown apps
return "audio-volume-high"
def _format_app_name(self, name):
"""Format application name with proper capitalization"""
if not name:
return "Unknown"
# Handle common app names specially
special_names = {
"spotify": "Spotify",
"firefox": "Firefox",
"chromium": "Chromium",
"chrome": "Chrome",
"vlc": "VLC",
"discord": "Discord",
"steam": "Steam",
"zen": "Zen Browser",
"code": "VS Code",
"telegram": "Telegram",
}
name_lower = name.lower()
if name_lower in special_names:
return special_names[name_lower]
# Default: capitalize first letter
return name.capitalize()
def _populate_apps(self):
"""Populate the widget with current audio applications"""
# Clear existing widgets
self.apps_container.children = []
self._app_widgets.clear()
# Use fabric audio service for applications
if not audio_service:
self._show_no_apps_message()
return
applications = getattr(audio_service, "applications", [])
if applications:
for i, app in enumerate(applications):
if hasattr(app, "name") and hasattr(app, "volume"):
app_widget = self._create_app_control(app)
self.apps_container.children = list(
self.apps_container.children
) + [app_widget]
# Add separator between apps (except for last one)
if i < len(applications) - 1:
separator = Separator(
orientation="horizontal",
style_classes="app-volume-separator",
)
self.apps_container.children = list(
self.apps_container.children
) + [separator]
self._app_widgets[app.name] = (app_widget, app)
else:
self._show_no_apps_message()
def _show_no_apps_message(self):
"""Show message when no apps are playing audio"""
message = Label(
label="No applications currently using audio",
style_classes="subtitle",
h_align="center",
v_align="center",
)
self.apps_container.children = [message]
def _create_app_control(self, app):
"""Create compact volume control for a single application"""
# Format app name (shorter for compact layout)
app_name = self._format_app_name(app.name)
# Get current volume from fabric audio service
current_volume = getattr(app, "volume", 0.0)
max_vol = getattr(audio_service, "max_volume", 100) if audio_service else 100
# Ensure volume is in valid range
volume_percent = max(0, min(current_volume, max_vol))
# App icon - use the same approach as expanded player
icon_name = self._get_app_icon(app)
app_icon = Image(
icon_name=icon_name,
name="app-icon",
icon_size=16,
style_classes="app-icon",
)
# App name label
name_label = Label(
label=app_name,
style_classes="app-name-compact",
justification="left",
h_align="start",
max_chars_width=10,
ellipsization="end",
)
# Volume scale (smaller and horizontal)
volume_scale = Scale(
value=volume_percent,
min_value=0,
max_value=max_vol,
increments=(1, 5),
name="compact-volume-slider",
size=20,
h_expand=True,
)
# Connect volume change handler
volume_scale.connect(
"change-value",
lambda scale, scroll_type, value, app=app: self._set_app_volume(app, value),
)
# Create horizontal compact layout
app_control = Box(
orientation="horizontal",
spacing=5,
h_expand=True,
style_classes="compact-app-volume-item",
children=[
Box(
orientation="h",
spacing=3,
v_expand=True,
v_align="center",
name="app-control-box",
children=[
app_icon,
name_label,
],
),
volume_scale,
],
)
# Add separator after the control
return app_control
def _set_app_volume(self, app, volume_value):
"""Set volume for a specific application using fabric audio service"""
if app.name in self._updating_volumes:
return
self._updating_volumes.add(app.name)
try:
# Get max volume from audio service
max_vol = (
getattr(audio_service, "max_volume", 100) if audio_service else 100
)
# Ensure volume is within bounds
volume_value = max(0, min(volume_value, max_vol))
# Set volume directly - fabric expects the actual volume value, not percentage
app.volume = volume_value
except Exception as e:
print(f"Error setting volume for {app.name}: {e}")
finally:
GLib.timeout_add(100, lambda: self._updating_volumes.discard(app.name))
def _on_stream_changed(self, *_):
"""Handle when audio streams are added or removed"""
GLib.idle_add(self._populate_apps)
def refresh(self):
"""Manually refresh the application list"""
self._populate_apps()
def destroy(self):
"""Clean up resources"""
if hasattr(self, "_refresh_timer"):
GLib.source_remove(self._refresh_timer)
# Disconnect fabric audio service signals
if audio_service:
for connection in self._signal_connections:
try:
audio_service.disconnect(connection)
except Exception:
pass
self._signal_connections.clear()
self._app_widgets.clear()
self._updating_volumes.clear()
super().destroy()
File diff suppressed because it is too large. Load diff
+706
View File
@@ -0,0 +1,706 @@
import subprocess
from fabric.utils import Gdk, GLib, Gtk
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.centerbox import CenterBox
from fabric.widgets.image import Image
from fabric.widgets.label import Label
from fabric.widgets.revealer import Revealer
from fabric.widgets.scrolledwindow import ScrolledWindow
from fabric.widgets.separator import Separator
from services.network import NetworkClient
from utils.functions import get_wifi_connecting_icon, get_wifi_icon_for_strength
from utils.utils import svg_file
from shared.dialogs.wifi_password_dialog import WiFiPasswordDialog
from gi.repository import NM
class WifiNetworkSlot(Box):
def __init__(self, access_point, wifi_service, parent=None, **kwargs):
super().__init__(name="wifi-network-slot", **kwargs)
self.access_point = access_point
self.wifi_service = wifi_service
self.parent = parent # Reference to control center
# Get network info from AccessPoint object
self.ssid = access_point.ssid
self.bssid = access_point.bssid
self.strength = access_point.strength
self.icon_name = access_point.icon
# Check if this network is currently connected
self.is_connected = access_point.is_active
# Initialize styles based on connection state
self.styles = [
"connected" if self.is_connected else "",
]
# Create connection status indicator using dynamic WiFi icon based on signal strength
wifi_icon_path = get_wifi_icon_for_strength(self.strength)
self.dimage = svg_file(wifi_icon_path, size=28)
self.wifi_icon_box = Box(
children=[self.dimage],
style_classes=["wifi-icon-box"],
)
# if self.is_connected:
# self.wifi_icon_box.remove_style_class("wifi-icon-box")
# self.wifi_icon_box.add_style_class("wifi-icon-box-connected")
#
# Set initial style classes
if self.is_connected:
self.dimage.add_style_class("connected")
self.network_label = Label(
label=self.ssid, name="wifi-network-name", h_align="start", h_expand=True
)
# Create lock icon for secured networks
self.lock_icon = None
if self.access_point.requires_password:
self.lock_icon = Image(
icon_name="changes-prevent-symbolic",
size=12,
name="wifi-lock-icon",
)
# Initialize password dialog
self.password_dialog = None
# Create the start section with WiFi icon and network name
start_box = Box(
orientation="h",
spacing=8,
)
start_box.children = [self.wifi_icon_box, self.network_label]
if self.lock_icon:
start_box.children.append(self.lock_icon)
# Create end section with lock icon if needed
self.children = [
Button(
child=start_box,
h_expand=True,
name="wifi-network-button",
on_clicked=lambda *_: self.toggle_connecting(),
)
]
# Emit initial change to update display
self.on_changed()
def toggle_connecting(self):
# Check if this network is currently connected
is_currently_connected = self.access_point.is_active
if is_currently_connected:
# Show disconnecting state using connecting icon
connecting_icon = get_wifi_connecting_icon()
self.dimage.set_from_file(connecting_icon)
self.dimage.add_style_class("disconnecting")
# Disconnect from network
self.wifi_service.disconnect_wifi()
self.is_connected = False
# Remove disconnecting state after a short delay to show feedback
GLib.timeout_add(500, lambda: self._reset_disconnect_state())
else:
# Try to connect - check if password is required
if self.access_point.requires_password:
# Show password dialog immediately
self._show_password_dialog()
else:
# Try to connect without password (for open networks)
connecting_icon = get_wifi_connecting_icon()
self.dimage.set_from_file(connecting_icon)
self.dimage.add_style_class("connecting")
def on_open_connection_result(success, message):
"""Handle the connection result for open networks"""
if success:
self.is_connected = True
# Remove connecting state after a short delay
GLib.timeout_add(500, lambda: self._reset_connect_state())
else:
# Connection failed
self._reset_connect_state()
# Update display after connection attempt
self.on_changed()
try:
self.wifi_service.connect_to_wifi(
self.access_point, callback=on_open_connection_result
)
except Exception:
# Handle any connection errors gracefully
self._reset_connect_state()
self.on_changed()
# Update display after connection attempt
self.on_changed()
def _reset_disconnect_state(self):
"""Reset visual state after disconnect operation"""
self.dimage.remove_style_class("disconnecting")
wifi_icon_path = get_wifi_icon_for_strength(self.strength)
self.dimage.set_from_file(wifi_icon_path)
self.on_changed()
return False # Remove timeout
def _reset_connect_state(self):
"""Reset visual state after connect operation"""
self.dimage.remove_style_class("connecting")
wifi_icon_path = get_wifi_icon_for_strength(self.strength)
self.dimage.set_from_file(wifi_icon_path)
self.on_changed()
return False # Remove timeout
def on_changed(self, *_):
# Check if this network is currently connected using the access point's is_active property
self.is_connected = self.access_point.is_active
self.styles = [
"connected" if self.is_connected else "",
]
# Update style classes for SVG widget
if self.is_connected:
self.wifi_icon_box.add_style_class("wifi-icon-box-connected")
else:
self.wifi_icon_box.remove_style_class("wifi-icon-box-connected")
return
def _show_password_dialog(self):
"""Show the WiFi password dialog"""
# Close the control center first
if self.parent and hasattr(self.parent, "hide_controlcenter"):
self.parent.hide_controlcenter()
# Create a new dialog each time to ensure clean state
if self.password_dialog:
self.password_dialog.destroy_dialog()
self.password_dialog = WiFiPasswordDialog(
ssid=self.ssid,
on_connect_callback=self._on_password_connect,
on_cancel_callback=self._on_password_cancel,
)
self.password_dialog.show_dialog()
def _on_password_connect(self, ssid, password):
"""Handle password dialog connect action"""
if password.strip():
# Show connecting state using connecting icon
connecting_icon = get_wifi_connecting_icon()
self.dimage.set_from_file(connecting_icon)
self.dimage.add_style_class("connecting")
# Try to connect with password using callback
def on_connection_result(success, message):
"""Handle the connection result"""
if success:
self.is_connected = True
# Remove connecting state after a short delay
GLib.timeout_add(500, lambda: self._reset_connect_state())
# Clear any timeout in the password dialog
if (
self.password_dialog
and self.password_dialog.connection_timeout_id
):
GLib.source_remove(self.password_dialog.connection_timeout_id)
self.password_dialog.connection_timeout_id = None
self.password_dialog.is_connecting = False
else:
# Connection failed - show error in dialog
self._reset_connect_state()
if self.password_dialog:
self._show_connection_error(message)
# Update display after connection attempt
self.on_changed()
try:
self.wifi_service.connect_to_wifi(
self.access_point, password, callback=on_connection_result
)
except Exception:
# Handle any connection errors gracefully
self._reset_connect_state()
if self.password_dialog:
self._show_connection_error("Connection failed. Please try again.")
self.on_changed()
def _show_connection_error(self, message="Incorrect password. Please try again."):
"""Show connection error in a separate thread to prevent UI blocking"""
if self.password_dialog:
self.password_dialog.show_error(message)
return False # Don't repeat if called from GLib.timeout_add
def _on_password_cancel(self):
"""Handle password dialog cancel action"""
# Reset any connecting state
self._reset_connect_state()
class WifiConnections(Box):
def __init__(self, parent, show_back_button=True, **kwargs):
super().__init__(
spacing=8,
orientation="vertical",
name="wifi-connections",
**kwargs,
)
self.parent = parent
self.network_service = NetworkClient()
self.wifi_service = None
self.is_scanning = False # Track scanning state
self.refresh_timer = None # Timer for periodic network refresh
self._update_in_progress = False # Prevent concurrent updates
self._destroyed = False # Track if widget is destroyed
# Wait for network service to be ready
self.network_service.connect("wifi-device-added", self.on_network_ready)
# Create pull-to-refresh indicator
self.refresh_indicator = Label(
name="wifi-refresh-indicator",
label="↓ Pull to scan for networks",
h_align="center",
visible=False,
style="color: #fff; font-size: 12px; padding: 5px;",
)
# Create title with optional back button
title_children = []
if show_back_button:
title_children.append(
Button(
image=Image(icon_name="back", size=10),
on_clicked=lambda *_: self.parent.close_wifi(),
)
)
title_children.append(Label("Wi-Fi", name="wifi-title"))
self.title = Box(
orientation="h",
children=title_children,
)
self.toggle_button = Gtk.Switch(visible=True, name="toggle-button")
# Create Known Network section
self.known_networks_label = Label(
label="Known Network", h_align="start", name="networks-title"
)
self.known_networks = Box(
spacing=4, orientation="vertical", name="known-networks"
)
# Create "No networks available" message
self.no_networks_label = Label(
label="No networks available",
h_align="center",
name="no-networks-label",
visible=False,
)
# Create Other Networks section with clickable title
self.other_networks_button = Button(
child=Label("Other Networks", h_align="start"),
name="wifi-other-button",
on_clicked=self.toggle_other_networks,
)
self.other_networks = Box(spacing=4, orientation="vertical")
# Create scrolled window for other networks
self.other_networks_scrolled = ScrolledWindow(
min_content_size=(303, 150),
child=self.other_networks,
overlay_scroll=True,
)
# Add pull-to-refresh functionality to scrolled window
self.setup_pull_to_refresh()
# Create revealer for Other Networks section
self.other_networks_revealer = Revealer(
child=self.other_networks_scrolled,
transition_type="slide-down",
transition_duration=100,
child_revealed=False,
)
# Create More Settings button (same style as Other Networks button)
self.more_settings_button = Button(
child=Label("More Settings", h_align="start"),
name="wifi-other-button",
on_clicked=self.open_network_settings,
)
self.children = [
CenterBox(
start_children=self.title,
end_children=self.toggle_button,
name="wifi-widget-top",
),
self.refresh_indicator,
Separator(orientation="h", name="separator"),
self.known_networks_label,
self.known_networks,
self.no_networks_label,
Separator(orientation="h", name="separator"),
self.other_networks_button,
self.other_networks_revealer,
Separator(orientation="h", name="separator"),
self.more_settings_button,
]
# Connect cleanup on destroy
self.connect("destroy", self.on_destroy)
# Start periodic network monitoring for real-time updates
self.start_network_monitoring()
def toggle_other_networks(self, *_):
"""Toggle the visibility of other networks section"""
current_state = self.other_networks_revealer.child_revealed
self.other_networks_revealer.child_revealed = not current_state
# Update button text based on state
if self.other_networks_revealer.child_revealed:
# Trigger a scan when revealing other networks and force refresh
if self.wifi_service:
self.wifi_service.scan()
# Also force an immediate network refresh to catch any missed connections
self.force_network_refresh()
def on_network_ready(self, *_):
"""Called when network service is ready"""
self.wifi_service = self.network_service.wifi_device
if self.wifi_service:
# Set up WiFi toggle
self.toggle_button.set_active(self.wifi_service.wireless_enabled)
self.toggle_button.connect("notify::active", self.on_toggle_changed)
# Connect to WiFi service signals
self.wifi_service.connect(
"notify::wireless-enabled", self.on_wifi_enabled_changed
)
self.wifi_service.connect("changed", self.update_networks)
self.wifi_service.connect("ap-added", self.update_networks)
self.wifi_service.connect("ap-removed", self.update_networks)
# Initial network update
self.update_networks()
def on_toggle_changed(self, toggle_button, *_):
"""Handle WiFi toggle button changes"""
if self.wifi_service:
new_state = toggle_button.get_active()
self.wifi_service.wireless_enabled = new_state
def on_wifi_enabled_changed(self, *_):
"""Handle WiFi enabled state changes"""
if self.wifi_service:
self.toggle_button.set_active(self.wifi_service.wireless_enabled)
def open_network_settings(self, *_):
"""Open NetworkManager connection editor"""
try:
subprocess.Popen(["nm-connection-editor"], start_new_session=True)
if self.parent and hasattr(self.parent, "hide_controlcenter"):
self.parent.hide_controlcenter()
except FileNotFoundError:
pass
except Exception:
pass
def update_networks(self, *_):
"""Update the list of available networks"""
# Prevent concurrent updates and check if destroyed
if self._update_in_progress or self._destroyed or not self.wifi_service:
return
self._update_in_progress = True
try:
# Store current network SSIDs to detect changes
current_known_ssids = {
child.ssid
for child in self.known_networks.get_children()
if hasattr(child, "ssid")
}
current_other_ssids = {
child.ssid
for child in self.other_networks.get_children()
if hasattr(child, "ssid")
}
# Get current networks
access_points = self.wifi_service.access_points
known_networks = []
other_networks = []
new_known_ssids = set()
new_other_ssids = set()
for access_point in access_points:
try:
if access_point.ssid and access_point.ssid != "Unknown":
# Categorize networks: connected or saved networks go to "Known Network"
# All others go to "Other Networks"
if access_point.is_active or self._is_saved_network(
access_point
):
known_networks.append(access_point)
new_known_ssids.add(access_point.ssid)
else:
other_networks.append(access_point)
new_other_ssids.add(access_point.ssid)
except Exception:
continue
# Check if we need to update (networks added/removed)
known_changed = current_known_ssids != new_known_ssids
other_changed = current_other_ssids != new_other_ssids
# Only rebuild if something actually changed
if known_changed or other_changed:
# Clear existing networks safely
for child in list(self.known_networks.get_children()):
if not self._destroyed:
child.destroy()
for child in list(self.other_networks.get_children()):
if not self._destroyed:
child.destroy()
# Add known networks
for access_point in known_networks:
if not self._destroyed:
network_slot = WifiNetworkSlot(
access_point, self.wifi_service, parent=self.parent
)
self.known_networks.add(network_slot)
# Add other networks
for access_point in other_networks:
if not self._destroyed:
network_slot = WifiNetworkSlot(
access_point, self.wifi_service, parent=self.parent
)
self.other_networks.add(network_slot)
# Show/hide sections based on available networks
if not self._destroyed:
has_known_networks = len(known_networks) > 0
has_other_networks = len(other_networks) > 0
has_any_networks = has_known_networks or has_other_networks
# Show known networks section only if there are known networks
self.known_networks_label.set_visible(has_known_networks)
self.known_networks.set_visible(has_known_networks)
# Show "No networks available" message if no networks at all
self.no_networks_label.set_visible(not has_any_networks)
# Always show the other networks button, regardless of available networks
self.other_networks_button.set_visible(True) # Always visible
# Update all network connection states
self.refresh_network_states()
except Exception:
pass
finally:
self._update_in_progress = False
def _is_saved_network(self, access_point):
"""Check if a network is saved/known using NetworkManager connections"""
if not self.network_service or not self.network_service._client:
return False
try:
ssid = access_point.ssid
if not ssid or ssid == "Unknown":
return False
# Get all saved connections from NetworkManager
connections = self.network_service._client.get_connections()
for connection in connections:
# Check if this is a WiFi connection
if connection.get_connection_type() != "802-11-wireless":
continue
# Get the wireless setting
wifi_setting = connection.get_setting_wireless()
if not wifi_setting:
continue
# Compare SSIDs
connection_ssid_bytes = wifi_setting.get_ssid()
if connection_ssid_bytes:
connection_ssid = NM.utils_ssid_to_utf8(
connection_ssid_bytes.get_data()
)
if connection_ssid == ssid:
return True
except Exception:
pass
return False
def refresh_network_states(self, *_):
"""Refresh connection states for all network slots"""
# Refresh known networks
for child in self.known_networks.get_children():
if hasattr(child, "on_changed"):
child.on_changed()
# Refresh other networks
for child in self.other_networks.get_children():
if hasattr(child, "on_changed"):
child.on_changed()
def start_network_monitoring(self):
"""Start periodic monitoring for network changes"""
# Monitor for network changes every 5 seconds
# This helps catch networks that connect from external sources
self.refresh_timer = GLib.timeout_add_seconds(5, self.periodic_network_refresh)
def stop_network_monitoring(self):
"""Stop periodic monitoring"""
if self.refresh_timer:
GLib.source_remove(self.refresh_timer)
self.refresh_timer = None
def periodic_network_refresh(self):
"""Periodically refresh network list to catch external connections"""
# Skip if update in progress, destroyed, or wifi service not available
if (
self._update_in_progress
or self._destroyed
or not self.wifi_service
or not self.wifi_service.wireless_enabled
):
return True # Continue monitoring
try:
# Simple check - just trigger update_networks which has its own safety checks
self.update_networks()
except Exception:
pass
return True # Continue monitoring
def force_network_refresh(self):
"""Force an immediate refresh of the network list"""
if self._update_in_progress or self._destroyed:
return
try:
# Simply trigger update_networks which has its own safety checks
self.update_networks()
except Exception:
pass
def setup_pull_to_refresh(self):
"""Setup pull-to-refresh gesture for the scrolled window"""
# Get the scrolled window's vertical adjustment
self.vadjustment = self.other_networks_scrolled.get_vadjustment()
# Track gesture state
self.pull_start_y = 0
self.is_pulling = False
self.pull_threshold = 50 # pixels to trigger refresh
# Connect to scroll events
self.other_networks_scrolled.connect("scroll-event", self.on_scroll_event)
self.other_networks_scrolled.connect("button-press-event", self.on_button_press)
self.other_networks_scrolled.connect(
"button-release-event", self.on_button_release
)
self.other_networks_scrolled.connect(
"motion-notify-event", self.on_motion_notify
)
# Enable events
self.other_networks_scrolled.set_events(
Gdk.EventMask.SCROLL_MASK
| Gdk.EventMask.BUTTON_PRESS_MASK
| Gdk.EventMask.BUTTON_RELEASE_MASK
| Gdk.EventMask.POINTER_MOTION_MASK
)
def on_scroll_event(self, widget, event):
"""Handle scroll events for pull-to-refresh"""
# Only handle pull-to-refresh when at the top
if self.vadjustment.get_value() <= 0:
if event.direction == Gdk.ScrollDirection.UP:
# Scrolling up at the top - trigger scan and force refresh
if self.wifi_service:
self.wifi_service.scan()
self.force_network_refresh()
return True # Consume the event
return False # Let normal scrolling continue
def on_button_press(self, widget, event):
"""Handle button press for touch/drag gestures"""
if self.vadjustment.get_value() <= 0:
self.pull_start_y = event.y
self.is_pulling = True
return False
def on_button_release(self, widget, event):
"""Handle button release for touch/drag gestures"""
if self.is_pulling:
pull_distance = event.y - self.pull_start_y
if pull_distance > self.pull_threshold:
# Trigger scan and force refresh
if self.wifi_service:
self.wifi_service.scan()
self.force_network_refresh()
# Hide refresh indicator
self.refresh_indicator.set_visible(False)
self.refresh_indicator.remove_style_class("ready-to-refresh")
self.is_pulling = False
return False
def on_motion_notify(self, widget, event):
"""Handle motion events for visual feedback during pull"""
if self.is_pulling and self.vadjustment.get_value() <= 0:
pull_distance = event.y - self.pull_start_y
if pull_distance > 0:
# Show refresh indicator when pulling down
self.refresh_indicator.set_visible(True)
if pull_distance >= self.pull_threshold:
self.refresh_indicator.set_label("↑ Release to scan")
self.refresh_indicator.add_style_class("ready-to-refresh")
else:
self.refresh_indicator.set_label("↓ Pull to scan for networks")
self.refresh_indicator.remove_style_class("ready-to-refresh")
else:
self.refresh_indicator.set_visible(False)
return False
def on_destroy(self, widget):
"""Cleanup when widget is destroyed"""
# Mark as destroyed to prevent further updates
self._destroyed = True
# Stop monitoring
self.stop_network_monitoring()
# Make sure other networks revealer is collapsed when closing
try:
self.other_networks_revealer.child_revealed = False
except Exception:
pass # Widget might already be destroyed
def close_wifi(self):
"""Called when WiFi panel is being closed"""
# Collapse the other networks section when closing
self.other_networks_revealer.child_revealed = False