"""System tray application for DaemonControl. This module provides the SystemTrayApp class that creates a system tray icon with a context menu for controlling the DaemonControl daemon. """ import os import subprocess import sys from pathlib import Path from typing import Optional from PyQt6.QtCore import QObject, QTimer, pyqtSignal from PyQt6.QtGui import QAction, QIcon from PyQt6.QtWidgets import QApplication, QMenu, QSystemTrayIcon class SystemTrayApp(QObject): """System tray application for DaemonControl. Provides a system tray icon with menu to control the daemon, monitor status, and access logs. """ # Signals for thread-safe communication status_changed = pyqtSignal(str) # Emits: 'running', 'stopped', 'error' def __init__(self): """Initialize system tray application. Creates: - QSystemTrayIcon with icons - Context menu with actions - Main GUI window (hidden by default) Note: Daemon is managed by systemd, not by this GUI. Use 'systemctl --user status daemoncontrol' to check daemon status. """ super().__init__() # Import here to avoid circular import from .main_window import MainWindow # Create main window (hidden by default) self.main_window = MainWindow() self.main_window.hide() # Connect signals self.main_window.jobs_modified.connect(self._on_jobs_modified) # Initialize database for fleet status monitoring from core.database import DatabaseManager self.db = DatabaseManager() # Initialize components self._init_tray_icon() self._init_menu() # Start fleet status monitoring (lightweight DB reads every 30 seconds) self._init_fleet_status_check() def _init_tray_icon(self) -> None: """Create system tray icon. Loads icons from gui/resources/ directory and sets up the tray icon with initial state (inactive/stopped). """ self.tray_icon = QSystemTrayIcon(self) # Load icons resources_dir = Path(__file__).parent / 'resources' self.icon_active = QIcon(str(resources_dir / 'icon_active.png')) self.icon_inactive = QIcon(str(resources_dir / 'icon_inactive.png')) self.icon_error = QIcon(str(resources_dir / 'icon_error.png')) # Set initial state (icon inactive = grey, daemon managed by systemd) self.tray_icon.setIcon(self.icon_inactive) self.tray_icon.setToolTip("DaemonControl GUI - Click to open dashboard") # Connect click signal to open GUI self.tray_icon.activated.connect(self._on_tray_activated) def _init_menu(self) -> None: """Create context menu with actions. Menu structure: - Open Dashboard - ───────────── (separator) - Open Logs Folder - ───────────── (separator) - Exit Note: Menu is created on-demand via _on_tray_activated() to work around Qt6 rendering issues on some Linux desktop environments. Daemon control (Start/Stop/Reload) has been removed since the daemon is now managed by systemd. Use systemctl commands to manage the daemon. """ # Don't set context menu here - we'll show it manually on right-click def _create_context_menu(self) -> QMenu: """Create and return context menu. Creates menu fresh each time to work around Qt6 rendering bugs on some Linux desktop environments (especially Wayland). Simplified menu without daemon controls (daemon managed by systemd). Returns: QMenu with all actions configured """ menu = QMenu() # Open Dashboard action open_gui_action = QAction("📊 Open Dashboard", menu) open_gui_action.triggered.connect(self.show_main_window) menu.addAction(open_gui_action) menu.addSeparator() # Open Logs Folder logs_action = QAction("📂 Open Logs Folder", menu) logs_action.triggered.connect(self.open_logs_folder) menu.addAction(logs_action) menu.addSeparator() # Exit action (closes GUI only, daemon keeps running via systemd) exit_action = QAction("Exit", menu) exit_action.triggered.connect(self.exit_app) menu.addAction(exit_action) return menu # =========================================================================== # DISABLED: Daemon control methods - daemon is now managed by systemd # =========================================================================== # # These methods are commented out because the daemon is now controlled by # systemd user services. To manage the daemon, use systemctl commands: # # systemctl --user start daemoncontrol # Start daemon # systemctl --user stop daemoncontrol # Stop daemon # systemctl --user restart daemoncontrol # Restart daemon # systemctl --user status daemoncontrol # Check status # # Keeping these methods here for reference in case manual control is needed # in the future (e.g., for development/testing). # =========================================================================== # def start_daemon(self) -> None: # """Start the daemon process. # # Starts start_daemon.py as a subprocess in the background. # Updates UI state and shows notification on success or failure. # """ # if self.daemon_running: # self._show_notification( # "Already Running", # "Daemon is already running" # ) # return # # try: # # Get path to start_daemon.py # daemon_script = Path(__file__).parent.parent / 'start_daemon.py' # # if not daemon_script.exists(): # raise FileNotFoundError(f"Daemon script not found: {daemon_script}") # # # Start daemon as subprocess # self.daemon_process = subprocess.Popen( # [sys.executable, str(daemon_script)], # stdout=subprocess.PIPE, # stderr=subprocess.PIPE, # stdin=subprocess.DEVNULL # Prevent blocking on input # ) # # # Give it a moment to potentially fail # import time # time.sleep(0.5) # # # Check if it died immediately # if self.daemon_process.poll() is not None: # stderr = self.daemon_process.stderr.read().decode('utf-8') # raise Exception(f"Daemon exited immediately: {stderr[:200]}") # # self.daemon_running = True # self._update_status('running') # self._show_notification( # "Daemon Started", # "DaemonControl is now running" # ) # # except Exception as e: # self._update_status('error') # self._show_notification( # "Startup Failed", # f"Failed to start daemon: {str(e)[:100]}" # ) # # def stop_daemon(self) -> None: # """Stop the daemon process gracefully. # # Sends SIGTERM for graceful shutdown, waits up to 10 seconds, # then force kills if necessary. Updates UI and shows notification. # """ # if not self.daemon_running or not self.daemon_process: # self._show_notification( # "Not Running", # "Daemon is not running" # ) # return # # try: # # Send SIGTERM for graceful shutdown # self.daemon_process.terminate() # # # Wait up to 10 seconds for graceful shutdown # try: # self.daemon_process.wait(timeout=10) # except subprocess.TimeoutExpired: # # Force kill if needed # self.daemon_process.kill() # self.daemon_process.wait() # # self.daemon_running = False # self.daemon_process = None # self._update_status('stopped') # self._show_notification( # "Daemon Stopped", # "DaemonControl has stopped" # ) # # except Exception as e: # self._update_status('error') # self._show_notification( # "Error", # f"Failed to stop daemon: {str(e)}" # ) # # def reload_jobs(self) -> None: # """Reload jobs without full daemon restart. # # Currently implemented as stop + restart after 2 seconds. # In future, this could use IPC to signal the daemon to reload. # """ # if not self.daemon_running: # self._show_notification( # "Not Running", # "Daemon must be running to reload jobs" # ) # return # # self._show_notification( # "Reloading Jobs", # "Restarting daemon to reload jobs..." # ) # # # Stop daemon # self.stop_daemon() # # # Restart after 2 seconds # QTimer.singleShot(2000, self.start_daemon) def open_logs_folder(self) -> None: """Open logs folder in system file manager. Uses xdg-open on Linux to open the logs directory. """ try: from core import ConfigManager config = ConfigManager() log_file = config.get('logging', 'daemon_log_file') log_dir = Path(log_file).expanduser().parent # Ensure directory exists if not log_dir.exists(): self._show_notification( "Logs Not Found", f"Log directory does not exist: {log_dir}" ) return # Open in file manager (Linux) subprocess.Popen(['xdg-open', str(log_dir)]) except Exception as e: self._show_notification( "Error", f"Failed to open logs folder: {str(e)}" ) def exit_app(self) -> None: """Exit the application. Stops daemon if running, then exits the Qt application. """ if self.daemon_running: self.stop_daemon() QApplication.quit() # DISABLED: Automatic status checking removed for performance # def _check_daemon_status(self) -> None: # """Check if daemon process is still running. # # Called by timer every 5 seconds. Updates UI if status changed. # """ # if self.daemon_process: # poll_result = self.daemon_process.poll() # if poll_result is not None: # # Process ended # self.daemon_running = False # self.daemon_process = None # self._update_status('stopped') # else: # if self.daemon_running: # # Inconsistent state # self.daemon_running = False # self._update_status('stopped') # def _update_status(self, status: str) -> None: # """Update UI based on status. # # Args: # status: One of 'running', 'stopped', or 'error' # # Updates: # - Tray icon # - Tooltip # - Menu actions (enable/disable) - only if they exist # - Status action text # """ # # Store current status for menu creation # self.current_status = status # # if status == 'running': # self.tray_icon.setIcon(self.icon_active) # self.tray_icon.setToolTip("DaemonControl - Running") # if self.status_action: # self.status_action.setText("Status: Running ✓") # if self.start_action: # self.start_action.setEnabled(False) # if self.stop_action: # self.stop_action.setEnabled(True) # if self.reload_action: # self.reload_action.setEnabled(True) # # elif status == 'stopped': # self.tray_icon.setIcon(self.icon_inactive) # self.tray_icon.setToolTip("DaemonControl - Stopped") # if self.status_action: # self.status_action.setText("Status: Stopped") # if self.start_action: # self.start_action.setEnabled(True) # if self.stop_action: # self.stop_action.setEnabled(False) # if self.reload_action: # self.reload_action.setEnabled(False) # # elif status == 'error': # self.tray_icon.setIcon(self.icon_error) # self.tray_icon.setToolTip("DaemonControl - Error") # if self.status_action: # self.status_action.setText("Status: Error ✗") # if self.start_action: # self.start_action.setEnabled(True) # if self.stop_action: # self.stop_action.setEnabled(False) # if self.reload_action: # self.reload_action.setEnabled(False) # # # Emit signal for potential listeners # self.status_changed.emit(status) def _show_notification(self, title: str, message: str) -> None: """Show system notification. Args: title: Notification title message: Notification message Displays a system tray notification for 3 seconds. """ self.tray_icon.showMessage( title, message, QSystemTrayIcon.MessageIcon.Information, 3000 # 3 seconds ) def show(self) -> None: """Show the tray icon. Makes the tray icon visible in the system tray. """ self.tray_icon.show() def show_main_window(self) -> None: """Show main GUI window. Opens the main window and brings it to front. """ self.main_window.show() self.main_window.raise_() self.main_window.activateWindow() def _on_tray_activated(self, reason) -> None: """Handle tray icon activation (click). Args: reason: Activation reason from Qt Opens main window on double-click. Shows context menu on right-click (manually created to fix Qt6/Wayland issues). """ from PyQt6.QtWidgets import QSystemTrayIcon from PyQt6.QtGui import QCursor if reason == QSystemTrayIcon.ActivationReason.DoubleClick: self.show_main_window() elif reason == QSystemTrayIcon.ActivationReason.Trigger: # Left click - show main window self.show_main_window() elif reason == QSystemTrayIcon.ActivationReason.Context: # Right click - show menu manually # Create fresh menu each time (fixes Qt6/Wayland rendering bug) menu = self._create_context_menu() # Try to position menu near tray icon, fall back to cursor position geometry = self.tray_icon.geometry() if geometry.isValid(): # Position menu below tray icon menu.exec(geometry.bottomLeft()) else: # Fallback to cursor position menu.exec(QCursor.pos()) def _on_jobs_modified(self) -> None: """Handle jobs modified from GUI. If daemon is running, reload it to pick up changes. """ if self.daemon_running: self._show_notification( "Jobs Modified", "Reloading daemon to apply changes..." ) self.reload_jobs() def _init_fleet_status_check(self) -> None: """Setup lightweight fleet status monitoring from database. Reads overall fleet status from database every 30 seconds and updates tray icon color accordingly. This is very lightweight (single DB query, no network/docker calls). """ # Check immediately on startup self._update_icon_from_db() # Then check every 30 seconds self.fleet_status_timer = QTimer() self.fleet_status_timer.timeout.connect(self._update_icon_from_db) self.fleet_status_timer.start(30000) # 30 seconds def _update_icon_from_db(self) -> None: """Update tray icon based on latest overall fleet status from database. Icon colors: - Green (active): All systems healthy - Red (error): Critical issues detected - Grey (inactive): Warning, unknown status, or stale data """ try: status_data = self.db.get_latest_fleet_status() overall = status_data.get('overall', {}) status = overall.get('status', 'unknown') age_seconds = overall.get('age_seconds') # Check if data is stale (daemon might be stopped) if age_seconds is None or age_seconds > 600: # 10 minutes self.tray_icon.setIcon(self.icon_inactive) self.tray_icon.setToolTip("DaemonControl - No recent data (daemon stopped?)") return # Update icon based on fleet status if status == 'healthy': self.tray_icon.setIcon(self.icon_active) # Green self.tray_icon.setToolTip("DaemonControl - All Systems Healthy ✓") elif status == 'critical': self.tray_icon.setIcon(self.icon_error) # Red self.tray_icon.setToolTip("DaemonControl - Critical Issues Detected!") else: # warning or unknown self.tray_icon.setIcon(self.icon_inactive) # Grey self.tray_icon.setToolTip(f"DaemonControl - Status: {status.title()}") except Exception as e: # On any error, set grey icon print(f"Error updating tray icon from fleet status: {e}") self.tray_icon.setIcon(self.icon_inactive) self.tray_icon.setToolTip("DaemonControl - Status check error")