diff --git a/src/window/controlcenter/battery.py b/src/window/controlcenter/battery.py deleted file mode 100644 index f2ddda1..0000000 --- a/src/window/controlcenter/battery.py +++ /dev/null @@ -1,654 +0,0 @@ -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._destroyed = False - self._signal_ids = [] - self._signal_ids.append(self.device.connect("changed", self.on_changed)) - self._signal_ids.append( - 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 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"An error occurred: {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): - self.device.emit("changed") - self.device.set_connecting(not self.device.connected) - - def on_changed(self, *_): - if self._destroyed or self.device is None: - return - 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_signal_ids = [] # Track BluetoothClient signal IDs - - 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 — track IDs for disconnect on destroy - 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()) - ) - - # Connect to device changes - 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) - ) - - # Connect to additional signals for better real-time monitoring - 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" - ) - - # 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() - - self.connect("map", lambda *_: self.start_device_monitoring()) - self.connect("unmap", lambda *_: self.on_hide()) - - 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 as e: - logger.error(f"An error occurred: {e}") - - 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 as e: - logger.error(f"An error occurred: {e}") - finally: - self._update_in_progress = False - - def on_hide(self, *_): - """Called when the widget is hidden""" - self.stop_device_monitoring() - - 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""" - # If destroyed, remove the GLib source by returning False - if self._destroyed: - self.refresh_timer = None - return False - - # Skip if update in progress or client not available/enabled - if self._update_in_progress or not self.client or not self.client.enabled: - return True # Continue monitoring - - try: - self.update_devices() - except Exception as e: - logger.error(f"An error occurred: {e}") - - 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 as e: - logger.error(f"An error occurred: {e}") - - 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 timer - self.stop_device_monitoring() - # Disconnect all BluetoothClient signals - for sig_id in self._client_signal_ids: - try: - self.client.disconnect(sig_id) - except Exception as e: - logger.error(f"An error occurred: {e}") - self._client_signal_ids.clear() - # Collapse other devices revealer - try: - self.other_devices_revealer.child_revealed = False - except Exception as e: - logger.error(f"An error occurred: {e}") - - 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()