aggiunta schermata per configurazione logger con salvataggio preferenze
This commit is contained in:
VALLONGOL
parent
3bfa5edf88
commit
b3a79d44f1
3
logger_prefs.json
Normal file
3
logger_prefs.json
Normal file
@ -0,0 +1,3 @@
|
||||
{
|
||||
"saved_levels": {}
|
||||
}
|
||||
@ -3,7 +3,7 @@
|
||||
"scan_limit": 60,
|
||||
"max_range": 100,
|
||||
"geometry": "1599x1089+501+84",
|
||||
"last_selected_scenario": "scenario3",
|
||||
"last_selected_scenario": null,
|
||||
"connection": {
|
||||
"target": {
|
||||
"type": "sfp",
|
||||
|
||||
@ -16,9 +16,14 @@ if project_root not in sys.path:
|
||||
sys.path.insert(0, project_root)
|
||||
|
||||
from target_simulator.gui.main_view import MainView
|
||||
from utils.logger import setup_basic_logging, get_logger, add_tkinter_handler
|
||||
from target_simulator.utils.logger import (
|
||||
setup_basic_logging,
|
||||
get_logger,
|
||||
add_tkinter_handler,
|
||||
apply_saved_logger_levels,
|
||||
)
|
||||
|
||||
from config import LOGGING_CONFIG
|
||||
from target_simulator.config import LOGGING_CONFIG
|
||||
|
||||
|
||||
def main():
|
||||
@ -32,6 +37,13 @@ def main():
|
||||
# Setup the global logging system, connecting it to the GUI's main loop
|
||||
setup_basic_logging(app, LOGGING_CONFIG)
|
||||
|
||||
# Apply any saved logger levels so preferences take effect even if the
|
||||
# LoggerPanel is never opened.
|
||||
try:
|
||||
apply_saved_logger_levels()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Now that the logging system is active, add the handler for the GUI widget
|
||||
add_tkinter_handler(app.log_text_widget, LOGGING_CONFIG)
|
||||
|
||||
|
||||
@ -6,6 +6,9 @@ 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),
|
||||
@ -29,17 +32,44 @@ class LoggerPanel(tk.Toplevel):
|
||||
|
||||
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: list of logger names
|
||||
# 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))
|
||||
@ -61,6 +91,7 @@ class LoggerPanel(tk.Toplevel):
|
||||
|
||||
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)
|
||||
|
||||
@ -74,6 +105,7 @@ class LoggerPanel(tk.Toplevel):
|
||||
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]:
|
||||
@ -95,12 +127,71 @@ class LoggerPanel(tk.Toplevel):
|
||||
names = sorted(set(names))
|
||||
return names
|
||||
|
||||
def _populate_logger_list(self):
|
||||
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 = self._gather_logger_names()
|
||||
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:
|
||||
@ -122,7 +213,41 @@ class LoggerPanel(tk.Toplevel):
|
||||
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)
|
||||
messagebox.showinfo("Logger level set", f"Logger '{name}' set to {lvl_name}.")
|
||||
# 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()
|
||||
@ -131,7 +256,41 @@ class LoggerPanel(tk.Toplevel):
|
||||
return
|
||||
logging.getLogger(name).setLevel(logging.NOTSET)
|
||||
self.level_var.set('NOTSET')
|
||||
messagebox.showinfo("Logger reset", f"Logger '{name}' reset to 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()
|
||||
@ -139,17 +298,116 @@ class LoggerPanel(tk.Toplevel):
|
||||
return
|
||||
# If exists in list, select it
|
||||
try:
|
||||
idx = self.logger_names.index(name)
|
||||
idx = self._all_logger_names.index(name)
|
||||
except ValueError:
|
||||
# Add to list
|
||||
self.logger_names.append(name)
|
||||
self.logger_names.sort()
|
||||
# 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:
|
||||
@ -157,3 +415,37 @@ class LoggerPanel(tk.Toplevel):
|
||||
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
|
||||
|
||||
@ -87,6 +87,8 @@ class MainView(tk.Tk):
|
||||
self._create_menubar()
|
||||
self._create_main_layout()
|
||||
self._create_statusbar()
|
||||
# Id for scheduled status clear; used by show_status_message
|
||||
self._status_after_id = None
|
||||
|
||||
# --- Post-UI Initialization ---
|
||||
self._initialize_communicators()
|
||||
@ -383,6 +385,36 @@ class MainView(tk.Tk):
|
||||
side=tk.LEFT, fill=tk.X, expand=True, padx=5
|
||||
)
|
||||
|
||||
def show_status_message(self, text: str, timeout_ms: int = 3000):
|
||||
"""Show a transient status message in the main status bar.
|
||||
|
||||
If another message is scheduled to clear, cancel it and schedule the
|
||||
new message to be cleared after timeout_ms.
|
||||
"""
|
||||
try:
|
||||
# Cancel previous scheduled clear if any
|
||||
try:
|
||||
if self._status_after_id is not None:
|
||||
self.after_cancel(self._status_after_id)
|
||||
except Exception:
|
||||
pass
|
||||
# Set message
|
||||
self.status_var.set(text)
|
||||
# Schedule clear back to Ready
|
||||
def _clear():
|
||||
try:
|
||||
self.status_var.set("Ready")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self._status_after_id = self.after(timeout_ms, _clear)
|
||||
except Exception:
|
||||
# As a fallback, log the status
|
||||
try:
|
||||
self.logger.info(text)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _draw_status_indicator(self, canvas, color):
|
||||
canvas.delete("all")
|
||||
canvas.create_oval(2, 2, 14, 14, fill=color, outline="black")
|
||||
@ -1116,14 +1148,44 @@ class MainView(tk.Tk):
|
||||
if self.is_simulation_running.get():
|
||||
self._on_stop_simulation()
|
||||
|
||||
settings_to_save = {
|
||||
"scan_limit": self.scan_limit,
|
||||
"max_range": self.max_range,
|
||||
"geometry": self.winfo_geometry(),
|
||||
"last_selected_scenario": self.current_scenario_name,
|
||||
}
|
||||
self.config_manager.save_general_settings(settings_to_save)
|
||||
self.config_manager.save_connection_settings(self.connection_config)
|
||||
# Merge current runtime general settings with any existing saved
|
||||
# settings so we don't clobber unrelated keys (e.g., logger_panel).
|
||||
try:
|
||||
existing = self.config_manager.get_general_settings() or {}
|
||||
except Exception:
|
||||
existing = {}
|
||||
|
||||
settings_to_save = dict(existing)
|
||||
settings_to_save.update(
|
||||
{
|
||||
"scan_limit": self.scan_limit,
|
||||
"max_range": self.max_range,
|
||||
"geometry": self.winfo_geometry(),
|
||||
"last_selected_scenario": self.current_scenario_name,
|
||||
}
|
||||
)
|
||||
|
||||
# Save merged general settings and connection settings separately
|
||||
try:
|
||||
self.config_manager.save_general_settings(settings_to_save)
|
||||
except Exception:
|
||||
# Fallback: try to write the minimal dict if merge/save fails
|
||||
try:
|
||||
self.config_manager.save_general_settings(
|
||||
{
|
||||
"scan_limit": self.scan_limit,
|
||||
"max_range": self.max_range,
|
||||
"geometry": self.winfo_geometry(),
|
||||
"last_selected_scenario": self.current_scenario_name,
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
self.config_manager.save_connection_settings(self.connection_config)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if self.target_communicator:
|
||||
if hasattr(self.target_communicator, 'remove_connection_state_callback'):
|
||||
|
||||
@ -149,7 +149,21 @@ class ConfigManager:
|
||||
|
||||
def save_general_settings(self, data: Dict[str, Any]):
|
||||
"""Saves the general settings."""
|
||||
self._settings["general"] = data
|
||||
# Merge incoming general settings with existing ones to avoid
|
||||
# overwriting unrelated keys (for example: logger_panel saved prefs).
|
||||
existing = self._settings.get("general", {}) if isinstance(self._settings, dict) else {}
|
||||
try:
|
||||
# Shallow merge: if a key contains a dict, update its contents
|
||||
for k, v in (data or {}).items():
|
||||
if isinstance(v, dict) and k in existing and isinstance(existing.get(k), dict):
|
||||
existing[k].update(v)
|
||||
else:
|
||||
existing[k] = v
|
||||
except Exception:
|
||||
# Fallback: replace entirely if merge fails
|
||||
existing = data
|
||||
|
||||
self._settings["general"] = existing
|
||||
self._save_settings()
|
||||
|
||||
def get_scenario_names(self) -> List[str]:
|
||||
|
||||
@ -7,6 +7,9 @@ from queue import Queue, Empty as QueueEmpty
|
||||
from typing import Optional, Dict, Any
|
||||
from contextlib import contextmanager
|
||||
from logging import Logger
|
||||
from target_simulator.utils.config_manager import ConfigManager
|
||||
import os
|
||||
import json
|
||||
|
||||
# Module-level logger for utils.logging helpers
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -210,3 +213,50 @@ def shutdown_logging_system():
|
||||
# Final flush of the queue
|
||||
_process_global_log_queue()
|
||||
logging.shutdown()
|
||||
|
||||
|
||||
def apply_saved_logger_levels():
|
||||
"""Apply saved logger levels from ConfigManager at startup.
|
||||
|
||||
Reads `general.logger_panel.saved_levels` from settings.json and sets
|
||||
each configured logger to the saved level name.
|
||||
"""
|
||||
try:
|
||||
cfg = ConfigManager()
|
||||
except Exception:
|
||||
return
|
||||
|
||||
try:
|
||||
# Prefer a dedicated logger_prefs.json next to the main settings file
|
||||
prefs_path = None
|
||||
cfg_path = getattr(cfg, "filepath", None)
|
||||
if cfg_path:
|
||||
prefs_path = os.path.join(os.path.dirname(cfg_path), "logger_prefs.json")
|
||||
|
||||
saved = {}
|
||||
if prefs_path and os.path.exists(prefs_path):
|
||||
try:
|
||||
with open(prefs_path, "r", encoding="utf-8") as f:
|
||||
jp = json.load(f)
|
||||
if isinstance(jp, dict):
|
||||
saved = jp.get("saved_levels", {}) or {}
|
||||
except Exception:
|
||||
saved = {}
|
||||
else:
|
||||
# Fallback to settings.json general.logger_panel
|
||||
try:
|
||||
gen = 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 = {}
|
||||
|
||||
for name, lvl_name in (saved or {}).items():
|
||||
try:
|
||||
lvl_val = logging.getLevelName(lvl_name)
|
||||
if isinstance(lvl_val, int):
|
||||
logging.getLogger(name).setLevel(lvl_val)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
117
tests/test_logger_persistence.py
Normal file
117
tests/test_logger_persistence.py
Normal file
@ -0,0 +1,117 @@
|
||||
import os
|
||||
import json
|
||||
import logging
|
||||
|
||||
from target_simulator.utils.config_manager import ConfigManager
|
||||
from target_simulator.utils.logger import apply_saved_logger_levels
|
||||
|
||||
|
||||
def _prefs_path(cfg: ConfigManager) -> str:
|
||||
return os.path.join(os.path.dirname(cfg.filepath), "logger_prefs.json")
|
||||
|
||||
|
||||
def test_apply_saved_logger_levels_from_prefs_file(tmp_path):
|
||||
cfg = ConfigManager()
|
||||
settings_path = cfg.filepath
|
||||
prefs = {"saved_levels": {"test.logger.persistence": "DEBUG"}}
|
||||
prefs_path = _prefs_path(cfg)
|
||||
|
||||
# Backup existing settings.json and prefs if present
|
||||
if os.path.exists(settings_path):
|
||||
with open(settings_path, "r", encoding="utf-8") as f:
|
||||
orig_settings = f.read()
|
||||
else:
|
||||
orig_settings = None
|
||||
|
||||
if os.path.exists(prefs_path):
|
||||
with open(prefs_path, "r", encoding="utf-8") as f:
|
||||
orig_prefs = f.read()
|
||||
else:
|
||||
orig_prefs = None
|
||||
|
||||
try:
|
||||
# Ensure the logger starts at INFO
|
||||
lg = logging.getLogger("test.logger.persistence")
|
||||
lg.setLevel(logging.INFO)
|
||||
|
||||
# Write prefs file and apply
|
||||
with open(prefs_path, "w", encoding="utf-8") as f:
|
||||
json.dump(prefs, f, indent=2)
|
||||
|
||||
apply_saved_logger_levels()
|
||||
|
||||
# After applying, the logger should be DEBUG
|
||||
assert logging.getLogger("test.logger.persistence").level == logging.DEBUG
|
||||
|
||||
# Now change prefs to WARNING and re-apply
|
||||
prefs["saved_levels"]["test.logger.persistence"] = "WARNING"
|
||||
with open(prefs_path, "w", encoding="utf-8") as f:
|
||||
json.dump(prefs, f, indent=2)
|
||||
|
||||
apply_saved_logger_levels()
|
||||
assert logging.getLogger("test.logger.persistence").level == logging.WARNING
|
||||
|
||||
finally:
|
||||
# Restore original files
|
||||
if orig_settings is not None:
|
||||
with open(settings_path, "w", encoding="utf-8") as f:
|
||||
f.write(orig_settings)
|
||||
if orig_prefs is not None:
|
||||
with open(prefs_path, "w", encoding="utf-8") as f:
|
||||
f.write(orig_prefs)
|
||||
else:
|
||||
try:
|
||||
if os.path.exists(prefs_path):
|
||||
os.remove(prefs_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def test_apply_saved_logger_levels_fallback_to_settings(tmp_path):
|
||||
cfg = ConfigManager()
|
||||
settings_path = cfg.filepath
|
||||
prefs_path = _prefs_path(cfg)
|
||||
|
||||
# Backup
|
||||
if os.path.exists(settings_path):
|
||||
with open(settings_path, "r", encoding="utf-8") as f:
|
||||
orig_settings = f.read()
|
||||
else:
|
||||
orig_settings = None
|
||||
if os.path.exists(prefs_path):
|
||||
with open(prefs_path, "r", encoding="utf-8") as f:
|
||||
orig_prefs = f.read()
|
||||
else:
|
||||
orig_prefs = None
|
||||
|
||||
try:
|
||||
# Remove prefs file to force fallback
|
||||
if os.path.exists(prefs_path):
|
||||
os.remove(prefs_path)
|
||||
|
||||
# Prepare settings.json general.logger_panel
|
||||
cfg_gen = cfg.get_general_settings() or {}
|
||||
cfg_gen["logger_panel"] = {"saved_levels": {"test.logger.fallback": "ERROR"}}
|
||||
cfg.save_general_settings(cfg_gen)
|
||||
|
||||
# Ensure logger initial level is lower
|
||||
logging.getLogger("test.logger.fallback").setLevel(logging.DEBUG)
|
||||
|
||||
apply_saved_logger_levels()
|
||||
|
||||
assert logging.getLogger("test.logger.fallback").level == logging.ERROR
|
||||
|
||||
finally:
|
||||
# Restore originals
|
||||
if orig_settings is not None:
|
||||
with open(settings_path, "w", encoding="utf-8") as f:
|
||||
f.write(orig_settings)
|
||||
if orig_prefs is not None:
|
||||
with open(prefs_path, "w", encoding="utf-8") as f:
|
||||
f.write(orig_prefs)
|
||||
else:
|
||||
try:
|
||||
if os.path.exists(prefs_path):
|
||||
os.remove(prefs_path)
|
||||
except Exception:
|
||||
pass
|
||||
Loading…
Reference in New Issue
Block a user