"""Service de surveillance des ressources système du Raspberry Pi.""" from __future__ import annotations import logging import subprocess from pathlib import Path try: import psutil HAS_PSUTIL = True except ImportError: HAS_PSUTIL = False logging.warning("psutil non disponible — stats système limitées") logger = logging.getLogger(__name__) class SystemService: def get_stats(self) -> dict: stats: dict = {} if not HAS_PSUTIL: return {"error": "psutil non installé"} # CPU stats["cpu_percent"] = psutil.cpu_percent(interval=None) # RAM mem = psutil.virtual_memory() stats["ram_total_mb"] = round(mem.total / 1024 / 1024) stats["ram_used_mb"] = round(mem.used / 1024 / 1024) stats["ram_percent"] = mem.percent stats["ram_available_mb"] = round(mem.available / 1024 / 1024) # Disque (racine) disk = psutil.disk_usage("/") stats["disk_total_gb"] = round(disk.total / 1024 ** 3, 1) stats["disk_used_gb"] = round(disk.used / 1024 ** 3, 1) stats["disk_percent"] = disk.percent # Température CPU (Raspberry Pi) stats["cpu_temp"] = self._get_cpu_temp() # Uptime import time boot_time = psutil.boot_time() uptime_sec = int(time.time() - boot_time) stats["uptime"] = self._format_uptime(uptime_sec) return stats def _get_cpu_temp(self) -> float | None: # Méthode 1 : fichier thermal du Pi try: temp_path = Path("/sys/class/thermal/thermal_zone0/temp") if temp_path.exists(): return round(int(temp_path.read_text().strip()) / 1000, 1) except Exception: pass # Méthode 2 : vcgencmd (Pi OS) try: result = subprocess.run( ["vcgencmd", "measure_temp"], capture_output=True, text=True, timeout=3 ) if result.returncode == 0: # output: "temp=47.0'C" val = result.stdout.strip().replace("temp=", "").replace("'C", "") return float(val) except Exception: pass # Méthode 3 : psutil sensors (si disponible) if hasattr(psutil, "sensors_temperatures"): try: temps = psutil.sensors_temperatures() if temps: first = next(iter(temps.values())) if first: return round(first[0].current, 1) except Exception: pass return None def _format_uptime(self, seconds: int) -> str: days = seconds // 86400 hours = (seconds % 86400) // 3600 minutes = (seconds % 3600) // 60 if days > 0: return f"{days}j {hours:02d}h {minutes:02d}m" return f"{hours:02d}h {minutes:02d}m" def get_services_status(self) -> list[dict]: """Vérifie le statut des services systemd utiles.""" services = [ ("photobooth-app", "Photobooth App"), ("jh-photomaton", "JH Photomaton"), ("cups", "CUPS (Impression)"), ] result = [] for svc_name, label in services: active = self._is_service_active(svc_name) result.append({"name": svc_name, "label": label, "active": active}) return result def _is_service_active(self, service: str) -> bool: try: r = subprocess.run( ["systemctl", "is-active", service], capture_output=True, text=True, timeout=3 ) return r.stdout.strip() == "active" except Exception: return False