658 lines
26 KiB
Python
658 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Robuste System-Control-Funktionen für wartungsfreien Produktionsbetrieb
|
|
Bietet sichere Restart-, Shutdown- und Kiosk-Verwaltungsfunktionen
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import subprocess
|
|
import time
|
|
import signal
|
|
import psutil
|
|
import threading
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
from pathlib import Path
|
|
import logging
|
|
import json
|
|
from contextlib import contextmanager
|
|
from enum import Enum
|
|
|
|
# Logging-Setup
|
|
try:
|
|
from utils.logging_config import get_logger
|
|
system_logger = get_logger("system_control")
|
|
except ImportError:
|
|
logging.basicConfig(level=logging.INFO)
|
|
system_logger = logging.getLogger("system_control")
|
|
|
|
|
|
class SystemOperation(Enum):
|
|
"""Verfügbare System-Operationen"""
|
|
RESTART = "restart"
|
|
SHUTDOWN = "shutdown"
|
|
KIOSK_RESTART = "kiosk_restart"
|
|
KIOSK_ENABLE = "kiosk_enable"
|
|
KIOSK_DISABLE = "kiosk_disable"
|
|
SERVICE_RESTART = "service_restart"
|
|
EMERGENCY_STOP = "emergency_stop"
|
|
|
|
|
|
class SystemControlManager:
|
|
"""
|
|
Zentraler Manager für alle System-Control-Operationen.
|
|
Bietet sichere und robuste Funktionen für wartungsfreien Betrieb.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.is_windows = os.name == 'nt'
|
|
self.pending_operations: Dict[str, Dict] = {}
|
|
self.operation_history: List[Dict] = []
|
|
self.lock = threading.Lock()
|
|
|
|
# Konfiguration
|
|
self.config = {
|
|
"restart_delay": 60, # Sekunden
|
|
"shutdown_delay": 30, # Sekunden
|
|
"kiosk_restart_delay": 10, # Sekunden
|
|
"max_operation_history": 100,
|
|
"safety_checks": True,
|
|
"require_confirmation": True
|
|
}
|
|
|
|
# Service-Namen für verschiedene Plattformen
|
|
self.services = {
|
|
"https": "myp-https.service",
|
|
"kiosk": "myp-kiosk.service",
|
|
"watchdog": "kiosk-watchdog.service"
|
|
}
|
|
|
|
system_logger.info("🔧 System-Control-Manager initialisiert")
|
|
|
|
def is_safe_to_operate(self) -> Tuple[bool, str]:
|
|
"""
|
|
Prüft ob System-Operationen sicher ausgeführt werden können.
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (is_safe, reason)
|
|
"""
|
|
try:
|
|
# Prüfe Systemlast
|
|
load_avg = psutil.getloadavg()[0] if hasattr(psutil, 'getloadavg') else 0
|
|
if load_avg > 2.0:
|
|
return False, f"Hohe Systemlast: {load_avg:.2f}"
|
|
|
|
# Prüfe verfügbaren Speicher
|
|
memory = psutil.virtual_memory()
|
|
if memory.percent > 90:
|
|
return False, f"Wenig verfügbarer Speicher: {memory.percent:.1f}% belegt"
|
|
|
|
# Prüfe aktive Drucker-Jobs
|
|
try:
|
|
from models import get_db_session, Job
|
|
db_session = get_db_session()
|
|
active_jobs = db_session.query(Job).filter(
|
|
Job.status.in_(["printing", "queued", "preparing"])
|
|
).count()
|
|
db_session.close()
|
|
|
|
if active_jobs > 0:
|
|
return False, f"Aktive Druckjobs: {active_jobs}"
|
|
except Exception as e:
|
|
system_logger.warning(f"Job-Prüfung fehlgeschlagen: {e}")
|
|
|
|
# Prüfe kritische Prozesse
|
|
critical_processes = ["chromium", "firefox", "python"]
|
|
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):
|
|
try:
|
|
if any(crit in proc.info['name'].lower() for crit in critical_processes):
|
|
if proc.info['cpu_percent'] > 80:
|
|
return False, f"Kritischer Prozess unter hoher Last: {proc.info['name']}"
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
continue
|
|
|
|
return True, "System ist sicher für Operationen"
|
|
|
|
except Exception as e:
|
|
system_logger.error(f"Fehler bei Sicherheitsprüfung: {e}")
|
|
return False, f"Sicherheitsprüfung fehlgeschlagen: {e}"
|
|
|
|
def schedule_operation(self,
|
|
operation: SystemOperation,
|
|
delay_seconds: int = None,
|
|
user_id: str = None,
|
|
reason: str = None,
|
|
force: bool = False) -> Dict[str, Any]:
|
|
"""
|
|
Plant eine System-Operation mit Verzögerung.
|
|
|
|
Args:
|
|
operation: Art der Operation
|
|
delay_seconds: Verzögerung in Sekunden (None = Standard)
|
|
user_id: ID des anfragenden Benutzers
|
|
reason: Grund für die Operation
|
|
force: Sicherheitsprüfungen überspringen
|
|
|
|
Returns:
|
|
Dict mit Operation-Details
|
|
"""
|
|
with self.lock:
|
|
# Sicherheitsprüfung (außer bei Force)
|
|
if not force and self.config["safety_checks"]:
|
|
is_safe, safety_reason = self.is_safe_to_operate()
|
|
if not is_safe:
|
|
return {
|
|
"success": False,
|
|
"error": f"Operation abgelehnt: {safety_reason}",
|
|
"safety_check": False
|
|
}
|
|
|
|
# Standard-Verzögerung setzen
|
|
if delay_seconds is None:
|
|
delay_seconds = {
|
|
SystemOperation.RESTART: self.config["restart_delay"],
|
|
SystemOperation.SHUTDOWN: self.config["shutdown_delay"],
|
|
SystemOperation.KIOSK_RESTART: self.config["kiosk_restart_delay"],
|
|
SystemOperation.KIOSK_ENABLE: 5,
|
|
SystemOperation.KIOSK_DISABLE: 5,
|
|
SystemOperation.SERVICE_RESTART: 10,
|
|
SystemOperation.EMERGENCY_STOP: 0
|
|
}.get(operation, 30)
|
|
|
|
# Operations-ID generieren
|
|
operation_id = f"{operation.value}_{int(time.time())}"
|
|
scheduled_time = datetime.now() + timedelta(seconds=delay_seconds)
|
|
|
|
# Operation speichern
|
|
operation_data = {
|
|
"id": operation_id,
|
|
"operation": operation.value,
|
|
"scheduled_time": scheduled_time,
|
|
"delay_seconds": delay_seconds,
|
|
"user_id": user_id,
|
|
"reason": reason or "Keine Begründung angegeben",
|
|
"force": force,
|
|
"created_at": datetime.now(),
|
|
"status": "scheduled"
|
|
}
|
|
|
|
self.pending_operations[operation_id] = operation_data
|
|
|
|
# Operation in separatem Thread ausführen
|
|
thread = threading.Thread(
|
|
target=self._execute_delayed_operation,
|
|
args=(operation_id,),
|
|
daemon=True
|
|
)
|
|
thread.start()
|
|
|
|
system_logger.info(f"🕐 Operation geplant: {operation.value} in {delay_seconds}s")
|
|
|
|
return {
|
|
"success": True,
|
|
"operation_id": operation_id,
|
|
"scheduled_time": scheduled_time.isoformat(),
|
|
"delay_seconds": delay_seconds,
|
|
"message": f"Operation '{operation.value}' geplant für {scheduled_time.strftime('%H:%M:%S')}"
|
|
}
|
|
|
|
def _execute_delayed_operation(self, operation_id: str):
|
|
"""
|
|
Führt geplante Operation nach Verzögerung aus.
|
|
|
|
Args:
|
|
operation_id: ID der auszuführenden Operation
|
|
"""
|
|
try:
|
|
operation_data = self.pending_operations.get(operation_id)
|
|
if not operation_data:
|
|
return
|
|
|
|
# Warten bis zur geplanten Zeit
|
|
scheduled_time = operation_data["scheduled_time"]
|
|
wait_time = (scheduled_time - datetime.now()).total_seconds()
|
|
|
|
if wait_time > 0:
|
|
time.sleep(wait_time)
|
|
|
|
# Status aktualisieren
|
|
operation_data["status"] = "executing"
|
|
operation_data["executed_at"] = datetime.now()
|
|
|
|
# Operation ausführen
|
|
operation = SystemOperation(operation_data["operation"])
|
|
result = self._execute_operation(operation, operation_data)
|
|
|
|
# Ergebnis speichern
|
|
operation_data["result"] = result
|
|
operation_data["status"] = "completed" if result.get("success") else "failed"
|
|
operation_data["completed_at"] = datetime.now()
|
|
|
|
# In Historie verschieben
|
|
self._move_to_history(operation_id)
|
|
|
|
except Exception as e:
|
|
system_logger.error(f"Fehler bei verzögerter Operation {operation_id}: {e}")
|
|
if operation_id in self.pending_operations:
|
|
self.pending_operations[operation_id]["status"] = "error"
|
|
self.pending_operations[operation_id]["error"] = str(e)
|
|
self._move_to_history(operation_id)
|
|
|
|
def _execute_operation(self, operation: SystemOperation, operation_data: Dict) -> Dict[str, Any]:
|
|
"""
|
|
Führt die eigentliche System-Operation aus.
|
|
|
|
Args:
|
|
operation: Art der Operation
|
|
operation_data: Operation-Daten
|
|
|
|
Returns:
|
|
Dict mit Ergebnis
|
|
"""
|
|
try:
|
|
system_logger.info(f"▶️ Führe Operation aus: {operation.value}")
|
|
|
|
if operation == SystemOperation.RESTART:
|
|
return self._restart_system(operation_data)
|
|
elif operation == SystemOperation.SHUTDOWN:
|
|
return self._shutdown_system(operation_data)
|
|
elif operation == SystemOperation.KIOSK_RESTART:
|
|
return self._restart_kiosk(operation_data)
|
|
elif operation == SystemOperation.KIOSK_ENABLE:
|
|
return self._enable_kiosk(operation_data)
|
|
elif operation == SystemOperation.KIOSK_DISABLE:
|
|
return self._disable_kiosk(operation_data)
|
|
elif operation == SystemOperation.SERVICE_RESTART:
|
|
return self._restart_services(operation_data)
|
|
elif operation == SystemOperation.EMERGENCY_STOP:
|
|
return self._emergency_stop(operation_data)
|
|
else:
|
|
return {"success": False, "error": f"Unbekannte Operation: {operation.value}"}
|
|
|
|
except Exception as e:
|
|
system_logger.error(f"Fehler bei Operation {operation.value}: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def _restart_system(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Startet das System neu."""
|
|
try:
|
|
system_logger.warning("🔄 System-Neustart wird ausgeführt...")
|
|
|
|
# Cleanup vor Neustart
|
|
self._cleanup_before_restart()
|
|
|
|
# System-Neustart je nach Plattform
|
|
if self.is_windows:
|
|
subprocess.run(["shutdown", "/r", "/t", "0"], check=True)
|
|
else:
|
|
subprocess.run(["sudo", "systemctl", "reboot"], check=True)
|
|
|
|
return {"success": True, "message": "System-Neustart initiiert"}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"success": False, "error": f"Neustart fehlgeschlagen: {e}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Unerwarteter Fehler: {e}"}
|
|
|
|
def _shutdown_system(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Fährt das System herunter."""
|
|
try:
|
|
system_logger.warning("🛑 System-Shutdown wird ausgeführt...")
|
|
|
|
# Cleanup vor Shutdown
|
|
self._cleanup_before_restart()
|
|
|
|
# System-Shutdown je nach Plattform
|
|
if self.is_windows:
|
|
subprocess.run(["shutdown", "/s", "/t", "0"], check=True)
|
|
else:
|
|
subprocess.run(["sudo", "systemctl", "poweroff"], check=True)
|
|
|
|
return {"success": True, "message": "System-Shutdown initiiert"}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"success": False, "error": f"Shutdown fehlgeschlagen: {e}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Unerwarteter Fehler: {e}"}
|
|
|
|
def _restart_kiosk(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Startet nur den Kiosk-Modus neu."""
|
|
try:
|
|
system_logger.info("🖥️ Kiosk-Neustart wird ausgeführt...")
|
|
|
|
success_count = 0
|
|
errors = []
|
|
|
|
# Kiosk-Service neustarten
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "restart", self.services["kiosk"]],
|
|
check=True, timeout=30)
|
|
success_count += 1
|
|
system_logger.info("✅ Kiosk-Service neugestartet")
|
|
except Exception as e:
|
|
errors.append(f"Kiosk-Service: {e}")
|
|
|
|
# Watchdog-Service neustarten (falls vorhanden)
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "restart", self.services["watchdog"]],
|
|
check=True, timeout=30)
|
|
success_count += 1
|
|
system_logger.info("✅ Watchdog-Service neugestartet")
|
|
except Exception as e:
|
|
errors.append(f"Watchdog-Service: {e}")
|
|
|
|
# X11-Session neustarten
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "restart", "getty@tty1.service"],
|
|
check=True, timeout=30)
|
|
success_count += 1
|
|
system_logger.info("✅ X11-Session neugestartet")
|
|
except Exception as e:
|
|
errors.append(f"X11-Session: {e}")
|
|
|
|
if success_count > 0:
|
|
return {
|
|
"success": True,
|
|
"message": f"Kiosk neugestartet ({success_count} Services)",
|
|
"errors": errors if errors else None
|
|
}
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"error": "Alle Kiosk-Neustarts fehlgeschlagen",
|
|
"details": errors
|
|
}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Kiosk-Neustart fehlgeschlagen: {e}"}
|
|
|
|
def _enable_kiosk(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Aktiviert den Kiosk-Modus."""
|
|
try:
|
|
system_logger.info("🖥️ Kiosk-Modus wird aktiviert...")
|
|
|
|
# Kiosk-Service aktivieren und starten
|
|
subprocess.run(["sudo", "systemctl", "enable", self.services["kiosk"]],
|
|
check=True, timeout=30)
|
|
subprocess.run(["sudo", "systemctl", "start", self.services["kiosk"]],
|
|
check=True, timeout=30)
|
|
|
|
# Watchdog aktivieren
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "enable", self.services["watchdog"]],
|
|
check=True, timeout=30)
|
|
subprocess.run(["sudo", "systemctl", "start", self.services["watchdog"]],
|
|
check=True, timeout=30)
|
|
except Exception as e:
|
|
system_logger.warning(f"Watchdog-Aktivierung fehlgeschlagen: {e}")
|
|
|
|
return {"success": True, "message": "Kiosk-Modus aktiviert"}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"success": False, "error": f"Kiosk-Aktivierung fehlgeschlagen: {e}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Unerwarteter Fehler: {e}"}
|
|
|
|
def _disable_kiosk(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Deaktiviert den Kiosk-Modus."""
|
|
try:
|
|
system_logger.info("🖥️ Kiosk-Modus wird deaktiviert...")
|
|
|
|
# Kiosk-Service stoppen und deaktivieren
|
|
subprocess.run(["sudo", "systemctl", "stop", self.services["kiosk"]],
|
|
check=True, timeout=30)
|
|
subprocess.run(["sudo", "systemctl", "disable", self.services["kiosk"]],
|
|
check=True, timeout=30)
|
|
|
|
# Watchdog stoppen
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "stop", self.services["watchdog"]],
|
|
check=True, timeout=30)
|
|
subprocess.run(["sudo", "systemctl", "disable", self.services["watchdog"]],
|
|
check=True, timeout=30)
|
|
except Exception as e:
|
|
system_logger.warning(f"Watchdog-Deaktivierung fehlgeschlagen: {e}")
|
|
|
|
return {"success": True, "message": "Kiosk-Modus deaktiviert"}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"success": False, "error": f"Kiosk-Deaktivierung fehlgeschlagen: {e}"}
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Unerwarteter Fehler: {e}"}
|
|
|
|
def _restart_services(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Startet wichtige Services neu."""
|
|
try:
|
|
system_logger.info("🔄 Services werden neugestartet...")
|
|
|
|
success_count = 0
|
|
errors = []
|
|
|
|
# HTTPS-Service neustarten
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "restart", self.services["https"]],
|
|
check=True, timeout=60)
|
|
success_count += 1
|
|
system_logger.info("✅ HTTPS-Service neugestartet")
|
|
except Exception as e:
|
|
errors.append(f"HTTPS-Service: {e}")
|
|
|
|
# NetworkManager neustarten (falls nötig)
|
|
try:
|
|
subprocess.run(["sudo", "systemctl", "restart", "NetworkManager"],
|
|
check=True, timeout=30)
|
|
success_count += 1
|
|
system_logger.info("✅ NetworkManager neugestartet")
|
|
except Exception as e:
|
|
errors.append(f"NetworkManager: {e}")
|
|
|
|
if success_count > 0:
|
|
return {
|
|
"success": True,
|
|
"message": f"Services neugestartet ({success_count})",
|
|
"errors": errors if errors else None
|
|
}
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"error": "Alle Service-Neustarts fehlgeschlagen",
|
|
"details": errors
|
|
}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Service-Neustart fehlgeschlagen: {e}"}
|
|
|
|
def _emergency_stop(self, operation_data: Dict) -> Dict[str, Any]:
|
|
"""Notfall-Stopp aller Services."""
|
|
try:
|
|
system_logger.warning("🚨 Notfall-Stopp wird ausgeführt...")
|
|
|
|
# Flask-App stoppen
|
|
try:
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
except Exception as e:
|
|
system_logger.error(f"Flask-Stopp fehlgeschlagen: {e}")
|
|
|
|
return {"success": True, "message": "Notfall-Stopp initiiert"}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": f"Notfall-Stopp fehlgeschlagen: {e}"}
|
|
|
|
def _cleanup_before_restart(self):
|
|
"""Führt Cleanup-Operationen vor Neustart/Shutdown aus."""
|
|
try:
|
|
system_logger.info("🧹 Cleanup vor Neustart/Shutdown...")
|
|
|
|
# Shutdown-Manager verwenden falls verfügbar
|
|
try:
|
|
from utils.shutdown_manager import get_shutdown_manager
|
|
shutdown_manager = get_shutdown_manager()
|
|
shutdown_manager.shutdown(exit_code=0)
|
|
except ImportError:
|
|
system_logger.warning("Shutdown-Manager nicht verfügbar")
|
|
|
|
# Datenbank-Cleanup
|
|
try:
|
|
from utils.database_cleanup import safe_database_cleanup
|
|
safe_database_cleanup(force_mode_switch=False)
|
|
except ImportError:
|
|
system_logger.warning("Database-Cleanup nicht verfügbar")
|
|
|
|
# Cache leeren
|
|
self._clear_caches()
|
|
|
|
except Exception as e:
|
|
system_logger.error(f"Cleanup fehlgeschlagen: {e}")
|
|
|
|
def _clear_caches(self):
|
|
"""Leert alle Caches."""
|
|
try:
|
|
# User-Cache leeren
|
|
from app import clear_user_cache, clear_printer_status_cache
|
|
clear_user_cache()
|
|
clear_printer_status_cache()
|
|
|
|
# System-Cache leeren
|
|
if not self.is_windows:
|
|
subprocess.run(["sudo", "sync"], timeout=10)
|
|
subprocess.run(["sudo", "echo", "3", ">", "/proc/sys/vm/drop_caches"],
|
|
shell=True, timeout=10)
|
|
|
|
except Exception as e:
|
|
system_logger.warning(f"Cache-Clearing fehlgeschlagen: {e}")
|
|
|
|
def _move_to_history(self, operation_id: str):
|
|
"""Verschiebt abgeschlossene Operation in Historie."""
|
|
with self.lock:
|
|
if operation_id in self.pending_operations:
|
|
operation_data = self.pending_operations.pop(operation_id)
|
|
self.operation_history.append(operation_data)
|
|
|
|
# Historie begrenzen
|
|
if len(self.operation_history) > self.config["max_operation_history"]:
|
|
self.operation_history = self.operation_history[-self.config["max_operation_history"]:]
|
|
|
|
def cancel_operation(self, operation_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Bricht geplante Operation ab.
|
|
|
|
Args:
|
|
operation_id: ID der abzubrechenden Operation
|
|
|
|
Returns:
|
|
Dict mit Ergebnis
|
|
"""
|
|
with self.lock:
|
|
if operation_id not in self.pending_operations:
|
|
return {"success": False, "error": "Operation nicht gefunden"}
|
|
|
|
operation_data = self.pending_operations[operation_id]
|
|
if operation_data["status"] == "executing":
|
|
return {"success": False, "error": "Operation bereits in Ausführung"}
|
|
|
|
operation_data["status"] = "cancelled"
|
|
operation_data["cancelled_at"] = datetime.now()
|
|
self._move_to_history(operation_id)
|
|
|
|
system_logger.info(f"❌ Operation abgebrochen: {operation_id}")
|
|
|
|
return {"success": True, "message": "Operation erfolgreich abgebrochen"}
|
|
|
|
def get_pending_operations(self) -> List[Dict]:
|
|
"""Gibt alle geplanten Operationen zurück."""
|
|
with self.lock:
|
|
return list(self.pending_operations.values())
|
|
|
|
def get_operation_history(self, limit: int = 20) -> List[Dict]:
|
|
"""Gibt Operation-Historie zurück."""
|
|
with self.lock:
|
|
return self.operation_history[-limit:] if limit else self.operation_history
|
|
|
|
def get_system_status(self) -> Dict[str, Any]:
|
|
"""Gibt aktuellen System-Status zurück."""
|
|
try:
|
|
# Service-Status prüfen
|
|
service_status = {}
|
|
for name, service in self.services.items():
|
|
try:
|
|
result = subprocess.run(
|
|
["sudo", "systemctl", "is-active", service],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
service_status[name] = result.stdout.strip()
|
|
except Exception as e:
|
|
service_status[name] = f"error: {e}"
|
|
|
|
# System-Metriken
|
|
memory = psutil.virtual_memory()
|
|
disk = psutil.disk_usage('/')
|
|
|
|
# Aktive Operations
|
|
pending_ops = len(self.pending_operations)
|
|
|
|
return {
|
|
"success": True,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"services": service_status,
|
|
"system_metrics": {
|
|
"memory_percent": memory.percent,
|
|
"memory_available_gb": memory.available / (1024**3),
|
|
"disk_percent": disk.percent,
|
|
"disk_free_gb": disk.free / (1024**3),
|
|
"load_average": psutil.getloadavg()[0] if hasattr(psutil, 'getloadavg') else 0
|
|
},
|
|
"operations": {
|
|
"pending": pending_ops,
|
|
"history_count": len(self.operation_history)
|
|
},
|
|
"is_safe": self.is_safe_to_operate()[0]
|
|
}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
# Globaler System-Control-Manager
|
|
_system_control_manager: Optional[SystemControlManager] = None
|
|
_control_lock = threading.Lock()
|
|
|
|
|
|
def get_system_control_manager() -> SystemControlManager:
|
|
"""
|
|
Singleton-Pattern für globalen System-Control-Manager.
|
|
|
|
Returns:
|
|
SystemControlManager: Globaler System-Control-Manager
|
|
"""
|
|
global _system_control_manager
|
|
|
|
with _control_lock:
|
|
if _system_control_manager is None:
|
|
_system_control_manager = SystemControlManager()
|
|
return _system_control_manager
|
|
|
|
|
|
# Convenience-Funktionen
|
|
def schedule_system_restart(delay_seconds: int = 60, user_id: str = None, reason: str = None, force: bool = False) -> Dict[str, Any]:
|
|
"""Plant System-Neustart."""
|
|
manager = get_system_control_manager()
|
|
return manager.schedule_operation(SystemOperation.RESTART, delay_seconds, user_id, reason, force)
|
|
|
|
|
|
def schedule_system_shutdown(delay_seconds: int = 30, user_id: str = None, reason: str = None, force: bool = False) -> Dict[str, Any]:
|
|
"""Plant System-Shutdown."""
|
|
manager = get_system_control_manager()
|
|
return manager.schedule_operation(SystemOperation.SHUTDOWN, delay_seconds, user_id, reason, force)
|
|
|
|
|
|
def restart_kiosk(delay_seconds: int = 10, user_id: str = None, reason: str = None) -> Dict[str, Any]:
|
|
"""Plant Kiosk-Neustart."""
|
|
manager = get_system_control_manager()
|
|
return manager.schedule_operation(SystemOperation.KIOSK_RESTART, delay_seconds, user_id, reason)
|
|
|
|
|
|
def get_system_status() -> Dict[str, Any]:
|
|
"""Gibt System-Status zurück."""
|
|
manager = get_system_control_manager()
|
|
return manager.get_system_status() |