522 lines
19 KiB
Python
522 lines
19 KiB
Python
# target_simulator/gui/logger_panel.py
|
|
"""
|
|
A small Toplevel UI to inspect and change logger levels at runtime.
|
|
"""
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
import logging
|
|
from typing import List
|
|
from target_simulator.utils.config_manager import ConfigManager
|
|
import os
|
|
import json
|
|
|
|
LEVELS = [
|
|
("NOTSET", logging.NOTSET),
|
|
("DEBUG", logging.DEBUG),
|
|
("INFO", logging.INFO),
|
|
("WARNING", logging.WARNING),
|
|
("ERROR", logging.ERROR),
|
|
("CRITICAL", logging.CRITICAL),
|
|
]
|
|
|
|
|
|
class LoggerPanel(tk.Toplevel):
|
|
"""Toplevel window that allows setting logger levels at runtime."""
|
|
|
|
def __init__(self, master=None):
|
|
super().__init__(master)
|
|
self.title("Logger Levels")
|
|
self.geometry("520x420")
|
|
self.transient(master)
|
|
self.grab_set()
|
|
|
|
self.logger_names = [] # type: List[str]
|
|
|
|
# Config manager for persistence
|
|
try:
|
|
self._cfg = ConfigManager()
|
|
except Exception:
|
|
self._cfg = None
|
|
|
|
self._create_widgets()
|
|
self._populate_logger_list()
|
|
|
|
# Load persisted selections and levels
|
|
try:
|
|
self._load_persisted()
|
|
except Exception:
|
|
pass
|
|
|
|
def _create_widgets(self):
|
|
top = ttk.Frame(self)
|
|
top.pack(fill=tk.BOTH, expand=True, padx=8, pady=8)
|
|
|
|
# Left: filter + list of logger names
|
|
left = ttk.Frame(top)
|
|
left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
|
|
|
ttk.Label(left, text="Filter:").pack(anchor=tk.W)
|
|
self.filter_var = tk.StringVar(value="")
|
|
self.filter_entry = ttk.Entry(left, textvariable=self.filter_var)
|
|
self.filter_entry.pack(fill=tk.X, padx=(0, 6), pady=(2, 4))
|
|
# Call _on_filter_change on text changes
|
|
try:
|
|
# Python 3.6+ trace_add preferred
|
|
self.filter_var.trace_add("write", lambda *args: self._on_filter_change())
|
|
except Exception:
|
|
# Fallback for older trace
|
|
try:
|
|
self.filter_var.trace("w", lambda *args: self._on_filter_change())
|
|
except Exception:
|
|
pass
|
|
|
|
ttk.Label(left, text="Available loggers:").pack(anchor=tk.W)
|
|
self.logger_listbox = tk.Listbox(left, exportselection=False)
|
|
self.logger_listbox.pack(fill=tk.BOTH, expand=True, padx=(0, 6), pady=(4, 0))
|
|
self.logger_listbox.bind("<<ListboxSelect>>", self._on_select_logger)
|
|
|
|
# Right: controls
|
|
right = ttk.Frame(top)
|
|
right.pack(side=tk.RIGHT, fill=tk.Y)
|
|
|
|
ttk.Label(right, text="Selected logger:").pack(anchor=tk.W)
|
|
self.selected_name_var = tk.StringVar(value="")
|
|
ttk.Label(right, textvariable=self.selected_name_var, foreground="blue").pack(
|
|
anchor=tk.W, pady=(0, 6)
|
|
)
|
|
|
|
ttk.Label(right, text="Level:").pack(anchor=tk.W)
|
|
self.level_var = tk.StringVar(value="INFO")
|
|
level_names = [n for n, v in LEVELS]
|
|
self.level_combo = ttk.Combobox(
|
|
right,
|
|
values=level_names,
|
|
textvariable=self.level_var,
|
|
state="readonly",
|
|
width=12,
|
|
)
|
|
self.level_combo.pack(anchor=tk.W, pady=(0, 6))
|
|
|
|
ttk.Button(right, text="Apply", command=self._apply_level).pack(
|
|
fill=tk.X, pady=(6, 4)
|
|
)
|
|
ttk.Button(right, text="Reset to NOTSET", command=self._reset_level).pack(
|
|
fill=tk.X
|
|
)
|
|
ttk.Button(right, text="Restore defaults", command=self._restore_defaults).pack(
|
|
fill=tk.X, pady=(6, 0)
|
|
)
|
|
|
|
ttk.Separator(self).pack(fill=tk.X, pady=6)
|
|
|
|
bottom = ttk.Frame(self)
|
|
bottom.pack(fill=tk.X, padx=8, pady=6)
|
|
|
|
ttk.Label(bottom, text="Add / open logger by name:").pack(anchor=tk.W)
|
|
self.new_logger_var = tk.StringVar()
|
|
entry = ttk.Entry(bottom, textvariable=self.new_logger_var)
|
|
entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 6))
|
|
ttk.Button(bottom, text="Open", command=self._open_named_logger).pack(
|
|
side=tk.LEFT
|
|
)
|
|
|
|
ttk.Button(self, text="Refresh", command=self._populate_logger_list).pack(
|
|
side=tk.RIGHT, padx=8, pady=(0, 8)
|
|
)
|
|
ttk.Button(self, text="Export", command=self._export_prefs).pack(
|
|
side=tk.RIGHT, padx=(0, 8), pady=(0, 8)
|
|
)
|
|
ttk.Button(self, text="Close", command=self._on_close).pack(
|
|
side=tk.RIGHT, pady=(0, 8)
|
|
)
|
|
|
|
def _gather_logger_names(self) -> List[str]:
|
|
# Gather logger names from the logging manager plus some defaults
|
|
manager = logging.root.manager
|
|
names = list(getattr(manager, "loggerDict", {}).keys())
|
|
# Add a few commonly useful module names if missing
|
|
defaults = [
|
|
"target_simulator",
|
|
"target_simulator.analysis.simulation_state_hub",
|
|
"target_simulator.gui.sfp_debug_window",
|
|
"target_simulator.gui.ppi_display",
|
|
"target_simulator.gui.payload_router",
|
|
"target_simulator.core.sfp_transport",
|
|
]
|
|
for d in defaults:
|
|
if d not in names:
|
|
names.append(d)
|
|
names = sorted(set(names))
|
|
return names
|
|
|
|
def _refresh_list(self):
|
|
"""Refresh listbox contents applying current filter against _all_logger_names."""
|
|
try:
|
|
filt = (self.filter_var.get() or "").strip().lower()
|
|
except Exception:
|
|
filt = ""
|
|
|
|
# Build filtered list
|
|
if hasattr(self, "_all_logger_names") and self._all_logger_names:
|
|
source = self._all_logger_names
|
|
else:
|
|
source = self.logger_names
|
|
|
|
if filt:
|
|
filtered = [n for n in source if filt in n.lower()]
|
|
else:
|
|
filtered = list(source)
|
|
|
|
# Remember current selection name to try to preserve it
|
|
try:
|
|
sel_idx = self.logger_listbox.curselection()
|
|
sel_name = self.logger_names[sel_idx[0]] if sel_idx else None
|
|
except Exception:
|
|
sel_name = None
|
|
|
|
# Update listbox
|
|
self.logger_listbox.delete(0, tk.END)
|
|
self.logger_names = sorted(filtered)
|
|
for n in self.logger_names:
|
|
self.logger_listbox.insert(tk.END, n)
|
|
|
|
# Restore selection if still present
|
|
if sel_name and sel_name in self.logger_names:
|
|
idx = self.logger_names.index(sel_name)
|
|
self.logger_listbox.select_set(idx)
|
|
self.logger_listbox.see(idx)
|
|
self._on_select_logger()
|
|
|
|
def _on_filter_change(self, *args):
|
|
"""Callback for filter text changes; refresh the list."""
|
|
try:
|
|
self._refresh_list()
|
|
except Exception:
|
|
pass
|
|
|
|
def _populate_logger_list(self):
|
|
self.logger_listbox.delete(0, tk.END)
|
|
# Gather and cache the full set of logger names, then refresh list
|
|
self._all_logger_names = self._gather_logger_names()
|
|
# Apply filter and populate listbox
|
|
self._refresh_list()
|
|
# If user previously selected a logger, select it
|
|
try:
|
|
if self._cfg:
|
|
gen = self._cfg.get_general_settings()
|
|
lp = gen.get("logger_panel", {}) if isinstance(gen, dict) else {}
|
|
last = lp.get("last_selected")
|
|
if last and last in self.logger_names:
|
|
idx = self.logger_names.index(last)
|
|
self.logger_listbox.select_set(idx)
|
|
self.logger_listbox.see(idx)
|
|
self._on_select_logger()
|
|
except Exception:
|
|
pass
|
|
|
|
def _on_select_logger(self, event=None):
|
|
sel = self.logger_listbox.curselection()
|
|
if not sel:
|
|
return
|
|
idx = sel[0]
|
|
name = self.logger_names[idx]
|
|
self.selected_name_var.set(name)
|
|
lg = logging.getLogger(name)
|
|
lvl = lg.getEffectiveLevel()
|
|
# Translate to name
|
|
lvl_name = logging.getLevelName(lvl)
|
|
self.level_var.set(lvl_name)
|
|
|
|
def _apply_level(self):
|
|
name = self.selected_name_var.get()
|
|
if not name:
|
|
messagebox.showwarning(
|
|
"No logger selected", "Select a logger from the list first."
|
|
)
|
|
return
|
|
lvl_name = self.level_var.get()
|
|
lvl = next((v for n, v in LEVELS if n == lvl_name), logging.INFO)
|
|
logging.getLogger(name).setLevel(lvl)
|
|
# Persist this selection into dedicated logger_prefs.json (single source of truth)
|
|
try:
|
|
cfg_path = getattr(self._cfg, "filepath", None) if self._cfg else None
|
|
prefs_path = (
|
|
os.path.join(os.path.dirname(cfg_path), "logger_prefs.json")
|
|
if cfg_path
|
|
else os.path.join(os.getcwd(), "logger_prefs.json")
|
|
)
|
|
# Read existing prefs
|
|
try:
|
|
with open(prefs_path, "r", encoding="utf-8") as f:
|
|
prefs = json.load(f) if f else {}
|
|
if not isinstance(prefs, dict):
|
|
prefs = {}
|
|
except Exception:
|
|
prefs = {}
|
|
|
|
saved = (
|
|
prefs.get("saved_levels", {})
|
|
if isinstance(prefs.get("saved_levels", {}), dict)
|
|
else {}
|
|
)
|
|
saved[name] = lvl_name
|
|
prefs["saved_levels"] = saved
|
|
prefs["last_selected"] = name
|
|
with open(prefs_path, "w", encoding="utf-8") as f:
|
|
json.dump(prefs, f, indent=4)
|
|
except Exception:
|
|
pass
|
|
|
|
# Post a status message to the application log instead of a modal dialog
|
|
try:
|
|
# Show a transient status in the main UI if available
|
|
parent = getattr(self, "master", None)
|
|
if parent and hasattr(parent, "show_status_message"):
|
|
parent.show_status_message(f"Logger '{name}' set to {lvl_name} (saved)")
|
|
else:
|
|
logging.getLogger(__name__).info(
|
|
"Logger '%s' set to %s (persisted)", name, lvl_name
|
|
)
|
|
except Exception:
|
|
try:
|
|
messagebox.showinfo(
|
|
"Logger level set", f"Logger '{name}' set to {lvl_name}."
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
def _reset_level(self):
|
|
name = self.selected_name_var.get()
|
|
if not name:
|
|
messagebox.showwarning(
|
|
"No logger selected", "Select a logger from the list first."
|
|
)
|
|
return
|
|
logging.getLogger(name).setLevel(logging.NOTSET)
|
|
self.level_var.set("NOTSET")
|
|
# Remove persisted level for this logger from logger_prefs.json
|
|
try:
|
|
cfg_path = getattr(self._cfg, "filepath", None) if self._cfg else None
|
|
prefs_path = (
|
|
os.path.join(os.path.dirname(cfg_path), "logger_prefs.json")
|
|
if cfg_path
|
|
else os.path.join(os.getcwd(), "logger_prefs.json")
|
|
)
|
|
try:
|
|
with open(prefs_path, "r", encoding="utf-8") as f:
|
|
prefs = json.load(f) if f else {}
|
|
if not isinstance(prefs, dict):
|
|
prefs = {}
|
|
except Exception:
|
|
prefs = {}
|
|
|
|
saved = (
|
|
prefs.get("saved_levels", {})
|
|
if isinstance(prefs.get("saved_levels", {}), dict)
|
|
else {}
|
|
)
|
|
if name in saved:
|
|
del saved[name]
|
|
prefs["saved_levels"] = saved
|
|
# if last_selected pointed to this logger, remove it
|
|
if prefs.get("last_selected") == name:
|
|
prefs.pop("last_selected", None)
|
|
with open(prefs_path, "w", encoding="utf-8") as f:
|
|
json.dump(prefs, f, indent=4)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
parent = getattr(self, "master", None)
|
|
if parent and hasattr(parent, "show_status_message"):
|
|
parent.show_status_message(f"Logger '{name}' reset to NOTSET (saved)")
|
|
else:
|
|
logging.getLogger(__name__).info(
|
|
"Logger '%s' reset to NOTSET (persisted)", name
|
|
)
|
|
except Exception:
|
|
try:
|
|
messagebox.showinfo("Logger reset", f"Logger '{name}' reset to NOTSET.")
|
|
except Exception:
|
|
pass
|
|
|
|
def _open_named_logger(self):
|
|
name = self.new_logger_var.get().strip()
|
|
if not name:
|
|
return
|
|
# If exists in list, select it
|
|
try:
|
|
idx = self._all_logger_names.index(name)
|
|
except ValueError:
|
|
# Add to list
|
|
# Add to cached master list and refresh
|
|
if not hasattr(self, "_all_logger_names"):
|
|
self._all_logger_names = []
|
|
if name not in self._all_logger_names:
|
|
self._all_logger_names.append(name)
|
|
self._all_logger_names.sort()
|
|
self._populate_logger_list()
|
|
idx = self.logger_names.index(name)
|
|
self.logger_listbox.select_clear(0, tk.END)
|
|
self.logger_listbox.select_set(idx)
|
|
self.logger_listbox.see(idx)
|
|
self._on_select_logger()
|
|
# Persist last selected into logger_prefs.json
|
|
try:
|
|
cfg_path = getattr(self._cfg, "filepath", None) if self._cfg else None
|
|
prefs_path = (
|
|
os.path.join(os.path.dirname(cfg_path), "logger_prefs.json")
|
|
if cfg_path
|
|
else os.path.join(os.getcwd(), "logger_prefs.json")
|
|
)
|
|
try:
|
|
with open(prefs_path, "r", encoding="utf-8") as f:
|
|
prefs = json.load(f) if f else {}
|
|
if not isinstance(prefs, dict):
|
|
prefs = {}
|
|
except Exception:
|
|
prefs = {}
|
|
prefs["last_selected"] = name
|
|
with open(prefs_path, "w", encoding="utf-8") as f:
|
|
json.dump(prefs, f, indent=4)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
parent = getattr(self, "master", None)
|
|
if parent and hasattr(parent, "show_status_message"):
|
|
parent.show_status_message(f"Selected logger: {name}")
|
|
except Exception:
|
|
pass
|
|
|
|
def _restore_defaults(self):
|
|
"""Clear all saved logger preferences and reset those loggers to NOTSET."""
|
|
if not self._cfg:
|
|
try:
|
|
messagebox.showinfo(
|
|
"Restore defaults", "Could not access settings to restore defaults."
|
|
)
|
|
except Exception:
|
|
pass
|
|
return
|
|
|
|
try:
|
|
gen = self._cfg.get_general_settings()
|
|
lp = gen.get("logger_panel", {}) if isinstance(gen, dict) else {}
|
|
saved = lp.get("saved_levels", {}) if isinstance(lp, dict) else {}
|
|
# Reset each saved logger to NOTSET
|
|
for name in list(saved.keys()):
|
|
try:
|
|
logging.getLogger(name).setLevel(logging.NOTSET)
|
|
except Exception:
|
|
pass
|
|
# Remove saved_levels
|
|
# Clear dedicated prefs file instead of touching settings.json
|
|
cfg_path = getattr(self._cfg, "filepath", None) if self._cfg else None
|
|
prefs_path = (
|
|
os.path.join(os.path.dirname(cfg_path), "logger_prefs.json")
|
|
if cfg_path
|
|
else os.path.join(os.getcwd(), "logger_prefs.json")
|
|
)
|
|
try:
|
|
with open(prefs_path, "w", encoding="utf-8") as f:
|
|
json.dump({"saved_levels": {}}, f, indent=4)
|
|
except Exception:
|
|
# fallback: attempt to remove the file
|
|
try:
|
|
if os.path.exists(prefs_path):
|
|
os.remove(prefs_path)
|
|
except Exception:
|
|
pass
|
|
logging.getLogger(__name__).info(
|
|
"LoggerPanel: restored defaults and cleared saved preferences"
|
|
)
|
|
# Refresh the list and selection
|
|
self._populate_logger_list()
|
|
except Exception:
|
|
try:
|
|
messagebox.showinfo("Restore defaults", "Failed to restore defaults.")
|
|
except Exception:
|
|
pass
|
|
|
|
def _export_prefs(self):
|
|
"""Export saved logger preferences to a separate JSON file for inspection."""
|
|
if not self._cfg:
|
|
try:
|
|
messagebox.showinfo(
|
|
"Export prefs", "Settings manager not available; cannot export."
|
|
)
|
|
except Exception:
|
|
pass
|
|
return
|
|
|
|
try:
|
|
gen = self._cfg.get_general_settings()
|
|
lp = gen.get("logger_panel", {}) if isinstance(gen, dict) else {}
|
|
saved = lp.get("saved_levels", {}) if isinstance(lp, dict) else {}
|
|
# Determine export path next to settings.json
|
|
cfg_path = getattr(self._cfg, "filepath", None)
|
|
export_dir = os.path.dirname(cfg_path) if cfg_path else os.getcwd()
|
|
export_path = os.path.join(export_dir, "logger_prefs.json")
|
|
with open(export_path, "w", encoding="utf-8") as f:
|
|
json.dump({"saved_levels": saved}, f, indent=4)
|
|
|
|
parent = getattr(self, "master", None)
|
|
if parent and hasattr(parent, "show_status_message"):
|
|
parent.show_status_message(f"Exported logger prefs to {export_path}")
|
|
else:
|
|
logging.getLogger(__name__).info(
|
|
"Exported logger prefs to %s", export_path
|
|
)
|
|
except Exception:
|
|
try:
|
|
messagebox.showinfo(
|
|
"Export prefs", "Failed to export logger preferences."
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
def _on_close(self):
|
|
try:
|
|
self.grab_release()
|
|
except Exception:
|
|
pass
|
|
self.destroy()
|
|
|
|
def _load_persisted(self):
|
|
"""Load persisted logger panel settings and apply saved levels."""
|
|
# Prefer reading the dedicated prefs file
|
|
cfg_path = getattr(self._cfg, "filepath", None) if self._cfg else None
|
|
prefs_path = (
|
|
os.path.join(os.path.dirname(cfg_path), "logger_prefs.json")
|
|
if cfg_path
|
|
else os.path.join(os.getcwd(), "logger_prefs.json")
|
|
)
|
|
saved = {}
|
|
try:
|
|
if os.path.exists(prefs_path):
|
|
with open(prefs_path, "r", encoding="utf-8") as f:
|
|
jp = json.load(f)
|
|
if isinstance(jp, dict):
|
|
saved = jp.get("saved_levels", {}) or {}
|
|
last = jp.get("last_selected")
|
|
if last:
|
|
# store last_selected in prefs for later UI use
|
|
pass
|
|
else:
|
|
# Fallback to settings.json
|
|
if self._cfg:
|
|
gen = self._cfg.get_general_settings()
|
|
lp = gen.get("logger_panel", {}) if isinstance(gen, dict) else {}
|
|
saved = lp.get("saved_levels", {}) if isinstance(lp, dict) else {}
|
|
except Exception:
|
|
saved = {}
|
|
|
|
# Apply saved levels to loggers
|
|
for name, lvl_name in (saved or {}).items():
|
|
try:
|
|
lvl = logging.getLevelName(lvl_name)
|
|
if isinstance(lvl, int):
|
|
logging.getLogger(name).setLevel(lvl)
|
|
except Exception:
|
|
pass
|