433 lines
15 KiB
Python
433 lines
15 KiB
Python
"""
|
|
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']}")
|