Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions docs/operator_guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,10 @@ This guide summarizes the live workflow for the desktop app. The original Word g
- Free-run mode captures without arming an edge trigger.
- Triggered mode arms the selected oscilloscope trigger channel.

After loading a reference MAT file, the right-side receipt shows the filename, directory, frequency coverage,
point count, whether phase data exists, whether reference correction is active, and whether the current sweep
range extends beyond the reference coverage. Out-of-range reference correction uses edge-clamped values.

## 4. Run A Sweep

1. Review hardware settings and safety limits.
Expand All @@ -38,6 +42,9 @@ This guide summarizes the live workflow for the desktop app. The original Word g
## 5. Output Files

Saving a measurement writes MAT, CSV, and TXT files. If plot figures are supplied, gain and gain-dB PNG files are also written.
After a manual save, the export receipt shows the output directory, artifact list, source, correction mode,
point count, timestamp, and whether the data came from the simulated no-hardware fixture path or a live-path run.
The receipt is an operator aid; it is not live hardware validation by itself.

The CSV columns are:

Expand Down
81 changes: 81 additions & 0 deletions src/app/application/services/calibration_receipts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

import numpy as np

from app.domain.models import AppSettings, ReferenceCurve


@dataclass(frozen=True, slots=True)
class ReferenceReceipt:
summary: str
warnings: tuple[str, ...] = ()


def build_reference_receipt(
*,
path: Path,
curve: ReferenceCurve,
settings: AppSettings | None,
calibration_enabled: bool,
) -> ReferenceReceipt:
freq = np.atleast_1d(np.asarray(curve.freq_hz, dtype=float).squeeze())
phase = None if curve.phase_deg is None else np.atleast_1d(np.asarray(curve.phase_deg, dtype=float).squeeze())

point_count = int(freq.size)
phase_present = phase is not None and phase.size > 0 and bool(np.any(np.isfinite(phase)))
coverage_text = "coverage unavailable"
warnings: list[str] = []

if point_count > 0:
finite_freq = freq[np.isfinite(freq)]
if finite_freq.size > 0:
ref_min = float(np.min(finite_freq))
ref_max = float(np.max(finite_freq))
coverage_text = f"{_format_frequency(ref_min)} - {_format_frequency(ref_max)}"
if ref_min == ref_max:
warnings.append("Reference has one frequency; all sweep points use the same reference value.")
if settings is not None:
sweep_min = min(settings.sweep.start_hz, settings.sweep.stop_hz)
sweep_max = max(settings.sweep.start_hz, settings.sweep.stop_hz)
if sweep_min < ref_min or sweep_max > ref_max:
warnings.append(
"Sweep range is outside reference coverage; out-of-range points will use edge-clamped reference values."
)
else:
warnings.append("Reference frequency data has no finite values.")
else:
warnings.append("Reference frequency data is empty.")

if settings is None:
warnings.append("Sweep settings unavailable; reference coverage was not checked against the current sweep.")

correction_text = "active" if calibration_enabled else "inactive"
run_mode_text = settings.run_mode.correction_mode.value if settings is not None else "unknown"
sweep_text = _format_sweep_range(settings) if settings is not None else "sweep unavailable"
dir_text = str(path.parent) if str(path.parent) else "."

summary = "\n".join(
[
f"Reference: {path.name}",
f"Dir: {dir_text}",
f"Coverage: {coverage_text} · {point_count} pts · phase {'yes' if phase_present else 'no'}",
f"Correction: {correction_text} · mode {run_mode_text}",
f"Sweep: {sweep_text}",
]
)
return ReferenceReceipt(summary=summary, warnings=tuple(warnings))


def _format_sweep_range(settings: AppSettings) -> str:
return f"{_format_frequency(settings.sweep.start_hz)} - {_format_frequency(settings.sweep.stop_hz)}"


def _format_frequency(freq_hz: float) -> str:
if abs(freq_hz) >= 1_000_000:
return f"{freq_hz / 1_000_000:.3g} MHz"
if abs(freq_hz) >= 1_000:
return f"{freq_hz / 1_000:.3g} kHz"
return f"{freq_hz:.3g} Hz"
66 changes: 66 additions & 0 deletions src/app/application/services/export_receipts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

from app.application.dto import SaveArtifacts
from app.domain.models import AppSettings, SweepResult


@dataclass(frozen=True, slots=True)
class ExportReceipt:
summary: str
artifacts: tuple[Path, ...]


def build_export_receipt(
*,
artifacts: SaveArtifacts,
settings: AppSettings,
result: SweepResult,
source_text: str,
fixture_badge_text: str,
saved_at: datetime | None = None,
) -> ExportReceipt:
saved_at = saved_at or datetime.now()
paths = _artifact_paths(artifacts)
artifact_names = ", ".join(path.name for path in paths)
boundary = _boundary_text(source_text=source_text, fixture_badge_text=fixture_badge_text)

