""" Main application entry point for the Cluster4NPU UI application. This module initializes the PyQt5 application, applies the theme, and launches the main dashboard window. It serves as the primary entry point for the modularized UI application. Main Components: - Application initialization and configuration - Theme application and font setup - Main window instantiation and display - Application event loop management Usage: python -m cluster4npu_ui.main # Or directly: from main import main main() """ import sys import os import tempfile from PyQt5.QtWidgets import QApplication, QMessageBox from PyQt5.QtGui import QFont from PyQt5.QtCore import Qt, QSharedMemory # Import fcntl only on Unix-like systems try: import fcntl HAS_FCNTL = True except ImportError: HAS_FCNTL = False # Add the parent directory to the path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config.theme import apply_theme from ui.windows.login import DashboardLogin class SingleInstance: """Ensure only one instance of the application can run.""" def __init__(self, app_name="Cluster4NPU"): self.app_name = app_name self.shared_memory = QSharedMemory(app_name) self.lock_file = None self.lock_fd = None def _cleanup_stale_lock(self): """Clean up stale lock files from previous crashes.""" try: lock_path = os.path.join(tempfile.gettempdir(), f"{self.app_name}.lock") if os.path.exists(lock_path): # Try to remove stale lock file if HAS_FCNTL: # On Unix systems, try to acquire lock to check if process is still alive try: test_fd = os.open(lock_path, os.O_RDWR) fcntl.lockf(test_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) # If we got the lock, previous process is dead os.close(test_fd) os.unlink(lock_path) except (OSError, IOError): # Lock is held by another process pass else: # On Windows, just try to remove the file # If it's locked by another process, this will fail try: os.unlink(lock_path) except OSError: pass except Exception: pass def is_running(self): """Check if another instance is already running.""" # First, clean up any stale locks self._cleanup_stale_lock() # Try to attach to existing shared memory if self.shared_memory.attach(): # Try to write to shared memory to verify it's valid try: # If we can attach but can't access, it might be stale self.shared_memory.detach() # Try to create new shared memory if self.shared_memory.create(1): # Successfully created, no other instance pass else: # Failed to create, another instance exists return True except: # Shared memory is stale, try to create new one if not self.shared_memory.create(1): return True else: # Try to create the shared memory if not self.shared_memory.create(1): # Failed to create, likely another instance exists return True # Also use file locking as backup try: self.lock_file = os.path.join(tempfile.gettempdir(), f"{self.app_name}.lock") if HAS_FCNTL: self.lock_fd = os.open(self.lock_file, os.O_CREAT | os.O_WRONLY, 0o644) fcntl.lockf(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) # Write PID to lock file os.write(self.lock_fd, str(os.getpid()).encode()) os.fsync(self.lock_fd) else: # On Windows, use exclusive create self.lock_fd = os.open(self.lock_file, os.O_CREAT | os.O_EXCL | os.O_RDWR) os.write(self.lock_fd, str(os.getpid()).encode()) except (OSError, IOError): # Another instance is running or we can't create lock self._cleanup_on_error() return True return False def _cleanup_on_error(self): """Clean up resources when instance check fails.""" try: if self.shared_memory.isAttached(): self.shared_memory.detach() if self.lock_fd: os.close(self.lock_fd) self.lock_fd = None except: pass def cleanup(self): """Clean up resources.""" try: if self.shared_memory.isAttached(): self.shared_memory.detach() if self.lock_fd: try: if HAS_FCNTL: fcntl.lockf(self.lock_fd, fcntl.LOCK_UN) os.close(self.lock_fd) if self.lock_file and os.path.exists(self.lock_file): os.unlink(self.lock_file) except Exception: pass finally: self.lock_fd = None except Exception: pass def setup_application(): """Initialize and configure the QApplication.""" # Enable high DPI support BEFORE creating QApplication QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) # Create QApplication if it doesn't exist if not QApplication.instance(): app = QApplication(sys.argv) else: app = QApplication.instance() # Set application properties app.setApplicationName("Cluster4NPU") app.setApplicationVersion("1.0.0") app.setOrganizationName("Cluster4NPU Team") # Set application font app.setFont(QFont("Arial", 9)) # Apply the harmonious theme apply_theme(app) return app def main(): """Main application entry point.""" single_instance = None try: # Create a minimal QApplication first for the message box temp_app = QApplication(sys.argv) if not QApplication.instance() else QApplication.instance() # Check for single instance single_instance = SingleInstance() if single_instance.is_running(): QMessageBox.warning( None, "Application Already Running", "Cluster4NPU is already running. Please check your taskbar or system tray.", ) single_instance.cleanup() sys.exit(0) # Setup the full application app = setup_application() # Create and show the main dashboard login window dashboard = DashboardLogin() dashboard.show() # Set up cleanup handlers app.aboutToQuit.connect(single_instance.cleanup) # Also handle system signals for cleanup import signal def signal_handler(signum, frame): print(f"Received signal {signum}, cleaning up...") single_instance.cleanup() sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # Start the application event loop exit_code = app.exec_() # Ensure cleanup even if aboutToQuit wasn't called single_instance.cleanup() sys.exit(exit_code) except Exception as e: print(f"Error starting application: {e}") import traceback traceback.print_exc() if single_instance: single_instance.cleanup() sys.exit(1) finally: # Final cleanup attempt if single_instance: single_instance.cleanup() if __name__ == '__main__': main()