[codex] Harden sweep cleanup follow-up

README.md

@@ -51,6 +51,7 @@ The UI and use cases do not call `src/equips.py` directly. That file is treated
 - For live instrument use:
   - supported AWG and oscilloscope models from `src/app/shared/mapping.py`
   - VISA access through `pyvisa` / `pyvisa-py`
+  - a working VISA backend for the connection type, such as NI-VISA / Keysight IO Libraries for LAN/USB/GPIB or the extra USB/GPIB libraries required by `pyvisa-py`
   - correct LAN/VISA addresses for the instruments
 
 Automated tests do not require AWG/OSC hardware.
@@ -83,12 +84,14 @@ auto-load-off-test
 python src/main.py
 ```
 
-Settings are stored at:
+Settings and auto-save data are rooted at the process working directory unless `AUTO_LOAD_OFF_TEST_ROOT` is set. From the repo root, settings are stored at:
 
 ```text
 __config__/settings.json
 ```
 
+For packaged installs or lab workstations, set `AUTO_LOAD_OFF_TEST_ROOT` to an explicit writable directory so settings and `__data__/measurement/` do not move when the app is launched from a different shell directory.
+
 ## Run Tests Without Hardware
 
 ```bash

docs/operator_guide.md

@@ -56,6 +56,7 @@ The files in `demo_data/` can be loaded through the measurement loader path to i
 
 - No resources visible: check VISA backend, LAN connectivity, USB/GPIB cable, or serial permissions.
 - Sweep fails immediately: verify model label, address, impedance/coupling combinations, and numeric settings.
+- Cleanup warning after Stop or window close: verify the AWG front-panel output state before touching the DUT or starting another sweep.
 - Flat or clipped waveform: reduce AWG amplitude or adjust oscilloscope range/offset.
 - Unexpected phase: verify reference channel, trigger mode, and cable/probe delays.
 - Save/load failure: confirm output directory permissions and supported file suffixes.

docs/safety.md

@@ -19,7 +19,9 @@ This project is not a certified production test platform. It does not replace la
 - The sweep loop checks that event between frequency points and emits `SweepStopped` with the partial result.
 - Runner shutdown signals stop and waits briefly for the worker thread before closing instrument ports.
 - AWG shutdown attempts to turn the configured output channel off before closing the port.
+- If the worker does not stop before the shutdown timeout, the runner emits `SHUTDOWN_TIMEOUT` and still attempts to turn AWG output off and close the known ports.
 - If output-off or port-close fails, the runner emits a `SweepWarning` so the UI/event log can surface the cleanup failure.
+- Treat any cleanup warning after Stop or window close as hardware-significant: verify the AWG front panel/output indicator and the DUT state before touching the setup or starting another sweep.
 
 ## Exception Behavior
 

pyproject.toml

@@ -9,24 +9,24 @@ description = "Desktop AWG/oscilloscope sweep measurement automation tool."
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
-  "numpy>=1.23",
-  "scipy>=1.10",
-  "matplotlib>=3.7",
-  "mplcursors>=0.5",
-  "pyvisa>=1.13",
-  "pyserial>=3.5",
-  "pyvisa-py>=0.7",
+  "numpy>=1.23,<3",
+  "scipy>=1.10,<2",
+  "matplotlib>=3.7,<4",
+  "mplcursors>=0.5,<1",
+  "pyvisa>=1.13,<2",
+  "pyserial>=3.5,<4",
+  "pyvisa-py>=0.7,<1",
 ]
 
 [project.scripts]
 auto-load-off-test = "main:main"
 
 [project.optional-dependencies]
 dev = [
-  "ruff>=0.4",
+  "ruff>=0.4,<1",
 ]
 build = [
-  "pyinstaller>=6",
+  "pyinstaller>=6,<7",
 ]
 
 [tool.setuptools]

requirements-dev.txt

@@ -1,3 +1,2 @@
 -r requirements.txt
-ruff>=0.4
-
+ruff>=0.4,<1

requirements.txt

@@ -1,7 +1,7 @@
-numpy>=1.23
-scipy>=1.10
-matplotlib>=3.7
-mplcursors>=0.5
-pyvisa>=1.13
-pyserial>=3.5
-pyvisa-py>=0.7
+numpy>=1.23,<3
+scipy>=1.10,<2
+matplotlib>=3.7,<4
+mplcursors>=0.5,<1
+pyvisa>=1.13,<2
+pyserial>=3.5,<4
+pyvisa-py>=0.7,<1