summary = "\n".join(
[
f"Saved: {artifacts.mat_path.name}",
f"Dir: {artifacts.mat_path.parent}",
f"Artifacts: {artifact_names}",
(
f"Metadata: source={source_text or 'unknown'}; "
f"correction={settings.run_mode.correction_mode.value}; "
f"points={len(result.points)}; "
f"saved={saved_at.strftime('%Y-%m-%d %H:%M:%S')}"
),
f"Boundary: {boundary}",
]
)
return ExportReceipt(summary=summary, artifacts=paths)


def _artifact_paths(artifacts: SaveArtifacts) -> tuple[Path, ...]:
paths = [
artifacts.mat_path,
artifacts.csv_path,
artifacts.txt_path,
artifacts.gain_plot_path,
artifacts.db_plot_path,
]
return tuple(path for path in paths if path is not None)


def _boundary_text(*, source_text: str, fixture_badge_text: str) -> str:
source_lower = source_text.lower()
fixture_lower = fixture_badge_text.lower()
if "fixture" in source_lower or "simulated" in fixture_lower or "no hardware" in fixture_lower:
return "No hardware simulated fixture; not live hardware validation"
if "loaded measurement" in source_lower:
return "Loaded file; live hardware state not implied"
return "Live path selected; export does not prove hardware validation"
72 changes: 68 additions & 4 deletions src/app/presentation/tk/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,15 @@
format_connection_receipt,
format_scan_receipt,
)
from app.application.services.calibration_receipts import build_reference_receipt
from app.application.services.export_receipts import build_export_receipt
from app.application.services.sweep_task_runner import SweepTaskRunner
from app.application.services.connection_monitor import ConnectionMonitor
from app.application.use_cases.load_measurement import LoadMeasurementUseCase
from app.application.use_cases.load_reference import LoadReferenceUseCase
from app.application.use_cases.save_measurement import SaveMeasurementUseCase
from app.application.use_cases.settings_use_case import SettingsUseCase
from app.domain.models import InstrumentEndpoint
from app.domain.models import InstrumentEndpoint, ReferenceCurve
from app.presentation.tk import dialogs
from app.presentation.tk.app_window import AppWindow
from app.presentation.tk.mapper import settings_to_vm, vm_to_settings
Expand Down Expand Up @@ -61,6 +63,9 @@ def __init__(

self._event_queue: queue.Queue[object] = queue.Queue()
self._reference_interpolator = None
self._reference_curve: ReferenceCurve | None = None
self._reference_path: Path | None = None
self._reference_trace_tokens: list[str] = []
self._paths = paths or AppPaths.default()
self._resolve_address = resolve_address
self._connection_target_lock = threading.Lock()
Expand Down Expand Up @@ -99,6 +104,7 @@ def initialize(self) -> None:
on_figure_change=self.on_figure_change,
on_mag_phase_change=self.on_mag_phase_change,
)
self._bind_reference_receipt_traces()

try:
settings = self.settings_use_case.load()
Expand All @@ -122,6 +128,9 @@ def on_start(self) -> None:
try:
settings = vm_to_settings(self.vm)
self._ui_handler.set_live_source()
self._refresh_reference_receipt(record=True)
if bool(self.vm.calibration_enabled.get()) and self._reference_interpolator is None:
self._ui_handler.record_event("Calibration enabled but no reference loaded", level="Warning")
self._refresh_connection_targets(settings)
self._task_runner.start(
settings=settings,
Expand Down Expand Up @@ -176,9 +185,17 @@ def on_save_data(self) -> None:
settings=settings,
target=dialogs_to_target(fp, self.window),
)
self._ui_handler.set_export_saved(path_name=artifacts.mat_path.name)
receipt = build_export_receipt(
artifacts=artifacts,
settings=settings,
result=self._ui_handler.latest_result,
source_text=self.vm.data_source_text.get(),
fixture_badge_text=self.vm.fixture_badge_text.get(),
)
self._ui_handler.set_export_saved(receipt_text=receipt.summary)
dialogs.show_info(self.window, f"Saved: {artifacts.mat_path.name}")
except Exception as exc: # noqa: BLE001
self._ui_handler.record_event(f"Save data failed: {exc}", level="Warning")
dialogs.show_warning(self.window, f"Failed to save data: {exc}")

def on_load_data(self) -> None:
Expand Down Expand Up @@ -222,12 +239,19 @@ def on_load_reference(self) -> None:
return

