VALLONGOL/S1005403_RisCC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

aggiunto misuratore di risorse nella status bar, sistemata dimensione status bar se si cambia la dimensione della schermata

Browse Source
This commit is contained in:
VALLONGOL 2025-11-03 15:11:32 +01:00
parent 76178e9888
commit d1315c714a
4 changed files with 161 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
logger_prefs.json
View File

@ -2,7 +2,8 @@
"saved_levels": {
"target_simulator.gui.payload_router": "INFO",
"target_simulator.analysis.simulation_state_hub": "INFO",
"target_simulator.gui.main_view": "INFO"
"target_simulator.gui.main_view": "INFO",
"target_simulator.core.sfp_transport": "INFO"
},
"last_selected": "target_simulator.gui.main_view"
"last_selected": "target_simulator.core.sfp_transport"
}

4
settings.json
View File

@ -2,8 +2,8 @@
"general": {
"scan_limit": 60,
"max_range": 100,
"geometry": "1599x1075+587+179",
"last_selected_scenario": "scenario_dritto",
"geometry": "1305x929+587+179",
"last_selected_scenario": "scenario2",
"connection": {
"target": {
"type": "sfp",

23
target_simulator/gui/main_view.py
View File

@ -603,7 +603,17 @@ class MainView(tk.Tk):
# Use the extracted StatusBar widget. Expose the same attributes
# MainView previously provided so callers elsewhere continue to work.
self.status_bar = StatusBar(self)
self.status_bar.pack(side=tk.BOTTOM, fill=tk.X)
# Try placing the status bar pinned to the bottom with a fixed height
# so it remains visible even when other widgets request large minimum
# sizes. Fall back to pack if place is not supported on some platforms
# or if an error occurs.
try:
# The StatusBar already configures its height (24px by default).
# Use place anchored to south-west so the bar stays at the bottom
# and spans the full width of the window.
self.status_bar.place(relx=0.0, rely=1.0, anchor="sw", relwidth=1.0, height=24)
except Exception:
self.status_bar.pack(side=tk.BOTTOM, fill=tk.X)
self.target_status_canvas = self.status_bar.target_status_canvas
self.lru_status_canvas = self.status_bar.lru_status_canvas
@ -1525,6 +1535,17 @@ class MainView(tk.Tk):
if self.lru_communicator and self.lru_communicator.is_open:
self.lru_communicator.disconnect()
# Stop resource monitor thread if present
try:
if hasattr(self, "status_bar") and self.status_bar:
try:
if hasattr(self.status_bar, "stop_resource_monitor"):
self.status_bar.stop_resource_monitor()
except Exception:
pass
except Exception:
pass
shutdown_logging_system()
self.destroy()

232
target_simulator/gui/status_bar.py
View File

@ -1,159 +1,195 @@
import os
import threading
import tkinter as tk
from tkinter import ttk
from typing import Optional
# Optional dependency: psutil provides portable CPU/memory info across OSes.
try:
import psutil # type: ignore
_HAS_PSUTIL = True
except Exception:
psutil = None # type: ignore
_HAS_PSUTIL = False
class StatusBar(ttk.Frame):
"""Status bar widget containing connection indicators and rate/status text.
"""Status bar widget containing connection indicators, status text,
small rate indicator and optional resource monitor.
Exposes small API so MainView can delegate status updates without managing
layout details.
Public attributes used elsewhere in the app:
- target_status_canvas
- lru_status_canvas
- status_var (tk.StringVar)
- rate_status_var (tk.StringVar) or None
- resource_var (tk.StringVar) or None
The resource monitor runs in a daemon thread and updates the UI via
`after(0, ...)` so updates are executed on the Tk mainloop.
"""
def __init__(self, parent):
def __init__(self, parent, resource_poll_s: float = 1.0, height: int = 24):
super().__init__(parent, relief=tk.SUNKEN)
ttk.Label(self, text="Target:").pack(side=tk.LEFT, padx=(5, 2))
# Keep the status bar a fixed small height so it remains visible on
# vertically-constrained windows. Prevent children from forcing the
# frame's size.
try:
self.configure(height=int(height))
self.pack_propagate(False)
except Exception:
pass
# Left area: connection indicators
ttk.Label(self, text="Target:").pack(side=tk.LEFT, padx=(6, 2))
self.target_status_canvas = tk.Canvas(self, width=16, height=16, highlightthickness=0)
self.target_status_canvas.pack(side=tk.LEFT, padx=(0, 10))
self.target_status_canvas.pack(side=tk.LEFT, padx=(0, 8))
self._draw_status_indicator(self.target_status_canvas, "#e74c3c")
ttk.Label(self, text="LRU:").pack(side=tk.LEFT, padx=(5, 2))
ttk.Label(self, text="LRU:").pack(side=tk.LEFT, padx=(6, 2))
self.lru_status_canvas = tk.Canvas(self, width=16, height=16, highlightthickness=0)
self.lru_status_canvas.pack(side=tk.LEFT, padx=(0, 10))
self.lru_status_canvas.pack(side=tk.LEFT, padx=(0, 8))
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c")
# Center: main status message
self.status_var = tk.StringVar(value="Ready")
ttk.Label(self, textvariable=self.status_var, anchor=tk.W).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
ttk.Label(self, textvariable=self.status_var, anchor=tk.W).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=6)
# Small rate indicator showing incoming real-state rate and PPI update rate
# Right: rate and resource indicators
try:
self.rate_status_var = tk.StringVar(value="")
ttk.Label(self, textvariable=self.rate_status_var, anchor=tk.E).pack(side=tk.RIGHT, padx=(4, 8))
ttk.Label(self, textvariable=self.rate_status_var, anchor=tk.E).pack(side=tk.RIGHT, padx=(6, 8))
except Exception:
self.rate_status_var = None
# Id for scheduled status clear; used by show_status_message
self._status_after_id: Optional[str] = None
# Resource usage (optional). We create the var even if psutil missing so
# callers can safely call getattr(..., 'resource_var', None).
try:
self.resource_var = tk.StringVar(value="")
ttk.Label(self, textvariable=self.resource_var, anchor=tk.E).pack(side=tk.RIGHT, padx=(6, 8))
except Exception:
self.resource_var = None
def _draw_status_indicator(self, canvas, color):
# Internal state
self._status_after_id: Optional[str] = None
self._res_stop_event = threading.Event()
self._res_thread: Optional[threading.Thread] = None
self._resource_poll_s = float(resource_poll_s)
# Start background monitor if psutil is available and we have a var
if _HAS_PSUTIL and self.resource_var is not None:
try:
self.start_resource_monitor(self._resource_poll_s)
except Exception:
# Don't fail construction if monitor can't start
pass
def _draw_status_indicator(self, canvas: tk.Canvas, color: str) -> None:
try:
canvas.delete("all")
canvas.create_oval(2, 2, 14, 14, fill=color, outline="black")
except Exception:
pass
def set_target_connected(self, is_connected: bool):
def set_target_connected(self, is_connected: bool) -> None:
color = "#2ecc40" if is_connected else "#e74c3c"
try:
self._draw_status_indicator(self.target_status_canvas, color)
except Exception:
pass
def set_lru_connected(self, is_connected: bool):
def set_lru_connected(self, is_connected: bool) -> None:
color = "#2ecc40" if is_connected else "#e74c3c"
try:
self._draw_status_indicator(self.lru_status_canvas, color)
except Exception:
pass
def show_status_message(self, text: str, timeout_ms: int = 3000):
def show_status_message(self, text: str, timeout_ms: int = 3000) -> None:
"""Show a transient status message in the main status bar.
If timeout_ms is 0 the message is sticky until changed again. This
method is thread-safe when called from the Tk mainloop; if you call it
from a worker thread, schedule it via `after(0, lambda: ... )`.
"""
try:
try:
if self._status_after_id is not None:
self.after_cancel(self._status_after_id)
try:
self.after_cancel(self._status_after_id)
except Exception:
pass
except Exception:
pass
self.status_var.set(text)
def _clear():
try:
self.status_var.set("Ready")
except Exception:
pass
self._status_after_id = self.after(timeout_ms, _clear)
except Exception:
# Fallback: set the var and do not schedule clear
try:
self.status_var.set(text)
except Exception:
pass
# Schedule clear back to Ready
if timeout_ms and timeout_ms > 0:
def _clear():
try:
self.status_var.set("Ready")
except Exception:
pass
class StatusBar(ttk.Frame):
"""Status bar widget containing connection indicators and rate/status text.
Exposes small API so MainView can delegate status updates without managing
layout details.
"""
def __init__(self, parent):
super().__init__(parent, relief=tk.SUNKEN)
ttk.Label(self, text="Target:").pack(side=tk.LEFT, padx=(5, 2))
self.target_status_canvas = tk.Canvas(self, width=16, height=16, highlightthickness=0)
self.target_status_canvas.pack(side=tk.LEFT, padx=(0, 10))
self._draw_status_indicator(self.target_status_canvas, "#e74c3c")
ttk.Label(self, text="LRU:").pack(side=tk.LEFT, padx=(5, 2))
self.lru_status_canvas = tk.Canvas(self, width=16, height=16, highlightthickness=0)
self.lru_status_canvas.pack(side=tk.LEFT, padx=(0, 10))
self._draw_status_indicator(self.lru_status_canvas, "#e74c3c")
self.status_var = tk.StringVar(value="Ready")
ttk.Label(self, textvariable=self.status_var, anchor=tk.W).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
# Small rate indicator showing incoming real-state rate and PPI update rate
try:
self.rate_status_var = tk.StringVar(value="")
ttk.Label(self, textvariable=self.rate_status_var, anchor=tk.E).pack(side=tk.RIGHT, padx=(4, 8))
except Exception:
self.rate_status_var = None
# Id for scheduled status clear; used by show_status_message
self._status_after_id: Optional[str] = None
def _draw_status_indicator(self, canvas, color):
try:
canvas.delete("all")
canvas.create_oval(2, 2, 14, 14, fill=color, outline="black")
except Exception:
pass
def set_target_connected(self, is_connected: bool):
color = "#2ecc40" if is_connected else "#e74c3c"
try:
self._draw_status_indicator(self.target_status_canvas, color)
except Exception:
pass
def set_lru_connected(self, is_connected: bool):
color = "#2ecc40" if is_connected else "#e74c3c"
try:
self._draw_status_indicator(self.lru_status_canvas, color)
except Exception:
pass
def show_status_message(self, text: str, timeout_ms: int = 3000):
try:
try:
if self._status_after_id is not None:
self.after_cancel(self._status_after_id)
except Exception:
pass
self.status_var.set(text)
def _clear():
try:
self.status_var.set("Ready")
self._status_after_id = self.after(timeout_ms, _clear)
except Exception:
self._status_after_id = None
except Exception:
# Swallow errors; status updates shouldn't crash the UI
pass
# ---- Resource monitor ----
def start_resource_monitor(self, poll_s: float = 1.0) -> None:
"""Start the background resource monitor (daemon thread)."""
if not _HAS_PSUTIL or self.resource_var is None:
return
# If already running, ignore
if self._res_thread and self._res_thread.is_alive():
return
self._res_stop_event.clear()
self._resource_poll_s = float(poll_s)
def _monitor_loop():
try:
proc = psutil.Process(os.getpid())
# Prime cpu_percent the first time
try:
proc.cpu_percent(None)
psutil.cpu_percent(None)
except Exception:
pass
self._status_after_id = self.after(timeout_ms, _clear)
except Exception:
# Fallback: set the var and do not schedule clear
try:
self.status_var.set(text)
while not self._res_stop_event.wait(self._resource_poll_s):
try:
cpu = psutil.cpu_percent(None)
rss = proc.memory_info().rss
rss_mb = rss / (1024.0 * 1024.0)
mem_pct = proc.memory_percent()
nthreads = proc.num_threads()
s = f"CPU {cpu:.0f}% · MEM {rss_mb:.0f}MB ({mem_pct:.1f}%) · Thr {nthreads}"
except Exception:
s = ""
try:
# Schedule UI update on main thread
self.after(0, lambda val=s: self.resource_var.set(val))
except Exception:
pass
except Exception:
# If psutil unexpectedly fails, stop quietly
pass
self._res_thread = threading.Thread(target=_monitor_loop, daemon=True)
self._res_thread.start()
def stop_resource_monitor(self) -> None:
try:
self._res_stop_event.set()
except Exception:
pass