✨ feat(switcher): add window previews using Glace for application switcher

This commit is contained in:
sankalp committed 2025-08-08 09:46:36 +05:45
1 parent 643d39b38b
commit 803b489dfa
2 files changed
+289 -13

No files matched your search

+165
View File
@@ -0,0 +1,165 @@
#!/usr/bin/env python3
"""
Real-time memory monitor for debugging expanded player memory leaks.
This module provides functions to track memory usage in real-time.
"""
import psutil
import os
import gc
import threading
import time
from loguru import logger
class MemoryMonitor:
"""Real-time memory monitoring for debugging memory leaks."""
def __init__(self):
self.process = psutil.Process(os.getpid())
self.baseline_memory = None
self.last_memory = None
self.monitoring = False
self.monitor_thread = None
def get_memory_usage(self):
"""Get current memory usage in MB."""
return self.process.memory_info().rss / 1024 / 1024
def get_memory_details(self):
"""Get detailed memory information."""
memory_info = self.process.memory_info()
memory_percent = self.process.memory_percent()
return {
"rss_mb": memory_info.rss / 1024 / 1024,
"vms_mb": memory_info.vms / 1024 / 1024,
"percent": memory_percent,
"num_threads": self.process.num_threads(),
}
def set_baseline(self, label="Baseline"):
"""Set the baseline memory usage."""
self.baseline_memory = self.get_memory_usage()
logger.info(f"🎯 {label} memory: {self.baseline_memory:.1f} MB")
return self.baseline_memory
def log_memory_change(self, label="Memory Check", force_gc=True):
"""Log current memory usage and change from baseline."""
if force_gc:
gc.collect()
current_memory = self.get_memory_usage()
details = self.get_memory_details()
if self.baseline_memory:
delta = current_memory - self.baseline_memory
logger.info(
f"📊 {label}: {current_memory:.1f} MB (Δ: {delta:+.1f} MB) | Threads: {
details['num_threads']
}"
)
else:
logger.info(
f"📊 {label}: {current_memory:.1f} MB | Threads: {
details['num_threads']
}"
)
self.last_memory = current_memory
return current_memory
def log_memory_spike(self, threshold_mb=10):
"""Log if there's a significant memory increase."""
if self.last_memory:
current = self.get_memory_usage()
increase = current - self.last_memory
if increase > threshold_mb:
logger.warning(
f"🚨 MEMORY SPIKE: +{increase:.1f} MB (from {self.last_memory:.1f} to {current:.1f} MB)"
)
return True
return False
def start_continuous_monitoring(self, interval_seconds=2):
"""Start continuous background monitoring."""
if self.monitoring:
return
self.monitoring = True
def monitor_loop():
while self.monitoring:
try:
self.log_memory_change("Continuous Monitor", force_gc=False)
time.sleep(interval_seconds)
except Exception as e:
logger.error(f"Memory monitor error: {e}")
break
self.monitor_thread = threading.Thread(target=monitor_loop, daemon=True)
self.monitor_thread.start()
logger.info(
f"🔍 Started continuous memory monitoring (every {interval_seconds}s)"
)
def stop_continuous_monitoring(self):
"""Stop continuous monitoring."""
if self.monitoring:
self.monitoring = False
logger.info("🛑 Stopped continuous memory monitoring")
# Global memory monitor instance
memory_monitor = MemoryMonitor()
# Convenience functions for easy use
def set_memory_baseline(label="Baseline"):
"""Set memory baseline."""
return memory_monitor.set_baseline(label)
def log_memory(label="Memory Check"):
"""Log current memory usage."""
return memory_monitor.log_memory_change(label)
def start_memory_monitoring():
"""Start continuous memory monitoring."""
memory_monitor.start_continuous_monitoring()
def stop_memory_monitoring():
"""Stop continuous memory monitoring."""
memory_monitor.stop_continuous_monitoring()
def check_memory_spike():
"""Check for memory spike."""
return memory_monitor.log_memory_spike()
# Test function
def test_memory_monitor():
"""Test the memory monitor."""
print("Testing Memory Monitor...")
monitor = MemoryMonitor()
monitor.set_baseline("Test Start")
# Simulate some memory usage
big_list = [i for i in range(100000)]
monitor.log_memory_change("After creating big list")
del big_list
monitor.log_memory_change("After deleting big list")
print("Memory monitor test complete!")
if __name__ == "__main__":
test_memory_monitor()
+124 -13
View File
@@ -1,6 +1,7 @@
import json
from gi.repository import Gdk
import gi
from gi.repository import Gdk, Glace
import config.data as data
from fabric.hyprland.widgets import get_hyprland_connection
@@ -11,6 +12,8 @@ from fabric.widgets.label import Label
from utils.icon_resolver import IconResolver
from widgets.wayland import WaylandWindow as Window
gi.require_version("Glace", "0.1")
class ApplicationSwitcher(Window):
def __init__(self, **kwargs):
@@ -33,6 +36,12 @@ class ApplicationSwitcher(Window):
self.items_per_row = data.WINDOW_SWITCHER_ITEMS_PER_ROW
self.icon_size = 64
# Initialize Glace manager for window previews
self._manager = Glace.Manager()
self.preview_size = [150, 100] # Preview size for each window in switcher
self.glace_clients = {} # Map window addresses to Glace clients
self.window_previews = {} # Map window addresses to preview images
container = Box(
name="application-switcher-container",
orientation="v",
@@ -45,13 +54,18 @@ class ApplicationSwitcher(Window):
self.view = Box(
name="application-switcher-view",
orientation="v",
spacing=0,
spacing=12,
h_align="center",
v_align="center",
)
container.add(self.view)
self.connect("key-press-event", self.on_key_press)
self.connect("key-release-event", self.on_key_release)
# Connect to Glace manager signals to track clients
self._manager.connect("client-added", self._on_glace_client_added)
self._manager.connect("client-removed", self._on_glace_client_removed)
self.show_all()
self.hide()
@@ -68,6 +82,103 @@ class ApplicationSwitcher(Window):
self.hide()
self.ungrab_keyboard()
def _on_glace_client_added(self, _, client):
"""Handle when a Glace client is added"""
try:
# Map the client by its window address for later lookup
# We'll need to match this with Hyprland window data
client_id = client.get_id()
self.glace_clients[client_id] = client
except Exception as e:
print(f"Error adding Glace client: {e}")
def _on_glace_client_removed(self, _, client):
"""Handle when a Glace client is removed"""
try:
client_id = client.get_id()
if client_id in self.glace_clients:
del self.glace_clients[client_id]
except Exception as e:
print(f"Error removing Glace client: {e}")
def _find_glace_client_for_window(self, window):
"""Find the corresponding Glace client for a Hyprland window"""
try:
window_class = window.get("class", "").lower()
window_title = window.get("title", "")
# Try to match by app_id/class and title
for _, client in self.glace_clients.items():
try:
client_app_id = client.get_app_id()
client_title = client.get_title()
if (client_app_id and client_app_id.lower() == window_class and
client_title and client_title == window_title):
return client
except Exception:
continue
# Fallback: try to match by class only
for _, client in self.glace_clients.items():
try:
client_app_id = client.get_app_id()
if client_app_id and client_app_id.lower() == window_class:
return client
except Exception:
continue
except Exception as e:
print(f"Error finding Glace client: {e}")
return None
def create_preview_for_window(self, window):
"""Create a preview image for a specific window"""
glace_client = self._find_glace_client_for_window(window)
# Create a placeholder image first
preview_image = Image()
if glace_client:
def capture_callback(pbuf, _):
try:
scaled_pixbuf = pbuf.scale_simple(
self.preview_size[0],
self.preview_size[1],
2 # GdkPixbuf.InterpType.BILINEAR
)
preview_image.set_from_pixbuf(scaled_pixbuf)
except Exception as e:
print(f"Error setting preview image: {e}")
try:
self._manager.capture_client(
client=glace_client,
overlay_cursor=False,
callback=capture_callback,
user_data=None,
)
except Exception as e:
print(f"Error capturing client preview: {e}")
# Fallback to icon if preview fails
self._set_fallback_icon(preview_image, window)
else:
# Use icon as fallback if no Glace client found
self._set_fallback_icon(preview_image, window)
return preview_image
def _set_fallback_icon(self, image_widget, window):
"""Set a fallback icon when preview is not available"""
class_name = window.get("class", "").lower()
icon_img = self.icon_resolver.get_icon_pixbuf(class_name, self.icon_size)
if not icon_img:
icon_img = self.icon_resolver.get_icon_pixbuf(
"application-x-executable-symbolic", self.icon_size
)
image_widget.set_from_pixbuf(icon_img)
def _is_special_workspace(self, client):
"""Check if a client is in a special workspace"""
if "workspace" not in client:
@@ -135,16 +246,10 @@ class ApplicationSwitcher(Window):
self.view.add(current_row)
for i, window in enumerate(self.windows):
class_name = window.get("class", "").lower()
title = window.get("title", "")
icon_img = self.icon_resolver.get_icon_pixbuf(
class_name, self.icon_size
)
if not icon_img:
icon_img = self.icon_resolver.get_icon_pixbuf(
"application-x-executable-symbolic", self.icon_size
)
# Create preview image for this window
preview_image = self.create_preview_for_window(window)
button_content = Box(
name="switcher-button",
@@ -153,7 +258,13 @@ class ApplicationSwitcher(Window):
h_align="center",
v_align="center",
children=[
Image(pixbuf=icon_img),
Box(
name="switcher-preview-box",
style_classes=["window-basic", "sleek-border"],
children=[preview_image],
h_align="center",
v_align="center",
),
Label(
label=title[:15] + "..." if len(title) > 15 else title,
h_align="center",
@@ -186,7 +297,7 @@ class ApplicationSwitcher(Window):
except Exception as e:
print(f"Failed to update windows: {e}")
def on_key_press(self, widget, event):
def on_key_press(self, _, event):
keyval = event.keyval
state = event.state
alt_pressed = bool(state & Gdk.ModifierType.MOD1_MASK)
@@ -243,7 +354,7 @@ class ApplicationSwitcher(Window):
return False
def on_key_release(self, widget, event):
def on_key_release(self, _, event):
keyval = event.keyval
if keyval in (Gdk.KEY_Alt_L, Gdk.KEY_Alt_R):