373 lines
11 KiB
Python
373 lines
11 KiB
Python
"""
|
|
Esempi di utilizzo del modulo ResourceMonitor.
|
|
|
|
Questo file contiene diversi esempi pratici di come integrare il ResourceMonitor
|
|
nelle tue applicazioni Python, sia con che senza GUI Tkinter.
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# Aggiungi il path del progetto per poter importare il modulo
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent))
|
|
|
|
from target_simulator.utils.resource_monitor import (
|
|
ResourceMonitor,
|
|
TkinterResourceMonitor,
|
|
is_psutil_available
|
|
)
|
|
|
|
|
|
def esempio_1_console_semplice():
|
|
"""
|
|
Esempio 1: Monitoring console con stampa su terminale.
|
|
|
|
Questo è l'esempio più semplice: stampa le statistiche sul terminale
|
|
ogni secondo.
|
|
"""
|
|
print("\n" + "="*70)
|
|
print("ESEMPIO 1: Monitoring console semplice")
|
|
print("="*70)
|
|
|
|
if not is_psutil_available():
|
|
print("ERRORE: psutil non disponibile. Installa con: pip install psutil")
|
|
return
|
|
|
|
def print_to_console(stats: str):
|
|
print(f"\r{stats}", end="", flush=True)
|
|
|
|
monitor = ResourceMonitor(
|
|
update_callback=print_to_console,
|
|
poll_interval=1.0
|
|
)
|
|
|
|
if monitor.start():
|
|
print("Monitoraggio avviato per 5 secondi...\n")
|
|
try:
|
|
time.sleep(5)
|
|
except KeyboardInterrupt:
|
|
print("\nInterrotto")
|
|
finally:
|
|
monitor.stop()
|
|
print("\n\nMonitoring fermato.\n")
|
|
else:
|
|
print("Impossibile avviare il monitor")
|
|
|
|
|
|
def esempio_2_lettura_sincrona():
|
|
"""
|
|
Esempio 2: Lettura sincrona delle statistiche.
|
|
|
|
A volte non serve un monitoring continuo, ma solo una lettura "on-demand"
|
|
delle risorse correnti. Questo esempio mostra come fare.
|
|
"""
|
|
print("\n" + "="*70)
|
|
print("ESEMPIO 2: Lettura sincrona (senza thread)")
|
|
print("="*70)
|
|
|
|
if not is_psutil_available():
|
|
print("ERRORE: psutil non disponibile.")
|
|
return
|
|
|
|
# Non serve nemmeno un callback se usiamo solo get_current_stats()
|
|
monitor = ResourceMonitor(
|
|
update_callback=lambda s: None, # callback dummy
|
|
poll_interval=1.0
|
|
)
|
|
|
|
print("\nLettura singola delle statistiche:\n")
|
|
stats = monitor.get_current_stats()
|
|
|
|
if stats:
|
|
print(f" CPU: {stats['cpu_percent']:.1f}%")
|
|
print(f" Memoria: {stats['memory_mb']:.1f} MB ({stats['memory_percent']:.1f}%)")
|
|
print(f" Tipo mem: {stats['memory_type']}")
|
|
print(f" Thread: {stats['thread_count']}")
|
|
else:
|
|
print(" Impossibile leggere le statistiche")
|
|
|
|
print()
|
|
|
|
|
|
def esempio_3_callback_personalizzato():
|
|
"""
|
|
Esempio 3: Callback personalizzato con logging.
|
|
|
|
Questo esempio mostra come usare un callback personalizzato per fare
|
|
qualcosa di più complesso con i dati, come logging su file o invio
|
|
a un sistema di monitoring esterno.
|
|
"""
|
|
print("\n" + "="*70)
|
|
print("ESEMPIO 3: Callback personalizzato con analisi")
|
|
print("="*70)
|
|
|
|
if not is_psutil_available():
|
|
print("ERRORE: psutil non disponibile.")
|
|
return
|
|
|
|
# Lista per raccogliere le misure
|
|
samples = []
|
|
max_samples = 5
|
|
|
|
def analyze_and_log(stats_str: str):
|
|
"""Callback che analizza e logga le statistiche."""
|
|
# Stampa
|
|
print(f"\r{stats_str}", end="", flush=True)
|
|
|
|
# Potresti anche parsare la stringa o usare get_current_stats()
|
|
# per fare analisi più avanzate
|
|
samples.append(stats_str)
|
|
|
|
# Mantieni solo le ultime N misure
|
|
if len(samples) > max_samples:
|
|
samples.pop(0)
|
|
|
|
monitor = ResourceMonitor(
|
|
update_callback=analyze_and_log,
|
|
poll_interval=0.5
|
|
)
|
|
|
|
if monitor.start():
|
|
print("Monitoring per 3 secondi con callback personalizzato...\n")
|
|
try:
|
|
time.sleep(3)
|
|
except KeyboardInterrupt:
|
|
print("\nInterrotto")
|
|
finally:
|
|
monitor.stop()
|
|
print(f"\n\nRaccolte {len(samples)} misure.")
|
|
print("Ultime 3 misure:")
|
|
for i, s in enumerate(samples[-3:], 1):
|
|
print(f" {i}. {s}")
|
|
print()
|
|
|
|
|
|
def esempio_4_tkinter_gui():
|
|
"""
|
|
Esempio 4: Integrazione con GUI Tkinter.
|
|
|
|
Questo esempio mostra come integrare il monitor in una GUI Tkinter
|
|
usando la classe TkinterResourceMonitor che gestisce automaticamente
|
|
l'aggiornamento thread-safe della GUI.
|
|
"""
|
|
print("\n" + "="*70)
|
|
print("ESEMPIO 4: Integrazione Tkinter GUI")
|
|
print("="*70)
|
|
|
|
if not is_psutil_available():
|
|
print("ERRORE: psutil non disponibile.")
|
|
return
|
|
|
|
try:
|
|
import tkinter as tk
|
|
from tkinter import ttk
|
|
except ImportError:
|
|
print("ERRORE: Tkinter non disponibile in questa installazione Python")
|
|
return
|
|
|
|
# Crea finestra principale
|
|
root = tk.Tk()
|
|
root.title("Resource Monitor - Esempio Tkinter")
|
|
root.geometry("600x250")
|
|
|
|
# Frame principale
|
|
main_frame = ttk.Frame(root, padding=20)
|
|
main_frame.pack(fill=tk.BOTH, expand=True)
|
|
|
|
# Titolo
|
|
title = ttk.Label(
|
|
main_frame,
|
|
text="Monitor Risorse di Sistema",
|
|
font=("Arial", 16, "bold")
|
|
)
|
|
title.pack(pady=(0, 20))
|
|
|
|
# Label per le statistiche
|
|
stats_var = tk.StringVar(value="Avvio monitor...")
|
|
stats_label = ttk.Label(
|
|
main_frame,
|
|
textvariable=stats_var,
|
|
font=("Courier", 12),
|
|
relief=tk.SUNKEN,
|
|
padding=10
|
|
)
|
|
stats_label.pack(fill=tk.X, pady=10)
|
|
|
|
# Frame per i pulsanti
|
|
button_frame = ttk.Frame(main_frame)
|
|
button_frame.pack(pady=20)
|
|
|
|
# Variabile per tenere traccia dello stato
|
|
monitor_state = {"monitor": None, "is_running": False}
|
|
|
|
def start_monitor():
|
|
"""Avvia il monitoring."""
|
|
if not monitor_state["is_running"]:
|
|
monitor = TkinterResourceMonitor(
|
|
tk_widget=root,
|
|
string_var=stats_var,
|
|
poll_interval=0.5 # Update veloce per demo
|
|
)
|
|
if monitor.start():
|
|
monitor_state["monitor"] = monitor
|
|
monitor_state["is_running"] = True
|
|
start_btn.config(state="disabled")
|
|
stop_btn.config(state="normal")
|
|
status_label.config(text="Stato: ATTIVO", foreground="green")
|
|
|
|
def stop_monitor():
|
|
"""Ferma il monitoring."""
|
|
if monitor_state["is_running"] and monitor_state["monitor"]:
|
|
monitor_state["monitor"].stop()
|
|
monitor_state["is_running"] = False
|
|
start_btn.config(state="normal")
|
|
stop_btn.config(state="disabled")
|
|
stats_var.set("Monitor fermato")
|
|
status_label.config(text="Stato: FERMO", foreground="red")
|
|
|
|
def on_closing():
|
|
"""Handler per chiusura finestra."""
|
|
stop_monitor()
|
|
root.destroy()
|
|
|
|
# Pulsanti
|
|
start_btn = ttk.Button(button_frame, text="Avvia Monitor", command=start_monitor)
|
|
start_btn.pack(side=tk.LEFT, padx=5)
|
|
|
|
stop_btn = ttk.Button(button_frame, text="Ferma Monitor", command=stop_monitor)
|
|
stop_btn.pack(side=tk.LEFT, padx=5)
|
|
stop_btn.config(state="disabled")
|
|
|
|
# Label stato
|
|
status_label = ttk.Label(
|
|
main_frame,
|
|
text="Stato: FERMO",
|
|
font=("Arial", 10),
|
|
foreground="red"
|
|
)
|
|
status_label.pack(pady=10)
|
|
|
|
# Info
|
|
info_text = (
|
|
"Questo esempio mostra l'integrazione del ResourceMonitor in una GUI Tkinter.\n"
|
|
"Il monitor gira in un thread separato e aggiorna la GUI in modo thread-safe."
|
|
)
|
|
info_label = ttk.Label(
|
|
main_frame,
|
|
text=info_text,
|
|
font=("Arial", 9),
|
|
foreground="gray",
|
|
wraplength=550,
|
|
justify=tk.CENTER
|
|
)
|
|
info_label.pack(pady=(20, 0))
|
|
|
|
# Avvia automaticamente il monitor
|
|
root.after(500, start_monitor)
|
|
|
|
# Gestisci chiusura
|
|
root.protocol("WM_DELETE_WINDOW", on_closing)
|
|
|
|
print("Finestra GUI aperta. Chiudi la finestra per continuare.\n")
|
|
root.mainloop()
|
|
|
|
|
|
def esempio_5_monitoring_altro_processo():
|
|
"""
|
|
Esempio 5: Monitoring di un altro processo (per PID).
|
|
|
|
Il ResourceMonitor può anche monitorare un processo diverso da quello
|
|
corrente, specificando il PID.
|
|
"""
|
|
print("\n" + "="*70)
|
|
print("ESEMPIO 5: Monitoring di un altro processo")
|
|
print("="*70)
|
|
|
|
if not is_psutil_available():
|
|
print("ERRORE: psutil non disponibile.")
|
|
return
|
|
|
|
import psutil
|
|
|
|
# Ottieni il PID del processo corrente per l'esempio
|
|
current_pid = psutil.Process().pid
|
|
|
|
print(f"\nMonitoring del processo corrente (PID {current_pid})...")
|
|
print("In un'applicazione reale potresti specificare il PID di un altro processo.\n")
|
|
|
|
def print_stats(stats: str):
|
|
print(f"\rPID {current_pid}: {stats}", end="", flush=True)
|
|
|
|
monitor = ResourceMonitor(
|
|
update_callback=print_stats,
|
|
poll_interval=1.0,
|
|
process_pid=current_pid # Specifica esplicitamente il PID
|
|
)
|
|
|
|
if monitor.start():
|
|
try:
|
|
time.sleep(3)
|
|
except KeyboardInterrupt:
|
|
print("\nInterrotto")
|
|
finally:
|
|
monitor.stop()
|
|
print("\n\nMonitoring fermato.\n")
|
|
|
|
|
|
def main():
|
|
"""Menu principale per scegliere l'esempio da eseguire."""
|
|
print("\n" + "="*70)
|
|
print("ESEMPI DI UTILIZZO DEL RESOURCE MONITOR")
|
|
print("="*70)
|
|
|
|
if not is_psutil_available():
|
|
print("\n⚠️ ATTENZIONE: psutil non è installato!")
|
|
print(" Installa con: pip install psutil")
|
|
print(" Alcuni esempi non funzioneranno senza psutil.\n")
|
|
|
|
esempi = [
|
|
("Monitoring console semplice", esempio_1_console_semplice),
|
|
("Lettura sincrona (senza thread)", esempio_2_lettura_sincrona),
|
|
("Callback personalizzato", esempio_3_callback_personalizzato),
|
|
("Integrazione Tkinter GUI", esempio_4_tkinter_gui),
|
|
("Monitoring altro processo (PID)", esempio_5_monitoring_altro_processo),
|
|
]
|
|
|
|
print("\nScegli un esempio da eseguire:")
|
|
for i, (nome, _) in enumerate(esempi, 1):
|
|
print(f" {i}. {nome}")
|
|
print(f" {len(esempi) + 1}. Esegui tutti gli esempi")
|
|
print(" 0. Esci")
|
|
|
|
try:
|
|
scelta = input("\nScelta: ").strip()
|
|
|
|
if scelta == "0":
|
|
print("Uscita.")
|
|
return
|
|
|
|
scelta_num = int(scelta)
|
|
|
|
if scelta_num == len(esempi) + 1:
|
|
# Esegui tutti
|
|
for nome, func in esempi:
|
|
func()
|
|
time.sleep(1)
|
|
elif 1 <= scelta_num <= len(esempi):
|
|
# Esegui singolo esempio
|
|
esempi[scelta_num - 1][1]()
|
|
else:
|
|
print("Scelta non valida.")
|
|
|
|
except (ValueError, KeyboardInterrupt):
|
|
print("\nUscita.")
|
|
|
|
print("\n" + "="*70)
|
|
print("Fine esempi.")
|
|
print("="*70 + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|