Files
vna_system/vna_system/api/endpoints/settings.py
2025-09-30 18:09:50 +03:00

654 lines
26 KiB
Python

from typing import Any, List # pydantic response_model uses List
from fastapi import APIRouter, HTTPException
import vna_system.core.singletons as singletons
from vna_system.core.logging.logger import get_component_logger
from vna_system.core.settings.calibration_manager import CalibrationStandard
from vna_system.core.visualization.magnitude_chart import (
generate_standards_magnitude_plots,
)
from vna_system.api.models.settings import (
PresetModel,
CalibrationModel,
SettingsStatusModel,
SetPresetRequest,
StartCalibrationRequest,
CalibrateStandardRequest,
SaveCalibrationRequest,
SetCalibrationRequest,
RemoveStandardRequest,
WorkingCalibrationModel,
ReferenceModel,
CreateReferenceRequest,
SetReferenceRequest,
)
router = APIRouter(prefix="/api/v1/settings", tags=["settings"])
logger = get_component_logger(__file__)
@router.get("/status", response_model=SettingsStatusModel)
async def get_status() -> dict[str, Any]:
"""Get current settings status."""
try:
return singletons.settings_manager.get_status_summary()
except Exception as exc: # noqa: BLE001
logger.error("Failed to get settings status")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/presets", response_model=List[PresetModel])
async def get_presets() -> list[PresetModel]:
"""Get all available configuration presets, optionally filtered by mode."""
try:
presets = singletons.settings_manager.get_available_presets()
return [
PresetModel(
filename=p.filename,
mode=p.mode.value,
start_freq=p.start_freq,
stop_freq=p.stop_freq,
points=p.points,
bandwidth=p.bandwidth,
)
for p in presets
]
except Exception as exc: # noqa: BLE001
logger.error("Failed to list presets")
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/preset/set")
async def set_preset(request: SetPresetRequest) -> dict[str, Any]:
"""Set current configuration preset."""
try:
presets = singletons.settings_manager.get_available_presets()
preset = next((p for p in presets if p.filename == request.filename), None)
if preset is None:
raise HTTPException(status_code=404, detail=f"Preset not found: {request.filename}")
# Changing preset invalidates active calibration selection.
singletons.settings_manager.calibration_manager.clear_current_calibration()
singletons.settings_manager.set_current_preset(preset)
logger.info("Preset selected via API", filename=preset.filename, mode=preset.mode.value)
return {"success": True, "message": f"Preset set to {request.filename}"}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to set preset")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/preset/current", response_model=PresetModel | None)
async def get_current_preset():
"""Get currently selected configuration preset."""
try:
preset = singletons.settings_manager.get_current_preset()
if not preset:
return None
return PresetModel(
filename=preset.filename,
mode=preset.mode.value,
start_freq=preset.start_freq,
stop_freq=preset.stop_freq,
points=preset.points,
bandwidth=preset.bandwidth,
)
except Exception as exc: # noqa: BLE001
logger.error("Failed to get current preset")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/calibrations", response_model=List[CalibrationModel])
async def get_calibrations(preset_filename: str | None = None) -> list[CalibrationModel]:
"""Get available calibrations for current or specified preset."""
try:
preset = None
if preset_filename:
presets = singletons.settings_manager.get_available_presets()
preset = next((p for p in presets if p.filename == preset_filename), None)
if preset is None:
raise HTTPException(status_code=404, detail=f"Preset not found: {preset_filename}")
calibrations = singletons.settings_manager.get_available_calibrations(preset)
details: list[CalibrationModel] = []
current_preset = preset or singletons.settings_manager.get_current_preset()
if current_preset:
for name in calibrations:
info = singletons.settings_manager.get_calibration_info(name, current_preset)
standards = info.get("standards", {})
# Normalize standards into {standard: bool}
if isinstance(standards, list):
required = singletons.settings_manager.get_required_standards(current_preset.mode)
standards = {std.value: (std.value in standards) for std in required}
details.append(
CalibrationModel(
name=name,
is_complete=bool(info.get("is_complete", False)),
standards=standards,
)
)
return details
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to list calibrations")
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/calibration/start")
async def start_calibration(request: StartCalibrationRequest) -> dict[str, Any]:
"""Start new calibration for current or specified preset."""
try:
preset = None
if request.preset_filename:
presets = singletons.settings_manager.get_available_presets()
preset = next((p for p in presets if p.filename == request.preset_filename), None)
if preset is None:
raise HTTPException(status_code=404, detail=f"Preset not found: {request.preset_filename}")
calib = singletons.settings_manager.start_new_calibration(preset)
required = singletons.settings_manager.get_required_standards(calib.preset.mode)
return {
"success": True,
"message": "Calibration started",
"preset": calib.preset.filename,
"required_standards": [s.value for s in required],
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to start calibration")
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/calibration/add-standard")
async def add_calibration_standard(request: CalibrateStandardRequest) -> dict[str, Any]:
"""Add calibration standard from the latest sweep."""
try:
try:
standard = CalibrationStandard(request.standard)
except ValueError:
raise HTTPException(status_code=400, detail=f"Invalid calibration standard: {request.standard}")
sweep_no = singletons.settings_manager.capture_calibration_standard_from_acquisition(
standard, singletons.vna_data_acquisition_instance
)
working = singletons.settings_manager.get_current_working_calibration()
progress = working.get_progress() if working else (0, 0)
return {
"success": True,
"message": f"Added {standard.value} standard from sweep {sweep_no}",
"sweep_number": sweep_no,
"progress": f"{progress[0]}/{progress[1]}",
"is_complete": working.is_complete() if working else False,
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to add calibration standard")
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/calibration/save")
async def save_calibration(request: SaveCalibrationRequest) -> dict[str, Any]:
"""Save current working calibration set."""
try:
saved = singletons.settings_manager.save_calibration_set(request.name)
return {
"success": True,
"message": f"Calibration '{request.name}' saved successfully",
"preset": saved.preset.filename,
"standards": [s.value for s in saved.standards.keys()],
}
except Exception as exc: # noqa: BLE001
logger.error("Failed to save calibration")
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/calibration/set")
async def set_calibration(request: SetCalibrationRequest) -> dict[str, Any]:
"""Set current active calibration."""
try:
preset = None
if request.preset_filename:
presets = singletons.settings_manager.get_available_presets()
preset = next((p for p in presets if p.filename == request.preset_filename), None)
if preset is None:
raise HTTPException(status_code=404, detail=f"Preset not found: {request.preset_filename}")
singletons.settings_manager.set_current_calibration(request.name, preset)
return {"success": True, "message": f"Calibration set to '{request.name}'"}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to set calibration")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/working-calibration", response_model=WorkingCalibrationModel)
async def get_working_calibration() -> WorkingCalibrationModel:
"""Get current working calibration status."""
try:
working = singletons.settings_manager.get_current_working_calibration()
if not working:
return WorkingCalibrationModel(active=False)
completed, total = working.get_progress()
return WorkingCalibrationModel(
active=True,
preset=working.preset.filename,
progress=f"{completed}/{total}",
is_complete=working.is_complete(),
completed_standards=[s.value for s in working.standards.keys()],
missing_standards=[s.value for s in working.get_missing_standards()],
)
except Exception as exc: # noqa: BLE001
logger.error("Failed to get working calibration")
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/calibration/remove-standard")
async def remove_calibration_standard(request: RemoveStandardRequest) -> dict[str, Any]:
"""Remove calibration standard from current working set."""
try:
try:
standard = CalibrationStandard(request.standard)
except ValueError:
raise HTTPException(status_code=400, detail=f"Invalid calibration standard: {request.standard}")
singletons.settings_manager.remove_calibration_standard(standard)
working = singletons.settings_manager.get_current_working_calibration()
progress = working.get_progress() if working else (0, 0)
return {
"success": True,
"message": f"Removed {standard.value} standard",
"progress": f"{progress[0]}/{progress[1]}",
"is_complete": working.is_complete() if working else False,
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to remove calibration standard")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/calibration/current")
async def get_current_calibration() -> dict[str, Any]:
"""Get currently selected calibration details."""
try:
current = singletons.settings_manager.get_current_calibration()
if not current:
return {"active": False}
return {
"active": True,
"preset": {"filename": current.preset.filename, "mode": current.preset.mode.value},
"calibration_name": current.name,
"standards": [s.value for s in current.standards.keys()],
"is_complete": current.is_complete(),
}
except Exception as exc: # noqa: BLE001
logger.error("Failed to get current calibration")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/calibration/{calibration_name}/standards-plots")
async def get_calibration_standards_plots(
calibration_name: str,
preset_filename: str | None = None,
) -> dict[str, Any]:
"""Get magnitude plots for all standards in a calibration set."""
try:
# Resolve preset (explicit or current)
preset = None
if preset_filename:
presets = singletons.settings_manager.get_available_presets()
preset = next((p for p in presets if p.filename == preset_filename), None)
if preset is None:
raise HTTPException(status_code=404, detail=f"Preset not found: {preset_filename}")
else:
preset = singletons.settings_manager.get_current_preset()
if preset is None:
raise HTTPException(status_code=400, detail="No current preset selected")
# Resolve calibration directory (uses manager's internal layout)
calibration_manager = singletons.settings_manager.calibration_manager
calibration_dir = calibration_manager._get_preset_calibration_dir(preset) / calibration_name # noqa: SLF001
if not calibration_dir.exists():
raise HTTPException(status_code=404, detail=f"Calibration not found: {calibration_name}")
individual_plots = generate_standards_magnitude_plots(calibration_dir, preset)
return {
"calibration_name": calibration_name,
"preset": {"filename": preset.filename, "mode": preset.mode.value},
"individual_plots": individual_plots,
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to build calibration standards plots")
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/working-calibration/standards-plots")
async def get_working_calibration_standards_plots() -> dict[str, Any]:
"""Get magnitude plots for standards in the current working calibration."""
try:
working = singletons.settings_manager.get_current_working_calibration()
if not working:
raise HTTPException(status_code=404, detail="No working calibration active")
if not working.standards:
raise HTTPException(status_code=404, detail="No standards captured in working calibration")
from vna_system.core.visualization.magnitude_chart import generate_magnitude_plot_from_sweep_data
individual: dict[str, Any] = {}
standard_colors = {
"open": "#2ca02c",
"short": "#d62728",
"load": "#ff7f0e",
"through": "#1f77b4",
}
for standard, sweep in working.standards.items():
try:
fig = generate_magnitude_plot_from_sweep_data(sweep, working.preset)
if "error" not in fig and fig.get("data"):
fig["data"][0]["line"]["color"] = standard_colors.get(standard.value, "#1f77b4")
fig["data"][0]["name"] = f"{standard.value.upper()} Standard"
fig["layout"]["title"] = f"{standard.value.upper()} Standard Magnitude (Working)"
fig["raw_sweep_data"] = {
"sweep_number": sweep.sweep_number,
"timestamp": sweep.timestamp,
"total_points": sweep.total_points,
"points": sweep.points,
"file_path": None,
}
fig["frequency_info"] = {
"start_freq": working.preset.start_freq,
"stop_freq": working.preset.stop_freq,
"points": working.preset.points,
"bandwidth": working.preset.bandwidth,
}
individual[standard.value] = fig
except Exception as exc: # noqa: BLE001
individual[standard.value] = {"error": f"Failed to generate plot for {standard.value}: {exc}"}
if not individual:
raise HTTPException(status_code=404, detail="No valid plots generated for working calibration")
return {
"calibration_name": "Working Calibration",
"preset": {"filename": working.preset.filename, "mode": working.preset.mode.value},
"individual_plots": individual,
"is_working": True,
"is_complete": working.is_complete(),
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to build working calibration standards plots")
raise HTTPException(status_code=500, detail=str(exc))
# ================================================================ #
# Reference Management Endpoints
# ================================================================ #
@router.get("/references", response_model=List[ReferenceModel])
async def get_references(preset_filename: str | None = None) -> List[dict[str, Any]]:
"""Get available references for a preset."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
references = settings_manager.get_available_references(preset)
return [
{
"name": ref.name,
"timestamp": ref.timestamp.isoformat(),
"preset_filename": ref.preset_filename,
"description": ref.description,
"metadata": ref.metadata
}
for ref in references
]
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to get references", error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.get("/reference/current", response_model=ReferenceModel | None)
async def get_current_reference(preset_filename: str | None = None) -> dict[str, Any] | None:
"""Get the currently selected reference."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
current_ref = settings_manager.get_current_reference(preset)
if not current_ref:
return None
return {
"name": current_ref.name,
"timestamp": current_ref.timestamp.isoformat(),
"preset_filename": current_ref.preset_filename,
"description": current_ref.description,
"metadata": current_ref.metadata
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to get current reference", error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/reference/create")
async def create_reference(request: CreateReferenceRequest) -> dict[str, Any]:
"""Create a new reference from current sweep data."""
try:
settings_manager = singletons.settings_manager
data_acquisition = singletons.vna_data_acquisition_instance
if request.preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == request.preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{request.preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
# Create reference using the new capture method
reference_info = settings_manager.capture_reference_from_acquisition(
reference_name=request.name,
data_acquisition=data_acquisition,
description=request.description,
metadata=request.metadata,
preset=preset
)
return {
"success": True,
"message": f"Reference '{request.name}' created successfully",
"reference": {
"name": reference_info.name,
"timestamp": reference_info.timestamp.isoformat(),
"preset_filename": reference_info.preset_filename,
"description": reference_info.description,
"metadata": reference_info.metadata
}
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to create reference", name=request.name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.post("/reference/set")
async def set_current_reference(request: SetReferenceRequest) -> dict[str, Any]:
"""Set the current reference."""
try:
settings_manager = singletons.settings_manager
if request.preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == request.preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{request.preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
success = settings_manager.set_current_reference(request.name, preset)
if not success:
raise HTTPException(status_code=404, detail=f"Reference '{request.name}' not found")
return {
"success": True,
"message": f"Reference '{request.name}' set as current"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to set current reference", name=request.name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/reference/current")
async def clear_current_reference() -> dict[str, Any]:
"""Clear the current reference selection."""
try:
settings_manager = singletons.settings_manager
success = settings_manager.clear_current_reference()
if not success:
raise HTTPException(status_code=500, detail="Failed to clear current reference")
return {
"success": True,
"message": "Current reference cleared"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to clear current reference", error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/reference/{reference_name}")
async def delete_reference(reference_name: str, preset_filename: str | None = None) -> dict[str, Any]:
"""Delete a reference."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
success = settings_manager.delete_reference(reference_name, preset)
if not success:
raise HTTPException(status_code=404, detail=f"Reference '{reference_name}' not found")
return {
"success": True,
"message": f"Reference '{reference_name}' deleted successfully"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to delete reference", name=reference_name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))
@router.delete("/calibration/{calibration_name}")
async def delete_calibration(calibration_name: str, preset_filename: str | None = None) -> dict[str, Any]:
"""Delete a calibration set."""
try:
settings_manager = singletons.settings_manager
if preset_filename:
preset = next(
(p for p in settings_manager.get_available_presets() if p.filename == preset_filename),
None
)
if not preset:
raise HTTPException(status_code=404, detail=f"Preset '{preset_filename}' not found")
else:
preset = settings_manager.get_current_preset()
if not preset:
raise HTTPException(status_code=400, detail="No current preset set and no preset_filename provided")
settings_manager.delete_calibration(preset, calibration_name)
return {
"success": True,
"message": f"Calibration '{calibration_name}' deleted"
}
except HTTPException:
raise
except Exception as exc: # noqa: BLE001
logger.error("Failed to delete calibration", name=calibration_name, error=repr(exc))
raise HTTPException(status_code=500, detail=str(exc))