sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge branch 'macos' of https://github.com/tr1xem/Modus into macos

Browse Source
This commit is contained in:
tr1x_em committed 2025-07-30 15:49:56 +05:30
2 parents d0cce24bfb 33f464087c
commit c9326bc8af
14 files changed
+164 -610
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

config/assets/launcher.json
-12
View File
@@ -43,14 +43,6 @@
"description": "Quick Web Search - Fast web search with question mark",
"examples": ["? cats", "? google cats", "? youtube music"]
},
"kanban": {
"icon": "kanban",
"description": "Kanban - Task management with kanban-style columns"
},
"cal": {
"icon": "calendar",
"description": "Calendar - View calendar, navigate dates, and get date/time info"
},
"remind": {
"icon": "timer_on",
"description": "Reminders - Set time-based reminders with notifications"
@@ -67,10 +59,6 @@
"icon": "bookmark",
"description": "Bookmarks - Search and manage bookmarks"
},
"ps": {
"icon": "process",
"description": "Process Manager - View and manage running processes"
},
"script": {
"icon": "terminal",
"description": "Bash Scripts - Manage and execute bash scripts"
main.py
-6
View File
@@ -46,12 +46,6 @@ if __name__ == "__main__":
css_file = monitor_file(get_relative_path("styles"))
_ = css_file.connect("changed", lambda *_: set_css())
color_css_main_file = monitor_file(get_relative_path("main.css"))
_ = color_css_main_file.connect("changed", lambda *_: set_css())
color_css_file = monitor_file(get_relative_path("./styles/colors.css"))
_ = color_css_file.connect("changed", lambda *_: set_css())
# Make sure corners is added to the app
app = Application(f"{APP_NAME}", panel, dock, switcher, launcher, osd)
modules/launcher/plugin_manager.py
-1
View File
@@ -91,7 +91,6 @@ class PluginManager:
"otp",
"password",
"bookmarks",
"process",
"bash_scripts",
"tmux",
]
modules/launcher/plugins/bookmarks.py
+63 -12
View File
@@ -267,7 +267,7 @@ class BookmarksPlugin(PluginBase):
return [
Result(
title="No Bookmarks Found",
subtitle="Use 'add <title> <url>' to add your first bookmark",
subtitle="Use 'add <title> <url>' to add first bookmark",
icon_markup=icons.info,
action=lambda: None,
relevance=1.0,
@@ -297,7 +297,7 @@ class BookmarksPlugin(PluginBase):
return [
Result(
title="No Bookmarks Yet",
subtitle="Use 'add <title> <url>' to add your first bookmark",
subtitle="Use 'add <title> <url>' to add first bookmark",
icon_markup=icons.info,
action=lambda: None,
relevance=1.0,
@@ -378,10 +378,15 @@ class BookmarksPlugin(PluginBase):
)
if already_exists:
# Truncate URL for display to prevent launcher resize
display_url = normalized_url
if len(display_url) > 35:
display_url = display_url[:32] + "..."
return [
Result(
title="Bookmark Already Exists",
subtitle=f"A bookmark with URL '{normalized_url}' already exists",
subtitle=f"URL '{display_url}' already exists",
icon_markup=icons.alert,
action=lambda: None,
relevance=1.0,
@@ -392,13 +397,27 @@ class BookmarksPlugin(PluginBase):
# Show add action - will execute on Enter
domain = self.bookmark_manager._extract_domain(normalized_url)
subtitle = f"Click to add bookmark: {domain}"
# Truncate domain if too long
if len(domain) > 25:
domain = domain[:22] + "..."
subtitle = f"Click to add: {domain}"
if description:
# Truncate description to prevent launcher resize
max_desc_len = 35 - len(domain) # Account for domain + separator
if len(description) > max_desc_len:
description = description[:max_desc_len-3] + "..."
subtitle += f" • {description}"
# Truncate title for display
display_title = title
if len(display_title) > 25:
display_title = display_title[:22] + "..."
return [
Result(
title=f"Add bookmark for '{title}'",
title=f"Add bookmark '{display_title}'",
subtitle=subtitle,
icon_markup=icons.plus,
action=lambda: self._add_bookmark_action(title, url, description),
@@ -445,10 +464,19 @@ class BookmarksPlugin(PluginBase):
title = bookmark.get("title", "Untitled")
domain = self.bookmark_manager._extract_domain(bookmark.get("url", ""))
# Truncate title and domain for display
display_title = title
if len(display_title) > 25:
display_title = display_title[:22] + "..."
display_domain = domain
if len(display_domain) > 30:
display_domain = display_domain[:27] + "..."
return [
Result(
title=f"Remove bookmark '{title}'?",
subtitle=f"Click to confirm deletion: {domain}",
title=f"Remove '{display_title}'?",
subtitle=f"Click to confirm: {display_domain}",
icon_markup=icons.trash,
action=lambda: self._remove_bookmark_action(identifier),
relevance=1.0,
@@ -490,10 +518,20 @@ class BookmarksPlugin(PluginBase):
for bookmark in bookmarks[:3]:
title = bookmark.get("title", "Untitled")
domain = self._extract_domain(bookmark.get("url", ""))
# Truncate title and domain for consistent display
display_title = title
if len(display_title) > 20:
display_title = display_title[:17] + "..."
display_domain = domain
if len(display_domain) > 20:
display_domain = display_domain[:17] + "..."
results.append(
Result(
title=f"remove {title}",
subtitle=f"Click to remove: {title} ({domain})",
title=f"remove {display_title}",
subtitle=f"Click to remove: {display_title} ({display_domain})",
icon_markup=icons.trash,
action=lambda t=title: self._remove_bookmark_action(t),
relevance=0.8,
@@ -592,17 +630,30 @@ class BookmarksPlugin(PluginBase):
url = bookmark.get("url", "")
description = bookmark.get("description", "")
# Truncate long titles
if len(title) > 60:
title = title[:57] + "..."
# Truncate long titles to prevent launcher resize
if len(title) > 45:
title = title[:42] + "..."
# Create subtitle with domain and description
domain = self._extract_domain(url)
# Truncate domain if too long
if len(domain) > 30:
domain = domain[:27] + "..."
if description:
# Truncate description to prevent launcher resize
max_desc_len = 50 - len(domain) # Account for domain + separator
if len(description) > max_desc_len:
description = description[:max_desc_len-3] + "..."
subtitle = f"{domain} • {description}"
else:
subtitle = domain
# Final subtitle length check to ensure consistent launcher size
if len(subtitle) > 60:
subtitle = subtitle[:57] + "..."
return Result(
title=title,
subtitle=subtitle,
modules/launcher/plugins/otp.py
+3 -2
View File
@@ -82,10 +82,11 @@ class OTPPlugin(PluginBase):
"""Start background thread for auto-refreshing tokens."""
def refresh_loop():
while not self.stop_refresh.wait(1):
while not self.stop_refresh.wait(5):
current_time = time.time()
if current_time - self.last_update >= 1:
if current_time - self.last_update >= 5:
self.last_update = current_time
# Only refresh if we have secrets and launcher is likely active
if self.secrets:
try:
self._selective_force_refresh()
modules/launcher/plugins/process.py
-523
View File
@@ -1,523 +0,0 @@
import threading
import time
from typing import List
import psutil
import utils.icons as icons
from modules.launcher.plugin_base import PluginBase
from modules.launcher.result import Result
class ProcessPlugin(PluginBase):
"""
Plugin for viewing and managing running processes with live CPU/RAM updates.
"""
def __init__(self):
super().__init__()
self.display_name = "Process Manager"
self.description = "View and manage running processes"
# Threading for auto-refresh
self.refresh_thread = None
self.stop_refresh = threading.Event()
self.last_update = 0
# CPU monitoring cache to get meaningful readings
self._cpu_cache = {}
self._last_cpu_update = 0
def initialize(self):
"""Initialize the process plugin."""
self.set_triggers(["ps"])
# Pre-populate CPU cache for immediate meaningful readings
self._initialize_cpu_monitoring()
self._start_refresh_thread()
def _initialize_cpu_monitoring(self):
"""Initialize CPU monitoring for all processes to get meaningful readings immediately."""
try:
for proc in psutil.process_iter(["pid"]):
try:
pid = proc.info["pid"]
# Initialize CPU monitoring - this establishes baseline for future readings
proc.cpu_percent()
self._cpu_cache[pid] = proc
except (
psutil.NoSuchProcess,
psutil.AccessDenied,
psutil.ZombieProcess,
):
continue
self._last_cpu_update = time.time()
except Exception as e:
print(f"ProcessPlugin: Error initializing CPU monitoring: {e}")
def cleanup(self):
"""Cleanup the process plugin."""
self.stop_refresh.set()
if self.refresh_thread:
self.refresh_thread.join(timeout=1)
# Clear CPU cache
self._cpu_cache.clear()
def _get_live_process_data(self):
"""Get fresh live process data with CPU/RAM information, grouped by application."""
try:
current_time = time.time()
# Initialize CPU monitoring if this is the first call or enough time has passed
if current_time - self._last_cpu_update > 0.1: # Update every 100ms
# Pre-populate CPU cache for all processes
for proc in psutil.process_iter(["pid"]):
try:
pid = proc.info["pid"]
# Initialize CPU monitoring - this call establishes baseline
proc.cpu_percent()
self._cpu_cache[pid] = proc
except (
psutil.NoSuchProcess,
psutil.AccessDenied,
psutil.ZombieProcess,
):
continue
self._last_cpu_update = current_time
# Group processes by application name
app_groups = {}
for proc in psutil.process_iter(["pid", "name", "status", "ppid"]):
try:
info = proc.info
if info["status"] == psutil.STATUS_ZOMBIE:
continue
pid = info["pid"]
name = info["name"] or "Unknown"
# Get CPU percentage - use cached process if available for better readings
if pid in self._cpu_cache:
try:
cpu_percent = self._cpu_cache[pid].cpu_percent()
except (psutil.NoSuchProcess, psutil.AccessDenied):
cpu_percent = proc.cpu_percent()
self._cpu_cache[pid] = proc
else:
cpu_percent = proc.cpu_percent()
self._cpu_cache[pid] = proc
# Normalize CPU to 0-100% range (divide by number of cores)
cpu_cores = psutil.cpu_count()
if cpu_cores and cpu_percent > 0:
cpu_percent = min(100.0, cpu_percent / cpu_cores)
memory_info = proc.memory_info()
memory_percent = proc.memory_percent()
memory_mb = memory_info.rss / (1024 * 1024) if memory_info else 0
process_data = {
"pid": pid,
"ppid": info["ppid"],
"name": name,
"cpu_percent": cpu_percent,
"memory_percent": memory_percent,
"memory_mb": memory_mb,
"status": info["status"],
}
# Group processes by application name
app_name = self._get_app_name(name)
if app_name not in app_groups:
app_groups[app_name] = []
app_groups[app_name].append(process_data)
except (
psutil.NoSuchProcess,
psutil.AccessDenied,
psutil.ZombieProcess,
):
continue
# Create grouped results
grouped_processes = []
for app_name, app_processes in app_groups.items():
if len(app_processes) == 1:
# Single process - show as individual
proc = app_processes[0]
proc["process_count"] = 1
grouped_processes.append(proc)
else:
# Multiple processes - group them and sum resources
total_cpu = sum(p["cpu_percent"] for p in app_processes)
total_memory_mb = sum(p["memory_mb"] for p in app_processes)
total_memory_percent = sum(
p["memory_percent"] for p in app_processes
)
# Find the main process (usually the one with lowest PID)
main_process = min(app_processes, key=lambda x: x["pid"])
grouped_process = {
# Use main process PID for killing
"pid": main_process["pid"],
"name": app_name,
"cpu_percent": total_cpu,
"memory_percent": total_memory_percent,
"memory_mb": total_memory_mb,
"status": main_process["status"],
"process_count": len(app_processes),
"child_pids": [p["pid"] for p in app_processes],
}
grouped_processes.append(grouped_process)
# Clean up old entries from cache
all_pids = set()
for app_processes in app_groups.values():
all_pids.update(p["pid"] for p in app_processes)
self._cpu_cache = {
pid: proc for pid, proc in self._cpu_cache.items() if pid in all_pids
}
# Sort by combined CPU and memory usage for better display
grouped_processes.sort(
key=lambda x: (x["cpu_percent"] + x["memory_percent"]), reverse=True
)
return grouped_processes
except Exception as e:
print(f"ProcessPlugin: Error getting process data: {e}")
return []
def _get_app_name(self, process_name: str) -> str:
"""Extract application name from process name, handling common patterns."""
# Remove common suffixes and normalize
name = process_name.lower()
# Handle common browser patterns
if "firefox" in name:
return "Firefox"
elif "chrome" in name or "chromium" in name:
return "Chrome"
elif "zen" in name or name in [
"socket process",
"privileged cont",
"rdd process",
"isolated web co",
"web content",
"webextensions",
"utility process",
"isolated servic",
]:
return "Zen Browser"
elif "code" in name and ("helper" in name or "oss" in name or name == "code"):
return "VS Code"
elif name.startswith("python") and len(name) > 6:
return "Python"
elif name.startswith("node") and len(name) > 4:
return "Node.js"
elif "electron" in name:
return "Electron"
elif "java" in name:
return "Java"
elif "gnome" in name:
return "GNOME"
elif "gtk" in name:
return "GTK App"
# Remove common suffixes
suffixes = [
"-bin",
"-real",
"-wrapped",
".bin",
".exe",
"-helper",
"-gpu",
"-renderer",
]
for suffix in suffixes:
if name.endswith(suffix):
name = name[: -len(suffix)]
break
# Capitalize first letter
return name.capitalize()
def query(self, query_string: str) -> List[Result]:
"""Search for processes matching the query with live updates - ALWAYS FRESH DATA."""
# ALWAYS get completely fresh data - NO CACHING EVER
processes = self._get_live_process_data()
query = query_string.strip().lower()
results = []
# Filter processes based on query
filtered_processes = processes
if query:
filtered_processes = [
proc
for proc in processes
if query in proc["name"].lower() or query in str(proc["pid"])
]
# Limit to top 20 processes to avoid overwhelming the UI
for proc in filtered_processes[:20]:
result = self._create_process_result(proc)
if result:
results.append(result)
return results
def _create_process_result(self, proc_data: dict) -> Result:
"""Create a Result object for a process or grouped application."""
try:
pid = proc_data["pid"]
name = proc_data["name"]
cpu_percent = proc_data["cpu_percent"]
memory_percent = proc_data["memory_percent"]
memory_mb = proc_data["memory_mb"]
process_count = proc_data.get("process_count", 1)
# Format the title with process count if grouped
if process_count > 1:
title = f"{name} ({process_count} processes)"
else:
title = f"{name} (PID: {pid})"
# Add visual indicators for high usage
cpu_indicator = (
"🔥" if cpu_percent > 80 else "⚡" if cpu_percent > 50 else ""
)
mem_indicator = "💾" if memory_percent > 80 else ""
subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: {memory_mb:.1f}MB ({memory_percent:.1f}%){mem_indicator}"
# Choose icon based on CPU usage
if cpu_percent > 50:
icon_markup = icons.cpu
elif memory_percent > 50:
icon_markup = icons.memory
else:
icon_markup = icons.terminal
return Result(
title=title,
subtitle=subtitle,
icon_markup=icon_markup,
action=lambda: None, # No action on Enter - just show process info
relevance=min(1.0, (cpu_percent + memory_percent) / 100.0),
plugin_name=self.display_name,
data={
"type": "process",
"pid": pid,
"name": name,
"cpu_percent": cpu_percent,
"memory_percent": memory_percent,
"memory_mb": memory_mb,
"process_count": process_count,
"child_pids": proc_data.get("child_pids", [pid]),
"keep_launcher_open": True, # Keep launcher open to see live updates
"alt_action": lambda data=proc_data: self._kill_process_group(data),
},
)
except Exception as e:
print(f"ProcessPlugin: Error creating process result: {e}")
return None
def _kill_process_group(self, proc_data: dict):
"""Kill a process group (application with all its subprocesses)."""
try:
process_count = proc_data.get("process_count", 1)
child_pids = proc_data.get("child_pids", [proc_data["pid"]])
app_name = proc_data["name"]
if process_count == 1:
# Single process
self._kill_process(proc_data["pid"])
else:
# Multiple processes - kill all
killed_count = 0
for pid in child_pids:
if self._kill_process(pid, silent=True):
killed_count += 1
except Exception as e:
print(f"✗ Error killing process group: {e}")
def _kill_process(self, pid: int, silent: bool = False) -> bool:
"""Kill a process by PID. Returns True if successful."""
try:
proc = psutil.Process(pid)
proc_name = proc.name()
# Try graceful termination first
proc.terminate()
# Wait a bit for graceful termination
try:
proc.wait(timeout=3)
return True
except psutil.TimeoutExpired:
# Force kill if graceful termination failed
proc.kill()
if not silent:
print(f"✓ Force killed process '{proc_name}' (PID: {pid})")
return True
except psutil.NoSuchProcess:
if not silent:
print(f"✗ Process {pid} not found - may have already been terminated")
return False
except psutil.AccessDenied:
if not silent:
print(
f"✗ Access denied - cannot kill process {pid} (insufficient permissions)"
)
return False
except Exception as e:
if not silent:
print(f"✗ Error killing process {pid}: {e}")
return False
def _start_refresh_thread(self):
"""Start background thread for auto-refreshing process data in milliseconds."""
def refresh_loop():
# Check every 500ms for real-time updates
while not self.stop_refresh.wait(0.5):
current_time = time.time()
if current_time - self.last_update >= 0.5: # Update every 500ms
self.last_update = current_time
# Update existing process labels without full refresh
try:
self._selective_force_refresh()
except Exception:
pass
self.refresh_thread = threading.Thread(target=refresh_loop, daemon=True)
self.refresh_thread.start()
def _selective_force_refresh(self):
"""Update process data in existing result items (like OTP plugin)."""
try:
import gc
from gi.repository import GLib
def do_update():
try:
for obj in gc.get_objects():
if (
hasattr(obj, "__class__")
and obj.__class__.__name__ == "Launcher"
and hasattr(obj, "results_box")
and hasattr(obj, "visible")
and obj.visible
and hasattr(obj, "results")
and obj.results
):
has_process_results = any(
result.data and result.data.get("type") == "process"
for result in obj.results
if hasattr(result, "data") and result.data
)
if has_process_results:
self._update_existing_process_labels(obj.results_box)
return False
except Exception:
pass
return False
GLib.idle_add(do_update)
except Exception:
pass
def _update_existing_process_labels(self, results_box):
"""Update subtitle labels in existing ResultItem widgets with FRESH data (optimized)."""
try:
# Get fresh process data in background thread - optimized
def get_data_async():
return self._get_live_process_data()
# Run data collection in background to avoid blocking UI
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(get_data_async)
try:
current_processes = future.result(timeout=1.0) # 1 second timeout
process_dict = {proc["pid"]: proc for proc in current_processes}
for child in results_box.get_children():
if (
hasattr(child, "__class__")
and child.__class__.__name__ == "ResultItem"
and hasattr(child, "result")
and hasattr(child.result, "data")
and child.result.data
and child.result.data.get("type") == "process"
):
self._update_process_result_item(child, process_dict)
except concurrent.futures.TimeoutError:
# Skip update if it takes too long
pass
except Exception as e:
print(f"Error updating process labels: {e}")
def _update_process_result_item(self, result_item, process_dict):
"""Update both the title and subtitle of a specific process ResultItem."""
try:
pid = result_item.result.data.get("pid")
if not pid or pid not in process_dict:
return
proc_data = process_dict[pid]
cpu_percent = proc_data["cpu_percent"]
memory_percent = proc_data["memory_percent"]
memory_mb = proc_data["memory_mb"]
# CPU is already normalized in _get_live_process_data, so no need to normalize again
# Update the subtitle with live data
cpu_indicator = (
"🔥" if cpu_percent > 80 else "⚡" if cpu_percent > 50 else ""
)
mem_indicator = "💾" if memory_percent > 80 else ""
new_subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: {memory_mb:.1f}MB ({memory_percent:.1f}%){mem_indicator}"
# Find and update the subtitle label widget
self._find_and_update_subtitle_label(result_item, new_subtitle)
# Update the result data for consistency
result_item.result.data.update(
{
"cpu_percent": cpu_percent,
"memory_percent": memory_percent,
"memory_mb": memory_mb,
}
)
except Exception as e:
print(f"Error updating process result item: {e}")
def _find_and_update_subtitle_label(self, result_item, new_subtitle):
"""Find the subtitle label widget and update its text."""
def find_subtitle_label(widget):
if (
hasattr(widget, "get_name")
and widget.get_name() == "result-item-subtitle"
):
return widget
if hasattr(widget, "get_children"):
for child in widget.get_children():
found = find_subtitle_label(child)
if found:
return found
return None
subtitle_label = find_subtitle_label(result_item)
if subtitle_label and hasattr(subtitle_label, "set_text"):
subtitle_label.set_text(new_subtitle)
modules/launcher/plugins/tmux.py
+22 -8
View File
@@ -24,16 +24,18 @@ class TmuxPlugin(PluginBase):
# Cache for sessions to avoid repeated subprocess calls
self._sessions_cache = []
self._last_cache_update = 0
self._cache_ttl = 2 # Cache sessions for 2 seconds
self._cache_ttl = 10 # Cache sessions for 10 seconds (increased from 2)
# Threading for auto-refresh
# Threading for auto-refresh - only when actively used
self.refresh_thread = None
self.stop_refresh = threading.Event()
self._last_query_time = 0
self._active_refresh_timeout = 30 # Stop refreshing after 30 seconds of inactivity
def initialize(self):
"""Initialize the tmux plugin."""
self.set_triggers(["tmux"])
self._start_refresh_thread()
# Don't start refresh thread immediately - start it when first used
def cleanup(self):
"""Cleanup the tmux plugin."""
@@ -51,19 +53,24 @@ class TmuxPlugin(PluginBase):
self.refresh_thread.start()
def _refresh_sessions_background(self):
"""Background thread to refresh sessions cache."""
"""Background thread to refresh sessions cache only when actively used."""
while not self.stop_refresh.is_set():
try:
current_time = time.time()
# Stop refreshing if plugin hasn't been used recently
if current_time - self._last_query_time > self._active_refresh_timeout:
print("TmuxPlugin: Stopping background refresh due to inactivity")
break
if current_time - self._last_cache_update > self._cache_ttl:
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = current_time
# Sleep for a short interval
self.stop_refresh.wait(1)
self.stop_refresh.wait(5)
except Exception as e:
print(f"TmuxPlugin: Error in refresh thread: {e}")
self.stop_refresh.wait(5) # Wait longer on error
self.stop_refresh.wait(10) # Wait longer on error
def _get_tmux_sessions(self):
"""Get list of tmux sessions."""
@@ -92,8 +99,15 @@ class TmuxPlugin(PluginBase):
query = query_string.strip().lower()
results = []
# Get current sessions (use cache if recent)
# Track usage and start refresh thread if needed
current_time = time.time()
self._last_query_time = current_time
# Start refresh thread if not running and plugin is being used
if not self.refresh_thread or not self.refresh_thread.is_alive():
self._start_refresh_thread()
# Get current sessions (use cache if recent)
if current_time - self._last_cache_update > self._cache_ttl:
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = current_time
modules/panel/components/indicators.py
+7 -8
View File
@@ -1,14 +1,13 @@
from fabric.bluetooth import BluetoothClient
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from fabric.utils import get_relative_path
from fabric.widgets.svg import Svg
from gi.repository import GLib
import utils.icons as icons
from services.network import NetworkClient
from gi.repository import GLib
from fabric.bluetooth import BluetoothClient
from fabric.utils import get_relative_path
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.svg import Svg
from services.battery import Battery as BatteryService
from services.network import NetworkClient
class WifiIndicator(Button):
modules/panel/components/recording_indicator.py
+37 -7
View File
@@ -1,10 +1,12 @@
import subprocess
import time
from gi.repository import GLib
from fabric.utils import get_relative_path
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
import time
import subprocess
from fabric.widgets.box import Box
from gi.repository import GLib
from fabric.utils import get_relative_path
from fabric.widgets.svg import Svg
@@ -20,12 +22,18 @@ class RecordingIndicator(Button):
size=24,
svg_file=get_relative_path("../../../config/assets/icons/media-record.svg"),
)
self.time_label = Label(name="recording-time-label", markup="00:00")
self.time_label = Label(
name="recording-time-label",
markup="00:00",
max_width_chars=5,
ellipsize="none"
)
self.recording_box = Box(
orientation="h",
spacing=2,
children=self.recording_icon,
children=[self.recording_icon, self.time_label],
size=(80, -1),
)
self.add(self.recording_box)
@@ -75,3 +83,25 @@ class RecordingIndicator(Button):
self.recording_start_time = None
return True # Continue the timeout
def get_recording_start_time(self):
"""Get the recording start time from the file"""
try:
with open("/tmp/recording_start_time.txt", "r") as f:
return float(f.read().strip())
except Exception:
return None
def on_stop_recording(self, *args):
try:
subprocess.Popen([self.script_path, "record", "stop"])
except Exception as e:
print(f"Error stopping recording: {e}")
def _delayed_init(self):
try:
self.check_recording_status()
self.timeout_id = GLib.timeout_add(5000, self.check_recording_status)
except Exception as e:
print(f"[DEBUG] Error in delayed recording indicator init: {e}")
return False # Don't repeat this timeout
modules/panel/main.py
+4 -3
View File
@@ -9,11 +9,10 @@ from fabric.widgets.revealer import Revealer
from fabric.widgets.svg import Svg
from utils.wayland import WaylandWindow as Window
from modules.panel.components.enhanced_system_tray import (
apply_enhanced_system_tray,
) # ignore
from modules.panel.components.enhanced_system_tray import apply_enhanced_system_tray
from modules.panel.components.indicators import Indicators
from modules.panel.components.menubar import MenuBar
from modules.panel.components.recording_indicator import RecordingIndicator
# Apply enhanced system tray icon handling
apply_enhanced_system_tray()
@@ -89,6 +88,7 @@ class Panel(Window):
),
),
)
self.recording_indicator = RecordingIndicator()
self.children = CenterBox(
name="panel",
@@ -108,6 +108,7 @@ class Panel(Window):
spacing=4,
orientation="h",
children=[
self.recording_indicator,
self.tray_revealer,
self.chevron_button,
self.indicators,
services/battery.py
+2 -2
View File
@@ -130,8 +130,8 @@ class Battery(Service):
self._psutil_battery = psutil.sensors_battery()
if self._psutil_battery is None:
return # No battery found
# Start periodic updates for psutil fallback
GLib.timeout_add_seconds(5, self._update_psutil_battery)
# Start periodic updates for psutil fallback - increased interval
GLib.timeout_add_seconds(10, self._update_psutil_battery)
except Exception:
return # psutil battery not available either
styles/colors.css
+24 -24
View File
@@ -1,32 +1,32 @@
:vars {
--foreground: #e4e1e9;
--background: #131318;
--cursor: #e4e1e9;
--primary: #bfc2ff;
--on-primary: #272b60;
--secondary: #c5c4dd;
--on-secondary: #2e2f42;
--tertiary: #e8b9d4;
--on-tertiary: #46263b;
--surface: #131318;
--surface-bright: #39383f;
--foreground: #f1dfdc;
--background: #1a1110;
--cursor: #f1dfdc;
--primary: #ffb4a8;
--on-primary: #561e17;
--secondary: #e7bdb6;
--on-secondary: #442925;
--tertiary: #dec48c;
--on-tertiary: #3e2e04;
--surface: #1a1110;
--surface-bright: #423735;
--error: #ffb4ab;
--error-dim: #ff8678;
--on-error: #690005;
--error-container: #93000a;
--outline: #918f9a;
--outline: #a08c89;
--shadow: #000000;
--red: #ffb2b9;
--red-dim: #ff7f8b;
--green: #95d5a7;
--green-dim: #70c789;
--yellow: #b8cf84;
--yellow-dim: #a3c15f;
--blue: #bfc2ff;
--blue-dim: #8c91ff;
--magenta: #e4b7f3;
--magenta-dim: #d48bec;
--cyan: #82d3e2;
--cyan-dim: #59c4d8;
--red: #ffb4a8;
--red-dim: #ff8875;
--green: #b7d085;
--green-dim: #a1c260;
--yellow: #dec56e;
--yellow-dim: #d5b444;
--blue: #cebdfe;
--blue-dim: #a98bfd;
--magenta: #fcb0d5;
--magenta-dim: #fa7fbb;
--cyan: #84d5c4;
--cyan-dim: #5ec8b2;
--white: #82d3e0;
}
styles/menubar.css
+1 -1
View File
@@ -3,7 +3,7 @@
}
#dropdown-content {
background-color: var(--shadow);
background-color: alpha(var(--shadow), 0.6);
border-radius: 8px;
}
styles/panel.css
+1 -1
View File
@@ -1,5 +1,5 @@
#panel {
background-color: alpha(var(--surface), 0.6);
background-color: alpha(var(--shadow), 0.6);
}
#panel-icon {