src/app/application/errors.py

@@ -15,3 +15,11 @@ class InstrumentAppError(ApplicationError):
 
 class PersistenceAppError(ApplicationError):
     pass
+
+
+def describe_exception(exc: BaseException) -> str:
+    message = str(exc)
+    exc_type = type(exc).__name__
+    if message:
+        return f"{exc_type}: {message}"
+    return exc_type

src/app/application/services/sweep/instrument_configurator.py

@@ -21,7 +21,7 @@ def configure(self, settings: AppSettings) -> None:
             self._awg.reset()
             self._osc.reset()
 
-        self._awg.output_on(awg_ch)
+        self._awg.output_off(awg_ch)
         self._awg.set_impedance(setup.awg_settings.impedance.value, awg_ch)
         self._awg.set_amplitude_vpp(setup.awg_settings.amplitude_vpp, awg_ch)
 

src/app/application/services/sweep/waveform_acquirer.py

@@ -29,6 +29,7 @@ def acquire(self, *, target_freq_hz: float, settings: AppSettings) -> AcquiredPo
 
         warnings: list[SweepServiceWarning] = []
         awg_ch = setup.channels.awg_ch
+        self._awg.output_off(awg_ch)
         self._awg.set_frequency(float(target_freq_hz), awg_ch)
         actual_freq = self._awg.get_frequency(awg_ch)
         if not np.isclose(actual_freq, target_freq_hz, atol=1e-3, rtol=5e-6):
@@ -58,6 +59,7 @@ def acquire(self, *, target_freq_hz: float, settings: AppSettings) -> AcquiredPo
 
         self._osc.set_timebase(window_s)
         triggered = run_mode.trigger_mode == TriggerMode.TRIGGERED
+        self._awg.output_on(awg_ch)
         self._osc.single_acquire(triggered=triggered)
 
         test_ch = setup.channels.osc_test_ch

src/app/application/services/sweep_task_runner.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 
 from app.application.dto import SaveTarget, StartSweepCommand
+from app.application.errors import describe_exception
 from app.application.events import EventEmitter, SweepFailed, SweepWarning
 from app.application.ports.instruments import InstrumentPorts, InstrumentPortsFactory
 from app.application.use_cases.save_measurement import SaveMeasurementUseCase