try:
_curve, interpolator = self.load_reference_use_case.execute(str(fp))
curve, interpolator = self.load_reference_use_case.execute(str(fp))
self._reference_interpolator = interpolator
self._reference_curve = curve
self._reference_path = Path(fp)
self.vm.calibration_enabled.set(True)
self._ui_handler.set_reference_loaded(path_name=Path(fp).name)
warnings = self._refresh_reference_receipt(record=True)
if warnings:
self.vm.status_text.set("Reference loaded with coverage warning")
else:
self.vm.status_text.set("Reference loaded")
dialogs.show_info(self.window, "Reference loaded")
except Exception as exc: # noqa: BLE001
self._ui_handler.record_event(f"Reference load failed: {exc}", level="Warning")
dialogs.show_warning(self.window, f"Failed to load reference: {exc}")

def on_scan_resources(self) -> None:
Expand All @@ -237,6 +261,7 @@ def on_scan_resources(self) -> None:
self.vm.status_text.set("Resource scan completed")
except Exception as exc: # noqa: BLE001
self.vm.discovery_status_text.set(f"Resource scan failed: {exc}")
self._ui_handler.record_event(f"Resource scan failed: {exc}", level="Warning")
dialogs.show_warning(self.window, f"Resource scan failed: {exc}")

def on_test_connect(self) -> None:
Expand All @@ -248,6 +273,7 @@ def on_test_connect(self) -> None:
self.vm.status_text.set("Connection test completed")
except Exception as exc: # noqa: BLE001
self.vm.discovery_status_text.set(f"Connection test failed: {exc}")
self._ui_handler.record_event(f"Connection test failed: {exc}", level="Warning")
dialogs.show_warning(self.window, f"Connection test failed: {exc}")

def on_figure_change(self) -> None:
Expand Down Expand Up @@ -330,6 +356,44 @@ def _apply_connection_checks(self, checks) -> None:
else:
self.vm.osc_connection_text.set(text)

def _bind_reference_receipt_traces(self) -> None:
variables = [
self.vm.freq_unit,
self.vm.start_freq,
self.vm.stop_freq,
self.vm.calibration_enabled,
self.vm.correction_mode,
self.vm.trigger_mode,
]
for variable in variables:
token = variable.trace_add("write", self._on_reference_setting_changed)
self._reference_trace_tokens.append(token)

def _on_reference_setting_changed(self, *_args) -> None:
self._refresh_reference_receipt(record=False)

def _refresh_reference_receipt(self, *, record: bool) -> tuple[str, ...]:
if self._reference_curve is None or self._reference_path is None:
return ()

try:
settings = vm_to_settings(self.vm)
except Exception:
settings = None

receipt = build_reference_receipt(
path=self._reference_path,
curve=self._reference_curve,
settings=settings,
calibration_enabled=bool(self.vm.calibration_enabled.get()),
)
self._ui_handler.set_reference_loaded(
receipt_text=receipt.summary,
warnings=receipt.warnings,
record=record,
)
return receipt.warnings


def dialogs_to_target(path, window: AppWindow):
from app.application.dto import SaveTarget
Expand Down
10 changes: 8 additions & 2 deletions src/app/presentation/tk/run_panel.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,8 @@ def _build(self) -> None:
self._build_run_card(1)
self._build_export_card(2)
self._build_source_card(3)
self._build_safety_card(4)
self._build_event_card(4)
self._build_safety_card(5)

def _title(self, text: str, row: int) -> None:
label = tk.Label(
Expand Down Expand Up @@ -141,7 +142,8 @@ def _build_export_card(self, row: int) -> None:
self.btn_load_settings = ttk.Button(actions, text="Load Settings")
self.btn_load_settings.grid(row=2, column=1, sticky="ew")

self._wrapped_label(card, self._vm.export_receipt_text, 1, MUTED)
self._wrapped_label(card, self._vm.reference_receipt_text, 1, MUTED)
self._wrapped_label(card, self._vm.export_receipt_text, 2, MUTED)

def _build_source_card(self, row: int) -> None:
card = self._card("Source receipt", row)
Expand Down Expand Up @@ -173,6 +175,10 @@ def _build_safety_card(self, row: int) -> None:
tk.Label(line, text="□", bg=CARD_BG, fg=AMBER, width=2, anchor="w").pack(side=tk.LEFT)
tk.Label(line, text=item, bg=CARD_BG, fg=TEXT, anchor="w").pack(side=tk.LEFT, fill=tk.X, expand=True)

def _build_event_card(self, row: int) -> None:
card = self._card("Event history", row)
self._wrapped_label(card, self._vm.event_history_text, 0, MUTED)

def _metric(
self,
parent: tk.Misc,
Expand Down
Loading