""" Resource Monitor - Modulo riutilizzabile per il monitoring di CPU, RAM e Thread. Questo modulo fornisce una classe standalone per monitorare le risorse di sistema (CPU, memoria RAM, numero di thread) di un processo Python. Può essere facilmente integrato in qualsiasi applicazione Python, con o senza GUI Tkinter. Caratteristiche: - Monitoring in background tramite thread daemon - Aggiornamenti periodici configurabili - Supporto per callback personalizzati - Integrazione opzionale con Tkinter StringVar - Gestione sicura degli errori - Normalizzazione CPU per sistemi multi-core - Preferenza per USS (Unique Set Size) come metrica di memoria Esempio di utilizzo con Tkinter: import tkinter as tk from resource_monitor import ResourceMonitor root = tk.Tk() status_var = tk.StringVar() monitor = ResourceMonitor( update_callback=lambda stats: status_var.set(stats), poll_interval=1.0 ) monitor.start() # ... alla chiusura dell'app monitor.stop() Esempio di utilizzo senza GUI (callback personalizzato): from resource_monitor import ResourceMonitor def print_stats(stats_string): print(f"Risorse: {stats_string}") monitor = ResourceMonitor( update_callback=print_stats, poll_interval=2.0 ) monitor.start() # ... quando non serve più monitor.stop() Dipendenze: - psutil (opzionale ma raccomandato): pip install psutil Se psutil non è disponibile, il monitor non farà nulla ma non genererà errori. Author: Estratto da Target Simulator License: Same as parent project """ import os import threading import time from typing import Optional, Callable, Dict, Any # Optional dependency: psutil provides portable CPU/memory info across OSes try: import psutil # type: ignore HAS_PSUTIL = True except ImportError: psutil = None # type: ignore HAS_PSUTIL = False class ResourceMonitor: """ Monitor delle risorse di sistema (CPU, RAM, Thread) in background. Questa classe crea un thread daemon che monitora periodicamente le risorse del processo corrente e chiama un callback con le statistiche formattate. Attributes: poll_interval (float): Intervallo in secondi tra le letture delle risorse. is_running (bool): True se il monitor è attualmente attivo. Args: update_callback (Callable[[str], None]): Funzione chiamata ad ogni aggiornamento. Riceve una stringa formattata con le statistiche (es: "CPU 5% · MEM 120MB (2.5%) [USS] · Thr 12") poll_interval (float): Intervallo in secondi tra gli aggiornamenti (default: 1.0) process_pid (Optional[int]): PID del processo da monitorare (default: processo corrente) """ def __init__( self, update_callback: Callable[[str], None], poll_interval: float = 1.0, process_pid: Optional[int] = None ): """ Inizializza il monitor delle risorse. Args: update_callback: Funzione che riceve le statistiche formattate come stringa poll_interval: Secondi tra ogni lettura (default: 1.0) process_pid: PID del processo da monitorare (None = processo corrente) """ self._callback = update_callback self._poll_interval = float(poll_interval) self._pid = process_pid if process_pid is not None else os.getpid() # Threading control self._stop_event = threading.Event() self._thread: Optional[threading.Thread] = None # State self._is_running = False @property def is_running(self) -> bool: """Ritorna True se il monitor è attualmente attivo.""" return self._is_running and self._thread is not None and self._thread.is_alive() @property def poll_interval(self) -> float: """Ritorna l'intervallo di polling corrente in secondi.""" return self._poll_interval def start(self) -> bool: """ Avvia il monitoring in background. Returns: bool: True se il monitor è stato avviato con successo, False altrimenti (es: psutil non disponibile, già in esecuzione, ecc.) """ if not HAS_PSUTIL: # psutil non disponibile - non possiamo fare monitoring return False if self.is_running: # Già in esecuzione return False self._stop_event.clear() self._is_running = True self._thread = threading.Thread( target=self._monitor_loop, daemon=True, name="ResourceMonitor" ) self._thread.start() return True def stop(self) -> None: """ Ferma il monitoring in background. Questo metodo è thread-safe e può essere chiamato più volte senza problemi. Il thread terminerà al prossimo ciclo di polling. """ self._is_running = False self._stop_event.set() def get_current_stats(self) -> Optional[Dict[str, Any]]: """ Ottiene le statistiche correnti in modo sincrono (senza thread). Utile per ottenere una singola lettura senza avviare il monitoring continuo. Returns: Dict con le chiavi: - cpu_percent: Percentuale CPU (0-100, normalizzata per multi-core) - memory_mb: Memoria in MiB - memory_percent: Percentuale di memoria del sistema - memory_type: "USS" o "RSS" - thread_count: Numero di thread Oppure None se psutil non è disponibile o si verifica un errore. """ if not HAS_PSUTIL: return None try: proc = psutil.Process(self._pid) # CPU (normalizzato per multi-core) cpu_proc = proc.cpu_percent(interval=0.1) ncpu = psutil.cpu_count(logical=True) or 1 cpu = cpu_proc / ncpu # Memoria (preferenza per USS) try: mem_full = proc.memory_full_info() uss = getattr(mem_full, "uss", None) except Exception: uss = None if uss is not None and uss > 0: mem_bytes = uss mem_type = "USS" else: mem_bytes = proc.memory_info().rss mem_type = "RSS" mem_mb = mem_bytes / (1024.0 * 1024.0) mem_pct = proc.memory_percent() # Thread nthreads = proc.num_threads() return { "cpu_percent": cpu, "memory_mb": mem_mb, "memory_percent": mem_pct, "memory_type": mem_type, "thread_count": nthreads } except Exception: return None def _format_stats(self, stats: Dict[str, Any]) -> str: """ Formatta le statistiche in una stringa leggibile. Args: stats: Dizionario con le statistiche (output di get_current_stats) Returns: Stringa formattata, es: "CPU 5% · MEM 120MB (2.5%) [USS] · Thr 12" """ try: return ( f"CPU {stats['cpu_percent']:.0f}% · " f"MEM {stats['memory_mb']:.0f}MB ({stats['memory_percent']:.1f}%) " f"[{stats['memory_type']}] · " f"Thr {stats['thread_count']}" ) except Exception: return "" def _monitor_loop(self) -> None: """ Loop principale del thread di monitoring. Questo metodo gira in un thread daemon e continua fino a quando _stop_event non viene settato. """ try: proc = psutil.Process(self._pid) # Prime cpu_percent per la prima lettura # (psutil ha bisogno di una lettura iniziale per calcolare la differenza) try: proc.cpu_percent(None) except Exception: pass while not self._stop_event.wait(self._poll_interval): try: # Misura CPU del processo # psutil.Process.cpu_percent può ritornare valori >100 su # sistemi multi-core perché riporta la percentuale di tempo CPU # su tutti i core. Per presentare un valore comparabile a quello # del Task Manager (scala 0-100), normalizziamo per il numero # di CPU logiche. cpu_proc = proc.cpu_percent(None) ncpu = psutil.cpu_count(logical=True) or 1 cpu = cpu_proc / ncpu # Preferisco USS (unique set size) quando disponibile perché # rappresenta la memoria unica del processo (più vicino a quello # che Task Manager riporta come "private working set"). try: mem_full = proc.memory_full_info() uss = getattr(mem_full, "uss", None) except Exception: uss = None if uss is not None and uss > 0: mem_bytes = uss mem_tag = "USS" else: # Fallback a RSS (working set) se USS non disponibile mem_bytes = proc.memory_info().rss mem_tag = "RSS" # Converti in MiB per display (1024^2) mem_mb = mem_bytes / (1024.0 * 1024.0) mem_pct = proc.memory_percent() nthreads = proc.num_threads() stats_dict = { "cpu_percent": cpu, "memory_mb": mem_mb, "memory_percent": mem_pct, "memory_type": mem_tag, "thread_count": nthreads } stats_str = self._format_stats(stats_dict) # Chiama il callback con le statistiche formattate if self._callback and stats_str: try: self._callback(stats_str) except Exception: # Se il callback fallisce, non fermiamo il monitor pass except Exception: # Se qualcosa va storto nella lettura, continua comunque pass except Exception: # Se psutil fallisce inaspettatamente, ferma silenziosamente pass finally: self._is_running = False class TkinterResourceMonitor(ResourceMonitor): """ Versione specializzata di ResourceMonitor per integrazione con Tkinter. Questa classe gestisce automaticamente l'aggiornamento di una StringVar di Tkinter in modo thread-safe usando il metodo `after(0, ...)` del widget. Esempio: import tkinter as tk from resource_monitor import TkinterResourceMonitor root = tk.Tk() status_var = tk.StringVar() tk.Label(root, textvariable=status_var).pack() monitor = TkinterResourceMonitor( tk_widget=root, string_var=status_var, poll_interval=1.0 ) monitor.start() root.mainloop() monitor.stop() """ def __init__( self, tk_widget, # Any Tkinter widget with .after() method string_var, # tk.StringVar poll_interval: float = 1.0, process_pid: Optional[int] = None ): """ Inizializza il monitor per Tkinter. Args: tk_widget: Qualsiasi widget Tkinter (es: root, Frame, ecc.) che ha il metodo .after() per scheduling thread-safe string_var: tk.StringVar da aggiornare con le statistiche poll_interval: Secondi tra ogni lettura (default: 1.0) process_pid: PID del processo da monitorare (None = processo corrente) """ self._tk_widget = tk_widget self._string_var = string_var # Callback che aggiorna la StringVar in modo thread-safe def _tk_update_callback(stats_string: str): try: # Schedule update sul main thread di Tkinter self._tk_widget.after( 0, lambda: self._string_var.set(stats_string) ) except Exception: # Ignora errori se il widget è stato distrutto pass super().__init__( update_callback=_tk_update_callback, poll_interval=poll_interval, process_pid=process_pid ) def is_psutil_available() -> bool: """ Verifica se psutil è disponibile nel sistema. Returns: bool: True se psutil è installato e importabile, False altrimenti """ return HAS_PSUTIL # Esempio di utilizzo standalone (eseguibile con: python -m target_simulator.utils.resource_monitor) if __name__ == "__main__": import sys if not HAS_PSUTIL: print("ERRORE: psutil non è installato.") print("Installalo con: pip install psutil") sys.exit(1) print("=== Test ResourceMonitor ===") print("Monitoraggio risorse per 10 secondi...") print("Premi Ctrl+C per interrompere\n") def print_stats(stats: str): print(f"\r{stats}", end="", flush=True) monitor = ResourceMonitor( update_callback=print_stats, poll_interval=0.5 # Update ogni 500ms per il test ) if not monitor.start(): print("ERRORE: Impossibile avviare il monitor") sys.exit(1) try: # Simula lavoro time.sleep(10) except KeyboardInterrupt: print("\n\nInterrotto dall'utente") finally: monitor.stop() print("\n\nMonitor fermato.") # Mostra una lettura sincrona finale stats = monitor.get_current_stats() if stats: print("\nStatistiche finali:") print(f" CPU: {stats['cpu_percent']:.1f}%") print(f" Memoria: {stats['memory_mb']:.1f} MB ({stats['memory_percent']:.1f}%)") print(f" Tipo memoria: {stats['memory_type']}") print(f" Thread: {stats['thread_count']}")