diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3a2a78a..d3bbb45 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,16 +4,85 @@ on: push: branches: [main] pull_request: - branches: [main] + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true jobs: test: + name: Tests (${{ matrix.os }}, Python ${{ matrix.python-version }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + include: + - os: ubuntu-latest + python-version: "3.10" + - os: ubuntu-latest + python-version: "3.13" + - os: macos-latest + python-version: "3.11" + - os: windows-latest + python-version: "3.11" + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: python -m pip install --upgrade pip + - run: python -m pip install -e . + - run: python -m unittest discover -s tests + env: + PYTHONPATH: src + + lint: + name: Ruff runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 + - uses: actions/checkout@v7 + - uses: actions/setup-python@v6 with: python-version: "3.11" + cache: pip - run: python -m pip install --upgrade pip - - run: python -m pip install -r requirements.txt - - run: PYTHONPATH=src python -m unittest discover -s tests + - run: python -m pip install -e ".[dev]" + - run: python -m ruff check src/app tests scripts + + windows-package: + name: Windows one-folder package smoke + runs-on: windows-latest + needs: [test, lint] + timeout-minutes: 30 + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v6 + with: + python-version: "3.11" + cache: pip + - name: Build and run no-hardware package smoke + shell: powershell + run: >- + .\scripts\build_windows_onefolder.ps1 + -RuntimeRoot "$env:RUNNER_TEMP\AutoLoadOffTestRuntime" + - name: Upload smoke receipt + if: always() + uses: actions/upload-artifact@v7 + with: + name: windows-package-smoke-receipt + path: ${{ runner.temp }}/AutoLoadOffTestRuntime/__data__/package_smoke_receipt.json + if-no-files-found: warn + retention-days: 14 + - name: Upload one-folder package + if: github.event_name != 'pull_request' + uses: actions/upload-artifact@v7 + with: + name: AutoLoadOffTest-windows-onefolder + path: dist/AutoLoadOffTest + if-no-files-found: error + retention-days: 14 diff --git a/README.md b/README.md index 64895cb..e10f1e3 100644 --- a/README.md +++ b/README.md @@ -15,14 +15,14 @@ It turns a repetitive manual lab workflow into a layered application: ## Evidence Status -This table is intentionally conservative. Update it when draft work merges or live bench validation becomes available. +This table is intentionally conservative. See the detailed [validation matrix](docs/validation_matrix.md). -| Surface | Public status | Safe claim | +| Surface | Inspectable evidence | Boundary | | --- | --- | --- | -| `main` branch | Layered Tkinter desktop app with hardware-free tests, deterministic fixture demo, measurement export, operator guide, architecture notes, and safety notes. | Software engineering maturity for a lab-automation workflow; hardware-free validation only. | -| [Draft PR #6](https://github.com/lishehao/auto-load-off-test/pull/6) | Draft capability/discovery/packaging sprint with public CI success on the PR head. Not merged into `main` yet. | In-review follow-up work for model capability profiles, mockable discovery/test-connect, and Windows packaging notes. | -| [Draft PR #7](https://github.com/lishehao/auto-load-off-test/pull/7) | Draft calibration/export workflow receipt sprint stacked on PR #6. Main-target CI may not run until it is retargeted. Not merged into `main` yet. | In-review UI polish for reference coverage receipts, export artifact receipts, and non-modal workflow warnings. | -| Live instruments | No current public evidence of live AWG/oscilloscope validation. | Do not claim live instrument validation; use the project as hardware-free software and workflow evidence. | +| Hardware-free core | Cross-platform unit tests, fake instrument ports, capability-aware preflight, strict measurement/reference validation, and a deterministic fixture/reference/correction/export/reload workflow. | Validates software behavior and data contracts without instruments. | +| Operator console | Real Tkinter window capture with 72-point replay, log Bode display, progress, source/safety receipts, and MAT/CSV/TXT export. | The replay is simulated and explicitly labeled; production instrument adapters are not used. | +| Windows distribution | PyInstaller one-folder build plus an automated packaged fixture/export/reload smoke and machine-readable receipt. | Does not install VISA drivers, prove Windows GUI rendering on every host, or validate connected instruments. | +| Live instruments | Capability profiles and production adapters are present, but there is no current public bench-validation record. | Do not claim live AWG/oscilloscope, metrology, or production-system validation. | ## Demo @@ -30,14 +30,16 @@ This table is intentionally conservative. Update it when draft work merges or li [Watch the point-by-point operator console demo on YouTube](https://youtu.be/fYokRzNnm84) -This capture shows the real Tkinter operator console replaying a deterministic 72-point fixture -point by point. It demonstrates the UI, plotting, progress/status updates, source receipt, and export -workflow without connected instruments. +The published YouTube walkthrough shows the real Tkinter operator console replaying a deterministic 72-point +fixture point by point. The repository poster and local MP4 are a newer recapture of the same workflow; they +also show the log-frequency gain/phase display, progress and latest-frequency updates, neutral `AWG/OSC not used` +status, the loaded 72-point reference/coverage receipt, and an export receipt. It is labeled `No hardware - simulated fixture`: the production AWG/oscilloscope adapters are not used in this demo, and it is not live hardware validation. -For offline review, the same capture is available as a [local MP4 fallback](docs/images/auto-load-off-test-point-replay-demo.mp4). +For offline review, the recaptured current-UI artifact is available as a +[local MP4 fallback](docs/images/auto-load-off-test-point-replay-demo.mp4). ## Why It Exists @@ -53,8 +55,9 @@ src/ runtime/ runtime paths and environment helpers presentation/tk/ Tkinter UI and plotting application/ use cases, DTOs, events, ports - domain/ pure models, validation, sweep math, DSP - infrastructure/ instrument adapters and persistence + domain/ models, capability profiles, validation, sweep math, DSP + infrastructure/ adapter registry, discovery, instrument IO, persistence + demo/ deterministic no-hardware fixture and package smoke equips.py legacy vendor/instrument compatibility layer ``` @@ -68,14 +71,16 @@ flowchart LR APP --> PERSIST["Settings + Measurement Persistence"] ``` -The UI and use cases do not call `src/equips.py` directly. That file is treated as a legacy vendor compatibility layer and is wrapped by infrastructure adapters. +The UI and use cases do not call `src/equips.py` directly. That file is treated as a legacy vendor compatibility +layer and is wrapped by registered infrastructure adapters. Supported model metadata comes from the capability +registry rather than UI string dispatch. ## Requirements - Python 3.10 or newer - Tkinter, usually included with the Python installer on macOS/Windows - For live instrument use: - - supported AWG and oscilloscope models from `src/app/shared/mapping.py` + - a model with a registered capability profile and production adapter - 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 @@ -124,7 +129,10 @@ For packaged installs or lab workstations, set `AUTO_LOAD_OFF_TEST_ROOT` to an e PYTHONPATH=src python -m unittest discover -s tests ``` -The test suite uses pure domain tests and mocked instrument ports. It covers sweep generation, signal processing, settings serialization, measurement I/O, start-sweep event flow, and the sweep task runner. +The suite covers sweep generation, DSP, capability validation, adapter/discovery fakes, settings serialization, +strict measurement/reference schemas, deterministic calibration/export round trips, UI receipt state, task-runner +cleanup, and the no-hardware package smoke. CI runs the suite on Linux, macOS, and Windows and also builds the +Windows one-folder artifact. ## Output Files @@ -158,6 +166,7 @@ See [docs/safety.md](docs/safety.md) for stop/shutdown behavior and hardware ass ## Documentation - [Architecture](docs/architecture.md) +- [Validation Matrix](docs/validation_matrix.md) - [Operator Guide](docs/operator_guide.md) - [Safety Notes](docs/safety.md) - [Extending The Application](docs/extending.md) @@ -170,14 +179,19 @@ See [docs/safety.md](docs/safety.md) for stop/shutdown behavior and hardware ass ## Evidence Map - Architecture and code boundaries: [docs/architecture.md](docs/architecture.md) +- Hardware-free vs live validation boundary: [docs/validation_matrix.md](docs/validation_matrix.md) - Operator workflow: [docs/operator_guide.md](docs/operator_guide.md) - Hardware and safety boundary: [docs/safety.md](docs/safety.md) - No-hardware fixture/demo boundary: [docs/hyperframe_demo.md](docs/hyperframe_demo.md) - Deterministic demo data: [demo_data/README.md](demo_data/README.md) +- End-to-end fixture correction/export test: [tests/test_hardware_free_workflow.py](tests/test_hardware_free_workflow.py) +- Packaged no-hardware smoke: [src/app/demo/package_smoke.py](src/app/demo/package_smoke.py) - CI workflow: [.github/workflows/ci.yml](.github/workflows/ci.yml) ## Project Status The refactored app is local, single-process, and hardware-adapter based. Its strongest engineering signal is the separation between UI, use-case orchestration, pure domain logic, persistence, and instrument side effects. -Current public validation is hardware-free: unit tests, mocked/fake instrument paths, deterministic fixture loading/replay, export round trips, and documentation checks. Live hardware validation remains future work and should not be claimed from this repository alone. +Current public validation is hardware-free: cross-platform tests, mocked/fake instrument paths, strict data +contracts, deterministic fixture correction/replay, export round trips, a real Tk UI capture, and a packaged +Windows smoke. Live hardware validation remains future work and should not be claimed from this repository alone. diff --git a/demo_data/README.md b/demo_data/README.md index c2e177b..135b327 100644 --- a/demo_data/README.md +++ b/demo_data/README.md @@ -46,6 +46,10 @@ These files are sample data for review and local testing. They are not a substit The Hyperframe fixture is explicitly simulated no-hardware data. Do not describe it as a live hardware validation run. +`tests/test_hardware_free_workflow.py` uses the measurement/reference pair to reconstruct corrected gain and phase, +compare against checked-in expected arrays, export MAT/CSV/TXT, and reload MAT/CSV. This validates the deterministic +software workflow, not the physical plausibility or metrological accuracy of a connected bench. + Regenerate the deterministic Hyperframe fixture with: ```bash diff --git a/docs/architecture.md b/docs/architecture.md index d815f16..1c7c4dd 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -24,7 +24,8 @@ flowchart LR - Use-case orchestration for start/stop sweep, save/load, reference loading, and settings. - Emits typed events for UI; no Tk widgets or message boxes. - `app/domain` - - Pure dataclasses, enums, instrument capability profiles, validation, sweep generation, DSP, calibration, and export array shaping. + - Pure dataclasses, enums, instrument capability profiles, capability preflight, strict data validation, + sweep generation, DSP, calibration, plot-scale selection, and export array shaping. - `app/infrastructure` - Adapter registry and wrappers around `src/equips.py`. - JSON settings and MAT/CSV/TXT persistence. @@ -61,9 +62,10 @@ Forbidden: ## Instrument Access -- Instrument model and address resolution go through `equips_factory`. - Supported model metadata is declared in `domain/instrument_capabilities.py`. -- Adapter construction goes through the explicit infrastructure adapter registry. +- Adapter construction goes through `infrastructure/instruments/adapter_registry.py`; unsupported model/role + combinations fail before the legacy vendor layer is entered. +- Address resolution remains isolated in infrastructure and is injected into the controller/discovery service. - AWG and OSC commands are executed through `AwgPort` and `OscPort` adapters. - Connection scanning is provided by `PyVisaResourceScanner`, `ConnectionMonitor`, and the discovery/test-connect service. Test-connect uses short `*IDN?` probes and does not start a sweep. @@ -77,6 +79,29 @@ Forbidden: Runtime locations are centralized through `AppPaths` in `app/runtime/paths.py`. +Measurement and reference loaders normalize data through `domain/data_validation.py`. Frequencies must be finite, +positive, unique, and strictly increasing; gain arrays must be finite and aligned; phase may be absent but cannot +contain infinity. Export validates the `SweepResult` again before writing. + +## Hardware-Free Evidence Flow + +```mermaid +flowchart LR + FIXTURE["Deterministic fixture"] --> LOAD["Strict measurement loader"] + REF["Reference fixture"] --> CAL["Reference interpolator"] + LOAD --> CAL + CAL --> EXPECTED["Expected corrected gain and phase"] + EXPECTED --> EXPORT["MAT / CSV / TXT exporter"] + EXPORT --> RELOAD["MAT / CSV reload checks"] + RELOAD --> RECEIPT["Source and no-hardware receipt"] +``` + +The package smoke follows a smaller bundled-resource path: fixture/reference load, interpolation, export, reload, +and a JSON receipt. It intentionally does not initialize Tk, scan VISA resources, or construct production adapters. + ## Test Strategy -The automated tests stay hardware-free by using pure domain tests and fake instrument ports. Live instrument verification remains a manual/operator workflow. +The automated suite stays hardware-free through pure domain tests, fake ports, fake scanners/identity probes, +temporary export directories, and deterministic fixtures. CI exercises the suite on Linux, macOS, and Windows; +Windows additionally builds and runs the PyInstaller smoke. See [validation_matrix.md](validation_matrix.md) for the +exact claim boundary. Live instrument verification remains a separate future bench workflow. diff --git a/docs/case_study.md b/docs/case_study.md index cacfd1f..ddc1a14 100644 --- a/docs/case_study.md +++ b/docs/case_study.md @@ -2,42 +2,70 @@ ## Problem -Manual AWG/oscilloscope sweep measurement is repetitive and error-prone. An operator must configure generator output, oscilloscope channels, trigger mode, acquisition timing, calibration/reference behavior, and data export for each run. +Manual AWG/oscilloscope sweep measurement is repetitive and easy to misconfigure. An operator must coordinate +generator output, oscilloscope channels, trigger mode, acquisition timing, reference correction, progress/stop +behavior, and export for every run. The original implementation also coupled UI choices to a large vendor-driver +module, which made software changes hard to verify without the lab bench. ## Constraints -- The application controls physical instruments through VISA/LAN/serial paths. -- The UI must stay responsive while long sweeps run. -- Sweep math and signal processing should be testable without hardware. -- Instrument-specific commands should be isolated from application logic. -- Output data should be usable in analysis tools through MAT/CSV/TXT files. +- Physical instruments use VISA/LAN/serial paths and model-specific commands. +- Long sweeps must not block the Tk main thread. +- The available development environment has no live AWG/oscilloscope validation path. +- Existing low-level behavior should not be broadly rewritten without physical regression testing. +- MAT/CSV/TXT data must remain usable outside the application. +- Demo evidence must not make simulated data look like a live measurement. -## Architecture +## Engineering Decisions -The refactor separates the workflow into four main layers: +### Isolate, do not rewrite, the legacy driver -- `presentation/tk`: Tkinter controls, dialogs, event handling, and plots. -- `application`: use cases, events, DTOs, and ports. -- `domain`: settings models, validation, sweep generation, signal processing, calibration, and export shaping. -- `infrastructure`: instrument adapters, resource scanning, settings persistence, and measurement IO. +`src/equips.py` remains a vendor compatibility layer. Application use cases depend on `AwgPort` and `OscPort`; +registered infrastructure adapters are the only production path into the legacy module. This reduces coupling while +avoiding an unverified SCPI rewrite. -The legacy `src/equips.py` driver file remains as a vendor compatibility layer and is wrapped by infrastructure adapters. +### Separate model capabilities from construction -## Testing Strategy +Current models have explicit capability profiles for role, channels, known ranges, coupling/impedance/trigger modes, +transports, timeouts, validation status, and safety notes. A separate adapter registry resolves model/role pairs. +The UI consumes registry-backed model lists, and unsupported selections fail clearly. -The automated tests avoid physical instruments by using: +### Make data contracts strict at every boundary -- pure tests for sweep generation, signal processing, auto range, and serialization -- fake AWG/OSC ports for the start-sweep use case -- temporary directories for measurement export/load round trips -- task-runner tests around threading, auto-save, cleanup, and warnings +Measurement/reference loaders reject invalid frequencies, mismatched arrays, and non-finite values instead of +sorting, deduplicating, or filling missing gain silently. Export validates again and preserves source, correction +mode, point count, timestamp, and the simulated/live boundary. -This keeps the core behavior reviewable on any development machine. +### Treat source state as an operator concept -## Output +The console distinguishes `live`, `loaded`, and `fixture` states. Fixture replay switches to gain-dB/phase on a log +axis, shows `AWG/OSC not used`, and keeps `No hardware - simulated fixture` visible. Reference coverage, export +artifacts, warnings, and safety checks remain receipts rather than transient dialogs. -The app exports measurement data as MAT, CSV, and TXT files. Plot PNGs can be saved when the UI provides figure handles. +### Build evidence that does not require hardware + +The deterministic fixture contains raw, reference, and expected corrected gain/phase arrays. The end-to-end test +reconstructs correction, compares exact expected curves, exports MAT/CSV/TXT, reloads MAT/CSV, and checks metadata. +The Windows package smoke exercises bundled resources and the same persistence path without opening Tk or VISA. + +## Result + +- Layered presentation, application, domain, and infrastructure boundaries. +- Responsive event-driven sweep updates with stop/cleanup warnings. +- Registry-backed capability validation and mockable resource discovery/test-connect. +- Strict reference/measurement IO and deterministic correction/export evidence. +- A single-screen operator console with Bode plotting, source receipts, event history, and scroll-safe side panels. +- Cross-platform tests plus a Windows PyInstaller one-folder smoke. +- A reproducible real-Tk point replay capture with explicit no-hardware labeling. + +## Limitations + +This work proves software structure and hardware-free workflows. It does not prove real VISA enumeration, model +firmware compatibility, electrical shutdown timing, calibration uncertainty, code signing, or safe DUT operation. +Those claims require a documented physical bench matrix and remain intentionally out of scope. ## What This Demonstrates -This project demonstrates real-world engineering in a physical-system context: separating hardware side effects from testable logic, preserving a practical desktop workflow, and improving maintainability without pretending the tool is a certified lab platform. +The project is supporting evidence for engineering judgment in a physical-system context: preserving uncertain +hardware behavior behind adapters, making the rest of the system testable, surfacing operator safety state, and +documenting exactly where evidence stops. diff --git a/docs/extending.md b/docs/extending.md index 6adbc45..72eb1ba 100644 --- a/docs/extending.md +++ b/docs/extending.md @@ -14,7 +14,8 @@ This project is intentionally structured around extension seams rather than dire - Tk controller - runtime paths -`src/main.py` should stay thin. If the app later gains CLI, scripted, or simulated run modes, add a new composition function instead of pushing more wiring into UI classes. +`src/main.py` stays thin: it dispatches either the desktop composition root or the isolated `--package-smoke` path. +Additional modes should receive their own composition function instead of pushing wiring into UI classes. ## Runtime Paths @@ -52,7 +53,8 @@ For a new measurement format: 1. Add a loader/exporter implementation in infrastructure. 2. Keep `SweepResult` and `AppSettings` as the domain boundary. 3. Add round-trip tests with temporary directories. -4. Do not put file dialogs or Tk concerns in persistence code. +4. Route arrays through strict domain validation; do not silently sort, deduplicate, or fill invalid measurement data. +5. Do not put file dialogs or Tk concerns in persistence code. ## Adding A New UI Field diff --git a/docs/hyperframe_capture_plan.md b/docs/hyperframe_capture_plan.md index 5637838..51ae263 100644 --- a/docs/hyperframe_capture_plan.md +++ b/docs/hyperframe_capture_plan.md @@ -1,7 +1,6 @@ # Hyperframe Fixture Replay Capture Plan -This plan captures the deterministic fixture replay first. It does not add an app-level demo-mode banner -and does not imply live hardware validation. +This plan documents the reproducible deterministic fixture replay. It does not imply live hardware validation. Required visible label in the capture: @@ -15,14 +14,18 @@ or: Simulated no-hardware demo fixture ``` -Keep the label small and persistent, for example in the lower-left corner or as a compact title overlay. +The real Tk operator console owns this label; no external overlay is required. ## Current Tooling Status -No `hyperframe` CLI or repo-local capture script is available in this workspace. The capture itself should -be done in the coordinator's Hyperframe environment or by the person running the desktop capture. +The repository contains a real-window capture path: -Available local inputs: +- `scripts/capture_operator_console_point_replay.py`: primary 0/72 -> 72/72 replay. +- `scripts/capture_operator_console_demo.py`: shorter immediate-load capture helper and shared macOS capture code. +- `docs/images/auto-load-off-test-point-replay-demo.mp4`: current primary video. +- `docs/images/auto-load-off-test-point-replay-demo.png`: current poster. + +The point-replay helper uses only these deterministic local inputs: - `demo_data/hyperframe_simulated_fixture.mat` - `demo_data/hyperframe_simulated_fixture.csv` @@ -32,21 +35,21 @@ Available local inputs: ## Capture Storyboard -1. Setup frame - - Show the app purpose in one line: AWG/oscilloscope sweep measurement automation. +1. Ready frame + - Show the empty gain-dB/phase workbench at 0/72. - Show the persistent label: `No hardware - simulated fixture`. - - Show the fixture source: `source=mock_fixture`. + - Show `AWG/OSC not used` rather than offline/connected. 2. Sweep configuration frame - Show a conservative sweep setup: 1 kHz to 1 MHz, log-spaced fixture points. - Show correction mode as dual and trigger mode as triggered. + - Load the matching deterministic reference through the application use case and show its coverage receipt. - Keep the wording clear that this is fixture replay, not connected instruments. 3. Fixture replay frame - - Use `Load Demo Fixture` in the UI, or load `demo_data/hyperframe_simulated_fixture.mat` - through the normal load-measurement path. - - Show the plot after load. - - If Hyperframe supports animation, reveal points progressively from the fixture CSV. + - Activate `Load Demo Fixture` in the real Tk window. + - Feed each point through the existing result/view-model plotting path. + - Show progress, point count, and latest frequency changing with the curve. 4. Gain and phase frame - Show gain dB with a plausible roll-off and mild noise. @@ -54,15 +57,15 @@ Available local inputs: - Optional: show raw/reference/corrected fields from the CSV as a small data callout. 5. Export/data frame - - Show MAT/CSV/TXT artifacts already present in `demo_data/`. - - Highlight that the same shape is accepted by the existing loader/export paths. + - Run a real MAT/CSV/TXT export into the capture's temporary directory. + - Show the resulting artifact receipt without presenting it as a hardware measurement. -6. Evidence frame - - Show: no-hardware tests, architecture boundaries, and live hardware validation boundary. - - Recommended copy: +6. README context outside the video + - Link the validation matrix and architecture notes next to the capture. + - Keep the supporting copy concise: `Core sweep math and persistence are testable without instruments; live hardware validation remains manual.` -## Exact Manual Capture Steps +## Exact Capture Steps 1. Regenerate fixture data if needed: @@ -70,48 +73,31 @@ Available local inputs: python scripts/generate_hyperframe_fixture.py ``` -2. Start the desktop app only in the capture environment: +2. Install the macOS capture-only dependencies: ```bash - python src/main.py - ``` - -3. In the UI, choose `Load Demo Fixture`. - - If that action is unavailable in an older build, choose the load-measurement action and open: - - ```text - demo_data/hyperframe_simulated_fixture.mat + python -m pip install -e ".[capture]" ``` -4. Capture the loaded plot and settings area. - -5. In Hyperframe, add a small persistent label: +3. Run the point replay: - ```text - No hardware - simulated fixture - ``` - -6. Add one short data callout using `demo_data/hyperframe_simulated_fixture_metadata.json`: - - ```text - source=mock_fixture; not live hardware validation + ```bash + PYTHONPATH=src python scripts/capture_operator_console_point_replay.py ``` -7. Add one short engineering callout: +4. Inspect early, middle, and final frames. Confirm 0/few points, a partial curve, 72/72, source badge, neutral + hardware state, reference coverage receipt, and the final export receipt. - ```text - Hardware side effects are isolated behind adapters; fixture replay exercises loader, plotting, and export shape. - ``` +5. Verify the poster/MP4 paths and keep the YouTube description explicit about simulated no-hardware data. ## Do Not Claim - Do not say the fixture is a live AWG/oscilloscope run. - Do not say the demo validates connected hardware. -- Do not add an app-level demo-mode banner before the first capture. - Do not edit production instrument adapters for this capture. +- Do not capture an unverified desktop region; use the resolved Tk CGWindowID. -## If Hyperframe Automation Is Later Added +## Implementation Boundary -Prefer a separate capture helper that consumes the existing fixture files. Keep it outside production -instrument code, and preserve the same visible no-hardware label in every exported clip. +Capture automation remains under `scripts/`. It consumes checked-in fixture files and updates the existing +Tk view-model/plot path. It never constructs production instrument ports or presents the temporary export as live data. diff --git a/docs/hyperframe_demo.md b/docs/hyperframe_demo.md index fa0ab38..b087f15 100644 --- a/docs/hyperframe_demo.md +++ b/docs/hyperframe_demo.md @@ -14,14 +14,18 @@ or: Simulated no-hardware demo fixture ``` -## What To Show +## Current Capture -1. Configure a sweep in the desktop UI with conservative AWG/oscilloscope settings. -2. Use the UI's `Load Demo Fixture` action, or manually load `demo_data/hyperframe_simulated_fixture.mat`. -3. Show the plotted gain and phase response. -4. Show the exported CSV/MAT/TXT fields, including raw, reference, and corrected values. -5. Close with a short architecture/testing frame: - UI -> application use case -> domain DSP -> persistence/instrument ports. +The checked-in point replay starts at 0/72, feeds one deterministic point per frame, and ends at 72/72 with: + +- gain dB and phase on a log-frequency axis +- latest frequency and progress updates +- `AWG/OSC not used` state +- the loaded 72-point reference, coverage, phase, and dual-correction receipt +- source and no-hardware validation receipts +- a real temporary MAT/CSV/TXT export receipt + +The capture is the actual Tkinter window, not a concept mockup or browser recreation. For the first fixture-replay capture, use the concrete storyboard in [`docs/hyperframe_capture_plan.md`](hyperframe_capture_plan.md). @@ -45,7 +49,10 @@ oscilloscope, VISA/LAN access, DUT limits, and operator safety checks. ## Regenerate ```bash -python scripts/generate_hyperframe_fixture.py +python -m pip install -e ".[capture]" +PYTHONPATH=src python scripts/capture_operator_console_point_replay.py ``` -The generator uses a fixed seed so the fixture remains reproducible across machines. +On macOS, the helper resolves the Tk CGWindowID through Quartz, rejects suspicious partial frames, flattens window +alpha before H.264 encoding, and falls back to a bundled ffmpeg when the system executable is unusable. Fixture data +itself can still be regenerated with `python scripts/generate_hyperframe_fixture.py`; the generator uses a fixed seed. diff --git a/docs/images/README.md b/docs/images/README.md index ec87c0d..e3b62af 100644 --- a/docs/images/README.md +++ b/docs/images/README.md @@ -1,15 +1,12 @@ # Screenshot Capture Notes -Expected portfolio screenshots: +Current review assets: -- `main_ui.png`: the configured desktop app before a sweep. - `sweep_result.png`: a completed or loaded sweep result. The current file is generated from `demo_data/Demo(2).mat`. -- `auto-load-off-test-demo-capture.mp4`: real Tk operator console capture showing the no-hardware - Hyperframe fixture load flow. -- `auto-load-off-test-demo-capture.png`: poster frame from that real UI capture. - `auto-load-off-test-point-replay-demo.mp4`: real Tk operator console capture showing the - deterministic no-hardware fixture replayed point by point. + deterministic no-hardware fixture from 0/72 through 72/72, a loaded reference receipt, and a temporary export receipt. - `auto-load-off-test-point-replay-demo.png`: poster frame from the point-by-point replay capture. +- `auto-load-off-test-demo-capture.mp4` and `.png`: older immediate-load capture retained for history/fallback. Capture these from the real Tk desktop app. Do not replace them with generated mockups, because the value of this project is that it controls a real lab workflow. @@ -17,20 +14,22 @@ For Hyperframe application-material capture without instruments, use a visible l `No hardware - simulated fixture` when showing `demo_data/hyperframe_simulated_fixture.*`. This demonstrates the UI/data workflow but must not be described as live hardware validation. -The reproducible local capture helper is: +Install capture-only dependencies and run the primary helper: ```bash -PYTHONPATH=src python scripts/capture_operator_console_demo.py +python -m pip install -e ".[capture]" +PYTHONPATH=src python scripts/capture_operator_console_point_replay.py ``` -It starts the real Tk app, loads the deterministic fixture through the UI/controller path, captures -the Tk window with macOS `screencapture`, and writes the mp4/poster files above. +It starts the real Tk app at 1440x810, applies the fixture's 1 kHz-1 MHz/72-point/log settings, loads the matching +reference through the application use case, and captures only the resolved Tk CGWindowID. Quartz is preferred; +`screencapture -l` is the precise-window fallback. Every PNG is flattened to RGB and rejected/retried if it looks +like a partial black frame. -For the point-by-point replay capture: +The shorter immediate-load helper remains available: ```bash -PYTHONPATH=src python scripts/capture_operator_console_point_replay.py +PYTHONPATH=src python scripts/capture_operator_console_demo.py ``` -That helper keeps the replay inside the demo/capture boundary: it feeds partial fixture results to the -existing Tk plot/view-model path and never calls production instrument adapters. +Both helpers stay inside the demo/capture boundary and never call production instrument adapters. diff --git a/docs/images/auto-load-off-test-point-replay-demo.mp4 b/docs/images/auto-load-off-test-point-replay-demo.mp4 index 00c52c1..82295b4 100644 Binary files a/docs/images/auto-load-off-test-point-replay-demo.mp4 and b/docs/images/auto-load-off-test-point-replay-demo.mp4 differ diff --git a/docs/images/auto-load-off-test-point-replay-demo.png b/docs/images/auto-load-off-test-point-replay-demo.png index 33481da..7f76d9a 100644 Binary files a/docs/images/auto-load-off-test-point-replay-demo.png and b/docs/images/auto-load-off-test-point-replay-demo.png differ diff --git a/docs/operator_guide.md b/docs/operator_guide.md index 4283659..91a4cc1 100644 --- a/docs/operator_guide.md +++ b/docs/operator_guide.md @@ -1,6 +1,7 @@ # Operator Guide -This guide summarizes the live workflow for the desktop app. The original Word guide in `UserGuide/` can remain as a detailed operator artifact, but this Markdown version is readable directly on GitHub. +This guide summarizes live, loaded-file, and simulated-fixture workflows for the desktop app. The original Word guide +in `UserGuide/` can remain as a detailed operator artifact, but this Markdown version is readable directly on GitHub. ## 1. Connect Instruments @@ -10,6 +11,12 @@ This guide summarizes the live workflow for the desktop app. The original Word g 4. For triggered operation, connect or select the trigger channel. 5. Confirm VISA/LAN visibility with the instrument scanner or external VISA tooling. +Use `Scan Resources` to list addresses and `Test Connect` to issue the short identity probe. A connected receipt is +useful setup evidence, but a successful identity query is not a sweep or safety validation. + +For a no-hardware review, use `Load Demo Fixture`. The UI changes the source state to fixture, labels the run +`No hardware - simulated fixture`, and marks both instruments `not used`. + ## 2. Configure Sweep Parameters - Start/stop frequency define the sweep range. @@ -18,6 +25,8 @@ This guide summarizes the live workflow for the desktop app. The original Word g - AWG amplitude is configured in Vpp. - Oscilloscope range and offset define the vertical acquisition window. - Coupling and impedance should match the probe, DUT, and measurement setup. +- The plot X-axis control supports `auto`, `linear`, and `log`. Fixture replay defaults to a log-frequency gain-dB and + phase view; an empty requested log plot remains safe until the first positive frequency arrives. ## 3. Choose Correction And Trigger Mode @@ -48,6 +57,9 @@ The receipt is an operator aid; it is not live hardware validation by itself. The CSV columns are: +- `source` +- `validation_boundary` +- `correction_mode` - `freq_hz` - `gain_linear` - `gain_db` @@ -68,11 +80,12 @@ The files in `demo_data/` can be loaded through the measurement loader path to i - Unexpected phase: verify reference channel, trigger mode, and cable/probe delays. - Save/load failure: confirm output directory permissions and supported file suffixes. -## 8. Screenshots +## 8. Review Artifacts -For portfolio documentation, capture: +The current primary review artifacts are: -- `docs/images/main_ui.png`: app configured before a sweep. -- `docs/images/sweep_result.png`: completed sweep with plotted result. +- `docs/images/auto-load-off-test-point-replay-demo.png`: current real-Tk poster. +- `docs/images/auto-load-off-test-point-replay-demo.mp4`: current point-by-point replay. +- `docs/images/sweep_result.png`: older loaded-data plot example. Screenshots should be captured from the real desktop app rather than mocked or generated images. diff --git a/docs/packaging.md b/docs/packaging.md index a768b59..c66596c 100644 --- a/docs/packaging.md +++ b/docs/packaging.md @@ -16,7 +16,8 @@ powershell -ExecutionPolicy Bypass -File scripts/build_windows_onefolder.ps1 ``` The script installs the local package with the optional `build` dependency and -then runs: +then runs PyInstaller and an automated no-hardware smoke against the packaged +executable: ```powershell python -m PyInstaller packaging/pyinstaller/auto_load_off_test_onefolder.spec --clean --noconfirm @@ -28,6 +29,18 @@ Expected output: dist/AutoLoadOffTest/AutoLoadOffTest.exe ``` +The smoke invokes `AutoLoadOffTest.exe --package-smoke`. It loads the bundled +72-point fixture and reference, exercises interpolation, exports MAT/CSV/TXT, +reloads MAT/CSV, and writes this machine-readable receipt under the configured +runtime root: + +```text +__data__/package_smoke_receipt.json +``` + +The receipt explicitly records `live_hardware_used: false`. Use `-SkipSmoke` +only when diagnosing the build itself. + ## External Prerequisites The package does not bundle lab driver runtimes. A live-hardware workstation @@ -55,9 +68,19 @@ The app writes: under that root. -## No-Hardware Packaging Smoke +## No-Hardware Packaging Validation + +CI runs the Python test suite on Linux, macOS, and Windows, then builds the +Windows one-folder distribution and executes the same packaged smoke. Mainline +CI uploads the one-folder build as a short-retention workflow artifact; pull +requests upload the smoke receipt only. -Before using a packaged artifact as portfolio/demo evidence: +The automated smoke proves that bundled Python dependencies, fixture/reference +resources, validation, export, and reload work together without opening Tk or +VISA resources. It does not prove that the GUI renders correctly on every +Windows host. + +Complete this manual UI smoke before using a build as visual demo evidence: 1. Launch `dist/AutoLoadOffTest/AutoLoadOffTest.exe`. 2. Confirm the operator console opens without a Python traceback. @@ -70,6 +93,8 @@ Before using a packaged artifact as portfolio/demo evidence: This smoke check validates packaged UI/data workflow only. It is not live hardware validation. +See [validation_matrix.md](validation_matrix.md) for the broader software, UI, package, and live-bench evidence split. + ## Live-Hardware Packaging Smoke Do this only on a real lab workstation: diff --git a/docs/safety.md b/docs/safety.md index a2a5c53..0faae61 100644 --- a/docs/safety.md +++ b/docs/safety.md @@ -8,13 +8,17 @@ This project is not a certified production test platform. It does not replace la ## Hardware Assumptions -- Supported model labels are defined in `src/app/shared/mapping.py`. +- Supported model metadata is defined in `src/app/domain/instrument_capabilities.py`; the UI model list is derived + from that registry. - Model capability profiles provide software preflight checks for supported channels, modes, and known limits. They are not a substitute for instrument manuals or live bench validation. - Live operation uses VISA/LAN/serial access through `src/equips.py` via infrastructure adapters. - Default settings are conservative examples, not a guarantee that a connected DUT is safe. - The operator must verify AWG amplitude, frequency range, impedance, coupling mode, oscilloscope vertical range, and trigger configuration before starting a sweep. +Capability preflight prevents known-invalid software selections. It cannot detect probe attenuation, cabling, +termination errors, DUT limits, stale instrument firmware behavior, or incorrect capability source data. + ## Stop And Shutdown Behavior - Pressing Stop sets a shared stop event. @@ -43,6 +47,9 @@ Before live measurement: Automated tests use mocked ports and do not validate real hardware behavior. +See [validation_matrix.md](validation_matrix.md) for the distinction between fake-port cleanup tests and electrical +output-off validation, which has not been performed. + ## Runtime File Location Settings and auto-save output default to the process working directory. Set `AUTO_LOAD_OFF_TEST_ROOT` to use an explicit writable runtime directory on lab machines or packaged installs. diff --git a/docs/validation_matrix.md b/docs/validation_matrix.md new file mode 100644 index 0000000..7ec6442 --- /dev/null +++ b/docs/validation_matrix.md @@ -0,0 +1,58 @@ +# Validation Matrix + +This matrix separates software evidence from claims that require physical instruments, a calibrated bench, or a +target workstation. The project currently has no public live-instrument validation record. + +## Evidence Levels + +- **Automated hardware-free**: runs with deterministic data, pure domain logic, temporary files, or fake ports. +- **Manual real-UI**: uses the actual Tkinter application window but no AWG/oscilloscope hardware. +- **Not validated**: requires physical instruments, electrical measurements, or target-machine certification. + +## Matrix + +| Surface | Level | Current evidence | What it does not prove | +| --- | --- | --- | --- | +| Sweep generation and DSP | Automated hardware-free | Domain tests cover linear/log sweeps, gain/phase, auto-range, and correction behavior. | Instrument timing, acquisition fidelity, or DUT response. | +| Capability preflight | Automated hardware-free | Registry tests cover current model profiles, channels, ranges, coupling, impedance, trigger modes, and unsupported selections. | That every documented limit has been re-verified on a physical unit. | +| Adapter selection | Automated hardware-free | Registry/factory tests resolve known production and fake adapters and reject unsupported models clearly. | SCPI compatibility with connected firmware revisions. | +| Resource discovery and test-connect | Automated hardware-free | Fake scanners and identity probes cover empty, offline, unsupported, and connected receipt states. | Live VISA backend enumeration or real `*IDN?` responses. | +| Measurement/reference input | Automated hardware-free | Strict tests reject missing fields, length mismatches, NaN/inf gain, non-positive/duplicate/unsorted frequencies, and invalid references. | Provenance or metrological quality of third-party files. | +| Reference correction | Automated hardware-free | The 72-point fixture reconstructs corrected gain/phase from raw plus reference curves and compares against checked-in expected arrays. | Calibration traceability, uncertainty, or bench accuracy. | +| Export and reload | Automated hardware-free | MAT/CSV/TXT export, metadata boundary, and MAT/CSV reload are verified end to end. | Compatibility with every external analysis tool/version. | +| Tk workflow state | Automated hardware-free | Pure view-model/event-handler tests cover fixture/live/loaded source states, receipts, Bode defaults, and neutral hardware status. | Pixel-perfect rendering on every OS/theme. | +| Operator demo | Manual real-UI | The checked-in poster/MP4 captures the actual Tk window from 0/72 through 72/72, a loaded reference/coverage receipt, and a temporary export receipt. | A live sweep or connected-instrument validation. | +| Windows one-folder package | Automated hardware-free | CI builds PyInstaller output and runs bundled fixture/reference/export/reload through `--package-smoke`. | Driver installation, manual Windows GUI smoke, code signing, or live VISA access. | +| Stop/output-off behavior | Automated hardware-free only | Fake-port task-runner tests cover stop, cleanup attempts, timeout, and warning events. | Electrical output-off latency or fail-safe behavior on real hardware. | +| Live AWG/oscilloscope workflow | Not validated | Production adapters and capability profiles are inspectable in code. | Discovery, configure, trigger, sweep, stop, calibration, or export on a physical bench. | +| Metrology and safety certification | Not validated | Operator checks and software preflight are documented. | Measurement uncertainty, calibration certification, DUT protection, or production safety certification. | + +## Reproducible Checks + +```bash +PYTHONPATH=src python -m unittest discover -s tests +python -m ruff check src/app tests scripts +``` + +The Windows packaging job additionally runs: + +```text +AutoLoadOffTest.exe --package-smoke +``` + +It writes `__data__/package_smoke_receipt.json` with `live_hardware_used: false` and verifies exported artifacts. + +## Safe Public Wording + +Safe: + +> Refactored a Python/Tkinter AWG-oscilloscope workflow into layered application, domain, persistence, and adapter +> boundaries; added deterministic hardware-free calibration/export tests, mocked discovery, operator receipts, and +> cross-platform CI with a Windows packaged smoke. + +Not supported by current evidence: + +- "Validated on live AWG and oscilloscope hardware." +- "Calibrated or metrology-grade measurement system." +- "Production-certified test platform" or "fail-safe hardware shutdown." +- "Works with any VISA instrument" or any unregistered model. diff --git a/packaging/pyinstaller/auto_load_off_test_onefolder.spec b/packaging/pyinstaller/auto_load_off_test_onefolder.spec index 3bbd1bf..3d246b7 100644 --- a/packaging/pyinstaller/auto_load_off_test_onefolder.spec +++ b/packaging/pyinstaller/auto_load_off_test_onefolder.spec @@ -4,7 +4,7 @@ from pathlib import Path -ROOT = Path(__file__).resolve().parents[2] +ROOT = Path(SPECPATH).resolve().parents[1] datas = [ (str(ROOT / "demo_data"), "demo_data"), diff --git a/pyproject.toml b/pyproject.toml index 65bc99f..1bf608f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,10 @@ dev = [ build = [ "pyinstaller>=6,<7", ] +capture = [ + "imageio-ffmpeg>=0.5,<1", + "pyobjc-framework-Quartz>=10.3; sys_platform == 'darwin'", +] [tool.setuptools] package-dir = {"" = "src"} diff --git a/scripts/build_windows_onefolder.ps1 b/scripts/build_windows_onefolder.ps1 index 86f6727..722201f 100644 --- a/scripts/build_windows_onefolder.ps1 +++ b/scripts/build_windows_onefolder.ps1 @@ -1,5 +1,6 @@ param( - [string]$RuntimeRoot = "$env:LOCALAPPDATA\AutoLoadOffTest" + [string]$RuntimeRoot = "$env:LOCALAPPDATA\AutoLoadOffTest", + [switch]$SkipSmoke ) $ErrorActionPreference = "Stop" @@ -20,11 +21,39 @@ python -m pip install -e ".[build]" $env:AUTO_LOAD_OFF_TEST_ROOT = $RuntimeRoot python -m PyInstaller packaging/pyinstaller/auto_load_off_test_onefolder.spec --clean --noconfirm +$exePath = Join-Path $repoRoot "dist\AutoLoadOffTest\AutoLoadOffTest.exe" +if (-not (Test-Path $exePath)) { + throw "PyInstaller did not create $exePath" +} + +if (-not $SkipSmoke) { + $process = Start-Process -FilePath $exePath -ArgumentList "--package-smoke" -Wait -PassThru + if ($process.ExitCode -ne 0) { + throw "Packaged no-hardware smoke failed with exit code $($process.ExitCode)" + } + + $receiptPath = Join-Path $RuntimeRoot "__data__\package_smoke_receipt.json" + if (-not (Test-Path $receiptPath)) { + throw "Packaged smoke did not create receipt: $receiptPath" + } + + $receipt = Get-Content $receiptPath -Raw | ConvertFrom-Json + if ($receipt.status -ne "passed" -or $receipt.live_hardware_used -ne $false) { + throw "Packaged smoke receipt did not preserve the hardware-free boundary" + } + foreach ($artifact in $receipt.artifacts) { + if (-not (Test-Path $artifact)) { + throw "Packaged smoke artifact is missing: $artifact" + } + } + Write-Host "Automated no-hardware package smoke passed: $receiptPath" +} + Write-Host "" Write-Host "Built: dist\AutoLoadOffTest\AutoLoadOffTest.exe" Write-Host "Runtime root for smoke testing: $env:AUTO_LOAD_OFF_TEST_ROOT" Write-Host "" -Write-Host "No-hardware smoke checklist:" +Write-Host "Manual UI smoke checklist:" Write-Host "1. Launch dist\AutoLoadOffTest\AutoLoadOffTest.exe" Write-Host "2. Click Load Demo Fixture" Write-Host "3. Confirm the plot and 'No hardware - simulated fixture' label are visible" diff --git a/scripts/capture_operator_console_demo.py b/scripts/capture_operator_console_demo.py index 8700ba7..635db60 100644 --- a/scripts/capture_operator_console_demo.py +++ b/scripts/capture_operator_console_demo.py @@ -31,6 +31,7 @@ def main() -> None: app = build_desktop_app(paths=AppPaths.from_root(ROOT)) app.window.geometry("1366x768+40+60") app.window.update() + _raise_window(app.window) window_id = _find_window_id(app.window.title()) print(f"window_id: {window_id}") with tempfile.TemporaryDirectory(prefix="auto-load-off-test-capture-") as td: @@ -55,7 +56,7 @@ def run(self) -> None: def _start(self) -> None: _raise_window(self.window) - self._capture_frames(14, self._press_demo_button) + self.window.after(500, lambda: self._capture_frames(14, self._press_demo_button)) def _press_demo_button(self) -> None: self.window.run_panel.btn_load_demo_fixture.configure(relief="sunken") @@ -90,12 +91,19 @@ def _finish(self) -> None: def _require_tools() -> None: - missing = [tool for tool in ("screencapture", "ffmpeg") if shutil.which(tool) is None] - if missing: - raise RuntimeError(f"Missing capture tool(s): {', '.join(missing)}") + if shutil.which("screencapture") is None: + raise RuntimeError("Missing capture tool: screencapture") + _ffmpeg_executable() def _raise_window(window) -> None: + try: + from AppKit import NSApplicationActivateIgnoringOtherApps, NSRunningApplication + + current_app = NSRunningApplication.runningApplicationWithProcessIdentifier_(os.getpid()) + current_app.activateWithOptions_(NSApplicationActivateIgnoringOtherApps) + except Exception: + pass try: subprocess.run( [ @@ -114,22 +122,35 @@ def _raise_window(window) -> None: window.attributes("-topmost", True) window.update() time.sleep(0.2) + window.attributes("-topmost", False) window.update() def _find_window_id(title: str) -> int: + quartz_window_id = _find_window_id_with_quartz(title) + if quartz_window_id is not None: + return quartz_window_id + script = _window_list_script_path() last_stdout = "" last_stderr = "" for _attempt in range(12): - result = subprocess.run( - ["swift", str(script), title], - check=False, - text=True, - capture_output=True, - ) + try: + result = subprocess.run( + ["swift", str(script), title], + check=False, + text=True, + capture_output=True, + ) + except FileNotFoundError as exc: + raise RuntimeError("Precise window capture requires Quartz or the Swift CLI") from exc last_stdout = result.stdout last_stderr = result.stderr + if "xcode license" in result.stderr.lower(): + raise RuntimeError( + "Precise window capture requires the optional capture dependencies when Swift is unavailable. " + "Install with: python -m pip install -e '.[capture]'" + ) for line in result.stdout.splitlines(): line = line.strip() if line.isdigit(): @@ -141,6 +162,28 @@ def _find_window_id(title: str) -> int: ) +def _find_window_id_with_quartz(title: str) -> int | None: + try: + import Quartz + except ImportError: + return None + + windows = Quartz.CGWindowListCopyWindowInfo( + Quartz.kCGWindowListOptionAll, + Quartz.kCGNullWindowID, + ) + for info in windows: + if int(info.get(Quartz.kCGWindowOwnerPID, -1)) != os.getpid(): + continue + name = str(info.get(Quartz.kCGWindowName, "")) + bounds = info.get(Quartz.kCGWindowBounds, {}) + width = float(bounds.get("Width", 0)) + height = float(bounds.get("Height", 0)) + if (name == title or "Auto-Load-off-Test" in name) and width > 0 and height > 0: + return int(info[Quartz.kCGWindowNumber]) + return None + + def _window_list_script_path() -> Path: script = Path(tempfile.gettempdir()) / "auto_load_off_test_window_id.swift" script.write_text( @@ -181,19 +224,76 @@ def _window_list_script_path() -> Path: return script -def _capture_window(window_id: int | None, output_path: Path) -> None: - if window_id is None: - raise RuntimeError("Capture window id is not set") - subprocess.run( - ["screencapture", "-x", "-l", str(window_id), str(output_path)], - check=True, +def _capture_window(window_id: int, output_path: Path) -> None: + for _attempt in range(3): + output_path.unlink(missing_ok=True) + if _capture_window_with_quartz(window_id, output_path): + _flatten_capture(output_path) + if _capture_has_full_frame(output_path): + return + + command = ["screencapture", "-x", "-o", "-l", str(window_id), str(output_path)] + result = subprocess.run(command, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + if result.returncode == 0: + _flatten_capture(output_path) + if _capture_has_full_frame(output_path): + return + time.sleep(0.05) + raise RuntimeError(f"Could not capture verified Tk window id {window_id}") + + +def _capture_window_with_quartz(window_id: int, output_path: Path) -> bool: + try: + import Quartz + from Foundation import NSURL + except ImportError: + return False + + image = Quartz.CGWindowListCreateImage( + Quartz.CGRectNull, + Quartz.kCGWindowListOptionIncludingWindow, + window_id, + Quartz.kCGWindowImageBoundsIgnoreFraming, ) + if image is None: + return False + destination = Quartz.CGImageDestinationCreateWithURL( + NSURL.fileURLWithPath_(str(output_path)), + "public.png", + 1, + None, + ) + if destination is None: + return False + Quartz.CGImageDestinationAddImage(destination, image, None) + return bool(Quartz.CGImageDestinationFinalize(destination)) + + +def _flatten_capture(output_path: Path) -> None: + from PIL import Image + + with Image.open(output_path) as source: + rgba = source.convert("RGBA") + opaque = Image.new("RGB", rgba.size, "white") + opaque.paste(rgba, mask=rgba.getchannel("A")) + opaque.save(output_path) + + +def _capture_has_full_frame(output_path: Path) -> bool: + from PIL import Image + + with Image.open(output_path) as source: + sample = source.convert("RGB") + sample.thumbnail((180, 120)) + pixels = list(sample.getdata()) + near_black = sum(1 for red, green, blue in pixels if max(red, green, blue) < 12) + return bool(pixels) and near_black / len(pixels) < 0.12 def _write_video(frame_dir: Path) -> None: subprocess.run( [ - "ffmpeg", + _ffmpeg_executable(), "-y", "-framerate", str(FRAME_RATE), @@ -213,5 +313,26 @@ def _write_video(frame_dir: Path) -> None: ) +def _ffmpeg_executable() -> str: + system_ffmpeg = shutil.which("ffmpeg") + if system_ffmpeg is not None: + probe = subprocess.run( + [system_ffmpeg, "-version"], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if probe.returncode == 0: + return system_ffmpeg + try: + import imageio_ffmpeg + except ImportError as exc: + raise RuntimeError( + "No working ffmpeg executable found. Install capture dependencies with: " + "python -m pip install -e '.[capture]'" + ) from exc + return imageio_ffmpeg.get_ffmpeg_exe() + + if __name__ == "__main__": main() diff --git a/scripts/capture_operator_console_point_replay.py b/scripts/capture_operator_console_point_replay.py index 3fe6714..303ad62 100644 --- a/scripts/capture_operator_console_point_replay.py +++ b/scripts/capture_operator_console_point_replay.py @@ -11,20 +11,26 @@ SRC = ROOT / "src" OUT_DIR = ROOT / "docs" / "images" FIXTURE = ROOT / "demo_data" / "hyperframe_simulated_fixture.mat" +REFERENCE = ROOT / "demo_data" / "hyperframe_reference_fixture.mat" POSTER = OUT_DIR / "auto-load-off-test-point-replay-demo.png" VIDEO = OUT_DIR / "auto-load-off-test-point-replay-demo.mp4" FRAME_RATE = 12 -DEMO_SIZE = "1366x768+40+60" +DEMO_SIZE = "1440x810+20+40" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) +from app.application.dto import SaveTarget # noqa: E402 +from app.application.services.export_receipts import build_export_receipt # noqa: E402 from app.bootstrap import build_desktop_app # noqa: E402 +from app.demo.hyperframe_fixture import DEMO_LABEL, build_fixture_settings # noqa: E402 from app.domain.models import SweepResult # noqa: E402 +from app.presentation.tk.mapper import settings_to_vm # noqa: E402 from app.runtime.paths import AppPaths # noqa: E402 from capture_operator_console_demo import ( # noqa: E402 _capture_window, + _ffmpeg_executable, _find_window_id, _raise_window, _require_tools, @@ -34,15 +40,21 @@ def main() -> None: _require_tools() _require_file(FIXTURE) + _require_file(REFERENCE) OUT_DIR.mkdir(parents=True, exist_ok=True) app = build_desktop_app(paths=AppPaths.from_root(ROOT)) + fixture_settings = build_fixture_settings() + fixture_settings.freq_unit = "KHz" + settings_to_vm(fixture_settings, app.window.vm) app.window.geometry(DEMO_SIZE) app.window.update() + _raise_window(app.window) window_id = _find_window_id(app.window.title()) print(f"window_id: {window_id}") loaded = app.controller.load_measurement_use_case.execute(str(FIXTURE)) + app.controller.load_reference_from_path(REFERENCE) with tempfile.TemporaryDirectory(prefix="auto-load-off-test-point-replay-") as td: session = PointReplayCaptureSession( app=app, @@ -76,7 +88,7 @@ def run(self) -> None: def _capture_start(self) -> None: _raise_window(self.window) - self._capture_frames(12, self._press_demo_button) + self.window.after(500, lambda: self._capture_frames(12, self._press_demo_button)) def _press_demo_button(self) -> None: self.window.run_panel.btn_load_demo_fixture.configure(relief="sunken") @@ -85,11 +97,14 @@ def _press_demo_button(self) -> None: def _start_replay(self) -> None: self.window.run_panel.btn_load_demo_fixture.configure(relief="flat") self.replay_index = 0 + self.app.controller._ui_handler.set_result(SweepResult(), refresh_plot=False) + self.app.controller._ui_handler.set_fixture_source(label=DEMO_LABEL, path_name=FIXTURE.name) self._set_replay_state(0) self._capture_replay_step() def _capture_replay_step(self) -> None: if self.replay_index >= self.total_points: + self._record_demo_export() self._capture_frames(24, self._finish) return @@ -108,16 +123,23 @@ def _capture_frames(self, remaining: int, done: Callable[[], None]) -> None: self.window.after(int(1000 / FRAME_RATE), lambda: self._capture_frames(remaining - 1, done)) def _set_initial_state(self) -> None: - self.app.controller._ui_handler.set_result(SweepResult(), refresh_plot=True) - self.vm.data_source_text.set("Live instrument path") - self.vm.fixture_badge_text.set("") - self.vm.validation_receipt_text.set("Live run requires operator hardware checks") + self.app.controller._ui_handler.set_result(SweepResult(), refresh_plot=False) + self.vm.source_mode.set("fixture") + self.vm.figure_mode.set("gain_db") + self.vm.magnitude_phase_mode.set("magnitude_phase") + self.vm.plot_scale.set("log") + self.vm.data_source_text.set(f"Fixture replay ready · {FIXTURE.name}") + self.vm.fixture_badge_text.set("No hardware - simulated fixture") + self.vm.validation_receipt_text.set("Ready for simulated point replay; not live hardware validation") self.vm.export_receipt_text.set("No export yet") - self.vm.run_state_text.set("Idle") - self.vm.progress_text.set("0 / 0") - self.vm.point_count_text.set("0 points") + self.vm.run_state_text.set("Ready to replay") + self.vm.progress_text.set(f"0 / {self.total_points}") + self.vm.point_count_text.set(f"0 / {self.total_points} points") self.vm.latest_frequency_text.set("-") - self.vm.status_text.set("Ready") + self.vm.status_text.set("Ready to replay simulated fixture (no hardware)") + self.window.set_connection_idle() + self.window.plot_widget.set_mode("gain_db") + self.app.controller._ui_handler.refresh_plot() def _set_replay_state(self, point_count: int) -> None: partial = SweepResult( @@ -125,11 +147,6 @@ def _set_replay_state(self, point_count: int) -> None: meta=dict(self.full_result.meta), ) self.app.controller._ui_handler.set_result(partial, refresh_plot=True) - self.vm.data_source_text.set(f"Fixture replay · {FIXTURE.name}") - self.vm.fixture_badge_text.set("No hardware - simulated fixture") - self.vm.validation_receipt_text.set( - "Point-by-point simulated fixture replay; not live hardware validation" - ) self.vm.export_receipt_text.set("Replaying deterministic fixture points; no hardware connected") self.vm.run_state_text.set("Replaying fixture" if point_count < self.total_points else "Fixture ready") self.vm.progress_text.set(f"{point_count} / {self.total_points}") @@ -141,6 +158,24 @@ def _set_replay_state(self, point_count: int) -> None: self.vm.latest_frequency_text.set("-") self.vm.status_text.set(f"Fixture replay point {point_count}/{self.total_points} (no hardware)") + def _record_demo_export(self) -> None: + settings = build_fixture_settings() + artifacts = self.app.controller.save_measurement_use_case.execute( + result=self.app.controller._ui_handler.latest_result, + settings=settings, + target=SaveTarget(base_path=self.frame_dir / "demo_capture_export", figures={}), + ) + receipt = build_export_receipt( + artifacts=artifacts, + settings=settings, + result=self.app.controller._ui_handler.latest_result, + source_text=self.vm.data_source_text.get(), + fixture_badge_text=self.vm.fixture_badge_text.get(), + ) + artifact_names = " · ".join(path.name for path in receipt.artifacts) + self.vm.export_receipt_text.set(f"Export verified · {artifact_names}\nTemporary capture output · no hardware") + self.vm.status_text.set("Fixture replay complete; export verified (no hardware)") + def _finish(self) -> None: _capture_window(self.window_id, POSTER) _write_video(self.frame_dir) @@ -157,7 +192,7 @@ def _finish(self) -> None: def _write_video(frame_dir: Path) -> None: subprocess.run( [ - "ffmpeg", + _ffmpeg_executable(), "-y", "-framerate", str(FRAME_RATE), diff --git a/src/app/application/dto.py b/src/app/application/dto.py index 2a00ed7..abfe5ea 100644 --- a/src/app/application/dto.py +++ b/src/app/application/dto.py @@ -4,8 +4,6 @@ from pathlib import Path from typing import Any -import numpy as np - from app.domain.models import AppSettings, SweepResult @@ -35,4 +33,4 @@ class SaveArtifacts: @dataclass(slots=True) class LoadedMeasurement: result: SweepResult - raw_payload: dict[str, np.ndarray] + raw_payload: dict[str, Any] diff --git a/src/app/demo/package_smoke.py b/src/app/demo/package_smoke.py new file mode 100644 index 0000000..6f63964 --- /dev/null +++ b/src/app/demo/package_smoke.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import json +from pathlib import Path +import sys +from typing import Any + +import numpy as np + +from app.application.dto import SaveTarget +from app.demo.hyperframe_fixture import POINT_COUNT, SOURCE, build_fixture_settings +from app.domain.calibration import build_reference_interpolator +from app.infrastructure.persistence.measurement_exporter import MeasurementExporter +from app.infrastructure.persistence.measurement_loader import MeasurementLoader +from app.infrastructure.persistence.reference_repo_mat import MatReferenceRepository +from app.runtime.paths import AppPaths + + +VALIDATION_BOUNDARY = "No hardware - simulated fixture; not live hardware validation" +RECEIPT_NAME = "package_smoke_receipt.json" + + +def run_package_smoke( + *, + runtime_root: Path | None = None, + resource_root: Path | None = None, +) -> Path: + """Exercise bundled fixture IO and export without opening Tk or VISA resources.""" + resources = (resource_root or bundled_resource_root()).resolve() + paths = AppPaths.from_root(runtime_root) if runtime_root is not None else AppPaths.default() + fixture_path = resources / "demo_data" / "hyperframe_simulated_fixture.mat" + reference_path = resources / "demo_data" / "hyperframe_reference_fixture.mat" + + loaded = MeasurementLoader().load(str(fixture_path)) + reference = MatReferenceRepository().load_reference(str(reference_path)) + reference_values = build_reference_interpolator(reference)(loaded.result.freq_array()) + + if len(loaded.result.points) != POINT_COUNT: + raise RuntimeError(f"Expected {POINT_COUNT} fixture points, got {len(loaded.result.points)}") + if loaded.result.meta.get("source") != SOURCE: + raise RuntimeError(f"Expected fixture source {SOURCE!r}, got {loaded.result.meta.get('source')!r}") + if reference_values.shape != (POINT_COUNT,) or not np.all(np.isfinite(reference_values)): + raise RuntimeError("Bundled reference interpolation produced invalid values") + + output_dir = paths.measurement_dir / "package_smoke" + artifacts = MeasurementExporter().export( + loaded.result, + build_fixture_settings(), + SaveTarget(base_path=output_dir / "simulated_fixture_export", figures={}), + ) + reloaded_mat = MeasurementLoader().load(str(artifacts.mat_path)) + reloaded_csv = MeasurementLoader().load(str(artifacts.csv_path)) + + for label, result in (("MAT", reloaded_mat.result), ("CSV", reloaded_csv.result)): + if len(result.points) != POINT_COUNT: + raise RuntimeError(f"{label} export reload returned {len(result.points)} points") + if result.meta.get("source") != SOURCE: + raise RuntimeError(f"{label} export lost fixture source metadata") + if "not live hardware validation" not in str(result.meta.get("validation_boundary", "")): + raise RuntimeError(f"{label} export lost the no-hardware validation boundary") + + artifact_paths = [artifacts.mat_path, artifacts.csv_path, artifacts.txt_path] + missing = [str(path) for path in artifact_paths if not path.is_file() or path.stat().st_size == 0] + if missing: + raise RuntimeError(f"Package smoke export artifacts are missing or empty: {missing}") + + receipt_path = paths.data_dir / RECEIPT_NAME + receipt_path.parent.mkdir(parents=True, exist_ok=True) + receipt: dict[str, Any] = { + "status": "passed", + "source": SOURCE, + "point_count": POINT_COUNT, + "fixture": str(fixture_path), + "reference": str(reference_path), + "artifacts": [str(path.resolve()) for path in artifact_paths], + "validation_boundary": VALIDATION_BOUNDARY, + "live_hardware_used": False, + } + receipt_path.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return receipt_path + + +def bundled_resource_root() -> Path: + frozen_root = getattr(sys, "_MEIPASS", None) + if frozen_root: + return Path(frozen_root) + return Path(__file__).resolve().parents[3] diff --git a/src/app/domain/calibration.py b/src/app/domain/calibration.py index d188b28..4897fba 100644 --- a/src/app/domain/calibration.py +++ b/src/app/domain/calibration.py @@ -5,29 +5,21 @@ import numpy as np from scipy.interpolate import make_interp_spline +from app.domain.data_validation import normalize_reference_curve from app.domain.models import ReferenceCurve, SweepPoint def build_reference_interpolator(curve: ReferenceCurve) -> Callable[[np.ndarray], np.ndarray]: - 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") + normalized = normalize_reference_curve(curve) + freq = normalized.freq_hz + gain_db = normalized.gain_db + phase = normalized.phase_deg if phase is None or phase.size == 0: href = 10 ** (gain_db / 20.0) else: href = 10 ** (gain_db / 20.0) * np.exp(1j * np.deg2rad(phase)) - order = np.argsort(freq) - freq = freq[order] - href = href[order] - - freq, unique_idx = np.unique(freq, return_index=True) - href = href[unique_idx] - if freq.size == 1: h0 = href[0] diff --git a/src/app/domain/data_validation.py b/src/app/domain/data_validation.py new file mode 100644 index 0000000..ca26fcb --- /dev/null +++ b/src/app/domain/data_validation.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np + +from app.domain.models import ReferenceCurve, SweepResult + + +class DataValidationError(ValueError): + """Raised when imported or exported measurement data is structurally unsafe.""" + + +@dataclass(frozen=True, slots=True) +class MeasurementArrays: + freq_hz: np.ndarray + gain_linear: np.ndarray + gain_db: np.ndarray + phase_deg: np.ndarray | None + + +def normalize_measurement_arrays( + *, + freq_hz: object, + gain_linear: object | None, + gain_db: object | None, + phase_deg: object | None, +) -> MeasurementArrays: + freq = _vector("frequency", freq_hz) + if freq.size == 0: + raise DataValidationError("Measurement frequency data is empty") + if not np.all(np.isfinite(freq)): + raise DataValidationError("Measurement frequency contains NaN or infinity") + if np.any(freq <= 0.0): + raise DataValidationError("Measurement frequency values must be positive") + if freq.size > 1 and np.any(np.diff(freq) <= 0.0): + raise DataValidationError("Measurement frequency values must be strictly increasing without duplicates") + + linear = None if gain_linear is None else _vector("gain_linear", gain_linear) + db = None if gain_db is None else _vector("gain_db", gain_db) + if linear is None and db is None: + raise DataValidationError("Measurement requires gain_linear or gain_db data") + + if linear is not None: + _require_length("gain_linear", linear, freq.size) + if not np.all(np.isfinite(linear)): + raise DataValidationError("Measurement gain_linear contains NaN or infinity") + if np.any(linear <= 0.0): + raise DataValidationError("Measurement gain_linear values must be positive") + + if db is not None: + _require_length("gain_db", db, freq.size) + if not np.all(np.isfinite(db)): + raise DataValidationError("Measurement gain_db contains NaN or infinity") + + if linear is None: + assert db is not None + linear = np.power(10.0, db / 20.0) + if db is None: + db = 20.0 * np.log10(linear) + + phase = None if phase_deg is None else _vector("phase_deg", phase_deg) + if phase is not None: + _require_length("phase_deg", phase, freq.size) + if np.any(np.isinf(phase)): + raise DataValidationError("Measurement phase_deg contains infinity") + if np.all(np.isnan(phase)): + phase = None + + return MeasurementArrays(freq_hz=freq, gain_linear=linear, gain_db=db, phase_deg=phase) + + +def normalize_reference_curve(curve: ReferenceCurve) -> ReferenceCurve: + freq = _vector("reference frequency", curve.freq_hz) + gain_db = _vector("reference gain_db", curve.gain_db) + if freq.size == 0: + raise DataValidationError("Reference frequency data is empty") + _require_length("reference gain_db", gain_db, freq.size) + if not np.all(np.isfinite(freq)): + raise DataValidationError("Reference frequency contains NaN or infinity") + if np.any(freq <= 0.0): + raise DataValidationError("Reference frequency values must be positive") + if freq.size > 1 and np.any(np.diff(freq) <= 0.0): + raise DataValidationError("Reference frequency values must be strictly increasing without duplicates") + if not np.all(np.isfinite(gain_db)): + raise DataValidationError("Reference gain_db contains NaN or infinity") + + phase = None if curve.phase_deg is None else _vector("reference phase_deg", curve.phase_deg) + if phase is not None: + _require_length("reference phase_deg", phase, freq.size) + if not np.all(np.isfinite(phase)): + raise DataValidationError("Reference phase_deg contains NaN or infinity") + + return ReferenceCurve(freq_hz=freq, gain_db=gain_db, phase_deg=phase) + + +def validate_sweep_result(result: SweepResult) -> MeasurementArrays: + phase = np.array( + [np.nan if point.phase_deg is None else float(point.phase_deg) for point in result.points], + dtype=float, + ) + return normalize_measurement_arrays( + freq_hz=np.array([point.freq_hz for point in result.points], dtype=float), + gain_linear=np.array([point.gain_linear for point in result.points], dtype=float), + gain_db=np.array([point.gain_db for point in result.points], dtype=float), + phase_deg=phase, + ) + + +def _vector(name: str, value: object) -> np.ndarray: + try: + array = np.asarray(value, dtype=float).squeeze() + except (TypeError, ValueError) as exc: + raise DataValidationError(f"{name} is not numeric") from exc + if array.ndim > 1: + raise DataValidationError(f"{name} must be a one-dimensional array") + return np.atleast_1d(array) + + +def _require_length(name: str, values: np.ndarray, expected: int) -> None: + if values.size != expected: + raise DataValidationError(f"{name} length {values.size} does not match frequency length {expected}") diff --git a/src/app/domain/plotting.py b/src/app/domain/plotting.py new file mode 100644 index 0000000..679a5cd --- /dev/null +++ b/src/app/domain/plotting.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import numpy as np + + +def choose_frequency_scale( + freq_hz: object, + *, + requested: str = "auto", + sweep_is_log: bool = False, +) -> str: + if requested not in {"auto", "linear", "log"}: + raise ValueError(f"Unsupported frequency scale: {requested}") + freq = np.atleast_1d(np.asarray(freq_hz, dtype=float).squeeze()) + if freq.size == 0: + return "linear" + if requested != "auto": + return requested + + if np.any(~np.isfinite(freq)) or np.any(freq <= 0.0): + return "linear" + if sweep_is_log: + return "log" + if freq.size < 3: + return "linear" + + linear_steps = np.diff(freq) + log_steps = np.diff(np.log10(freq)) + linear_cv = _coefficient_of_variation(linear_steps) + log_cv = _coefficient_of_variation(log_steps) + return "log" if log_cv <= 0.05 and log_cv < linear_cv else "linear" + + +def _coefficient_of_variation(values: np.ndarray) -> float: + mean = float(np.mean(np.abs(values))) + if mean == 0.0: + return float("inf") + return float(np.std(values) / mean) diff --git a/src/app/infrastructure/persistence/measurement_exporter.py b/src/app/infrastructure/persistence/measurement_exporter.py index cc1ba27..975a95a 100644 --- a/src/app/infrastructure/persistence/measurement_exporter.py +++ b/src/app/infrastructure/persistence/measurement_exporter.py @@ -2,19 +2,21 @@ import csv import json -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path import numpy as np from scipy.io import savemat from app.application.dto import SaveArtifacts, SaveTarget +from app.domain.data_validation import validate_sweep_result from app.domain.exporters import result_to_arrays, settings_to_metadata from app.domain.models import AppSettings, SweepResult class MeasurementExporter: def export(self, result: SweepResult, settings: AppSettings, target: SaveTarget) -> SaveArtifacts: + validate_sweep_result(result) directory = target.base_path.parent directory.mkdir(parents=True, exist_ok=True) @@ -27,9 +29,23 @@ def export(self, result: SweepResult, settings: AppSettings, target: SaveTarget) txt_path = directory / f"{file_base}.txt" arrays = result_to_arrays(result) + source = str(result.meta.get("source") or "unknown") + validation_boundary = _validation_boundary(result) + metadata = settings_to_metadata(settings) + metadata["result"] = dict(result.meta) + metadata["export"] = { + "point_count": len(result.points), + "source": source, + "validation_boundary": validation_boundary, + "exported_at_utc": datetime.now(timezone.utc).isoformat(), + } payload: dict[str, object] = { "schema_version": settings.schema_version, - "metadata_json": json.dumps(settings_to_metadata(settings), ensure_ascii=True), + "metadata_json": json.dumps(metadata, ensure_ascii=True, default=_json_default), + "source": source, + "point_count": len(result.points), + "correction_mode": settings.run_mode.correction_mode.value, + "validation_boundary": validation_boundary, } payload.update(arrays) savemat(mat_path, payload) @@ -39,13 +55,24 @@ def export(self, result: SweepResult, settings: AppSettings, target: SaveTarget) gain_db = arrays.get("gain_db", np.array([], dtype=float)) phase = arrays.get("phase_deg", np.array([], dtype=float)) - headers = ["freq_hz", "gain_linear", "gain_db", "phase_deg"] + headers = [ + "source", + "validation_boundary", + "correction_mode", + "freq_hz", + "gain_linear", + "gain_db", + "phase_deg", + ] with csv_path.open("w", newline="", encoding="utf-8") as fh: writer = csv.writer(fh) writer.writerow(headers) for idx in range(len(freq)): writer.writerow( [ + source, + validation_boundary, + settings.run_mode.correction_mode.value, float(freq[idx]), float(gain_linear[idx]) if idx < len(gain_linear) else "", float(gain_db[idx]) if idx < len(gain_db) else "", @@ -61,7 +88,8 @@ def export(self, result: SweepResult, settings: AppSettings, target: SaveTarget) phase if len(phase) == len(freq) else np.full(len(freq), np.nan), ] ) - np.savetxt(txt_path, rows, delimiter="\t", header="\t".join(headers), comments="") + numeric_headers = ["freq_hz", "gain_linear", "gain_db", "phase_deg"] + np.savetxt(txt_path, rows, delimiter="\t", header="\t".join(numeric_headers), comments="") gain_plot_path = None db_plot_path = None @@ -90,3 +118,23 @@ def _optional_float(values: np.ndarray, idx: int) -> float | str: if np.isnan(value): return "" return value + + +def _validation_boundary(result: SweepResult) -> str: + explicit = str(result.meta.get("validation_boundary") or "").strip() + if explicit: + return explicit + source = str(result.meta.get("source") or "").lower() + if "fixture" in source or "mock" in source: + return "No hardware - simulated fixture; not live hardware validation" + return "Export artifact; live hardware validation is not implied" + + +def _json_default(value: object) -> object: + if isinstance(value, np.ndarray): + return value.tolist() + if isinstance(value, np.generic): + return value.item() + if isinstance(value, Path): + return str(value) + return str(value) diff --git a/src/app/infrastructure/persistence/measurement_loader.py b/src/app/infrastructure/persistence/measurement_loader.py index 822184f..f1eed9d 100644 --- a/src/app/infrastructure/persistence/measurement_loader.py +++ b/src/app/infrastructure/persistence/measurement_loader.py @@ -2,11 +2,13 @@ import csv from pathlib import Path +from typing import Any import numpy as np from scipy.io import loadmat from app.application.dto import LoadedMeasurement +from app.domain.data_validation import DataValidationError, normalize_measurement_arrays from app.domain.models import SweepPoint, SweepResult @@ -18,60 +20,106 @@ def load(self, file_path: str) -> LoadedMeasurement: if suffix == ".mat": payload = loadmat(path) freq = self._get_array(payload, ["freq_hz", "freq"]) # type: ignore[arg-type] - gain_db = self._get_array(payload, ["gain_db", "gain_db_raw", "gain_db_corr"]) + gain_db = self._get_array(payload, ["gain_db", "gain_db_raw", "gain_db_corr"], required=False) gain_linear = self._get_array(payload, ["gain_linear", "gain_raw"], required=False) - if gain_linear is None: - gain_linear = np.power(10.0, gain_db / 20.0) phase = self._get_array(payload, ["phase_deg", "phase", "phase_deg_corr", "phase_corr"], required=False) elif suffix == ".csv": - freq_l: list[float] = [] - gain_l: list[float] = [] - gain_db_l: list[float] = [] - phase_l: list[float] = [] - with path.open("r", encoding="utf-8") as fh: - reader = csv.DictReader(fh) - for row in reader: - freq_l.append(float(row.get("freq_hz", "0") or 0.0)) - gain_l.append(float(row.get("gain_linear", "0") or 0.0)) - gain_db_l.append(float(row.get("gain_db", "0") or 0.0)) - phase_value = row.get("phase_deg") - phase_l.append(float(phase_value) if phase_value not in (None, "") else np.nan) - freq = np.array(freq_l, dtype=float) - gain_linear = np.array(gain_l, dtype=float) - gain_db = np.array(gain_db_l, dtype=float) - phase_values = np.array(phase_l, dtype=float) if phase_l else None - phase = phase_values if phase_values is not None and np.any(~np.isnan(phase_values)) else None - payload = {"freq_hz": freq, "gain_linear": gain_linear, "gain_db": gain_db} - if phase is not None: - payload["phase_deg"] = phase + payload, freq, gain_linear, gain_db, phase = self._read_csv(path) else: raise ValueError(f"Unsupported file type: {suffix}") + arrays = normalize_measurement_arrays( + freq_hz=freq, + gain_linear=gain_linear, + gain_db=gain_db, + phase_deg=phase, + ) points: list[SweepPoint] = [] - for idx in range(len(freq)): - phase_deg = _optional_phase(phase, idx) + for idx in range(len(arrays.freq_hz)): + phase_deg = _optional_phase(arrays.phase_deg, idx) points.append( SweepPoint( - freq_hz=float(freq[idx]), - gain_linear=float(gain_linear[idx]) if idx < len(gain_linear) else 0.0, - gain_db=float(gain_db[idx]) if idx < len(gain_db) else 0.0, + freq_hz=float(arrays.freq_hz[idx]), + gain_linear=float(arrays.gain_linear[idx]), + gain_db=float(arrays.gain_db[idx]), phase_deg=phase_deg, gain_complex=( complex( - float(gain_linear[idx]) * np.cos(np.deg2rad(phase_deg)), - float(gain_linear[idx]) * np.sin(np.deg2rad(phase_deg)), + float(arrays.gain_linear[idx]) * np.cos(np.deg2rad(phase_deg)), + float(arrays.gain_linear[idx]) * np.sin(np.deg2rad(phase_deg)), ) - if phase_deg is not None and idx < len(gain_linear) + if phase_deg is not None else None ), ) ) - return LoadedMeasurement(result=SweepResult(points=points), raw_payload=payload) + return LoadedMeasurement( + result=SweepResult(points=points, meta=_result_meta(payload, path=path, point_count=len(points))), + raw_payload=payload, + ) + + def _read_csv( + self, + path: Path, + ) -> tuple[dict[str, Any], np.ndarray, np.ndarray | None, np.ndarray | None, np.ndarray | None]: + with path.open("r", encoding="utf-8-sig", newline="") as fh: + reader = csv.DictReader(fh) + fieldnames = set(reader.fieldnames or []) + if "freq_hz" not in fieldnames: + raise DataValidationError("CSV is missing required column: freq_hz") + if "gain_linear" not in fieldnames and "gain_db" not in fieldnames: + raise DataValidationError("CSV requires gain_linear or gain_db column") + rows = list(reader) + + if not rows: + raise DataValidationError("Measurement CSV has no data rows") + + freq = np.array( + [_required_float(row, "freq_hz", row_number) for row_number, row in enumerate(rows, start=2)], + dtype=float, + ) + gain_linear = ( + np.array( + [_required_float(row, "gain_linear", row_number) for row_number, row in enumerate(rows, start=2)], + dtype=float, + ) + if "gain_linear" in fieldnames + else None + ) + gain_db = ( + np.array( + [_required_float(row, "gain_db", row_number) for row_number, row in enumerate(rows, start=2)], + dtype=float, + ) + if "gain_db" in fieldnames + else None + ) + phase = ( + np.array( + [_optional_float(row, "phase_deg", row_number) for row_number, row in enumerate(rows, start=2)], + dtype=float, + ) + if "phase_deg" in fieldnames + else None + ) + + payload: dict[str, Any] = {"freq_hz": freq} + if gain_linear is not None: + payload["gain_linear"] = gain_linear + if gain_db is not None: + payload["gain_db"] = gain_db + if phase is not None: + payload["phase_deg"] = phase + for key in ("source", "demo_label", "correction_mode", "validation_boundary"): + value = (rows[0].get(key) or "").strip() + if value: + payload[key] = value + return payload, freq, gain_linear, gain_db, phase def _get_array( self, - payload: dict[str, np.ndarray], + payload: dict[str, Any], keys: list[str], *, required: bool = True, @@ -81,10 +129,30 @@ def _get_array( if isinstance(value, np.ndarray): return np.atleast_1d(np.asarray(value, dtype=float).squeeze()) if required: - raise ValueError(f"Missing required keys: {keys}") + raise DataValidationError(f"Missing required measurement keys: {keys}") return None +def _required_float(row: dict[str, str | None], field: str, row_number: int) -> float: + raw = row.get(field) + if raw is None or not raw.strip(): + raise DataValidationError(f"CSV row {row_number} has no value for {field}") + try: + return float(raw) + except ValueError as exc: + raise DataValidationError(f"CSV row {row_number} has non-numeric {field}: {raw!r}") from exc + + +def _optional_float(row: dict[str, str | None], field: str, row_number: int) -> float: + raw = row.get(field) + if raw is None or not raw.strip(): + return float("nan") + try: + return float(raw) + except ValueError as exc: + raise DataValidationError(f"CSV row {row_number} has non-numeric {field}: {raw!r}") from exc + + def _optional_phase(phase: np.ndarray | None, idx: int) -> float | None: if phase is None or idx >= len(phase): return None @@ -92,3 +160,28 @@ def _optional_phase(phase: np.ndarray | None, idx: int) -> float | None: if np.isnan(value): return None return value + + +def _result_meta(payload: dict[str, Any], *, path: Path, point_count: int) -> dict[str, Any]: + meta: dict[str, Any] = {"source_file": path.name, "point_count": point_count} + for key in ("source", "demo_label", "correction_mode", "trigger_mode", "validation_boundary"): + value = _payload_text(payload.get(key)) + if value: + meta[key] = value + if meta.get("source") == "mock_fixture": + meta.setdefault("validation_boundary", "No hardware - simulated fixture; not live hardware validation") + return meta + + +def _payload_text(value: object) -> str: + if value is None: + return "" + array = np.asarray(value) + if array.dtype.kind in {"U", "S"}: + if array.ndim == 2 and array.shape[0] == 1: + return "".join(str(item) for item in array[0]).strip() + return "".join(str(item) for item in array.ravel()).strip() + squeezed = array.squeeze() + if squeezed.shape == () and isinstance(squeezed.item(), str): + return str(squeezed.item()).strip() + return "" diff --git a/src/app/infrastructure/persistence/reference_repo_mat.py b/src/app/infrastructure/persistence/reference_repo_mat.py index 8d2b4dc..65ac0db 100644 --- a/src/app/infrastructure/persistence/reference_repo_mat.py +++ b/src/app/infrastructure/persistence/reference_repo_mat.py @@ -3,6 +3,7 @@ import numpy as np from scipy.io import loadmat +from app.domain.data_validation import DataValidationError, normalize_reference_curve from app.domain.models import ReferenceCurve @@ -14,7 +15,7 @@ def load_reference(self, file_path: str) -> ReferenceCurve: gain_db = self._get_array(payload, ["gain_db", "gain_db_raw", "gain_db_corr"]) phase = self._get_array(payload, ["phase_deg", "phase", "phase_deg_corr"], required=False) - return ReferenceCurve(freq_hz=freq, gain_db=gain_db, phase_deg=phase) + return normalize_reference_curve(ReferenceCurve(freq_hz=freq, gain_db=gain_db, phase_deg=phase)) def _get_array( self, @@ -26,7 +27,7 @@ def _get_array( for key in keys: value = payload.get(key) if isinstance(value, np.ndarray): - return np.asarray(value, dtype=float).squeeze() + return np.atleast_1d(np.asarray(value, dtype=float).squeeze()) if required: - raise ValueError(f"Missing required keys: {keys}") + raise DataValidationError(f"Missing required reference keys: {keys}") return None diff --git a/src/app/presentation/tk/app_window.py b/src/app/presentation/tk/app_window.py index 9bc537a..cb3fca0 100644 --- a/src/app/presentation/tk/app_window.py +++ b/src/app/presentation/tk/app_window.py @@ -26,7 +26,7 @@ def __init__(self, vm: ViewModel | None = None) -> None: container.pack(fill=tk.BOTH, expand=True, padx=12, pady=12) container.grid_columnconfigure(0, weight=0, minsize=295) container.grid_columnconfigure(1, weight=1, minsize=640) - container.grid_columnconfigure(2, weight=0, minsize=340) + container.grid_columnconfigure(2, weight=0, minsize=355) container.grid_rowconfigure(0, weight=1) left = tk.Frame(container, bg="#f4f6f8", width=300) @@ -42,13 +42,16 @@ def __init__(self, vm: ViewModel | None = None) -> None: right = tk.Frame(container, bg="#f4f6f8") right.grid(row=0, column=2, sticky="nsew") + right.grid_rowconfigure(0, weight=1) + right.grid_columnconfigure(0, weight=1) self.control_panel = self._build_sidebar(left) self.plot_widget = PlotWidget(center, self.vm) self.plot_widget.frame.grid(row=0, column=0, sticky="nsew") - self.run_panel = RunPanel(right, self.vm) + right_content = self._build_scrollable_content(right, width=340) + self.run_panel = RunPanel(right_content, self.vm) self.run_panel.pack(fill=tk.BOTH, expand=True) self._alias_control_widgets() @@ -92,6 +95,7 @@ def bind_actions( on_close, on_figure_change, on_mag_phase_change, + on_plot_scale_change, ) -> None: self.control_panel.bind_actions( on_save_settings=on_save_settings, @@ -112,15 +116,25 @@ def bind_actions( self.plot_widget.bind_controls( on_figure_change=on_figure_change, on_mag_phase_change=on_mag_phase_change, + on_plot_scale_change=on_plot_scale_change, ) self._on_close = on_close def set_connection_status(self, awg_connected: bool, osc_connected: bool) -> None: + if self.vm.source_mode.get() != "live": + self.set_connection_idle() + return self.canvas_awg.itemconfig(self.awg_light, fill="green" if awg_connected else "red") self.canvas_osc.itemconfig(self.osc_light, fill="green" if osc_connected else "red") self.vm.awg_connection_text.set("AWG online" if awg_connected else "AWG offline") self.vm.osc_connection_text.set("OSC online" if osc_connected else "OSC offline") + def set_connection_idle(self) -> None: + self.canvas_awg.itemconfig(self.awg_light, fill="#94a3b8") + self.canvas_osc.itemconfig(self.osc_light, fill="#94a3b8") + self.vm.awg_connection_text.set("AWG not used") + self.vm.osc_connection_text.set("OSC not used") + def on_close(self) -> None: if self._on_close is not None: self._on_close() @@ -140,9 +154,16 @@ def _alias_control_widgets(self) -> None: self.btn_test_connect = self.control_panel.btn_test_connect self.cmb_figure = self.plot_widget.cmb_figure self.cmb_mag_phase = self.plot_widget.cmb_mag_phase + self.cmb_plot_scale = self.plot_widget.cmb_plot_scale def _build_sidebar(self, parent: tk.Frame) -> ControlPanel: - canvas = tk.Canvas(parent, bg="#f4f6f8", highlightthickness=0, width=300) + content = self._build_scrollable_content(parent, width=300) + panel = ControlPanel(content, self.vm) + panel.pack(fill=tk.BOTH, expand=True) + return panel + + def _build_scrollable_content(self, parent: tk.Frame, *, width: int) -> tk.Frame: + canvas = tk.Canvas(parent, bg="#f4f6f8", highlightthickness=0, width=width) scrollbar = ttk.Scrollbar(parent, orient=tk.VERTICAL, command=canvas.yview) content = tk.Frame(canvas, bg="#f4f6f8") window_id = canvas.create_window((0, 0), window=content, anchor="nw") @@ -157,10 +178,7 @@ def resize_content(_event=None) -> None: content.bind("", resize_content) canvas.bind("", resize_content) - - panel = ControlPanel(content, self.vm) - panel.pack(fill=tk.BOTH, expand=True) - return panel + return content def _configure_style(self) -> None: style = ttk.Style(self) diff --git a/src/app/presentation/tk/control_panel.py b/src/app/presentation/tk/control_panel.py index 452c906..da99525 100644 --- a/src/app/presentation/tk/control_panel.py +++ b/src/app/presentation/tk/control_panel.py @@ -214,4 +214,6 @@ def _connection_chip_colors(text: str) -> tuple[str, str]: return "#ecfdf5", GREEN if "checking" in normalized or "connecting" in normalized or "scan" in normalized: return "#fff7ed", AMBER + if "not used" in normalized or "not evaluated" in normalized: + return "#f1f5f9", MUTED return "#fef2f2", RED diff --git a/src/app/presentation/tk/controller.py b/src/app/presentation/tk/controller.py index 0d1ae64..a46778c 100644 --- a/src/app/presentation/tk/controller.py +++ b/src/app/presentation/tk/controller.py @@ -103,6 +103,7 @@ def initialize(self) -> None: on_close=self.on_close, on_figure_change=self.on_figure_change, on_mag_phase_change=self.on_mag_phase_change, + on_plot_scale_change=self.on_plot_scale_change, ) self._bind_reference_receipt_traces() @@ -239,21 +240,28 @@ def on_load_reference(self) -> None: return try: - 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) - 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") + self.load_reference_from_path(fp) 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 load_reference_from_path(self, path: str | Path, *, record: bool = True) -> tuple[str, ...]: + """Load a reference without a file dialog for tests and reproducible demo tooling.""" + reference_path = Path(path) + curve, interpolator = self.load_reference_use_case.execute(str(reference_path)) + self._reference_interpolator = interpolator + self._reference_curve = curve + self._reference_path = reference_path + self.window.plot_widget.set_reference_coverage( + float(np.min(curve.freq_hz)), + float(np.max(curve.freq_hz)), + ) + self.vm.calibration_enabled.set(True) + warnings = self._refresh_reference_receipt(record=record) + self.vm.status_text.set("Reference loaded with coverage warning" if warnings else "Reference loaded") + return warnings + def on_scan_resources(self) -> None: try: scan = self._discovery_service.scan_resources() @@ -282,6 +290,9 @@ def on_figure_change(self) -> None: def on_mag_phase_change(self) -> None: self._ui_handler.refresh_plot() + def on_plot_scale_change(self) -> None: + self._ui_handler.refresh_plot() + def _load_measurement_from_path(self, path: Path, *, force_fixture: bool = False) -> None: loaded = self.load_measurement_use_case.execute(str(path)) self._ui_handler.set_result(loaded.result) @@ -405,7 +416,7 @@ def dialogs_to_target(path, window: AppWindow): ) -def _describe_loaded_source(path: Path, raw_payload: dict[str, np.ndarray]) -> dict[str, object]: +def _describe_loaded_source(path: Path, raw_payload: dict[str, object]) -> dict[str, object]: source = _payload_text(raw_payload, "source") label = _payload_text(raw_payload, "demo_label") or "Loaded fixture" is_fixture = source == "mock_fixture" or "hyperframe_simulated_fixture" in path.name @@ -415,7 +426,7 @@ def _describe_loaded_source(path: Path, raw_payload: dict[str, np.ndarray]) -> d } -def _payload_text(payload: dict[str, np.ndarray], key: str) -> str: +def _payload_text(payload: dict[str, object], key: str) -> str: value = payload.get(key) if value is None: return "" diff --git a/src/app/presentation/tk/plot_widget.py b/src/app/presentation/tk/plot_widget.py index 0edcbdc..8357ee1 100644 --- a/src/app/presentation/tk/plot_widget.py +++ b/src/app/presentation/tk/plot_widget.py @@ -7,6 +7,7 @@ from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg from matplotlib.figure import Figure +from app.domain.plotting import choose_frequency_scale from app.domain.models import SweepResult from app.presentation.tk.view_model import ViewModel from app.shared.cvt_tools import CvtTools @@ -24,6 +25,8 @@ class PlotWidget: def __init__(self, parent: tk.Misc, vm: ViewModel) -> None: self._vm = vm + self._reference_coverage_hz: tuple[float, float] | None = None + self._reference_spans: list[object] = [] self.frame = tk.Frame(parent, bg=PANEL_BG) self.frame.grid_rowconfigure(1, weight=1) self.frame.grid_columnconfigure(0, weight=1) @@ -31,9 +34,10 @@ def __init__(self, parent: tk.Misc, vm: ViewModel) -> None: self._build_header() self._build_plots() - def bind_controls(self, *, on_figure_change, on_mag_phase_change) -> None: + def bind_controls(self, *, on_figure_change, on_mag_phase_change, on_plot_scale_change) -> None: self.cmb_figure.bind("<>", lambda _e: on_figure_change()) self.cmb_mag_phase.bind("<>", lambda _e: on_mag_phase_change()) + self.cmb_plot_scale.bind("<>", lambda _e: on_plot_scale_change()) def _build_header(self) -> None: header = tk.Frame(self.frame, bg=CARD_BG, highlightbackground="#d9e0e8", highlightthickness=1) @@ -95,6 +99,21 @@ def _build_header(self) -> None: ) self.cmb_mag_phase.grid(row=1, column=1, sticky="ew", pady=(5, 0)) + tk.Label(controls, text="X axis", bg=CARD_BG, fg=MUTED).grid( + row=2, + column=0, + sticky="e", + padx=(0, 6), + ) + self.cmb_plot_scale = ttk.Combobox( + controls, + textvariable=self._vm.plot_scale, + values=["auto", "linear", "log"], + width=16, + state="readonly", + ) + self.cmb_plot_scale.grid(row=2, column=1, sticky="ew", pady=(5, 0)) + tk.Label( controls, textvariable=self._vm.point_count_text, @@ -102,7 +121,7 @@ def _build_header(self) -> None: fg=BLUE, font=("TkDefaultFont", 10, "bold"), anchor="e", - ).grid(row=2, column=0, columnspan=2, sticky="ew", pady=(7, 0)) + ).grid(row=3, column=0, columnspan=2, sticky="ew", pady=(7, 0)) def _build_plots(self) -> None: plot_card = tk.Frame(self.frame, bg=CARD_BG, highlightbackground="#d9e0e8", highlightthickness=1) @@ -161,6 +180,9 @@ def set_mode(self, mode: str) -> None: self._canvas_db.get_tk_widget().grid_remove() self._canvas_gain.get_tk_widget().grid(row=0, column=0, sticky="nsew", padx=10, pady=10) + def set_reference_coverage(self, minimum_hz: float, maximum_hz: float) -> None: + self._reference_coverage_hz = (min(minimum_hz, maximum_hz), max(minimum_hz, maximum_hz)) + def update_result(self, result: SweepResult, freq_unit: str, mag_phase_mode: str) -> None: freq_hz = np.array([p.freq_hz for p in result.points], dtype=float) gain = np.array([p.gain_linear for p in result.points], dtype=float) @@ -183,6 +205,14 @@ def update_result(self, result: SweepResult, freq_unit: str, mag_phase_mode: str self._line_phase_gain.set_data(px, py) self._line_phase_db.set_data(px, py) + x_scale = choose_frequency_scale( + freq_hz, + requested=self._vm.plot_scale.get(), + sweep_is_log=bool(self._vm.is_log.get()), + ) + self._ax_gain.set_xscale(x_scale) + self._ax_db.set_xscale(x_scale) + if mag_phase_mode == "magnitude": self._line_gain.set_visible(True) self._line_db.set_visible(True) @@ -202,6 +232,7 @@ def update_result(self, result: SweepResult, freq_unit: str, mag_phase_mode: str self._ax_gain.set_xlabel(f"Frequency ({freq_unit})") self._ax_db.set_xlabel(f"Frequency ({freq_unit})") + self._update_reference_spans(freq_hz=freq_hz, unit_scale=scale) self._autoscale() self._canvas_gain.draw_idle() self._canvas_db.draw_idle() @@ -211,5 +242,29 @@ def _autoscale(self) -> None: ax.relim() ax.autoscale_view() + def _update_reference_spans(self, *, freq_hz: np.ndarray, unit_scale: float) -> None: + for span in self._reference_spans: + span.remove() + self._reference_spans.clear() + if self._reference_coverage_hz is None or freq_hz.size == 0: + return + + data_min = float(np.min(freq_hz)) / unit_scale + data_max = float(np.max(freq_hz)) / unit_scale + ref_min = self._reference_coverage_hz[0] / unit_scale + ref_max = self._reference_coverage_hz[1] / unit_scale + left_end = min(ref_min, data_max) + right_start = max(ref_max, data_min) + + for axis in (self._ax_gain, self._ax_db): + if data_min < left_end: + self._reference_spans.append( + axis.axvspan(data_min, left_end, color="#f59e0b", alpha=0.10, zorder=0) + ) + if right_start < data_max: + self._reference_spans.append( + axis.axvspan(right_start, data_max, color="#f59e0b", alpha=0.10, zorder=0) + ) + def figures(self) -> dict[str, Figure]: return {"gain": self._fig_gain, "db": self._fig_db} diff --git a/src/app/presentation/tk/ui_event_handler.py b/src/app/presentation/tk/ui_event_handler.py index 58ab4a2..950f475 100644 --- a/src/app/presentation/tk/ui_event_handler.py +++ b/src/app/presentation/tk/ui_event_handler.py @@ -51,12 +51,17 @@ def set_result(self, result: SweepResult, *, refresh_plot: bool = True) -> None: self.refresh_plot() def set_live_source(self) -> None: + self._vm.source_mode.set("live") self._vm.data_source_text.set("Live instrument path") self._vm.fixture_badge_text.set("") self._vm.validation_receipt_text.set("Live run requires operator hardware checks") def set_fixture_source(self, *, label: str, path_name: str) -> None: point_count = len(self._latest_result.points) + self._vm.source_mode.set("fixture") + self._vm.figure_mode.set("gain_db") + self._vm.magnitude_phase_mode.set("magnitude_phase") + self._vm.plot_scale.set("log") self._vm.data_source_text.set(f"Fixture replay · {path_name}") self._vm.fixture_badge_text.set("No hardware - simulated fixture") self._vm.validation_receipt_text.set(f"{label}; not live hardware validation") @@ -66,14 +71,19 @@ def set_fixture_source(self, *, label: str, path_name: str) -> None: self._vm.progress_text.set(f"{point_count} / {point_count}") self._vm.latest_frequency_text.set(_format_frequency(self._latest_result.points[-1].freq_hz)) self._vm.elapsed_text.set("00:00") + self._window.set_connection_idle() + self._window.plot_widget.set_mode("gain_db") + self.refresh_plot() self.record_event(f"Loaded simulated fixture: {path_name}") def set_loaded_source(self, *, path_name: str) -> None: + self._vm.source_mode.set("loaded") self._vm.data_source_text.set(f"Loaded measurement · {path_name}") self._vm.fixture_badge_text.set("") self._vm.validation_receipt_text.set("Loaded file; live hardware state not implied") self._vm.export_receipt_text.set("Loaded measurement; Save Data exports the current result") self._vm.run_state_text.set("Data loaded") + self._window.set_connection_idle() self.record_event(f"Loaded measurement: {path_name}") def set_reference_loaded( diff --git a/src/app/presentation/tk/view_model.py b/src/app/presentation/tk/view_model.py index 380fff6..e95459a 100644 --- a/src/app/presentation/tk/view_model.py +++ b/src/app/presentation/tk/view_model.py @@ -47,6 +47,8 @@ def __init__(self, root: tk.Misc) -> None: self.figure_mode = tk.StringVar(root, value="gain") self.magnitude_phase_mode = tk.StringVar(root, value="magnitude") + self.plot_scale = tk.StringVar(root, value="auto") + self.source_mode = tk.StringVar(root, value="live") self.status_text = tk.StringVar(root, value="Ready") self.run_state_text = tk.StringVar(root, value="Idle") diff --git a/src/main.py b/src/main.py index 0c9cacc..585684e 100644 --- a/src/main.py +++ b/src/main.py @@ -1,11 +1,49 @@ from __future__ import annotations +import json +import os +from pathlib import Path +import sys +import traceback + + +def main() -> int: + if "--package-smoke" in sys.argv[1:]: + try: + from app.demo.package_smoke import run_package_smoke + + run_package_smoke() + except Exception as exc: + _write_package_smoke_failure(exc) + return 1 + return 0 -def main() -> None: from app.bootstrap import run_desktop_app run_desktop_app() + return 0 + + +def _write_package_smoke_failure(exc: Exception) -> None: + root = Path(os.environ.get("AUTO_LOAD_OFF_TEST_ROOT", Path.cwd())).resolve() + receipt_path = root / "__data__" / "package_smoke_receipt.json" + receipt_path.parent.mkdir(parents=True, exist_ok=True) + receipt_path.write_text( + json.dumps( + { + "status": "failed", + "error": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc(), + "validation_boundary": "No hardware - simulated fixture; not live hardware validation", + "live_hardware_used": False, + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/tests/test_data_validation.py b/tests/test_data_validation.py new file mode 100644 index 0000000..3718972 --- /dev/null +++ b/tests/test_data_validation.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import sys +from pathlib import Path +import tempfile +import unittest + +from scipy.io import savemat + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.application.dto import SaveTarget +from app.domain.data_validation import DataValidationError, normalize_measurement_arrays +from app.domain.models import SweepPoint, SweepResult +from app.infrastructure.persistence.measurement_exporter import MeasurementExporter +from app.infrastructure.persistence.measurement_loader import MeasurementLoader +from app.infrastructure.persistence.reference_repo_mat import MatReferenceRepository +from app.infrastructure.persistence.settings_defaults import DefaultSettingsFactory + + +class DataValidationTests(unittest.TestCase): + def test_measurement_frequency_must_be_positive_and_strictly_increasing(self) -> None: + for freq, message in ( + ([0.0, 1_000.0], "must be positive"), + ([1_000.0, 1_000.0], "strictly increasing"), + ([2_000.0, 1_000.0], "strictly increasing"), + ): + with self.subTest(freq=freq): + with self.assertRaisesRegex(DataValidationError, message): + normalize_measurement_arrays( + freq_hz=freq, + gain_linear=[1.0, 1.0], + gain_db=[0.0, 0.0], + phase_deg=None, + ) + + def test_measurement_arrays_must_have_matching_lengths_and_finite_gain(self) -> None: + with self.assertRaisesRegex(DataValidationError, "does not match frequency length"): + normalize_measurement_arrays( + freq_hz=[1_000.0, 2_000.0], + gain_linear=[1.0], + gain_db=[0.0, 0.0], + phase_deg=None, + ) + + with self.assertRaisesRegex(DataValidationError, "gain_db contains NaN"): + normalize_measurement_arrays( + freq_hz=[1_000.0, 2_000.0], + gain_linear=None, + gain_db=[0.0, float("nan")], + phase_deg=None, + ) + + def test_csv_requires_frequency_and_a_gain_column(self) -> None: + loader = MeasurementLoader() + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "bad.csv" + path.write_text("gain_db\n0.0\n", encoding="utf-8") + with self.assertRaisesRegex(DataValidationError, "missing required column: freq_hz"): + loader.load(str(path)) + + path.write_text("freq_hz,phase_deg\n1000,0\n", encoding="utf-8") + with self.assertRaisesRegex(DataValidationError, "requires gain_linear or gain_db"): + loader.load(str(path)) + + def test_csv_can_compute_linear_gain_from_db_without_silent_zero_fill(self) -> None: + loader = MeasurementLoader() + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "db_only.csv" + path.write_text( + "source,validation_boundary,freq_hz,gain_db,phase_deg\n" + "mock_fixture,No hardware fixture,1000,0,\n" + "mock_fixture,No hardware fixture,2000,6.020599913279624,10\n", + encoding="utf-8", + ) + loaded = loader.load(str(path)) + + self.assertAlmostEqual(loaded.result.points[0].gain_linear, 1.0, delta=1e-9) + self.assertAlmostEqual(loaded.result.points[1].gain_linear, 2.0, delta=1e-9) + self.assertIsNone(loaded.result.points[0].phase_deg) + self.assertEqual(loaded.result.meta["source"], "mock_fixture") + self.assertIn("No hardware", loaded.result.meta["validation_boundary"]) + + def test_csv_blank_required_gain_value_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "blank.csv" + path.write_text("freq_hz,gain_db\n1000,\n", encoding="utf-8") + with self.assertRaisesRegex(DataValidationError, "row 2 has no value for gain_db"): + MeasurementLoader().load(str(path)) + + def test_mat_measurement_and_reference_reject_length_or_order_errors(self) -> None: + with tempfile.TemporaryDirectory() as td: + root = Path(td) + measurement = root / "bad_measurement.mat" + savemat(measurement, {"freq_hz": [1_000.0, 2_000.0], "gain_db": [0.0]}) + with self.assertRaisesRegex(DataValidationError, "does not match frequency length"): + MeasurementLoader().load(str(measurement)) + + reference = root / "bad_reference.mat" + savemat(reference, {"freq_hz": [1_000.0, 1_000.0], "gain_db": [0.0, 0.0]}) + with self.assertRaisesRegex(DataValidationError, "strictly increasing"): + MatReferenceRepository().load_reference(str(reference)) + + def test_export_rejects_invalid_result_before_writing_artifacts(self) -> None: + result = SweepResult( + points=[ + SweepPoint(freq_hz=2_000.0, gain_linear=1.0, gain_db=0.0), + SweepPoint(freq_hz=1_000.0, gain_linear=1.0, gain_db=0.0), + ] + ) + with tempfile.TemporaryDirectory() as td: + target = SaveTarget(base_path=Path(td) / "invalid", figures={}) + with self.assertRaisesRegex(DataValidationError, "strictly increasing"): + MeasurementExporter().export(result, DefaultSettingsFactory().create(), target) + self.assertFalse((Path(td) / "invalid.mat").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hardware_free_workflow.py b/tests/test_hardware_free_workflow.py new file mode 100644 index 0000000..e320b92 --- /dev/null +++ b/tests/test_hardware_free_workflow.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path +import tempfile +import unittest + +import numpy as np +from scipy.io import loadmat + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.application.dto import SaveTarget +from app.application.services.export_receipts import build_export_receipt +from app.demo.hyperframe_fixture import POINT_COUNT, build_fixture_settings +from app.domain.calibration import apply_reference_to_point, build_reference_interpolator +from app.domain.models import SweepPoint, SweepResult +from app.infrastructure.persistence.measurement_exporter import MeasurementExporter +from app.infrastructure.persistence.measurement_loader import MeasurementLoader +from app.infrastructure.persistence.reference_repo_mat import MatReferenceRepository + + +class HardwareFreeWorkflowTests(unittest.TestCase): + def test_fixture_reference_correction_export_and_reload_end_to_end(self) -> None: + demo_dir = Path(__file__).resolve().parents[1] / "demo_data" + fixture_path = demo_dir / "hyperframe_simulated_fixture.mat" + reference_path = demo_dir / "hyperframe_reference_fixture.mat" + + loaded = MeasurementLoader().load(str(fixture_path)) + reference = MatReferenceRepository().load_reference(str(reference_path)) + interpolator = build_reference_interpolator(reference) + raw = loaded.raw_payload + + freq = _array(raw, "freq_hz") + raw_gain_db = _array(raw, "gain_db_raw") + raw_phase_deg = _array(raw, "phase_deg_raw") + expected_gain_db = _array(raw, "gain_db_corrected") + expected_phase_deg = _array(raw, "phase_deg_corrected") + + corrected_points: list[SweepPoint] = [] + for freq_hz, gain_db, phase_deg in zip(freq, raw_gain_db, raw_phase_deg, strict=True): + gain_linear = float(10.0 ** (gain_db / 20.0)) + gain_complex = gain_linear * np.exp(1j * np.deg2rad(phase_deg)) + raw_point = SweepPoint( + freq_hz=float(freq_hz), + gain_linear=gain_linear, + gain_db=float(gain_db), + phase_deg=float(phase_deg), + gain_complex=complex(gain_complex), + ) + corrected_points.append( + apply_reference_to_point( + raw_point, + interpolator(np.array([freq_hz], dtype=float))[0], + use_phase=True, + ) + ) + + corrected = SweepResult(points=corrected_points, meta=dict(loaded.result.meta)) + corrected.meta["reference_file"] = reference_path.name + corrected.meta["correction_mode"] = "dual" + + self.assertEqual(len(corrected.points), POINT_COUNT) + np.testing.assert_allclose( + np.array([point.gain_db for point in corrected.points]), + expected_gain_db, + atol=1e-8, + ) + phase_delta = _phase_delta( + np.array([point.phase_deg for point in corrected.points], dtype=float), + expected_phase_deg, + ) + np.testing.assert_allclose(phase_delta, np.zeros_like(phase_delta), atol=1e-8) + + settings = build_fixture_settings() + with tempfile.TemporaryDirectory() as td: + artifacts = MeasurementExporter().export( + corrected, + settings, + SaveTarget(base_path=Path(td) / "corrected_fixture", figures={}), + ) + reloaded_mat = MeasurementLoader().load(str(artifacts.mat_path)) + reloaded_csv = MeasurementLoader().load(str(artifacts.csv_path)) + mat_payload = loadmat(artifacts.mat_path) + txt_header = artifacts.txt_path.read_text(encoding="utf-8").splitlines()[0] + + receipt = build_export_receipt( + artifacts=artifacts, + settings=settings, + result=corrected, + source_text="Fixture replay · hyperframe_simulated_fixture.mat", + fixture_badge_text="No hardware - simulated fixture", + ) + + for reloaded in (reloaded_mat, reloaded_csv): + self.assertEqual(len(reloaded.result.points), POINT_COUNT) + np.testing.assert_allclose( + np.array([point.gain_db for point in reloaded.result.points]), + expected_gain_db, + atol=1e-8, + ) + self.assertEqual(reloaded.result.meta["source"], "mock_fixture") + self.assertIn("not live hardware validation", reloaded.result.meta["validation_boundary"]) + + metadata = json.loads(_mat_text(mat_payload["metadata_json"])) + self.assertEqual(metadata["export"]["source"], "mock_fixture") + self.assertEqual(metadata["export"]["point_count"], POINT_COUNT) + self.assertIn("not live hardware validation", metadata["export"]["validation_boundary"]) + self.assertEqual(txt_header, "freq_hz\tgain_linear\tgain_db\tphase_deg") + self.assertIn("No hardware simulated fixture", receipt.summary) + + +def _array(payload: dict[str, object], key: str) -> np.ndarray: + return np.atleast_1d(np.asarray(payload[key], dtype=float).squeeze()) + + +def _phase_delta(actual: np.ndarray, expected: np.ndarray) -> np.ndarray: + return (actual - expected + 180.0) % 360.0 - 180.0 + + +def _mat_text(value: object) -> str: + array = np.asarray(value) + if array.dtype.kind in {"U", "S"}: + if array.ndim == 2 and array.shape[0] == 1: + return "".join(str(item) for item in array[0]).strip() + return "".join(str(item) for item in array.ravel()).strip() + return str(array.squeeze()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_main_cli.py b/tests/test_main_cli.py new file mode 100644 index 0000000..0776dac --- /dev/null +++ b/tests/test_main_cli.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +import main + + +class MainCliTests(unittest.TestCase): + def test_package_smoke_failure_returns_nonzero_and_writes_receipt(self) -> None: + with tempfile.TemporaryDirectory() as td: + with ( + patch.dict(os.environ, {"AUTO_LOAD_OFF_TEST_ROOT": td}), + patch.object(sys, "argv", ["main.py", "--package-smoke"]), + patch("app.demo.package_smoke.run_package_smoke", side_effect=RuntimeError("test failure")), + ): + exit_code = main.main() + + receipt_path = Path(td) / "__data__" / "package_smoke_receipt.json" + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + + self.assertEqual(exit_code, 1) + self.assertEqual(receipt["status"], "failed") + self.assertIn("RuntimeError: test failure", receipt["error"]) + self.assertFalse(receipt["live_hardware_used"]) + self.assertIn("not live hardware validation", receipt["validation_boundary"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_package_smoke.py b/tests/test_package_smoke.py new file mode 100644 index 0000000..8200464 --- /dev/null +++ b/tests/test_package_smoke.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import json +from pathlib import Path +import sys +import tempfile +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.demo.package_smoke import RECEIPT_NAME, VALIDATION_BOUNDARY, run_package_smoke + + +class PackageSmokeTests(unittest.TestCase): + def test_package_smoke_exports_and_reloads_fixture_without_hardware(self) -> None: + repo_root = Path(__file__).resolve().parents[1] + with tempfile.TemporaryDirectory() as td: + runtime_root = Path(td) + receipt_path = run_package_smoke(runtime_root=runtime_root, resource_root=repo_root) + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + + self.assertEqual(receipt_path, (runtime_root / "__data__" / RECEIPT_NAME).resolve()) + self.assertEqual(receipt["status"], "passed") + self.assertEqual(receipt["source"], "mock_fixture") + self.assertEqual(receipt["point_count"], 72) + self.assertFalse(receipt["live_hardware_used"]) + self.assertEqual(receipt["validation_boundary"], VALIDATION_BOUNDARY) + self.assertEqual(len(receipt["artifacts"]), 3) + for artifact in receipt["artifacts"]: + path = Path(artifact) + self.assertTrue(path.is_file()) + self.assertGreater(path.stat().st_size, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_plotting.py b/tests/test_plotting.py new file mode 100644 index 0000000..6c261d6 --- /dev/null +++ b/tests/test_plotting.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import sys +from pathlib import Path +import unittest + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.domain.plotting import choose_frequency_scale + + +class PlottingPolicyTests(unittest.TestCase): + def test_auto_detects_log_spaced_fixture_frequency(self) -> None: + freq = np.geomspace(1_000.0, 1_000_000.0, 72) + self.assertEqual(choose_frequency_scale(freq), "log") + + def test_auto_keeps_linear_spaced_frequency_linear(self) -> None: + freq = np.linspace(1_000.0, 10_000.0, 20) + self.assertEqual(choose_frequency_scale(freq), "linear") + + def test_sweep_mode_and_explicit_selection_override_detection(self) -> None: + freq = np.linspace(1_000.0, 10_000.0, 20) + self.assertEqual(choose_frequency_scale(freq, sweep_is_log=True), "log") + self.assertEqual(choose_frequency_scale(freq, requested="log"), "log") + self.assertEqual(choose_frequency_scale(np.geomspace(1.0, 100.0, 10), requested="linear"), "linear") + + def test_invalid_auto_data_falls_back_to_linear(self) -> None: + self.assertEqual(choose_frequency_scale([0.0, 1.0, 2.0]), "linear") + with self.assertRaisesRegex(ValueError, "Unsupported frequency scale"): + choose_frequency_scale([1.0, 2.0], requested="decade") + + def test_requested_log_waits_for_first_frequency(self) -> None: + self.assertEqual(choose_frequency_scale([], requested="log"), "linear") + self.assertEqual(choose_frequency_scale([1_000.0], requested="log"), "log") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflow_receipts.py b/tests/test_workflow_receipts.py index fcd7569..5760081 100644 --- a/tests/test_workflow_receipts.py +++ b/tests/test_workflow_receipts.py @@ -34,6 +34,40 @@ def __init__(self) -> None: self.reference_receipt_text = FakeVar("No reference loaded") self.export_receipt_text = FakeVar("No export yet") self.event_history_text = FakeVar("No warnings or workflow events") + self.source_mode = FakeVar("live") + self.figure_mode = FakeVar("gain") + self.magnitude_phase_mode = FakeVar("magnitude") + self.plot_scale = FakeVar("auto") + self.data_source_text = FakeVar("Live instrument path") + self.fixture_badge_text = FakeVar("") + self.validation_receipt_text = FakeVar("") + self.run_state_text = FakeVar("Idle") + self.progress_text = FakeVar("0 / 0") + self.latest_frequency_text = FakeVar("-") + self.elapsed_text = FakeVar("00:00") + self.point_count_text = FakeVar("0 points") + self.freq_unit = FakeVar("Hz") + + +class FakePlotWidget: + def __init__(self) -> None: + self.mode = "gain" + self.updated = False + + def set_mode(self, mode: str) -> None: + self.mode = mode + + def update_result(self, *_args) -> None: + self.updated = True + + +class FakeWindow: + def __init__(self) -> None: + self.plot_widget = FakePlotWidget() + self.connection_idle = False + + def set_connection_idle(self) -> None: + self.connection_idle = True class WorkflowReceiptTests(unittest.TestCase): @@ -136,6 +170,30 @@ def test_ui_handler_updates_reference_export_and_event_history_state(self) -> No self.assertIn("outside reference coverage", vm.event_history_text.get()) self.assertIn("Export artifacts saved", vm.event_history_text.get()) + def test_fixture_source_uses_bode_defaults_and_neutral_connection_state(self) -> None: + vm = ReceiptViewModel() + window = FakeWindow() + handler = UiEventHandler(window=window, vm=vm) + handler.set_result( + SweepResult( + points=[SweepPoint(freq_hz=1_000.0, gain_linear=1.0, gain_db=0.0, phase_deg=0.0)] + ), + refresh_plot=False, + ) + + handler.set_fixture_source( + label="Simulated no-hardware demo fixture", + path_name="hyperframe_simulated_fixture.mat", + ) + + self.assertEqual(vm.source_mode.get(), "fixture") + self.assertEqual(vm.figure_mode.get(), "gain_db") + self.assertEqual(vm.magnitude_phase_mode.get(), "magnitude_phase") + self.assertEqual(vm.plot_scale.get(), "log") + self.assertTrue(window.connection_idle) + self.assertEqual(window.plot_widget.mode, "gain_db") + self.assertTrue(window.plot_widget.updated) + if __name__ == "__main__": unittest.main()