246 lines
8.1 KiB
Python
246 lines
8.1 KiB
Python
"""
|
|
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() |