Files
Modus/src/window/controlcenter/bluetooth.py
T
sankalp beab79a405 refactor: extract shared mixins, fix cpu waste, resolve browser icons
- Extract CloseButtonRevealerMixin, PullToRefreshMixin, BasePlayerBoxStack
- Replace mpris.py 2s+5s D-Bus polling with Seeked signal subscription
- Remove redundant desktop/main.py os.listdir() poll (FileMonitor covers)
- Remove switcher/main.py 200ms capture timer, chain from frame-ready
- Resolve browser MPRIS icons via desktop file lookup, fallback to app name
- Fix CodeQL py/clear-text-storage by rounding lat/lon and stripping from cache
2026-07-31 01:35:25 +05:45

691 lines
25 KiB
Python

from enum import Enum, auto
from fabric.utils import GLib, Gtk, idle_add, 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.separator import Separator
from services.bluetooth import BluetoothClient, BluetoothDevice
from shared.widgets.pull_to_refresh import PullToRefreshMixin
from shared.widgets.smooth_switch import SmoothSwitch
from shared.window.animated_scrollwindow import AnimatedScrollable
from utils.functions import spawn_detached
from utils.gtk_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:
if hasattr(client, "set_enabled"):
client.set_enabled(enabled)
else:
client.enabled = enabled
except Exception as e:
logger.warning(f"Bluetooth set_enabled({enabled}) failed: {e}")
class BTState(Enum):
IDLE = auto()
CONNECTED = auto()
CONNECTING = auto()
FAILED = auto()
class BluetoothDeviceSlot(Box):
def __init__(self, device: BluetoothDevice, **kwargs):
super().__init__(h_expand=True, name="device-button", **kwargs)
self.device = device
self._destroyed = False
self._state = BTState.IDLE
self._signal_ids = []
self._signal_ids.append(self.device.connect("changed", self.on_changed))
self._signal_ids.append(
self.device.connect(
"notify::closed",
lambda *_: getattr(self.device, "closed", False) and self.destroy(),
)
)
self.styles = [
"connected" if self.device.connected else "",
"paired" if self.device.paired else "",
]
self.dimage = Image(
icon_name=self._get_icon_for_device(self.device),
pixel_size=20,
name="device-icon",
style_classes=" ".join(self.styles),
)
self.status_icon = Image(
icon_name="process-working-symbolic",
pixel_size=16,
name="device-icon",
)
self.status_icon.set_visible(False)
self.battery_icon = None
self.battery_label = None
if hasattr(device, "battery_percentage") and device.battery_percentage > 0:
self.battery_icon = svg_file(
get_battery_icon_file(device.battery_percentage, False), size=20
)
self.battery_label = Label(
label=f"{device.battery_percentage:.0f}%", name="battery-label"
)
button_children = [
self.dimage,
Label(label=device.name),
Box(h_expand=True), # Spacer
self.status_icon,
]
if self.battery_icon and self.battery_label:
button_children.insert(-1, self.battery_icon)
button_children.insert(-1, self.battery_label)
self.device_button = Button(
on_clicked=lambda *_: self.toggle_connecting(),
h_expand=True,
h_align="fill",
child=Box(
orientation="h",
h_expand=True,
spacing=8,
children=button_children,
),
)
self.children = [self.device_button]
self.device_button.connect("enter-notify-event", self.on_button_enter)
self.device_button.connect("leave-notify-event", self.on_button_leave)
# Right-click for pair / trust / remove actions
self.device_button.connect("button-press-event", self._on_button_press)
self._refresh_state()
self.device.emit("changed")
def _get_icon_for_device(self, device) -> str:
return getattr(device, "type", None) or getattr(
device, "icon_name", "bluetooth"
)
def destroy(self):
"""Clean up device signal connections to prevent memory leaks."""
if self._destroyed:
return
self._destroyed = True
for sig_id in self._signal_ids:
try:
self.device.disconnect(sig_id)
except Exception as e:
logger.error(f"[Bluetooth] Failed to disconnect signal {sig_id}: {e}")
self._signal_ids.clear()
self.device = None
super().destroy()
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):
if self._destroyed:
return
if self._state == BTState.CONNECTED:
self.device.connecting = False
self._set_state(BTState.IDLE)
elif self._state != BTState.CONNECTING:
self._set_state(BTState.CONNECTING)
self.device.connecting = True
self.device.emit("changed")
def on_changed(self, *_):
if self._destroyed or self.device is None:
return
if getattr(self.device, "connected", False):
self._set_state(BTState.CONNECTED)
elif self._state == BTState.CONNECTING:
self._set_state(BTState.FAILED)
else:
self._set_state(BTState.IDLE)
try:
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 as e:
logger.error(
f"[Bluetooth] Failed to update styles for {self.device.name}: {e}"
)
if (
hasattr(self.device, "battery_percentage")
and self.device.battery_percentage > 0
and self.device.connected
):
if self.battery_icon and self.battery_label:
self.battery_icon.set_visible(True)
self.battery_label.set_visible(True)
self.battery_icon.set_from_file(
get_battery_icon_file(self.device.battery_percentage, False)
)
self.battery_label.set_label(f"{self.device.battery_percentage:.0f}%")
else:
self.battery_icon = svg_file(
get_battery_icon_file(self.device.battery_percentage, False),
size=20,
)
self.battery_label = Label(
label=f"{self.device.battery_percentage:.0f}%",
name="battery-label",
)
btn_box = self.device_button.get_child()
if btn_box:
btn_box.children = [
*btn_box.children[:-1],
self.battery_icon,
self.battery_label,
btn_box.children[-1],
]
elif self.battery_icon or self.battery_label:
if self.battery_icon:
self.battery_icon.set_visible(False)
if self.battery_label:
self.battery_label.set_visible(False)
def _refresh_state(self):
self._set_state(
BTState.CONNECTED
if getattr(self.device, "connected", False)
else BTState.IDLE
)
def _set_state(self, state: BTState):
if self._destroyed:
return
self._state = state
for cls in ["active", "connecting", "failed"]:
self.remove_style_class(cls)
if state == BTState.CONNECTED:
self.add_style_class("active")
self.status_icon.set_visible(False)
elif state == BTState.CONNECTING:
self.add_style_class("connecting")
self.status_icon.set_visible(False)
elif state == BTState.FAILED:
self.add_style_class("failed")
self.status_icon.set_visible(False)
GLib.timeout_add(5000, self._reset_from_failed)
else:
self.status_icon.set_visible(False)
def _reset_from_failed(self):
if not self._destroyed:
self._set_state(BTState.IDLE)
return False
def _on_button_press(self, widget, event):
"""Right-click opens a context menu with pair/trust/remove actions."""
if self._destroyed or self.device is None:
return False
if event.button != 3: # only right-click
return False
menu = Gtk.Menu()
if not self.device.paired:
item_pair = Gtk.MenuItem(label="Pair")
item_pair.connect("activate", lambda *_: self.device.pair())
menu.append(item_pair)
else:
if self.device.trusted:
item_trust = Gtk.MenuItem(label="Untrust")
item_trust.connect("activate", lambda *_: self.device.untrust())
else:
item_trust = Gtk.MenuItem(label="Trust")
item_trust.connect("activate", lambda *_: self.device.trust())
menu.append(item_trust)
item_remove = Gtk.MenuItem(label="Remove / Forget")
item_remove.connect(
"activate",
lambda *_: self._remove_device(),
)
menu.append(item_remove)
menu.show_all()
menu.popup_at_pointer(event)
return True
def _remove_device(self):
"""Ask the client to remove (unpair/forget) this device."""
if self._destroyed or self.device is None:
return
try:
# Walk up to find a BluetoothConnections parent that holds the client
parent = self.get_parent()
while parent is not None:
if isinstance(parent, BluetoothConnections):
parent.client.remove_device(self.device)
return
parent = parent.get_parent()
# Fallback: use device.remove() which sets a flag
self.device.remove()
except Exception as e:
logger.warning(f"[Bluetooth] Failed to remove device: {e}")
class BluetoothConnections(PullToRefreshMixin, 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._update_in_progress = False # Prevent concurrent updates
self._destroyed = False # Track if widget is destroyed
self._client_signal_ids = [] # Track BluetoothClient signal IDs
self.client = BluetoothClient()
# 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;",
)
title_children = []
if show_back_button:
title_children.append(
Button(
image=svg_file("misc/chevron-left.svg", 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._switch_lock = False
self.toggle_button = SmoothSwitch(
name="toggle-button",
active=self.client.enabled,
on_user_toggle=self.handle_user_toggle,
)
self._client_signal_ids.append(
self.client.connect(
"notify::enabled",
lambda *_: self.toggle_button.set_active(self.client.enabled),
)
)
self._client_signal_ids.append(
self.client.connect("notify::scanning", lambda *_: self.update_scan_label())
)
self._client_signal_ids.append(
self.client.connect("device-added", self.update_devices)
)
self._client_signal_ids.append(
self.client.connect("device-removed", self.update_devices)
)
self._client_signal_ids.append(
self.client.connect("notify::connected-devices", self.update_devices)
)
self._client_signal_ids.append(
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"
)
self.paired_devices_scrolled = AnimatedScrollable(
min_content_size=(303, 0),
max_content_size=(303, 200),
child=self.paired_devices,
overlay_scroll=True,
animate=False,
)
self.no_devices_label = Label(
label="No devices available",
h_align="center",
name="no-networks-label",
visible=False,
)
self.other_devices_button = Button(
child=CenterBox(
start_children=Label("Other Devices", h_align="start"),
end_children=self.refresh_indicator,
),
name="wifi-other-button",
on_clicked=self.toggle_other_devices,
)
self.other_devices = Box(spacing=4, orientation="vertical")
self.other_devices_scrolled = AnimatedScrollable(
min_content_size=(303, 0),
max_content_size=(303, 300),
child=self.other_devices,
overlay_scroll=True,
animate=False,
)
self.other_devices.set_visible(False)
# Add pull-to-refresh functionality to scrolled window
self.setup_pull_to_refresh(self.other_devices_scrolled)
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",
),
Separator(orientation="h", name="separator"),
self.paired_devices_label,
self.paired_devices_scrolled,
self.no_devices_label,
Separator(orientation="h", name="separator"),
self.other_devices_button,
self.other_devices_scrolled,
Separator(orientation="h", name="separator"),
self.more_settings_button,
]
self.connect("destroy", self.on_destroy)
self.connect("unmap", self.on_hide)
self.client.notify("scanning")
self.client.notify("enabled")
self.update_devices()
def handle_user_toggle(self, active: bool):
self._switch_lock = True
set_bluetooth_enabled_with_fallback(self.client, active)
self.toggle_button.set_active(active)
GLib.timeout_add(1500, self._unlock_switch)
def _unlock_switch(self):
self._switch_lock = False
if not self._destroyed and getattr(self, "toggle_button", None):
self.toggle_button.set_active(self.client.enabled)
return False
def on_hide(self, *_):
"""Called when the widget is hidden (popup closed)"""
self._cancel_pending_refresh()
if self.other_devices.get_visible():
self.other_devices.set_visible(False)
self.other_devices_scrolled.snap_to_size(0)
if (
self.client
and self.client.scanning
and hasattr(self.client, "stop_scan")
):
self.client.stop_scan()
self.update_scan_label()
def toggle_other_devices(self, *_):
"""Toggle the visibility of other devices section"""
current_state = self.other_devices.get_visible()
self._cancel_pending_refresh()
if current_state:
self.other_devices.set_visible(False)
self.other_devices_scrolled.snap_to_size(0)
self.update_scan_label()
if (
self.client
and self.client.scanning
and hasattr(self.client, "stop_scan")
):
self.client.stop_scan()
else:
self.other_devices.set_visible(True)
self.update_scan_label()
if (
self.client
and not self.client.scanning
and hasattr(self.client, "scan")
):
self.client.scan()
# Defer refresh until after the height animation completes
idle_add(self._refresh_after_animation)
def open_bluetooth_settings(self, *_):
"""Open Blueman bluetooth manager"""
try:
spawn_detached(["blueman-manager"])
if self.parent and hasattr(self.parent, "hide_controlcenter"):
self.parent.hide_controlcenter()
except FileNotFoundError as e:
logger.warning(
f"[bluetooth] spawn_detached(['blueman-manager']) failed: {e}"
)
except Exception as e:
logger.error(f"[Bluetooth] Failed to open bluetooth settings: {e}")
def update_scan_label(self):
"""Update scanning state appearance"""
if self.client and self.client.scanning and self.other_devices.get_visible():
self.refresh_indicator.set_label("Scanning for devices...")
self.refresh_indicator.set_visible(True)
self.refresh_indicator.add_style_class("scanning")
else:
self.refresh_indicator.set_visible(False)
self.refresh_indicator.remove_style_class("scanning")
def update_devices(self, *_):
"""Update the list of available devices"""
if self._update_in_progress or self._destroyed or not self.client:
return
self._update_in_progress = True
try:
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")
}
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":
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 as e:
logger.warning(
f"[Bluetooth] Failed to process device {device}: {e}"
)
continue
paired_changed = current_paired_addresses != new_paired_addresses
other_changed = current_other_addresses != new_other_addresses
if paired_changed or other_changed:
existing_paired = {
child.device.address: child
for child in self.paired_devices.get_children()
if hasattr(child, "device")
}
existing_other = {
child.device.address: child
for child in self.other_devices.get_children()
if hasattr(child, "device")
}
for device in paired_devices:
if not self._destroyed:
if device.address in existing_paired:
slot = existing_paired.pop(device.address)
# update existing
slot.device = device
slot.device.emit("changed")
elif device.address in existing_other:
slot = existing_other.pop(device.address)
slot.device = device
self.other_devices.remove(slot)
self.paired_devices.add(slot)
else:
device_slot = BluetoothDeviceSlot(device)
self.paired_devices.add(device_slot)
for device in other_devices:
if not self._destroyed:
if device.address in existing_other:
slot = existing_other.pop(device.address)
slot.device = device
slot.device.emit("changed")
elif device.address in existing_paired:
slot = existing_paired.pop(device.address)
slot.device = device
self.paired_devices.remove(slot)
self.other_devices.add(slot)
else:
device_slot = BluetoothDeviceSlot(device)
self.other_devices.add(device_slot)
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
self.paired_devices_scrolled.set_visible(True)
self.no_devices_label.set_visible(not has_any_devices)
self.other_devices_button.set_visible(True) # Always visible
except Exception as e:
logger.error(f"[Bluetooth] Error during update_devices: {e}")
finally:
self._update_in_progress = False
def force_device_refresh(self):
"""Force an immediate refresh of the device list"""
if self._update_in_progress or self._destroyed:
return
try:
self.update_devices()
except Exception as e:
logger.error(f"[Bluetooth] Error during forced device refresh: {e}")
def on_client_changed(self, *_):
"""Handle when the bluetooth client state changes"""
if getattr(self, "toggle_button", None) and not self._switch_lock:
self.toggle_button.set_active(self.client.enabled)
self.update_scan_label()
self.update_devices()
def on_destroy(self, widget):
"""Cleanup when widget is destroyed"""
self._destroyed = True
self._cancel_pending_refresh()
for sig_id in self._client_signal_ids:
try:
self.client.disconnect(sig_id)
except Exception as e:
logger.error(
f"[Bluetooth] Failed to disconnect client signal {sig_id}: {e}"
)
self._client_signal_ids.clear()
def close_bluetooth(self):
"""Called when Bluetooth panel is being closed"""
if self.other_devices.get_visible():
self.other_devices.set_visible(False)
self.other_devices_scrolled.snap_to_size(0)
def _trigger_scan(self):
if not self.client.scanning and hasattr(self.client, "scan"):
self.client.scan()
def _trigger_refresh(self):
self.force_device_refresh()
def _get_pull_normal_label(self):
if self.client.scanning:
return "↓ Pull to stop scanning"
return "↓ Pull to scan for devices"
def _get_pull_ready_label(self):
if self.client.scanning:
return "↑ Release to stop scanning"
return "↑ Release to scan"
def on_device_added(self, client: BluetoothClient, address: str):
"""Handle when a new device is added"""
self.update_devices()