452 lines
18 KiB
Python
452 lines
18 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
|