import psutil import json import subprocess from typing import List, Optional, Dict, Any from pydantic import BaseModel class CPUUsage(BaseModel): percent: float average: float cores: int per_cpu: List[float] freq: float # GHz class MemoryUsage(BaseModel): percent: float used: int total: int available: int cached: int class DiskUsage(BaseModel): device: str total: int used: int free: int percent: float class NetworkUsage(BaseModel): bytes_sent: int bytes_recv: int packets_sent: int packets_recv: int class ProcessInfo(BaseModel): pid: int name: str status: str cpu_percent: float memory_percent: float memory_mb: float username: str class BlockDevicePartition(BaseModel): """Représente une partition d'un disque""" name: str type: str size: str used: str available: str percent_used: float mountpoint: Optional[str] = None class BlockDevice(BaseModel): """Représente un disque ou un périphérique bloc""" name: str type: str size: str used: str available: str percent_used: float mountpoint: Optional[str] = None partitions: List[BlockDevicePartition] = [] class BlockDevicesInfo(BaseModel): """Informations sur tous les disques et partitions""" devices: List[BlockDevice] total_size: str total_used: str total_available: str class SystemStats(BaseModel): cpu: CPUUsage memory: MemoryUsage disk: List[DiskUsage] network: NetworkUsage processes: List[ProcessInfo] class SystemService: """Service pour récupérer les informations système""" @staticmethod def get_cpu_usage() -> CPUUsage: """Récupère l'utilisation CPU""" per_cpu = psutil.cpu_percent(interval=0, percpu=True) avg_cpu = sum(per_cpu) / len(per_cpu) if per_cpu else 0 try: freq = psutil.cpu_freq().current / 1000 # GHz except: freq = 0 return CPUUsage( percent=avg_cpu, average=avg_cpu, cores=psutil.cpu_count(), per_cpu=per_cpu, freq=freq ) @staticmethod def get_memory_usage() -> MemoryUsage: """Récupère l'utilisation mémoire""" mem = psutil.virtual_memory() return MemoryUsage( percent=mem.percent, used=mem.used, total=mem.total, available=mem.available, cached=mem.cached or 0 ) @staticmethod def get_disk_usage() -> List[DiskUsage]: """Récupère l'utilisation disque""" disks = [] try: for partition in psutil.disk_partitions(all=False): try: usage = psutil.disk_usage(partition.mountpoint) disks.append(DiskUsage( device=partition.device, total=usage.total, used=usage.used, free=usage.free, percent=usage.percent )) except (PermissionError, OSError): continue except: pass return disks @staticmethod def get_network_usage() -> NetworkUsage: """Récupère les stats réseau""" net = psutil.net_io_counters() return NetworkUsage( bytes_sent=net.bytes_sent, bytes_recv=net.bytes_recv, packets_sent=net.packets_sent, packets_recv=net.packets_recv ) @staticmethod def get_top_processes(limit: int = 10) -> List[ProcessInfo]: """Récupère les processus les plus actifs""" processes = [] try: for proc in psutil.process_iter(['pid', 'name', 'status', 'username']): try: info = proc.info cpu_percent = proc.cpu_percent(interval=0) mem_percent = proc.memory_percent() mem_mb = proc.memory_info().rss / 1024 / 1024 # Ne garder que les processus avec activité if cpu_percent > 0.5 or mem_percent > 0.5: processes.append(ProcessInfo( pid=info['pid'], name=info['name'], status=info['status'], cpu_percent=cpu_percent, memory_percent=mem_percent, memory_mb=mem_mb, username=info['username'] or 'N/A' )) except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): continue # Trier par CPU + mémoire processes.sort(key=lambda x: (x.cpu_percent + x.memory_percent), reverse=True) return processes[:limit] except Exception: return [] @staticmethod def get_system_stats() -> SystemStats: """Récupère toutes les stats système""" return SystemStats( cpu=SystemService.get_cpu_usage(), memory=SystemService.get_memory_usage(), disk=SystemService.get_disk_usage(), network=SystemService.get_network_usage(), processes=SystemService.get_top_processes() ) @staticmethod def format_bytes(bytes_val: int) -> str: """Convertit les bytes en format lisible""" for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if bytes_val < 1024: return f"{bytes_val:.2f} {unit}" bytes_val /= 1024 return f"{bytes_val:.2f} PB" @staticmethod def get_block_devices() -> BlockDevicesInfo: """Récupère les disques et partitions avec lsblk""" try: # Exécuter lsblk avec sortie JSON result = subprocess.run( ['lsblk', '--json', '--bytes', '--output', 'NAME,TYPE,SIZE,FSTYPE,MOUNTPOINTS'], capture_output=True, text=True, timeout=10 ) if result.returncode != 0: raise Exception(f"lsblk failed: {result.stderr}") lsblk_data = json.loads(result.stdout) devices = [] total_size = 0 total_used = 0 total_available = 0 for block_device in lsblk_data.get('blockdevices', []): # Obtenir les informations d'utilisation pour ce disque device_name = f"/dev/{block_device['name']}" size = block_device.get('size', 0) used = 0 available = 0 percent_used = 0 mountpoint = None partitions = [] # Essayer d'obtenir les stats d'utilisation try: # Chercher le premier mountpoint if block_device.get('mountpoints'): mountpoint = block_device['mountpoints'][0] if mountpoint: disk_usage = psutil.disk_usage(mountpoint) used = disk_usage.used available = disk_usage.free percent_used = disk_usage.percent total_used += used total_available += available except: pass # Traiter les partitions if 'children' in block_device: for child in block_device['children']: child_device = f"/dev/{child['name']}" child_size = child.get('size', 0) child_used = 0 child_available = 0 child_percent = 0 child_mountpoint = None try: if child.get('mountpoints'): child_mountpoint = child['mountpoints'][0] if child_mountpoint: disk_usage = psutil.disk_usage(child_mountpoint) child_used = disk_usage.used child_available = disk_usage.free child_percent = disk_usage.percent except: pass partitions.append(BlockDevicePartition( name=child['name'], type=child.get('type', 'unknown'), size=SystemService.format_bytes(child_size), used=SystemService.format_bytes(child_used), available=SystemService.format_bytes(child_available), percent_used=child_percent, mountpoint=child_mountpoint )) total_size += size devices.append(BlockDevice( name=block_device['name'], type=block_device.get('type', 'unknown'), size=SystemService.format_bytes(size), used=SystemService.format_bytes(used), available=SystemService.format_bytes(available), percent_used=percent_used, mountpoint=mountpoint, partitions=partitions )) return BlockDevicesInfo( devices=devices, total_size=SystemService.format_bytes(total_size), total_used=SystemService.format_bytes(total_used), total_available=SystemService.format_bytes(total_available) ) except Exception as e: # Si lsblk échoue, retourner une liste vide return BlockDevicesInfo( devices=[], total_size="0 B", total_used="0 B", total_available="0 B" )