205 lines
7.4 KiB
Python
205 lines
7.4 KiB
Python
# radar_data_reader/utils/config_manager.py
|
|
|
|
"""
|
|
Configuration Manager for the application.
|
|
Handles loading and saving of application settings to a JSON file.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional, List
|
|
|
|
from . import logger
|
|
from ..core.export_profiles import ExportProfile, ExportField
|
|
|
|
log = logger.get_logger(__name__)
|
|
|
|
|
|
class ConfigManager:
|
|
"""Manages loading and saving configuration from/to a JSON file."""
|
|
|
|
def __init__(self, config_path: Path):
|
|
self.config_path: Path = config_path
|
|
self.config: Dict[str, Any] = {}
|
|
self.export_profiles: List[ExportProfile] = []
|
|
|
|
@staticmethod
|
|
def _get_default_config() -> Dict[str, Any]:
|
|
"""Provides the default configuration structure for the entire application."""
|
|
default_export_profile = ExportProfile(
|
|
name="Default",
|
|
fields=[
|
|
ExportField(column_name="batch_id", data_path="batch_id"),
|
|
ExportField(
|
|
column_name="timetag",
|
|
data_path="main_header.ge_header.signal_descr.ttag",
|
|
),
|
|
],
|
|
)
|
|
|
|
default_cpp_config = {
|
|
"cpp_executable_path": "C:/path/to/your/g_reconvert.exe",
|
|
"analyze_only": False,
|
|
"no_sign": False,
|
|
"use_fw_header": False,
|
|
"fix_dsp_bug": False,
|
|
"sync_signal_flows": False,
|
|
"extract_sw_only": False,
|
|
"extract_dsp_only": False,
|
|
"stop_on_discontinuity": False,
|
|
"dry_run": False,
|
|
"start_from_stby": False,
|
|
"fix_sata_date": False,
|
|
"examine_data": False,
|
|
"post_process": True,
|
|
"post_process_level": "1",
|
|
"output_format": "",
|
|
"json_config_file": "",
|
|
"max_batches": "",
|
|
"pri_number": "",
|
|
"rbin_number": "",
|
|
"video_out_to_rec": False,
|
|
"video_show": True,
|
|
"video_save": True,
|
|
"video_fix_framerate": False,
|
|
"gps_save_track": True,
|
|
"fix_hr_debug": False,
|
|
"sar_save_only": False,
|
|
"video_add_subtitles": False,
|
|
"verbose": False,
|
|
"debug_messages": False,
|
|
"quiet": False,
|
|
"mask_warnings": False,
|
|
"mask_errors": False,
|
|
"silent_overwrite": True,
|
|
"enable_python_worker_profiling": False,
|
|
}
|
|
|
|
segment_g_reconverter_options = default_cpp_config.copy()
|
|
segment_g_reconverter_options.update(
|
|
{
|
|
"post_process": False,
|
|
"video_show": False,
|
|
"video_save": False,
|
|
"gps_save_track": True,
|
|
"no_sign": False,
|
|
}
|
|
)
|
|
|
|
default_segment_export_config = {
|
|
"naming_options": {
|
|
"folder_name_template": "{Segment}_{StartBatch}-{EndBatch}",
|
|
},
|
|
"g_reconverter_options": segment_g_reconverter_options,
|
|
}
|
|
|
|
return {
|
|
"last_opened_out_file": "",
|
|
"last_opened_rec_file": "",
|
|
"last_out_output_dir": "",
|
|
"last_rec_output_dir": "",
|
|
"last_flight_folder": "",
|
|
"active_out_export_profile_name": "Default",
|
|
"export_profiles": [default_export_profile.to_dict()],
|
|
"cpp_converter_config": default_cpp_config,
|
|
"segment_export_config": default_segment_export_config,
|
|
"segment_processor_config": {
|
|
"last_output_dir": "",
|
|
"create_separate_folders": True,
|
|
},
|
|
"flight_analysis_options": {
|
|
"aggregate_by_scale": True,
|
|
"aggregate_by_waveform": True,
|
|
},
|
|
}
|
|
|
|
def load_config(self) -> None:
|
|
log.info(f"Attempting to load configuration from: {self.config_path}")
|
|
if self.config_path.is_file():
|
|
try:
|
|
with open(self.config_path, "r", encoding="utf-8") as f:
|
|
loaded_data = json.load(f)
|
|
|
|
default_conf = self._get_default_config()
|
|
|
|
self.config = default_conf
|
|
for key, value in loaded_data.items():
|
|
if isinstance(value, dict) and key in self.config:
|
|
self.config[key].update(value)
|
|
if (
|
|
key == "segment_export_config"
|
|
and "g_reconverter_options" in self.config[key]
|
|
):
|
|
main_exe_path = self.config.get(
|
|
"cpp_converter_config", {}
|
|
).get("cpp_executable_path")
|
|
if main_exe_path:
|
|
self.config[key]["g_reconverter_options"][
|
|
"cpp_executable_path"
|
|
] = main_exe_path
|
|
else:
|
|
self.config[key] = value
|
|
|
|
profiles_data = self.config.get("export_profiles", [])
|
|
self.export_profiles = [
|
|
ExportProfile.from_dict(p_data) for p_data in profiles_data
|
|
]
|
|
|
|
if not self.export_profiles:
|
|
self._load_default_profiles()
|
|
|
|
log.info(
|
|
f"Configuration loaded. Found {len(self.export_profiles)} export profiles."
|
|
)
|
|
|
|
except (json.JSONDecodeError, IOError, KeyError) as e:
|
|
log.error(f"Failed to load or parse config file: {e}. Using defaults.")
|
|
self.config = self._get_default_config()
|
|
self._load_default_profiles()
|
|
else:
|
|
log.warning("Configuration file not found. Using default settings.")
|
|
self.config = self._get_default_config()
|
|
self._load_default_profiles()
|
|
|
|
def _load_default_profiles(self):
|
|
default_conf = self._get_default_config()
|
|
profiles_data = default_conf.get("export_profiles", [])
|
|
self.export_profiles = [
|
|
ExportProfile.from_dict(p_data) for p_data in profiles_data
|
|
]
|
|
|
|
def save_config(self) -> None:
|
|
log.info(f"Saving configuration to: {self.config_path}")
|
|
self.config["export_profiles"] = [
|
|
profile.to_dict() for profile in self.export_profiles
|
|
]
|
|
try:
|
|
self.config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(self.config_path, "w", encoding="utf-8") as f:
|
|
json.dump(self.config, f, indent=4)
|
|
log.info("Configuration saved successfully.")
|
|
except IOError as e:
|
|
log.error(f"Failed to save configuration file: {e}")
|
|
|
|
def get(self, key: str, default: Optional[Any] = None) -> Any:
|
|
return self.config.get(key, default)
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
self.config[key] = value
|
|
|
|
def get_cpp_converter_config(self) -> Dict[str, Any]:
|
|
return self.config.get(
|
|
"cpp_converter_config", self._get_default_config()["cpp_converter_config"]
|
|
)
|
|
|
|
def save_cpp_converter_config(self, new_config: Dict[str, Any]):
|
|
self.config["cpp_converter_config"] = new_config
|
|
self.save_config()
|
|
|
|
def get_export_profiles(self) -> List[ExportProfile]:
|
|
return self.export_profiles
|
|
|
|
def save_export_profiles(self, profiles: List[ExportProfile]):
|
|
self.export_profiles = profiles
|
|
self.save_config()
|