@@ -48,22 +49,36 @@ def start(
         if self.is_running():
             return
 
-        ports = self._ports_factory(settings.setup)
-        stop_event = threading.Event()
-        self._stop_use_case = StopSweepUseCase(stop_event=stop_event)
-
-        cmd = StartSweepCommand(
-            settings=settings,
-            calibration_enabled=calibration_enabled,
-            reference_interpolator=reference_interpolator,
-        )
-        start_use_case = self._use_case_factory(awg=ports.awg, osc=ports.osc, stop_event=stop_event)
-
-        with self._ports_lock:
-            self._ports = ports
-            self._active_awg_channel = settings.setup.channels.awg_ch
-        self._sweep_thread = threading.Thread(target=self._run_sweep, args=(start_use_case, cmd), daemon=True)
-        self._sweep_thread.start()
+        awg_channel = settings.setup.channels.awg_ch
+        ports: InstrumentPorts | None = None
+        try:
+            ports = self._ports_factory(settings.setup)
+            stop_event = threading.Event()
+            self._stop_use_case = StopSweepUseCase(stop_event=stop_event)
+
+            cmd = StartSweepCommand(
+                settings=settings,
+                calibration_enabled=calibration_enabled,
+                reference_interpolator=reference_interpolator,
+            )
+            start_use_case = self._use_case_factory(awg=ports.awg, osc=ports.osc, stop_event=stop_event)
+
+            thread = threading.Thread(target=self._run_sweep, args=(start_use_case, cmd), daemon=True)
+            with self._ports_lock:
+                self._ports = ports
+                self._active_awg_channel = awg_channel
+                self._sweep_thread = thread
+            thread.start()
+        except Exception:
+            if ports is not None:
+                with self._ports_lock:
+                    if self._ports is ports:
+                        self._ports = None
+                        self._active_awg_channel = None
+                self._close_port_set(ports=ports, awg_channel=awg_channel)
+            self._stop_use_case = None
+            self._sweep_thread = None
+            raise
 
     def stop(self) -> None:
         if self._stop_use_case is not None:
@@ -79,26 +94,39 @@ def shutdown(self, timeout: float = 2.0) -> None:
         if self.is_running():
             self._emit_warning(
                 code="SHUTDOWN_TIMEOUT",
-                message="Sweep worker did not stop before shutdown timeout; ports will close when the worker exits.",
+                message=(
+                    "Sweep worker did not stop before shutdown timeout; forcing AWG output off "
+                    "and closing ports during shutdown."
+                ),
             )
+            self._close_ports()
             return
         self._close_ports()
 
     def _run_sweep(self, start_use_case: StartSweepUseCase, cmd: StartSweepCommand) -> None:
         try:
             result = start_use_case.run(cmd, self._emitter)
-            if not result.is_empty and cmd.settings.auto_save_data:
-                target = SaveTarget(
-                    base_path=self._auto_save_dir / "measurement",
-                    include_timestamp=True,
-                    figures={},
-                )
-                self._save_measurement_use_case.execute(result=result, settings=cmd.settings, target=target)
         except Exception as exc:  # noqa: BLE001
-            self._emitter.emit(SweepFailed(error_code="SWEEP_THREAD", message=str(exc)))
+            self._emitter.emit(SweepFailed(error_code="SWEEP_THREAD", message=describe_exception(exc)))
+        else:
+            self._auto_save_if_requested(result=result, cmd=cmd)
         finally:
             self._close_ports()
 
+    def _auto_save_if_requested(self, *, result, cmd: StartSweepCommand) -> None:
+        if result.is_empty or not cmd.settings.auto_save_data:
+            return
+
+        target = SaveTarget(
+            base_path=self._auto_save_dir / "measurement",
+            include_timestamp=True,
+            figures={},
+        )
+        try:
+            self._save_measurement_use_case.execute(result=result, settings=cmd.settings, target=target)
+        except Exception as exc:  # noqa: BLE001
+            self._emit_warning(code="AUTO_SAVE_FAILED", message=describe_exception(exc))
+
     def _close_ports(self) -> None:
         with self._ports_lock:
             ports = self._ports
@@ -109,22 +137,24 @@ def _close_ports(self) -> None:
         if ports is None:
             return
 
+        self._close_port_set(ports=ports, awg_channel=awg_channel)
+
+    def _close_port_set(self, *, ports: InstrumentPorts, awg_channel: int | None) -> None:
         if awg_channel is not None:
             try:
                 ports.awg.output_off(awg_channel)
             except Exception as exc:  # noqa: BLE001
-                self._emit_warning(code="AWG_OUTPUT_OFF_FAILED", message=str(exc))
+                self._emit_warning(code="AWG_OUTPUT_OFF_FAILED", message=describe_exception(exc))
 
         try:
             ports.awg.close()
         except Exception as exc:  # noqa: BLE001
-            self._emit_warning(code="AWG_CLOSE_FAILED", message=str(exc))
+            self._emit_warning(code="AWG_CLOSE_FAILED", message=describe_exception(exc))
 
         try:
             ports.osc.close()
         except Exception as exc:  # noqa: BLE001
-            self._emit_warning(code="OSC_CLOSE_FAILED", message=str(exc))
+            self._emit_warning(code="OSC_CLOSE_FAILED", message=describe_exception(exc))
 
     def _emit_warning(self, *, code: str, message: str) -> None:
         self._emitter.emit(SweepWarning(code=code, message=message))
-

src/app/application/use_cases/start_sweep.py

@@ -5,6 +5,7 @@
 from datetime import datetime, timezone
 
 from app.application.dto import StartSweepCommand
+from app.application.errors import describe_exception
 from app.application.events import (
     EventEmitter,
     SweepCompleted,
@@ -98,8 +99,8 @@ def run(self, cmd: StartSweepCommand, emitter: EventEmitter) -> SweepResult:
             return result
 
         except ValidationError as exc:
-            emitter.emit(SweepFailed(error_code="VALIDATION", message=str(exc)))
+            emitter.emit(SweepFailed(error_code="VALIDATION", message=describe_exception(exc)))
             return SweepResult()
         except Exception as exc:  # noqa: BLE001
-            emitter.emit(SweepFailed(error_code="SWEEP_RUNTIME", message=str(exc)))
+            emitter.emit(SweepFailed(error_code="SWEEP_RUNTIME", message=describe_exception(exc)))
             return SweepResult()

src/app/domain/calibration.py

@@ -9,9 +9,9 @@
 
 
 def build_reference_interpolator(curve: ReferenceCurve) -> Callable[[np.ndarray], np.ndarray]:
-    freq = np.asarray(curve.freq_hz, dtype=float).squeeze()
-    gain_db = np.asarray(curve.gain_db, dtype=float).squeeze()
-    phase = None if curve.phase_deg is None else np.asarray(curve.phase_deg, dtype=float).squeeze()
+    freq = np.atleast_1d(np.asarray(curve.freq_hz, dtype=float).squeeze())
+    gain_db = np.atleast_1d(np.asarray(curve.gain_db, dtype=float).squeeze())
+    phase = None if curve.phase_deg is None else np.atleast_1d(np.asarray(curve.phase_deg, dtype=float).squeeze())
 
     if freq.size == 0:
         raise ValueError("Reference frequency data is empty")

src/app/infrastructure/instruments/awg_adapter.py

@@ -32,7 +32,4 @@ def get_amplitude_vpp(self, channel: int) -> float:
         return float(self._inst.get_amp(ch=channel))
 
     def close(self) -> None:
-        try:
-            self._inst.inst_close()
-        except Exception:
-            pass
+        self._inst.inst_close()

src/app/infrastructure/instruments/osc_adapter.py

@@ -54,7 +54,4 @@ def get_sample_rate(self) -> float:
         return float(self._inst.get_sample_rate())
 
     def close(self) -> None:
-        try:
-            self._inst.inst_close()
-        except Exception:
-            pass
+        self._inst.inst_close()

src/equips.py

@@ -84,14 +84,12 @@ def inst_open(self):
                     self.Inst = ResourceBase.open_VisaRM().open_resource(self.VisaAddress)
         return self.Inst
 
-    def inst_close(self):
-        if self.Inst:
-            try:
-                self.Inst.close()
-            except:
-                pass
-            finally:
-                self.Inst = None
+    def inst_close(self):
+        if self.Inst:
+            try:
+                self.Inst.close()
+            finally:
+                self.Inst = None
 
     def callback_after_open(self): pass
 

tests/test_architecture_boundaries.py

@@ -29,7 +29,15 @@ def py_files(base: Path) -> list[Path]:
 
 class ArchitectureBoundaryTests(unittest.TestCase):
     def test_domain_stays_pure(self) -> None:
-        forbidden_prefixes = ("tkinter", "pyvisa", "serial", "matplotlib", "app.infrastructure", "app.presentation")
+        forbidden_prefixes = (
+            "tkinter",
+            "pyvisa",
+            "serial",
+            "matplotlib",
+            "equips",
+            "app.infrastructure",
+            "app.presentation",
+        )
         offenders = []
         for path in py_files(SRC_APP / "domain"):
             for module in imported_modules(path):
@@ -39,7 +47,14 @@ def test_domain_stays_pure(self) -> None:
         self.assertEqual(offenders, [])
 
     def test_application_does_not_import_infrastructure_or_presentation(self) -> None:
-        forbidden_prefixes = ("app.infrastructure", "app.presentation")
+        forbidden_prefixes = (
+            "app.infrastructure",
+            "app.presentation",
+            "equips",
+            "pyvisa",
+            "serial",
+            "tkinter",
+        )
         offenders = []
         for path in py_files(SRC_APP / "application"):
             for module in imported_modules(path):
@@ -49,15 +64,15 @@ def test_application_does_not_import_infrastructure_or_presentation(self) -> Non
         self.assertEqual(offenders, [])
 
     def test_presentation_does_not_import_infrastructure(self) -> None:
+        forbidden_prefixes = ("app.infrastructure", "equips", "pyvisa", "serial")
         offenders = []
         for path in py_files(SRC_APP / "presentation"):
             for module in imported_modules(path):
-                if module.startswith("app.infrastructure"):
+                if module.startswith(forbidden_prefixes):
                     offenders.append((path.relative_to(PROJECT_ROOT), module))
 
         self.assertEqual(offenders, [])
 
 
 if __name__ == "__main__":
     unittest.main()
-