From 3f0f2bd47cb44166cc4adb86ab9ddd70f9136eef Mon Sep 17 00:00:00 2001 From: ShehaoLi Date: Tue, 2 Jun 2026 02:47:32 -0700 Subject: [PATCH] Add capability discovery packaging sprint --- README.md | 1 + docs/architecture.md | 9 +- docs/extending.md | 15 +- docs/packaging.md | 85 ++++++++ docs/safety.md | 2 + .../auto_load_off_test_onefolder.spec | 66 ++++++ scripts/build_windows_onefolder.ps1 | 33 +++ src/app/application/ports/instruments.py | 4 + .../services/instrument_discovery.py | 124 +++++++++++ src/app/bootstrap.py | 3 +- src/app/domain/instrument_capabilities.py | 206 ++++++++++++++++++ src/app/domain/validators.py | 62 +++++- .../instruments/adapter_registry.py | 64 ++++++ .../instruments/equips_factory.py | 20 +- .../instruments/identity_probe.py | 25 +++ .../instruments/mock_adapters.py | 98 +++++++++ src/app/infrastructure/instruments/ports.py | 18 +- src/app/presentation/tk/app_window.py | 6 + src/app/presentation/tk/control_panel.py | 21 ++ src/app/presentation/tk/controller.py | 52 ++++- src/app/presentation/tk/view_model.py | 1 + src/app/shared/mapping.py | 6 +- tests/test_adapter_registry.py | 106 +++++++++ tests/test_instrument_capabilities.py | 113 ++++++++++ tests/test_instrument_discovery.py | 129 +++++++++++ 25 files changed, 1250 insertions(+), 19 deletions(-) create mode 100644 docs/packaging.md create mode 100644 packaging/pyinstaller/auto_load_off_test_onefolder.spec create mode 100644 scripts/build_windows_onefolder.ps1 create mode 100644 src/app/application/services/instrument_discovery.py create mode 100644 src/app/domain/instrument_capabilities.py create mode 100644 src/app/infrastructure/instruments/adapter_registry.py create mode 100644 src/app/infrastructure/instruments/identity_probe.py create mode 100644 src/app/infrastructure/instruments/mock_adapters.py create mode 100644 tests/test_adapter_registry.py create mode 100644 tests/test_instrument_capabilities.py create mode 100644 tests/test_instrument_discovery.py diff --git a/README.md b/README.md index 414b77d..f8c759e 100644 --- a/README.md +++ b/README.md @@ -150,6 +150,7 @@ See [docs/safety.md](docs/safety.md) for stop/shutdown behavior and hardware ass - [Operator Guide](docs/operator_guide.md) - [Safety Notes](docs/safety.md) - [Extending The Application](docs/extending.md) +- [Packaging](docs/packaging.md) - [Case Study](docs/case_study.md) - [Hyperframe Demo Fixture](docs/hyperframe_demo.md) - [Hyperframe Capture Plan](docs/hyperframe_capture_plan.md) diff --git a/docs/architecture.md b/docs/architecture.md index e18f105..d815f16 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -24,9 +24,9 @@ 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, validation, sweep generation, DSP, calibration, and export array shaping. + - Pure dataclasses, enums, instrument capability profiles, validation, sweep generation, DSP, calibration, and export array shaping. - `app/infrastructure` - - Adapter wrappers around `src/equips.py`. + - Adapter registry and wrappers around `src/equips.py`. - JSON settings and MAT/CSV/TXT persistence. ## Dependency Rules @@ -62,8 +62,11 @@ 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. - AWG and OSC commands are executed through `AwgPort` and `OscPort` adapters. -- Connection scanning is provided by `PyVisaResourceScanner` and `ConnectionMonitor`. +- 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. - `src/equips.py` is intentionally treated as a vendor compatibility layer. It contains legacy SCPI/serial behavior that should not be casually refactored without physical instrument verification. ## Persistence diff --git a/docs/extending.md b/docs/extending.md index 2e214ec..6adbc45 100644 --- a/docs/extending.md +++ b/docs/extending.md @@ -30,11 +30,18 @@ Use `AppPaths` instead of recomputing `Path(__file__).parents[...]` in new code. ## Adding A New Instrument -1. Add or verify the model label in `src/app/shared/mapping.py`. -2. Add the vendor driver mapping in `src/equips.py` only if the low-level SCPI behavior is known. -3. Prefer adding behavior through `app.infrastructure.instruments` adapters rather than calling `equips.py` from UI or use cases. +1. Add an `InstrumentCapability` profile in `src/app/domain/instrument_capabilities.py`. + Keep the profile conservative: include only supported channels, modes, limits, transports, + and safety notes that are known or explicitly marked as software assumptions. +2. Register the model in `app.infrastructure.instruments.adapter_registry`. + Current production adapters may wrap `src/equips.py`, but the UI/use cases should not call + `equips.py` directly. +3. Add or verify the vendor driver mapping in `src/equips.py` only when the low-level SCPI behavior + is known and can be bench-validated. 4. Keep `AwgPort` / `OscPort` as the application contract. -5. Add hardware-free tests with fake ports before doing live bench validation. +5. Add hardware-free tests with fake ports and capability-aware validation before live bench validation. +6. Add the model to live validation notes only after scan, IDN, configure, sweep, stop, and export have + been checked on the actual instrument. ## Adding A New Persistence Format diff --git a/docs/packaging.md b/docs/packaging.md new file mode 100644 index 0000000..a768b59 --- /dev/null +++ b/docs/packaging.md @@ -0,0 +1,85 @@ +# Packaging + +This project can be packaged for review or lab workstation setup, but packaged +artifacts are not live-hardware validated by default. Hardware use still depends +on the target machine's VISA backend, instrument drivers, cabling, and operator +safety checks. + +## Recommended First Target + +Use a Windows PyInstaller one-folder build first. One-folder output is easier to +debug than a one-file executable when bundling Tkinter, Matplotlib, SciPy, +PyVISA, and instrument-driver dependencies. + +```powershell +powershell -ExecutionPolicy Bypass -File scripts/build_windows_onefolder.ps1 +``` + +The script installs the local package with the optional `build` dependency and +then runs: + +```powershell +python -m PyInstaller packaging/pyinstaller/auto_load_off_test_onefolder.spec --clean --noconfirm +``` + +Expected output: + +```text +dist/AutoLoadOffTest/AutoLoadOffTest.exe +``` + +## External Prerequisites + +The package does not bundle lab driver runtimes. A live-hardware workstation +still needs one of these paths configured: + +- NI-VISA, Keysight IO Libraries, or another compatible VISA runtime. +- Or a working `pyvisa-py` backend plus any USB/GPIB/serial support libraries + required by the connected instruments. +- OS-level USB/GPIB/serial permissions where relevant. +- Verified instrument addresses and model labels. + +## Runtime Data Paths + +Set `AUTO_LOAD_OFF_TEST_ROOT` on packaged workstations so settings and saved +measurements do not depend on the launch directory: + +```powershell +$env:AUTO_LOAD_OFF_TEST_ROOT = "$env:LOCALAPPDATA\\AutoLoadOffTest" +``` + +The app writes: + +- `__config__/settings.json` +- `__data__/measurement/` + +under that root. + +## No-Hardware Packaging Smoke + +Before using a packaged artifact as portfolio/demo evidence: + +1. Launch `dist/AutoLoadOffTest/AutoLoadOffTest.exe`. +2. Confirm the operator console opens without a Python traceback. +3. Click `Load Demo Fixture`. +4. Confirm the plot is visible and labeled `No hardware - simulated fixture`. +5. Confirm the source receipt names `hyperframe_simulated_fixture.mat`. +6. Click `Save Data` and verify MAT/CSV/TXT files are written to a writable path. +7. Close the app and confirm no shutdown error appears. + +This smoke check validates packaged UI/data workflow only. It is not live +hardware validation. + +## Live-Hardware Packaging Smoke + +Do this only on a real lab workstation: + +1. Install/verify the VISA backend and drivers. +2. Set `AUTO_LOAD_OFF_TEST_ROOT` to a writable app-data folder. +3. Launch the packaged app. +4. Use `Scan Resources` and `Test Connect`. +5. Confirm IDN/address/model status before starting a sweep. +6. Run a short, conservative sweep into a safe load/DUT. +7. Verify Stop turns the AWG output off and warnings are visible. + +Record the instrument models, VISA backend, OS version, and result artifacts. diff --git a/docs/safety.md b/docs/safety.md index 206e74b..a2a5c53 100644 --- a/docs/safety.md +++ b/docs/safety.md @@ -9,6 +9,8 @@ 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`. +- 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. diff --git a/packaging/pyinstaller/auto_load_off_test_onefolder.spec b/packaging/pyinstaller/auto_load_off_test_onefolder.spec new file mode 100644 index 0000000..3bbd1bf --- /dev/null +++ b/packaging/pyinstaller/auto_load_off_test_onefolder.spec @@ -0,0 +1,66 @@ +# PyInstaller spec for a Windows one-folder build. +# Run from the repository root: +# python -m PyInstaller packaging/pyinstaller/auto_load_off_test_onefolder.spec --clean --noconfirm + +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] + +datas = [ + (str(ROOT / "demo_data"), "demo_data"), + (str(ROOT / "docs" / "images" / "auto-load-off-test-point-replay-demo.png"), "docs/images"), +] + +block_cipher = None + +a = Analysis( + [str(ROOT / "src" / "main.py")], + pathex=[str(ROOT / "src")], + binaries=[], + datas=datas, + hiddenimports=[ + "scipy.io", + "scipy.interpolate", + "matplotlib.backends.backend_tkagg", + "mplcursors", + "pyvisa", + "pyvisa_py", + "serial", + ], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + win_no_prefer_redirects=False, + win_private_assemblies=False, + cipher=block_cipher, + noarchive=False, +) +pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) +exe = EXE( + pyz, + a.scripts, + [], + exclude_binaries=True, + name="AutoLoadOffTest", + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + console=False, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +) +coll = COLLECT( + exe, + a.binaries, + a.zipfiles, + a.datas, + strip=False, + upx=True, + upx_exclude=[], + name="AutoLoadOffTest", +) diff --git a/scripts/build_windows_onefolder.ps1 b/scripts/build_windows_onefolder.ps1 new file mode 100644 index 0000000..86f6727 --- /dev/null +++ b/scripts/build_windows_onefolder.ps1 @@ -0,0 +1,33 @@ +param( + [string]$RuntimeRoot = "$env:LOCALAPPDATA\AutoLoadOffTest" +) + +$ErrorActionPreference = "Stop" + +$isWindowsPlatform = [System.Runtime.InteropServices.RuntimeInformation]::IsOSPlatform( + [System.Runtime.InteropServices.OSPlatform]::Windows +) +if (-not $isWindowsPlatform) { + Write-Warning "This packaging spike is intended for Windows. Continuing because PyInstaller may still validate the spec." +} + +$repoRoot = Resolve-Path (Join-Path $PSScriptRoot "..") +Set-Location $repoRoot + +python -m pip install --upgrade pip +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 + +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 "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" +Write-Host "4. Save Data to a writable folder and verify MAT/CSV/TXT files" +Write-Host "" +Write-Host "Live hardware still requires NI-VISA/Keysight IO Libraries or a working pyvisa backend plus drivers." diff --git a/src/app/application/ports/instruments.py b/src/app/application/ports/instruments.py index 1e44c2a..b06179a 100644 --- a/src/app/application/ports/instruments.py +++ b/src/app/application/ports/instruments.py @@ -41,6 +41,10 @@ class ResourceScannerPort(Protocol): def list_resources(self) -> tuple[str, ...]: ... +class InstrumentIdentityProbePort(Protocol): + def identify(self, address: str, timeout_ms: int | None = None) -> str: ... + + @dataclass(slots=True) class InstrumentPorts: awg: AwgPort diff --git a/src/app/application/services/instrument_discovery.py b/src/app/application/services/instrument_discovery.py new file mode 100644 index 0000000..61a996d --- /dev/null +++ b/src/app/application/services/instrument_discovery.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone + +from app.application.ports.instruments import InstrumentIdentityProbePort, ResourceScannerPort +from app.domain.instrument_capabilities import InstrumentRole, get_capability +from app.domain.models import InstrumentEndpoint, InstrumentSetup + + +@dataclass(frozen=True, slots=True) +class ResourceScan: + resources: tuple[str, ...] + scanned_at: str + + +@dataclass(frozen=True, slots=True) +class ConnectionCheck: + role: InstrumentRole + model: str + address: str + status: str + message: str + idn: str = "" + backend: str = "pyvisa" + last_seen: str = "" + + +class InstrumentDiscoveryService: + def __init__( + self, + *, + scanner: ResourceScannerPort, + identity_probe: InstrumentIdentityProbePort, + timeout_ms: int = 2_000, + ) -> None: + self._scanner = scanner + self._identity_probe = identity_probe + self._timeout_ms = timeout_ms + + def scan_resources(self) -> ResourceScan: + resources = self._scanner.list_resources() + return ResourceScan(resources=resources, scanned_at=_timestamp()) + + def test_setup(self, setup: InstrumentSetup, resolve_address) -> tuple[ConnectionCheck, ConnectionCheck]: + return ( + self.test_endpoint( + role=InstrumentRole.AWG, + endpoint=setup.awg, + address=resolve_address(setup.awg), + ), + self.test_endpoint( + role=InstrumentRole.OSC, + endpoint=setup.osc, + address=resolve_address(setup.osc), + ), + ) + + def test_endpoint(self, *, role: InstrumentRole, endpoint: InstrumentEndpoint, address: str) -> ConnectionCheck: + if not address: + return ConnectionCheck( + role=role, + model=endpoint.model, + address="", + status="address_empty", + message=f"{role.value.upper()} address empty", + ) + + try: + capability = get_capability(endpoint.model, role) + except ValueError: + return ConnectionCheck( + role=role, + model=endpoint.model, + address=address, + status="unsupported_model", + message=f"Unsupported {role.value.upper()} model: {endpoint.model}", + ) + + if endpoint.connect_mode not in capability.transports: + return ConnectionCheck( + role=role, + model=endpoint.model, + address=address, + status="unsupported_transport", + message=f"{endpoint.model} does not support {endpoint.connect_mode.value} connection mode", + ) + + try: + idn = self._identity_probe.identify(address, timeout_ms=self._timeout_ms) + except Exception as exc: # noqa: BLE001 + return ConnectionCheck( + role=role, + model=endpoint.model, + address=address, + status="offline", + message=f"{role.value.upper()} offline or unreachable: {exc}", + ) + + return ConnectionCheck( + role=role, + model=endpoint.model, + address=address, + status="connected", + message=f"{role.value.upper()} connected: {idn}", + idn=idn, + last_seen=_timestamp(), + ) + + +def format_scan_receipt(scan: ResourceScan, *, limit: int = 4) -> str: + if not scan.resources: + return f"No VISA resources found · {scan.scanned_at}" + visible = ", ".join(scan.resources[:limit]) + suffix = "" if len(scan.resources) <= limit else f", +{len(scan.resources) - limit} more" + return f"{len(scan.resources)} VISA resources · {visible}{suffix}" + + +def format_connection_receipt(checks: tuple[ConnectionCheck, ...]) -> str: + return " | ".join(check.message for check in checks) + + +def _timestamp() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") diff --git a/src/app/bootstrap.py b/src/app/bootstrap.py index c5f3e26..0bc493b 100644 --- a/src/app/bootstrap.py +++ b/src/app/bootstrap.py @@ -7,6 +7,7 @@ from app.application.use_cases.save_measurement import SaveMeasurementUseCase from app.application.use_cases.settings_use_case import SettingsUseCase from app.infrastructure.instruments.equips_factory import create_instrument_ports, resolve_visa_address +from app.infrastructure.instruments.identity_probe import PyVisaIdentityProbe from app.infrastructure.instruments.resource_scanner import PyVisaResourceScanner from app.infrastructure.persistence.measurement_repo_mat_csv import MatCsvMeasurementRepository from app.infrastructure.persistence.reference_repo_mat import MatReferenceRepository @@ -45,6 +46,7 @@ def build_desktop_app(paths: AppPaths | None = None) -> DesktopApp: load_measurement_use_case=LoadMeasurementUseCase(measurement_repo), load_reference_use_case=LoadReferenceUseCase(reference_repo), scanner=PyVisaResourceScanner(), + identity_probe=PyVisaIdentityProbe(), ports_factory=create_instrument_ports, resolve_address=resolve_visa_address, paths=app_paths, @@ -54,4 +56,3 @@ def build_desktop_app(paths: AppPaths | None = None) -> DesktopApp: def run_desktop_app(paths: AppPaths | None = None) -> None: build_desktop_app(paths=paths).run() - diff --git a/src/app/domain/instrument_capabilities.py b/src/app/domain/instrument_capabilities.py new file mode 100644 index 0000000..a888846 --- /dev/null +++ b/src/app/domain/instrument_capabilities.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + +from app.domain.enums import ConnectionMode, CouplingMode, ImpedanceMode, TriggerMode + + +class InstrumentRole(str, Enum): + AWG = "awg" + OSC = "osc" + + +class ValidationStatus(str, Enum): + LEGACY_SUPPORTED = "legacy_supported" + SOFTWARE_PROFILE = "software_profile" + TEST_ONLY = "test_only" + + +@dataclass(frozen=True, slots=True) +class NumericLimit: + minimum: float | None = None + maximum: float | None = None + unit: str = "" + source: str = "not specified" + + def contains(self, value: float) -> bool: + if self.minimum is not None and value < self.minimum: + return False + if self.maximum is not None and value > self.maximum: + return False + return True + + +@dataclass(frozen=True, slots=True) +class InstrumentCapability: + model: str + role: InstrumentRole + channel_count: int + supported_impedances: frozenset[ImpedanceMode] + supported_couplings: frozenset[CouplingMode] = frozenset() + supported_trigger_modes: frozenset[TriggerMode] = frozenset() + transports: frozenset[ConnectionMode] = frozenset((ConnectionMode.AUTO, ConnectionMode.LAN)) + frequency_hz: NumericLimit = NumericLimit(unit="Hz") + amplitude_vpp: NumericLimit = NumericLimit(unit="Vpp") + osc_full_scale_v: NumericLimit = NumericLimit(unit="V") + timeout_s: float = 15.0 + validation_status: ValidationStatus = ValidationStatus.SOFTWARE_PROFILE + validation_source: str = "software profile; not live-hardware validation" + safety_notes: tuple[str, ...] = () + visible_in_ui: bool = True + + +_CURRENT_PRODUCTION_CAPABILITIES: tuple[InstrumentCapability, ...] = ( + InstrumentCapability( + model="DSG4102", + role=InstrumentRole.AWG, + channel_count=2, + frequency_hz=NumericLimit(minimum=1e-6, maximum=100e6, unit="Hz", source="legacy driver/default sweep profile"), + amplitude_vpp=NumericLimit(minimum=0.001, maximum=10.0, unit="Vpp", source="conservative software preflight"), + supported_impedances=frozenset((ImpedanceMode.R50, ImpedanceMode.HIGH_Z)), + validation_status=ValidationStatus.LEGACY_SUPPORTED, + validation_source="legacy equips.py DG4102-compatible adapter; bench revalidation recommended", + safety_notes=( + "Software label maps to the legacy DG4102-compatible driver path.", + "Confirm real front-panel limits before live sweeps.", + ), + ), + InstrumentCapability( + model="DSG836", + role=InstrumentRole.AWG, + channel_count=1, + supported_impedances=frozenset((ImpedanceMode.R50,)), + validation_status=ValidationStatus.LEGACY_SUPPORTED, + validation_source="legacy equips.py RF-generator adapter; frequency/amplitude limits require bench/manual confirmation", + safety_notes=( + "Single-channel RF generator path; channel selection is ignored by the legacy adapter.", + "Keep output-level assumptions conservative until bench-validated.", + ), + ), + InstrumentCapability( + model="MDO34", + role=InstrumentRole.OSC, + channel_count=4, + supported_impedances=frozenset((ImpedanceMode.R50, ImpedanceMode.HIGH_Z)), + supported_couplings=frozenset((CouplingMode.AC, CouplingMode.DC)), + supported_trigger_modes=frozenset((TriggerMode.FREE_RUN, TriggerMode.TRIGGERED)), + validation_status=ValidationStatus.LEGACY_SUPPORTED, + validation_source="legacy equips.py Tektronix MDO3 adapter path", + safety_notes=("Verify probe attenuation, termination, and range before live acquisition.",), + ), + InstrumentCapability( + model="MDO3024", + role=InstrumentRole.OSC, + channel_count=4, + supported_impedances=frozenset((ImpedanceMode.R50, ImpedanceMode.HIGH_Z)), + supported_couplings=frozenset((CouplingMode.AC, CouplingMode.DC)), + supported_trigger_modes=frozenset((TriggerMode.FREE_RUN, TriggerMode.TRIGGERED)), + validation_status=ValidationStatus.LEGACY_SUPPORTED, + validation_source="legacy equips.py Tektronix MDO3 adapter path", + safety_notes=("Verify probe attenuation, termination, and range before live acquisition.",), + ), + InstrumentCapability( + model="DHO1202", + role=InstrumentRole.OSC, + channel_count=2, + supported_impedances=frozenset((ImpedanceMode.HIGH_Z,)), + supported_couplings=frozenset((CouplingMode.AC, CouplingMode.DC)), + supported_trigger_modes=frozenset((TriggerMode.FREE_RUN, TriggerMode.TRIGGERED)), + validation_status=ValidationStatus.LEGACY_SUPPORTED, + validation_source="legacy equips.py DHO1000 adapter; code path documents 1 MOhm-only termination", + safety_notes=("DHO1000 series adapter treats termination as 1 MOhm only; do not select 50 ohm.",), + ), + InstrumentCapability( + model="DHO1204", + role=InstrumentRole.OSC, + channel_count=4, + supported_impedances=frozenset((ImpedanceMode.HIGH_Z,)), + supported_couplings=frozenset((CouplingMode.AC, CouplingMode.DC)), + supported_trigger_modes=frozenset((TriggerMode.FREE_RUN, TriggerMode.TRIGGERED)), + validation_status=ValidationStatus.LEGACY_SUPPORTED, + validation_source="legacy equips.py DHO1000 adapter; code path documents 1 MOhm-only termination", + safety_notes=("DHO1000 series adapter treats termination as 1 MOhm only; do not select 50 ohm.",), + ), +) + + +_TEST_CAPABILITIES: tuple[InstrumentCapability, ...] = ( + InstrumentCapability( + model="MOCK_AWG", + role=InstrumentRole.AWG, + channel_count=2, + frequency_hz=NumericLimit(minimum=1.0, maximum=10e6, unit="Hz", source="test fake"), + amplitude_vpp=NumericLimit(minimum=0.001, maximum=5.0, unit="Vpp", source="test fake"), + supported_impedances=frozenset((ImpedanceMode.R50, ImpedanceMode.HIGH_Z)), + validation_status=ValidationStatus.TEST_ONLY, + validation_source="hardware-free fake adapter", + safety_notes=("Test-only profile; never present as live hardware.",), + visible_in_ui=False, + ), + InstrumentCapability( + model="MOCK_OSC", + role=InstrumentRole.OSC, + channel_count=4, + supported_impedances=frozenset((ImpedanceMode.R50, ImpedanceMode.HIGH_Z)), + supported_couplings=frozenset((CouplingMode.AC, CouplingMode.DC)), + supported_trigger_modes=frozenset((TriggerMode.FREE_RUN, TriggerMode.TRIGGERED)), + validation_status=ValidationStatus.TEST_ONLY, + validation_source="hardware-free fake adapter", + safety_notes=("Test-only profile; never present as live hardware.",), + visible_in_ui=False, + ), +) + + +ALL_CAPABILITIES: tuple[InstrumentCapability, ...] = _CURRENT_PRODUCTION_CAPABILITIES + _TEST_CAPABILITIES + + +def capabilities_for_role( + role: InstrumentRole, + *, + include_test: bool = False, + visible_only: bool = True, +) -> tuple[InstrumentCapability, ...]: + capabilities = [] + for capability in ALL_CAPABILITIES: + if capability.role != role: + continue + if capability.validation_status == ValidationStatus.TEST_ONLY and not include_test: + continue + if visible_only and not capability.visible_in_ui: + continue + capabilities.append(capability) + return tuple(capabilities) + + +def model_names_for_role(role: InstrumentRole, *, include_test: bool = False) -> tuple[str, ...]: + return tuple( + capability.model + for capability in capabilities_for_role( + role, + include_test=include_test, + visible_only=not include_test, + ) + ) + + +def get_capability(model: str, role: InstrumentRole | None = None, *, include_test: bool = False) -> InstrumentCapability: + for capability in ALL_CAPABILITIES: + if capability.model != model: + continue + if role is not None and capability.role != role: + continue + if capability.validation_status == ValidationStatus.TEST_ONLY and not include_test: + break + return capability + role_text = f" {role.value}" if role is not None else "" + raise ValueError(f"Unsupported{role_text} model: {model}") + + +def is_supported_model(model: str, role: InstrumentRole | None = None, *, include_test: bool = False) -> bool: + try: + get_capability(model, role, include_test=include_test) + except ValueError: + return False + return True diff --git a/src/app/domain/validators.py b/src/app/domain/validators.py index d8f0de7..d775218 100644 --- a/src/app/domain/validators.py +++ b/src/app/domain/validators.py @@ -1,6 +1,7 @@ from __future__ import annotations from app.domain.enums import CorrectionMode, CouplingMode, ImpedanceMode, TriggerMode +from app.domain.instrument_capabilities import InstrumentCapability, InstrumentRole, get_capability from app.domain.models import AppSettings, ChannelSelection, OscSettings, SweepSpec @@ -44,7 +45,48 @@ def validate_osc_settings(settings: OscSettings) -> None: raise ValidationError("50-ohm impedance does not support AC coupling") -def validate_settings(settings: AppSettings) -> None: +def validate_capabilities(settings: AppSettings, *, include_test: bool = False) -> None: + setup = settings.setup + run_mode = settings.run_mode + + awg_capability = get_capability(setup.awg.model, InstrumentRole.AWG, include_test=include_test) + osc_capability = get_capability(setup.osc.model, InstrumentRole.OSC, include_test=include_test) + + _validate_transport("AWG", setup.awg.connect_mode, awg_capability) + _validate_transport("OSC", setup.osc.connect_mode, osc_capability) + + if setup.channels.awg_ch > awg_capability.channel_count: + raise ValidationError(f"AWG channel {setup.channels.awg_ch} exceeds {setup.awg.model} channel count") + for label, channel in ( + ("test", setup.channels.osc_test_ch), + ("reference", setup.channels.osc_ref_ch), + ("trigger", setup.channels.osc_trig_ch), + ): + if channel is not None and channel > osc_capability.channel_count: + raise ValidationError(f"OSC {label} channel {channel} exceeds {setup.osc.model} channel count") + + if setup.awg_settings.impedance not in awg_capability.supported_impedances: + raise ValidationError( + f"{setup.awg.model} does not support AWG impedance {setup.awg_settings.impedance.value}" + ) + if setup.osc_settings.impedance not in osc_capability.supported_impedances: + raise ValidationError( + f"{setup.osc.model} does not support OSC impedance {setup.osc_settings.impedance.value}" + ) + if setup.osc_settings.coupling not in osc_capability.supported_couplings: + raise ValidationError( + f"{setup.osc.model} does not support OSC coupling {setup.osc_settings.coupling.value}" + ) + if run_mode.trigger_mode not in osc_capability.supported_trigger_modes: + raise ValidationError(f"{setup.osc.model} does not support trigger mode {run_mode.trigger_mode.value}") + + _validate_limit("start frequency", settings.sweep.start_hz, awg_capability.frequency_hz, setup.awg.model) + _validate_limit("stop frequency", settings.sweep.stop_hz, awg_capability.frequency_hz, setup.awg.model) + _validate_limit("AWG amplitude", setup.awg_settings.amplitude_vpp, awg_capability.amplitude_vpp, setup.awg.model) + _validate_limit("OSC full-scale range", setup.osc_settings.full_scale_v, osc_capability.osc_full_scale_v, setup.osc.model) + + +def validate_settings(settings: AppSettings, *, include_test: bool = False) -> None: validate_sweep_spec(settings.sweep) validate_osc_settings(settings.setup.osc_settings) validate_channels( @@ -52,3 +94,21 @@ def validate_settings(settings: AppSettings) -> None: settings.run_mode.correction_mode, settings.run_mode.trigger_mode, ) + validate_capabilities(settings, include_test=include_test) + + +def _validate_transport(label: str, mode, capability: InstrumentCapability) -> None: + if mode not in capability.transports: + raise ValidationError(f"{capability.model} does not support {label} connection mode {mode.value}") + + +def _validate_limit(label: str, value: float, limit, model: str) -> None: + if limit.minimum is None and limit.maximum is None: + return + if limit.contains(float(value)): + return + + lower = "-inf" if limit.minimum is None else f"{limit.minimum:g}" + upper = "inf" if limit.maximum is None else f"{limit.maximum:g}" + unit = f" {limit.unit}" if limit.unit else "" + raise ValidationError(f"{model} {label} {value:g}{unit} outside supported range {lower}..{upper}{unit}") diff --git a/src/app/infrastructure/instruments/adapter_registry.py b/src/app/infrastructure/instruments/adapter_registry.py new file mode 100644 index 0000000..907fcf1 --- /dev/null +++ b/src/app/infrastructure/instruments/adapter_registry.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from collections.abc import Callable + +from app.application.ports.instruments import AwgPort, InstrumentPorts, OscPort +from app.domain.instrument_capabilities import InstrumentRole, get_capability +from app.domain.models import InstrumentSetup +from app.infrastructure.instruments.awg_adapter import EquipsAwgAdapter +from app.infrastructure.instruments.mock_adapters import MockAwgAdapter, MockOscAdapter +from app.infrastructure.instruments.osc_adapter import EquipsOscAdapter + + +AwgAdapterFactory = Callable[[str, str], AwgPort] +OscAdapterFactory = Callable[[str, str], OscPort] + + +class AdapterRegistry: + def __init__(self) -> None: + self._awg_factories: dict[str, AwgAdapterFactory] = {} + self._osc_factories: dict[str, OscAdapterFactory] = {} + + def register_awg(self, model: str, factory: AwgAdapterFactory) -> None: + get_capability(model, InstrumentRole.AWG, include_test=True) + self._awg_factories[model] = factory + + def register_osc(self, model: str, factory: OscAdapterFactory) -> None: + get_capability(model, InstrumentRole.OSC, include_test=True) + self._osc_factories[model] = factory + + def create_awg(self, *, model: str, address: str) -> AwgPort: + try: + factory = self._awg_factories[model] + except KeyError as exc: + raise ValueError(f"No AWG adapter registered for model: {model}") from exc + return factory(model, address) + + def create_osc(self, *, model: str, address: str) -> OscPort: + try: + factory = self._osc_factories[model] + except KeyError as exc: + raise ValueError(f"No OSC adapter registered for model: {model}") from exc + return factory(model, address) + + def create_ports(self, *, setup: InstrumentSetup, awg_address: str, osc_address: str) -> InstrumentPorts: + awg = self.create_awg(model=setup.awg.model, address=awg_address) + osc = self.create_osc(model=setup.osc.model, address=osc_address) + return InstrumentPorts(awg=awg, osc=osc, awg_address=awg_address, osc_address=osc_address) + + +def build_production_adapter_registry() -> AdapterRegistry: + registry = AdapterRegistry() + for model in ("DSG4102", "DSG836"): + registry.register_awg(model, lambda registered_model, address: EquipsAwgAdapter(registered_model, address)) + for model in ("MDO34", "MDO3024", "DHO1202", "DHO1204"): + registry.register_osc(model, lambda registered_model, address: EquipsOscAdapter(registered_model, address)) + return registry + + +def build_mock_adapter_registry() -> AdapterRegistry: + registry = AdapterRegistry() + awg = MockAwgAdapter() + registry.register_awg("MOCK_AWG", lambda _model, _address: awg) + registry.register_osc("MOCK_OSC", lambda _model, _address: MockOscAdapter(awg)) + return registry diff --git a/src/app/infrastructure/instruments/equips_factory.py b/src/app/infrastructure/instruments/equips_factory.py index d563d77..56c0576 100644 --- a/src/app/infrastructure/instruments/equips_factory.py +++ b/src/app/infrastructure/instruments/equips_factory.py @@ -3,8 +3,10 @@ from app.application.ports.instruments import InstrumentPorts from app.domain.enums import ConnectionMode from app.domain.models import InstrumentEndpoint, InstrumentSetup -from app.infrastructure.instruments.awg_adapter import EquipsAwgAdapter -from app.infrastructure.instruments.osc_adapter import EquipsOscAdapter +from app.infrastructure.instruments.adapter_registry import AdapterRegistry, build_production_adapter_registry + + +_PRODUCTION_REGISTRY: AdapterRegistry | None = None def resolve_visa_address(endpoint: InstrumentEndpoint) -> str: @@ -15,9 +17,17 @@ def resolve_visa_address(endpoint: InstrumentEndpoint) -> str: def create_instrument_ports(setup: InstrumentSetup) -> InstrumentPorts: + return create_instrument_ports_with_registry(setup=setup, registry=_production_registry()) + + +def create_instrument_ports_with_registry(setup: InstrumentSetup, registry: AdapterRegistry) -> InstrumentPorts: awg_address = resolve_visa_address(setup.awg) osc_address = resolve_visa_address(setup.osc) + return registry.create_ports(setup=setup, awg_address=awg_address, osc_address=osc_address) + - awg = EquipsAwgAdapter(model=setup.awg.model, visa_address=awg_address) - osc = EquipsOscAdapter(model=setup.osc.model, visa_address=osc_address) - return InstrumentPorts(awg=awg, osc=osc, awg_address=awg_address, osc_address=osc_address) +def _production_registry() -> AdapterRegistry: + global _PRODUCTION_REGISTRY + if _PRODUCTION_REGISTRY is None: + _PRODUCTION_REGISTRY = build_production_adapter_registry() + return _PRODUCTION_REGISTRY diff --git a/src/app/infrastructure/instruments/identity_probe.py b/src/app/infrastructure/instruments/identity_probe.py new file mode 100644 index 0000000..948eef4 --- /dev/null +++ b/src/app/infrastructure/instruments/identity_probe.py @@ -0,0 +1,25 @@ +from __future__ import annotations + + +class PyVisaIdentityProbe: + def identify(self, address: str, timeout_ms: int | None = None) -> str: + try: + import pyvisa as visa + except ModuleNotFoundError as exc: + raise RuntimeError("pyvisa is not installed") from exc + + resource = None + rm = visa.ResourceManager() + try: + resource = rm.open_resource(address) + if timeout_ms is not None: + resource.timeout = int(timeout_ms) + response = resource.query("*IDN?") + return str(response).strip() + finally: + if resource is not None: + resource.close() + try: + rm.close() + except Exception: + pass diff --git a/src/app/infrastructure/instruments/mock_adapters.py b/src/app/infrastructure/instruments/mock_adapters.py new file mode 100644 index 0000000..bf0ca1c --- /dev/null +++ b/src/app/infrastructure/instruments/mock_adapters.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import numpy as np + + +class MockAwgAdapter: + def __init__(self) -> None: + self.frequency_hz = 1_000.0 + self.amplitude_vpp = 1.0 + self.output_channels: set[int] = set() + self.closed = False + + def reset(self) -> None: + self.output_channels.clear() + + def output_on(self, channel: int) -> None: + self.output_channels.add(channel) + + def output_off(self, channel: int) -> None: + self.output_channels.discard(channel) + + def set_impedance(self, mode: str, channel: int) -> None: + _ = (mode, channel) + + def set_frequency(self, hz: float, channel: int) -> None: + _ = channel + self.frequency_hz = float(hz) + + def get_frequency(self, channel: int) -> float: + _ = channel + return self.frequency_hz + + def set_amplitude_vpp(self, vpp: float, channel: int) -> None: + _ = channel + self.amplitude_vpp = float(vpp) + + def get_amplitude_vpp(self, channel: int) -> float: + _ = channel + return self.amplitude_vpp + + def close(self) -> None: + self.closed = True + + +class MockOscAdapter: + def __init__(self, awg: MockAwgAdapter | None = None) -> None: + self._awg = awg or MockAwgAdapter() + self._range_v = 1.0 + self._offset_v = 0.0 + self.closed = False + + def reset(self) -> None: + return None + + def output_on(self, channel: int) -> None: + _ = channel + + def set_timebase(self, window_s: float, offset_s: float | None = None) -> None: + _ = (window_s, offset_s) + + def set_vertical(self, channel: int, full_scale_v: float, offset_v: float) -> None: + _ = channel + self._range_v = float(full_scale_v) + self._offset_v = float(offset_v) + + def get_vertical(self, channel: int) -> tuple[float, float]: + _ = channel + return self._range_v, self._offset_v + + def set_coupling(self, channel: int, mode: str) -> None: + _ = (channel, mode) + + def set_impedance(self, channel: int, mode: str) -> None: + _ = (channel, mode) + + def arm_trigger(self, channel: int, level_v: float) -> None: + _ = (channel, level_v) + + def set_free_run(self) -> None: + return None + + def single_acquire(self, triggered: bool) -> None: + _ = triggered + + def read_waveform(self, channel: int, points: int | None) -> tuple[np.ndarray, np.ndarray]: + _ = channel + n = int(points or 4_000) + sample_rate = self.get_sample_rate() + times = np.arange(n, dtype=float) / sample_rate + peak = max(self._awg.amplitude_vpp * 0.25, 1e-6) + volts = peak * np.sin(2.0 * np.pi * self._awg.frequency_hz * times) + self._offset_v + return times, volts + + def get_sample_rate(self) -> float: + return 200_000.0 + + def close(self) -> None: + self.closed = True diff --git a/src/app/infrastructure/instruments/ports.py b/src/app/infrastructure/instruments/ports.py index c5fb368..27e1921 100644 --- a/src/app/infrastructure/instruments/ports.py +++ b/src/app/infrastructure/instruments/ports.py @@ -1,3 +1,17 @@ -from app.application.ports.instruments import AwgPort, InstrumentPorts, InstrumentPortsFactory, OscPort, ResourceScannerPort +from app.application.ports.instruments import ( + AwgPort, + InstrumentIdentityProbePort, + InstrumentPorts, + InstrumentPortsFactory, + OscPort, + ResourceScannerPort, +) -__all__ = ["AwgPort", "OscPort", "ResourceScannerPort", "InstrumentPorts", "InstrumentPortsFactory"] +__all__ = [ + "AwgPort", + "OscPort", + "ResourceScannerPort", + "InstrumentIdentityProbePort", + "InstrumentPorts", + "InstrumentPortsFactory", +] diff --git a/src/app/presentation/tk/app_window.py b/src/app/presentation/tk/app_window.py index 34557f6..9bc537a 100644 --- a/src/app/presentation/tk/app_window.py +++ b/src/app/presentation/tk/app_window.py @@ -87,6 +87,8 @@ def bind_actions( on_load_ref, on_save_settings, on_load_settings, + on_scan_resources, + on_test_connect, on_close, on_figure_change, on_mag_phase_change, @@ -94,6 +96,8 @@ def bind_actions( self.control_panel.bind_actions( on_save_settings=on_save_settings, on_load_settings=on_load_settings, + on_scan_resources=on_scan_resources, + on_test_connect=on_test_connect, ) self.run_panel.bind_actions( on_start=on_start, @@ -132,6 +136,8 @@ def _alias_control_widgets(self) -> None: self.btn_load_ref = self.run_panel.btn_load_ref self.btn_save_settings = self.control_panel.btn_save_settings self.btn_load_settings = self.control_panel.btn_load_settings + self.btn_scan_resources = self.control_panel.btn_scan_resources + 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 diff --git a/src/app/presentation/tk/control_panel.py b/src/app/presentation/tk/control_panel.py index 9aa5e28..452c906 100644 --- a/src/app/presentation/tk/control_panel.py +++ b/src/app/presentation/tk/control_panel.py @@ -28,9 +28,13 @@ def bind_actions( *, on_save_settings, on_load_settings, + on_scan_resources, + on_test_connect, ) -> None: self.btn_save_settings.configure(command=on_save_settings) self.btn_load_settings.configure(command=on_load_settings) + self.btn_scan_resources.configure(command=on_scan_resources) + self.btn_test_connect.configure(command=on_test_connect) def _build(self) -> None: self.grid_columnconfigure(0, weight=1) @@ -71,6 +75,23 @@ def _build_instruments(self, row: int) -> None: self.awg_connection_chip = self._status_chip(chips, self._vm.awg_connection_text, 0) self.osc_connection_chip = self._status_chip(chips, self._vm.osc_connection_text, 1) + actions = ttk.Frame(section) + actions.grid(row=9, column=0, columnspan=2, sticky="ew", pady=(7, 0)) + actions.grid_columnconfigure((0, 1), weight=1) + self.btn_scan_resources = ttk.Button(actions, text="Scan Resources") + self.btn_scan_resources.grid(row=0, column=0, sticky="ew", padx=(0, 6)) + self.btn_test_connect = ttk.Button(actions, text="Test Connect") + self.btn_test_connect.grid(row=0, column=1, sticky="ew") + tk.Label( + section, + textvariable=self._vm.discovery_status_text, + bg=CARD_BG, + fg=MUTED, + justify=tk.LEFT, + anchor="w", + wraplength=250, + ).grid(row=10, column=0, columnspan=2, sticky="ew", pady=(6, 0)) + def _build_sweep(self, row: int) -> None: section = self._section("Sweep", row) self._combo(section, "Unit", self._vm.freq_unit, Mapping.values_freq_unit, 0, width=10) diff --git a/src/app/presentation/tk/controller.py b/src/app/presentation/tk/controller.py index 2654499..ea8e1e4 100644 --- a/src/app/presentation/tk/controller.py +++ b/src/app/presentation/tk/controller.py @@ -8,13 +8,22 @@ import numpy as np from app.application.events import EventEmitter +from app.application.ports.instruments import ( + InstrumentIdentityProbePort, + InstrumentPortsFactory, + ResourceScannerPort, +) +from app.application.services.instrument_discovery import ( + InstrumentDiscoveryService, + format_connection_receipt, + format_scan_receipt, +) from app.application.services.sweep_task_runner import SweepTaskRunner from app.application.services.connection_monitor import ConnectionMonitor from app.application.use_cases.load_measurement import LoadMeasurementUseCase from app.application.use_cases.load_reference import LoadReferenceUseCase from app.application.use_cases.save_measurement import SaveMeasurementUseCase from app.application.use_cases.settings_use_case import SettingsUseCase -from app.application.ports.instruments import InstrumentPortsFactory, ResourceScannerPort from app.domain.models import InstrumentEndpoint from app.presentation.tk import dialogs from app.presentation.tk.app_window import AppWindow @@ -38,6 +47,7 @@ def __init__( load_measurement_use_case: LoadMeasurementUseCase, load_reference_use_case: LoadReferenceUseCase, scanner: ResourceScannerPort, + identity_probe: InstrumentIdentityProbePort, ports_factory: InstrumentPortsFactory, resolve_address: Callable[[InstrumentEndpoint], str], paths: AppPaths | None = None, @@ -59,6 +69,7 @@ def __init__( self._closing = False self._ui_handler = UiEventHandler(window=window, vm=vm) + self._discovery_service = InstrumentDiscoveryService(scanner=scanner, identity_probe=identity_probe) self._task_runner = SweepTaskRunner( emitter=self, save_measurement_use_case=save_measurement_use_case, @@ -82,6 +93,8 @@ def initialize(self) -> None: on_load_ref=self.on_load_reference, on_save_settings=self.on_save_settings, on_load_settings=self.on_load_settings, + on_scan_resources=self.on_scan_resources, + on_test_connect=self.on_test_connect, on_close=self.on_close, on_figure_change=self.on_figure_change, on_mag_phase_change=self.on_mag_phase_change, @@ -217,6 +230,26 @@ def on_load_reference(self) -> None: except Exception as exc: # noqa: BLE001 dialogs.show_warning(self.window, f"Failed to load reference: {exc}") + def on_scan_resources(self) -> None: + try: + scan = self._discovery_service.scan_resources() + self.vm.discovery_status_text.set(format_scan_receipt(scan)) + self.vm.status_text.set("Resource scan completed") + except Exception as exc: # noqa: BLE001 + self.vm.discovery_status_text.set(f"Resource scan failed: {exc}") + dialogs.show_warning(self.window, f"Resource scan failed: {exc}") + + def on_test_connect(self) -> None: + try: + settings = vm_to_settings(self.vm) + checks = self._discovery_service.test_setup(settings.setup, self._resolve_address) + self._apply_connection_checks(checks) + self.vm.discovery_status_text.set(format_connection_receipt(checks)) + self.vm.status_text.set("Connection test completed") + except Exception as exc: # noqa: BLE001 + self.vm.discovery_status_text.set(f"Connection test failed: {exc}") + dialogs.show_warning(self.window, f"Connection test failed: {exc}") + def on_figure_change(self) -> None: self.window.plot_widget.set_mode(self.vm.figure_mode.get()) @@ -280,6 +313,23 @@ def _get_cached_osc_target_address(self) -> str: with self._connection_target_lock: return self._osc_target_address + def _apply_connection_checks(self, checks) -> None: + for check in checks: + label = check.role.value.upper() + if check.status == "connected": + text = f"{label} connected" + elif check.status == "address_empty": + text = f"{label} address empty" + elif check.status == "unsupported_model": + text = f"{label} unsupported" + else: + text = f"{label} offline" + + if check.role.value == "awg": + self.vm.awg_connection_text.set(text) + else: + self.vm.osc_connection_text.set(text) + def dialogs_to_target(path, window: AppWindow): from app.application.dto import SaveTarget diff --git a/src/app/presentation/tk/view_model.py b/src/app/presentation/tk/view_model.py index 4d14123..9ccfb4c 100644 --- a/src/app/presentation/tk/view_model.py +++ b/src/app/presentation/tk/view_model.py @@ -60,3 +60,4 @@ def __init__(self, root: tk.Misc) -> None: self.export_receipt_text = tk.StringVar(root, value="No export yet") self.awg_connection_text = tk.StringVar(root, value="AWG offline") self.osc_connection_text = tk.StringVar(root, value="OSC offline") + self.discovery_status_text = tk.StringVar(root, value="Scan VISA resources or test current addresses") diff --git a/src/app/shared/mapping.py b/src/app/shared/mapping.py index 09247e2..5e9fbe1 100644 --- a/src/app/shared/mapping.py +++ b/src/app/shared/mapping.py @@ -1,5 +1,7 @@ from __future__ import annotations +from app.domain.instrument_capabilities import InstrumentRole, model_names_for_role + class Mapping: label_for_input_ui = "Input Control Panel" @@ -97,8 +99,8 @@ class Mapping: mapping_state_on = "ON" mapping_state_off = "OFF" - values_awg = [mapping_DSG_4102, mapping_DSG_836] - values_osc = [mapping_MDO_34, mapping_MDO_3024, mapping_DHO_1202, mapping_DHO_1204] + values_awg = list(model_names_for_role(InstrumentRole.AWG)) + values_osc = list(model_names_for_role(InstrumentRole.OSC)) values_device_type = [label_for_device_type_awg, label_for_device_type_osc] values_freq_unit = [mapping_hz, mapping_khz, mapping_mhz, mapping_ghz] values_device_num_list = [1, 2, 3, 4] diff --git a/tests/test_adapter_registry.py b/tests/test_adapter_registry.py new file mode 100644 index 0000000..3734ac3 --- /dev/null +++ b/tests/test_adapter_registry.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import sys +from pathlib import Path +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.domain.enums import ConnectionMode, CouplingMode, ImpedanceMode +from app.domain.models import ( + AwgSettings, + ChannelSelection, + InstrumentEndpoint, + InstrumentSetup, + OscSettings, +) +from app.infrastructure.instruments.adapter_registry import ( + AdapterRegistry, + build_mock_adapter_registry, +) +from app.infrastructure.instruments.equips_factory import create_instrument_ports_with_registry + + +class AdapterRegistryTests(unittest.TestCase): + def test_registered_models_resolve_through_explicit_registry(self) -> None: + registry = AdapterRegistry() + awg = object() + osc = object() + registry.register_awg("DSG4102", lambda model, address: (model, address, awg)) + registry.register_osc("MDO34", lambda model, address: (model, address, osc)) + + ports = create_instrument_ports_with_registry(_production_setup(), registry) + + self.assertEqual(ports.awg, ("DSG4102", "USB::AWG::INSTR", awg)) + self.assertEqual(ports.osc, ("MDO34", "USB::OSC::INSTR", osc)) + self.assertEqual(ports.awg_address, "USB::AWG::INSTR") + self.assertEqual(ports.osc_address, "USB::OSC::INSTR") + + def test_unsupported_model_fails_before_implicit_equips_mapping(self) -> None: + registry = AdapterRegistry() + + with self.assertRaisesRegex(ValueError, "Unsupported awg model"): + registry.register_awg("NOT_SUPPORTED", lambda model, address: object()) + + def test_missing_adapter_registration_fails_clearly(self) -> None: + registry = AdapterRegistry() + registry.register_awg("DSG4102", lambda model, address: object()) + + with self.assertRaisesRegex(ValueError, "No OSC adapter registered"): + create_instrument_ports_with_registry(_production_setup(), registry) + + def test_mock_registry_provides_hardware_free_ports(self) -> None: + ports = create_instrument_ports_with_registry(_mock_setup(), build_mock_adapter_registry()) + + ports.awg.set_frequency(2_000.0, 1) + ports.awg.set_amplitude_vpp(1.2, 1) + ports.awg.output_on(1) + times, volts = ports.osc.read_waveform(1, 1000) + + self.assertEqual(len(times), 1000) + self.assertEqual(len(volts), 1000) + self.assertGreater(float(volts.max() - volts.min()), 0.1) + + +def _production_setup() -> InstrumentSetup: + return InstrumentSetup( + awg=InstrumentEndpoint( + model="DSG4102", + connect_mode=ConnectionMode.AUTO, + visa_address="USB::AWG::INSTR", + ), + osc=InstrumentEndpoint( + model="MDO34", + connect_mode=ConnectionMode.AUTO, + visa_address="USB::OSC::INSTR", + ), + channels=ChannelSelection(awg_ch=1, osc_test_ch=1, osc_ref_ch=2, osc_trig_ch=2), + awg_settings=AwgSettings(amplitude_vpp=1.0, impedance=ImpedanceMode.R50), + osc_settings=OscSettings( + full_scale_v=1.0, + offset_v=0.0, + points=1000, + impedance=ImpedanceMode.R50, + coupling=CouplingMode.DC, + ), + ) + + +def _mock_setup() -> InstrumentSetup: + return InstrumentSetup( + awg=InstrumentEndpoint(model="MOCK_AWG", connect_mode=ConnectionMode.AUTO), + osc=InstrumentEndpoint(model="MOCK_OSC", connect_mode=ConnectionMode.AUTO), + channels=ChannelSelection(awg_ch=1, osc_test_ch=1, osc_ref_ch=2, osc_trig_ch=2), + awg_settings=AwgSettings(amplitude_vpp=1.0, impedance=ImpedanceMode.R50), + osc_settings=OscSettings( + full_scale_v=1.0, + offset_v=0.0, + points=1000, + impedance=ImpedanceMode.R50, + coupling=CouplingMode.DC, + ), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_instrument_capabilities.py b/tests/test_instrument_capabilities.py new file mode 100644 index 0000000..79a08f8 --- /dev/null +++ b/tests/test_instrument_capabilities.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import sys +from pathlib import Path +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.domain.enums import ConnectionMode, CorrectionMode, CouplingMode, ImpedanceMode, MagnitudePhaseMode, TriggerMode +from app.domain.instrument_capabilities import InstrumentRole, get_capability, model_names_for_role +from app.domain.models import ( + AppSettings, + AwgSettings, + ChannelSelection, + InstrumentEndpoint, + InstrumentSetup, + OscSettings, + RunMode, + SweepSpec, +) +from app.domain.validators import ValidationError, validate_settings +from app.infrastructure.persistence.settings_defaults import DefaultSettingsFactory + + +class InstrumentCapabilityTests(unittest.TestCase): + def test_registry_exposes_current_production_models_only_by_default(self) -> None: + self.assertEqual(model_names_for_role(InstrumentRole.AWG), ("DSG4102", "DSG836")) + self.assertEqual(model_names_for_role(InstrumentRole.OSC), ("MDO34", "MDO3024", "DHO1202", "DHO1204")) + self.assertNotIn("MOCK_AWG", model_names_for_role(InstrumentRole.AWG)) + self.assertIn("MOCK_AWG", model_names_for_role(InstrumentRole.AWG, include_test=True)) + + def test_dho_profile_documents_high_z_only_impedance(self) -> None: + capability = get_capability("DHO1202", InstrumentRole.OSC) + + self.assertEqual(capability.channel_count, 2) + self.assertEqual(capability.supported_impedances, frozenset((ImpedanceMode.HIGH_Z,))) + self.assertIn("1 MOhm", " ".join(capability.safety_notes)) + + def test_default_settings_pass_capability_validation(self) -> None: + validate_settings(DefaultSettingsFactory().create()) + + def test_unsupported_model_fails_clearly(self) -> None: + settings = DefaultSettingsFactory().create() + settings.setup.awg.model = "NEW_AWG" + + with self.assertRaisesRegex(ValueError, "Unsupported awg model"): + validate_settings(settings) + + def test_osc_channel_and_impedance_are_capability_aware(self) -> None: + settings = DefaultSettingsFactory().create() + settings.setup.osc.model = "DHO1202" + settings.setup.osc_settings.impedance = ImpedanceMode.R50 + + with self.assertRaisesRegex(ValidationError, "does not support OSC impedance"): + validate_settings(settings) + + settings.setup.osc_settings.impedance = ImpedanceMode.HIGH_Z + settings.setup.channels.osc_test_ch = 3 + with self.assertRaisesRegex(ValidationError, "channel 3 exceeds DHO1202"): + validate_settings(settings) + + def test_awg_channel_and_known_frequency_limits_are_capability_aware(self) -> None: + settings = DefaultSettingsFactory().create() + settings.setup.awg.model = "DSG836" + settings.setup.awg_settings.impedance = ImpedanceMode.R50 + settings.setup.channels.awg_ch = 2 + + with self.assertRaisesRegex(ValidationError, "AWG channel 2 exceeds DSG836"): + validate_settings(settings) + + settings = DefaultSettingsFactory().create() + settings.sweep.stop_hz = 200e6 + with self.assertRaisesRegex(ValidationError, "stop frequency"): + validate_settings(settings) + + def test_test_only_profiles_validate_only_when_explicitly_enabled(self) -> None: + settings = _mock_settings() + + with self.assertRaisesRegex(ValueError, "Unsupported awg model"): + validate_settings(settings) + + validate_settings(settings, include_test=True) + + +def _mock_settings() -> AppSettings: + return AppSettings( + schema_version=1, + freq_unit="Hz", + sweep=SweepSpec(start_hz=1_000.0, stop_hz=10_000.0, step_hz=1_000.0, step_count=None, is_log=False), + run_mode=RunMode( + correction_mode=CorrectionMode.NONE, + trigger_mode=TriggerMode.FREE_RUN, + auto_range=False, + auto_reset=False, + ), + setup=InstrumentSetup( + awg=InstrumentEndpoint(model="MOCK_AWG", connect_mode=ConnectionMode.AUTO), + osc=InstrumentEndpoint(model="MOCK_OSC", connect_mode=ConnectionMode.AUTO), + channels=ChannelSelection(awg_ch=1, osc_test_ch=1, osc_ref_ch=2, osc_trig_ch=2), + awg_settings=AwgSettings(amplitude_vpp=1.0, impedance=ImpedanceMode.R50), + osc_settings=OscSettings( + full_scale_v=1.0, + offset_v=0.0, + points=1_000, + impedance=ImpedanceMode.R50, + coupling=CouplingMode.DC, + ), + ), + magnitude_phase_mode=MagnitudePhaseMode.MAG, + auto_save_data=False, + ) +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_instrument_discovery.py b/tests/test_instrument_discovery.py new file mode 100644 index 0000000..023ce44 --- /dev/null +++ b/tests/test_instrument_discovery.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import sys +from pathlib import Path +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from app.application.services.instrument_discovery import ( + InstrumentDiscoveryService, + format_connection_receipt, + format_scan_receipt, +) +from app.domain.enums import ConnectionMode, CouplingMode, ImpedanceMode +from app.domain.instrument_capabilities import InstrumentRole +from app.domain.models import ( + AwgSettings, + ChannelSelection, + InstrumentEndpoint, + InstrumentSetup, + OscSettings, +) +from app.infrastructure.instruments.equips_factory import resolve_visa_address + + +class FakeScanner: + def __init__(self, resources: tuple[str, ...]) -> None: + self._resources = resources + + def list_resources(self) -> tuple[str, ...]: + return self._resources + + +class FakeIdentityProbe: + def __init__(self, responses: dict[str, str]) -> None: + self._responses = responses + + def identify(self, address: str, timeout_ms: int | None = None) -> str: + _ = timeout_ms + if address not in self._responses: + raise TimeoutError("not reachable") + return self._responses[address] + + +class InstrumentDiscoveryTests(unittest.TestCase): + def test_scan_resources_returns_receipt(self) -> None: + service = InstrumentDiscoveryService( + scanner=FakeScanner(("USB::AWG::INSTR", "TCPIP0::10.0.0.2::INSTR")), + identity_probe=FakeIdentityProbe({}), + ) + + receipt = format_scan_receipt(service.scan_resources()) + + self.assertIn("2 VISA resources", receipt) + self.assertIn("USB::AWG::INSTR", receipt) + + def test_test_setup_reports_connected_idn(self) -> None: + service = InstrumentDiscoveryService( + scanner=FakeScanner(()), + identity_probe=FakeIdentityProbe( + { + "USB::AWG::INSTR": "RIGOL,DG4102,123,1.0", + "USB::OSC::INSTR": "TEKTRONIX,MDO34,456,1.0", + } + ), + ) + + checks = service.test_setup(_setup(), resolve_visa_address) + receipt = format_connection_receipt(checks) + + self.assertTrue(all(check.status == "connected" for check in checks)) + self.assertIn("RIGOL,DG4102", checks[0].idn) + self.assertIn("AWG connected", receipt) + self.assertIn("OSC connected", receipt) + + def test_address_empty_and_offline_states_are_clear(self) -> None: + setup = _setup() + setup.awg.visa_address = "" + service = InstrumentDiscoveryService( + scanner=FakeScanner(()), + identity_probe=FakeIdentityProbe({"USB::OTHER::INSTR": "OTHER"}), + ) + + checks = service.test_setup(setup, resolve_visa_address) + + self.assertEqual(checks[0].status, "address_empty") + self.assertEqual(checks[1].status, "offline") + self.assertIn("not reachable", checks[1].message) + + def test_unsupported_model_state_does_not_probe_hardware(self) -> None: + endpoint = InstrumentEndpoint( + model="NEW_SCOPE", + connect_mode=ConnectionMode.AUTO, + visa_address="USB::NEW::INSTR", + ) + service = InstrumentDiscoveryService(scanner=FakeScanner(()), identity_probe=FakeIdentityProbe({})) + + check = service.test_endpoint(role=InstrumentRole.OSC, endpoint=endpoint, address="USB::NEW::INSTR") + + self.assertEqual(check.status, "unsupported_model") + self.assertIn("Unsupported OSC model", check.message) + + +def _setup() -> InstrumentSetup: + return InstrumentSetup( + awg=InstrumentEndpoint( + model="DSG4102", + connect_mode=ConnectionMode.AUTO, + visa_address="USB::AWG::INSTR", + ), + osc=InstrumentEndpoint( + model="MDO34", + connect_mode=ConnectionMode.AUTO, + visa_address="USB::OSC::INSTR", + ), + channels=ChannelSelection(awg_ch=1, osc_test_ch=1, osc_ref_ch=2, osc_trig_ch=2), + awg_settings=AwgSettings(amplitude_vpp=1.0, impedance=ImpedanceMode.R50), + osc_settings=OscSettings( + full_scale_v=1.0, + offset_v=0.0, + points=1000, + impedance=ImpedanceMode.R50, + coupling=CouplingMode.DC, + ), + ) + + +if __name__ == "__main__": + unittest.main()