diff --git a/README.md b/README.md index fa817cca..441853f0 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ WaveBench 是一个用 Python 编写的实验室自动测量台，面向电子设计竞赛调试和日常实验。它把仪器控制、实验步骤和采集证据放在同一条命令链中，支持先离线检查 plan，再决定是否连接硬件。 -当前仓库开发线为 `0.8.25`，最新稳定 tag 为 `v0.8.0`。不同版本的命令和能力可能不同，以对应 tag 中的文档为准。 +当前仓库开发线为 `0.8.26`，最新稳定 tag 为 `v0.8.0`。不同版本的命令和能力可能不同，以对应 tag 中的文档为准。 ## 🌟 特别鸣谢 @@ -67,6 +67,26 @@ flowchart LR dmm --> artifacts ``` +#### 示波器联合视图 + +`scope focus` 用一个事务配置一个或多个目标模拟通道，并可同时设置完整横向时间范围、各目标通道的 +V/div，以及是否隐藏插件 profile 声明的其他模拟通道： + +```bash +wavebench scope focus \ + --channel 1 \ + --channel 2 \ + --time-range 0.01 \ + --vertical-scale 1=0.2 \ + --vertical-scale 2=0.5 \ + --hide-others +``` + +Core 不定义仪器型号、通道数量或数值范围；这些 guard 由当前插件的 descriptor profile 声明。 +操作会先读取 profile 全部模拟通道及受保护的时基、位置和偏置字段。成功后保留目标视图；任一写入 +或回读失败时恢复完整 baseline 并重新查询，恢复不完整则停止该 session 的后续写入。该命令不启动 +采集、不调用 autoscale，也不修改耦合或输入终端。 + ## 先在没有仪器时跑通 下面的命令只生成和检查 plan，不会连接仪器，也不会打开输出。 diff --git a/docs/README.md b/docs/README.md index 4437383c..88a93a34 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ [English](README_EN.md) · 中文 -WaveBench 是一个用 Python 编写的实验室自动测量台，提供 CLI、实验性 TUI、显式 run plan、采集包和离线报告。当前开发线为 `0.8.25`，最新稳定 tag 为 `v0.8.0`。版本变化见 [更新日志](../CHANGELOG.md)；旧版本原始文档可切换到对应 Git tag 查看。 +WaveBench 是一个用 Python 编写的实验室自动测量台，提供 CLI、实验性 TUI、显式 run plan、采集包和离线报告。当前开发线为 `0.8.26`，最新稳定 tag 为 `v0.8.0`。版本变化见 [更新日志](../CHANGELOG.md)；旧版本原始文档可切换到对应 Git tag 查看。 > [!WARNING] > 部分命令会连接并控制真实仪器。示例会区分离线检查、连接读取和硬件写入；执行写入前，应确认接线和限制值。 @@ -26,6 +26,16 @@ wavebench run check --plan /tmp/wavebench-demo.toml - [配置文件格式](project/reference/WaveBench_配置文件格式.md)：TOML 查找顺序、字段和安全限制。 - 仪器型号命令和编程手册由 [仪器插件仓库](https://github.com/Scaxlibur/wavebench-instrument-plugins) 维护；本仓库只记录 WaveBench 的接入边界。 +### 配置示波器联合视图 + +`wavebench scope focus` 接受重复的 `--channel`、可选的 `--time-range`、重复的 +`--vertical-scale CHANNEL=V_PER_DIV` 和 `--hide-others`。Core 只定义可移植事务：插件 profile +声明模拟通道、数值范围、容差和 I/O 预算；Core 读取完整联合 baseline，成功后保留目标视图，失败时 +恢复并重新查询。未声明 `scope.focus_configure_v2` 的插件会在仪器 I/O 前拒绝操作。 + +该命令会修改仪器状态，但不会启动采集、调用 autoscale、修改耦合或切换输入终端。执行前应核对 +接线、输入状态和插件 capability。 + ### 使用 RF 信号源 `rf_source` 不复用普通 `source` 的 Vpp、offset 或数字 channel 模型。先从 [RF 信号源使用指南](project/guides/WaveBench_RF信号源使用指南.md) 确认当前 production capability 和端接声明；DSG830 已开放固定 profile 的调制输出，以及唯一受验证的后面板 `pulse_in_out` output 路径。后者不代表 Pulse input、`TRIGGER IN` 或同步能力。需要实现新型号或查看证据门时，再阅读 [领域设计](project/design/WaveBench_RF信号源设计.md) 与 [开发里程碑](project/design/WaveBench_RF信号源开发里程碑.md)。 @@ -68,7 +78,7 @@ wavebench run check --plan /tmp/wavebench-demo.toml | --- | --- | --- | | 离线 | `run schema`、`run template`、`run check`、`run report`、`capture inspect`、`tui --fake` | 不连接仪器；TUI 可能写本地日志 | | 连接读取 | `doctor`、`idn`、`status`、`run verify` | 是，仅读取或做预检 | -| 修改状态 | `scope fetch/capture/autoscale`、source/power setter、output、`run plan` | 是，可能写入、触发或改变输出 | +| 修改状态 | `scope focus/fetch/capture/autoscale`、source/power setter、output、`run plan` | 是，可能写入、触发或改变输出 | `run check` 不代表 plan 可以安全执行。它只检查 TOML 和字段；真正执行前，还要核对接线、scope coupling、输出状态、保护限值和 restore 条款。 diff --git a/docs/README_EN.md b/docs/README_EN.md index d607ec08..40ceb08a 100644 --- a/docs/README_EN.md +++ b/docs/README_EN.md @@ -2,7 +2,7 @@ [中文文档](README.md) · English -WaveBench is a Python measurement bench for laboratory debugging. It combines explicit instrument commands, run plans, capture packages, and offline reports. It requires Python 3.11 or newer. The current development line is `0.8.25`; the latest stable tag is `v0.8.0`. +WaveBench is a Python measurement bench for laboratory debugging. It combines explicit instrument commands, run plans, capture packages, and offline reports. It requires Python 3.11 or newer. The current development line is `0.8.26`; the latest stable tag is `v0.8.0`. > [!WARNING] > Some commands connect to and change real instruments. Check wiring, input impedance, output state, and voltage/current limits before running a hardware action. @@ -51,6 +51,19 @@ For the terminal UI, install `.[tui]` and run `wavebench tui --fake`. The fake m | TUI | Power, DMM, and source panels | Experimental manual control | | Plugins | `wavebench.instruments` drivers | Optional, explicitly selected extensions | +## Configure a joint oscilloscope view + +`wavebench scope focus` accepts repeated `--channel` arguments, optional `--time-range`, repeated +`--vertical-scale CHANNEL=V_PER_DIV`, and `--hide-others`. Core defines only the portable +transaction. Each plugin descriptor profile owns its analog-channel set, numeric request guards, +tolerances, and I/O budgets. Core reads the complete joint baseline, preserves the requested view +after success, and restores then freshly verifies that baseline after failure. A plugin that does +not declare `scope.focus_configure_v2` is rejected before instrument I/O. + +This command changes front-panel state. It does not start acquisition, invoke autoscale, change +coupling, or switch input termination. Wiring, input state, and the selected plugin capability must +be checked before execution. + ## Find a guide - Setup and configuration: [configuration format](project/reference/WaveBench_配置文件格式.md) @@ -69,7 +82,7 @@ Most detailed pages are currently maintained in Chinese. Commands, identifiers, | --- | --- | --- | | Offline | `run schema`, `run template`, `run check`, `run report`, `capture inspect`, `tui --fake` | No instrument I/O; TUI may write a local log | | Connected read/preflight | `doctor`, `idn`, `status`, `run verify` | Yes, for queries and checks | -| State-changing | `scope fetch/capture/autoscale`, source/power setters, output commands, `run plan` | Yes; may change setup, trigger acquisition, or switch output | +| State-changing | `scope focus/fetch/capture/autoscale`, source/power setters, output commands, `run plan` | Yes; may change setup, trigger acquisition, or switch output | WaveBench does not implicitly reset instruments, enable outputs, or change oscilloscope input impedance. `power set` and `power output` are separate operations. When enabled, source restoration covers only the documented basic fields; it is not a full channel snapshot. diff --git a/pyproject.toml b/pyproject.toml index 1a746ec1..b86598da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "wavebench" -version = "0.8.25" +version = "0.8.26" description = "Lightweight VISA/SCPI measurement bench for contest debugging" readme = "README.md" requires-python = ">=3.11" diff --git a/src/wavebench/cli.py b/src/wavebench/cli.py index 1de1752e..4d5215dc 100644 --- a/src/wavebench/cli.py +++ b/src/wavebench/cli.py @@ -81,7 +81,10 @@ from .instruments.registry import resolve_instrument_descriptor from .instruments.scope_extensions import ( ErrorCheckSpec, + ScopeChannelDisplayRequest, ScopeContinuousAcquisitionRequest, + ScopeFocusRequest, + ScopeFocusVerticalScale, ScopeScreenshot, ScopeScreenshotRequest, ScopeTraceData, @@ -602,6 +605,51 @@ def _scope_error_check(args: argparse.Namespace) -> ErrorCheckSpec | None: raise ConfigError(str(exc)) from exc +def _scope_channel_display_request(args: argparse.Namespace) -> ScopeChannelDisplayRequest: + try: + return ScopeChannelDisplayRequest( + channel=args.channel, + enabled=args.state == "on", + ) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + + +def _scope_focus_request(args: argparse.Namespace) -> ScopeFocusRequest: + raw_channels = tuple(args.channels) + if len(set(raw_channels)) != len(raw_channels): + raise ConfigError("scope focus channels must not contain duplicates") + scales: list[ScopeFocusVerticalScale] = [] + scale_channels: set[int] = set() + for raw in args.vertical_scales: + channel_text, separator, scale_text = raw.partition("=") + if not separator: + raise ConfigError("scope focus --vertical-scale must use CHANNEL=V_PER_DIV") + try: + channel = int(channel_text) + scale = float(scale_text) + except ValueError as exc: + raise ConfigError( + "scope focus --vertical-scale must use CHANNEL=V_PER_DIV" + ) from exc + if channel in scale_channels: + raise ConfigError("scope focus vertical-scale channels must not contain duplicates") + scale_channels.add(channel) + try: + scales.append(ScopeFocusVerticalScale(channel, scale)) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + try: + return ScopeFocusRequest( + channels=tuple(sorted(raw_channels)), + time_range_s=args.time_range, + vertical_scales=tuple(sorted(scales, key=lambda item: item.channel)), + hide_others=args.hide_others, + ) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + + def _scope_trace_ref(args: argparse.Namespace) -> ScopeTraceRef: try: return ScopeTraceRef( @@ -1882,6 +1930,20 @@ def _main(argv: list[str] | None = None) -> int: for item in service.errors(): print(item) return 0 + if args.command == "display": + result = service.configure_channel_display_v2( + _scope_channel_display_request(args), + error_check=_scope_error_check(args), + ) + _emit_scope_extension_result(result.as_dict(), json_mode=args.json) + return 0 + if args.command == "focus": + result = service.configure_focus_v2( + _scope_focus_request(args), + error_check=_scope_error_check(args), + ) + _emit_scope_extension_result(result.as_dict(), json_mode=args.json) + return 0 if args.command == "screenshot": if args.screenshot_command == "profile": result = service.screenshot_profile() diff --git a/src/wavebench/cli_parser.py b/src/wavebench/cli_parser.py index 6462451e..b73f1600 100644 --- a/src/wavebench/cli_parser.py +++ b/src/wavebench/cli_parser.py @@ -1202,6 +1202,49 @@ def build_parser() -> argparse.ArgumentParser: errors = scope_sub.add_parser("errors", help="Read SYST:ERR? until empty") add_runtime_options(errors) + display = scope_sub.add_parser( + "display", + help="Explicitly enable or disable one analog channel display", + ) + display.add_argument("--channel", type=int, required=True) + display.add_argument("state", choices=("on", "off")) + add_scope_error_options(display) + add_runtime_options(display) + + focus = scope_sub.add_parser( + "focus", + help="Configure one atomic multi-channel observation view", + ) + focus.add_argument( + "--channel", + dest="channels", + type=int, + action="append", + required=True, + help="Target analog channel; repeat for a multi-channel focus view", + ) + focus.add_argument( + "--time-range", + type=float, + default=None, + help="Optional full horizontal visible range in seconds", + ) + focus.add_argument( + "--vertical-scale", + dest="vertical_scales", + action="append", + default=[], + metavar="CHANNEL=V_PER_DIV", + help="Optional target-channel V/div; repeat for different target channels", + ) + focus.add_argument( + "--hide-others", + action="store_true", + help="Disable other analog channels declared by the driver profile", + ) + add_scope_error_options(focus) + add_runtime_options(focus) + status = scope_sub.add_parser( "status", help="Read a typed, non-mutating oscilloscope state snapshot", diff --git a/src/wavebench/instruments/scope_extension_capabilities.py b/src/wavebench/instruments/scope_extension_capabilities.py index 3bbfb033..2cb2a217 100644 --- a/src/wavebench/instruments/scope_extension_capabilities.py +++ b/src/wavebench/instruments/scope_extension_capabilities.py @@ -15,6 +15,7 @@ SCOPE_EXTENSIONS_MIN_CORE_VERSION = "0.8.23" SCOPE_WAVEFORM_BINARY_MIN_CORE_VERSION = "0.8.24" SCOPE_PORTABILITY_V2_MIN_CORE_VERSION = "0.8.24" +SCOPE_FOCUS_V2_MIN_CORE_VERSION = "0.8.26" SCOPE_STRICT_V2_CAPABILITIES = frozenset( { "scope.channel_input_state_v2", @@ -25,6 +26,8 @@ "scope.fft_status_v2", "scope.cursor_readout_v2", "scope.capture_average_v2", + "scope.channel_display_configure_v2", + "scope.focus_configure_v2", } ) @@ -92,6 +95,16 @@ "restore_average_capture_state", "verify_average_capture_state_restored", ), + "scope.channel_display_configure_v2": ( + "get_channel_display_state_v2", + "configure_channel_display_v2", + "restore_channel_display_v2", + ), + "scope.focus_configure_v2": ( + "get_focus_state_v2", + "configure_focus_v2", + "restore_focus_v2", + ), } ) @@ -138,8 +151,17 @@ def validate_scope_descriptor( "scope portability V2 capabilities require wavebench_min_version " f">= {SCOPE_PORTABILITY_V2_MIN_CORE_VERSION}" ) + if "scope.focus_configure_v2" in declared and minimum < Version( + SCOPE_FOCUS_V2_MIN_CORE_VERSION + ): + raise ConfigError( + "scope focus V2 capability requires wavebench_min_version " + f">= {SCOPE_FOCUS_V2_MIN_CORE_VERSION}" + ) dependencies = { "scope.acquisition_control": {"scope.acquisition_run_state"}, + "scope.channel_display_configure_v2": {"scope.idn"}, + "scope.focus_configure_v2": {"scope.idn"}, "scope.capture_average_v2": { "scope.idn", "scope.acquisition_status_v2", @@ -166,6 +188,8 @@ def validate_scope_descriptor( "scope.fft_status_v2": "fft_status_profile_v2", "scope.cursor_readout_v2": "cursor_readout_profile_v2", "scope.capture_average_v2": "average_capture_profile_v2", + "scope.channel_display_configure_v2": "channel_display_profile_v2", + "scope.focus_configure_v2": "focus_profile_v2", } for capability in sorted(declared): profile_name = profile_requirements.get(capability) diff --git a/src/wavebench/instruments/scope_extensions.py b/src/wavebench/instruments/scope_extensions.py index ffed43bc..95435b88 100644 --- a/src/wavebench/instruments/scope_extensions.py +++ b/src/wavebench/instruments/scope_extensions.py @@ -4,7 +4,7 @@ from collections.abc import Callable from dataclasses import dataclass -from math import isfinite +from math import isclose, isfinite import re from typing import Literal, Protocol, runtime_checkable import zlib @@ -1276,6 +1276,556 @@ def validate_result( raise ValueError("cursor readout V2 readable fields must have a value") +ScopeChannelDisplayField = Literal["scope.channel_display"] +_CHANNEL_DISPLAY_FIELDS = {"scope.channel_display"} + + +@dataclass(frozen=True, slots=True) +class ScopeChannelDisplayRequest: + channel: int + enabled: bool + + def __post_init__(self) -> None: + _strict_int(self.channel, label="channel display channel", minimum=1) + if not isinstance(self.enabled, bool): + raise TypeError("channel display enabled must be bool") + + +@dataclass(frozen=True, slots=True) +class ScopeChannelDisplayState: + channel: int + enabled: bool + + def __post_init__(self) -> None: + _strict_int(self.channel, label="channel display state channel", minimum=1) + if not isinstance(self.enabled, bool): + raise TypeError("channel display state enabled must be bool") + + +@dataclass(frozen=True, slots=True) +class ScopeChannelDisplayResult: + request: ScopeChannelDisplayRequest + before: ScopeChannelDisplayState + after: ScopeChannelDisplayState + write_performed: bool + + def __post_init__(self) -> None: + if not isinstance(self.request, ScopeChannelDisplayRequest): + raise TypeError("channel display result request has an invalid type") + if not isinstance(self.before, ScopeChannelDisplayState) or not isinstance( + self.after, + ScopeChannelDisplayState, + ): + raise TypeError("channel display result state has an invalid type") + if self.before.channel != self.request.channel or self.after.channel != self.request.channel: + raise ValueError("channel display result states must use the requested channel") + if self.after.enabled is not self.request.enabled: + raise ValueError("channel display result does not prove the requested state") + if not isinstance(self.write_performed, bool): + raise TypeError("channel display write_performed must be bool") + if self.write_performed != (self.before.enabled is not self.request.enabled): + raise ValueError("channel display write_performed disagrees with the observed change") + + +@dataclass(frozen=True, slots=True) +class ScopeChannelDisplayProfileV2: + analog_channels: tuple[int, ...] + snapshot_max_steps: int + configure_max_steps: int + restore_max_steps: int + verify_max_steps: int + + def __post_init__(self) -> None: + channels = _unique_tuple(self.analog_channels, label="analog_channels") + _strict_int(len(channels), label="analog channel count", minimum=1, maximum=64) + for channel in channels: + _strict_int( + channel, + label="analog channel", + minimum=1, + maximum=65_535, + ) + if tuple(sorted(channels)) != channels: + raise ValueError("analog_channels must use ascending channel order") + for label, value, minimum in ( + ("snapshot_max_steps", self.snapshot_max_steps, 1), + ("configure_max_steps", self.configure_max_steps, 2), + ("restore_max_steps", self.restore_max_steps, 1), + ("verify_max_steps", self.verify_max_steps, 1), + ): + _strict_int(value, label=label, minimum=minimum, maximum=64) + + def validate_request(self, request: ScopeChannelDisplayRequest) -> None: + if not isinstance(request, ScopeChannelDisplayRequest): + raise TypeError("channel display request has an invalid type") + if request.channel not in self.analog_channels: + raise ValueError("channel display request is outside the descriptor analog channels") + + def validate_state(self, state: ScopeChannelDisplayState, *, channel: int) -> None: + if not isinstance(state, ScopeChannelDisplayState): + raise TypeError("channel display driver returned an invalid state") + if channel not in self.analog_channels or state.channel != channel: + raise ValueError("channel display driver returned the wrong channel") + + +@dataclass(frozen=True, slots=True) +class ScopeChannelDisplayBaseline: + context_id: str + session_epoch: str + baseline_nonce: str + snapshot: ScopeChannelDisplayState + restore_order: tuple[ScopeChannelDisplayField, ...] + + def __post_init__(self) -> None: + _safe_token(self.context_id, label="channel display baseline context_id") + _safe_token(self.session_epoch, label="channel display baseline session_epoch") + _safe_token(self.baseline_nonce, label="channel display baseline nonce") + if not isinstance(self.snapshot, ScopeChannelDisplayState): + raise TypeError("channel display baseline snapshot has an invalid type") + if self.restore_order != ("scope.channel_display",): + raise ValueError("channel display restore order must contain its only state field") + + +@dataclass(frozen=True, slots=True) +class ScopeChannelDisplayRestoreResult: + status: Literal["completed", "failed", "not_attempted"] + attempted_fields: tuple[ScopeChannelDisplayField, ...] + restored_fields: tuple[ScopeChannelDisplayField, ...] + error_code: str | None = None + + def __post_init__(self) -> None: + _literal(self.status, {"completed", "failed", "not_attempted"}, label="restore status") + attempted = _unique_tuple(self.attempted_fields, label="attempted_fields") + restored = _unique_tuple(self.restored_fields, label="restored_fields") + if not set(attempted + restored) <= _CHANNEL_DISPLAY_FIELDS: + raise ValueError("channel display restore fields are invalid") + _optional_safe_token(self.error_code, label="error_code") + + def validate_for(self, baseline: ScopeChannelDisplayBaseline) -> None: + if not isinstance(baseline, ScopeChannelDisplayBaseline): + raise TypeError("channel display baseline has an invalid type") + _validate_prefix_and_subsequence( + expected=baseline.restore_order, + attempted=self.attempted_fields, + completed=self.restored_fields, + status=self.status, + ) + + +ScopeFocusField = Literal[ + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", +] +_SCOPE_FOCUS_FIELDS = { + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", +} + + +@dataclass(frozen=True, slots=True) +class ScopeFocusVerticalScale: + channel: int + scale_v_per_div: float + + def __post_init__(self) -> None: + _strict_int(self.channel, label="focus vertical scale channel", minimum=1) + scale = _finite(self.scale_v_per_div, label="focus vertical scale") + if scale <= 0: + raise ValueError("focus vertical scale must be > 0") + object.__setattr__(self, "scale_v_per_div", scale) + + +@dataclass(frozen=True, slots=True) +class ScopeFocusRequest: + channels: tuple[int, ...] + time_range_s: float | None = None + vertical_scales: tuple[ScopeFocusVerticalScale, ...] = () + hide_others: bool = False + + def __post_init__(self) -> None: + channels = _unique_tuple(self.channels, label="focus channels") + _strict_int(len(channels), label="focus channel count", minimum=1, maximum=64) + for channel in channels: + _strict_int(channel, label="focus channel", minimum=1, maximum=65_535) + if tuple(sorted(channels)) != channels: + raise ValueError("focus channels must use ascending order") + if self.time_range_s is not None: + time_range_s = _finite(self.time_range_s, label="focus time range") + if time_range_s <= 0: + raise ValueError("focus time range must be > 0") + object.__setattr__(self, "time_range_s", time_range_s) + scales = _unique_tuple(self.vertical_scales, label="focus vertical scales") + if not all(isinstance(item, ScopeFocusVerticalScale) for item in scales): + raise TypeError("focus vertical scales contain an invalid item") + scale_channels = tuple(item.channel for item in scales) + if tuple(sorted(scale_channels)) != scale_channels: + raise ValueError("focus vertical scales must use ascending channel order") + if not set(scale_channels) <= set(channels): + raise ValueError("focus vertical scales must target selected channels") + if not isinstance(self.hide_others, bool): + raise TypeError("focus hide_others must be bool") + + +@dataclass(frozen=True, slots=True) +class ScopeFocusChannelState: + channel: int + enabled: bool + range_v: float + scale_v_per_div: float + position: float + offset_v: float + + def __post_init__(self) -> None: + _strict_int(self.channel, label="focus state channel", minimum=1) + if not isinstance(self.enabled, bool): + raise TypeError("focus state enabled must be bool") + range_v = _finite(self.range_v, label="focus state vertical range") + if range_v <= 0: + raise ValueError("focus state vertical range must be > 0") + object.__setattr__(self, "range_v", range_v) + scale = _finite(self.scale_v_per_div, label="focus state vertical scale") + if scale <= 0: + raise ValueError("focus state vertical scale must be > 0") + object.__setattr__(self, "scale_v_per_div", scale) + object.__setattr__( + self, + "position", + _finite(self.position, label="focus state vertical position"), + ) + object.__setattr__( + self, + "offset_v", + _finite(self.offset_v, label="focus state vertical offset"), + ) + + +@dataclass(frozen=True, slots=True) +class ScopeFocusState: + time_range_s: float + time_position_s: float + channels: tuple[ScopeFocusChannelState, ...] + + def __post_init__(self) -> None: + time_range_s = _finite(self.time_range_s, label="focus state time range") + if time_range_s <= 0: + raise ValueError("focus state time range must be > 0") + object.__setattr__(self, "time_range_s", time_range_s) + object.__setattr__( + self, + "time_position_s", + _finite(self.time_position_s, label="focus state time position"), + ) + channels = _unique_tuple(self.channels, label="focus state channels") + if not channels or not all(isinstance(item, ScopeFocusChannelState) for item in channels): + raise TypeError("focus state channels contain an invalid item") + channel_numbers = tuple(item.channel for item in channels) + if tuple(sorted(channel_numbers)) != channel_numbers: + raise ValueError("focus state channels must use ascending channel order") + + +@dataclass(frozen=True, slots=True) +class ScopeFocusProfileV2: + analog_channels: tuple[int, ...] + time_range_min_s: float + time_range_max_s: float + time_range_abs_tolerance_s: float + vertical_scale_min_v_per_div: float + vertical_scale_max_v_per_div: float + vertical_scale_abs_tolerance_v_per_div: float + vertical_range_abs_tolerance_v: float + time_position_abs_tolerance_s: float + position_abs_tolerance: float + offset_abs_tolerance_v: float + snapshot_max_steps: int + configure_max_steps: int + restore_max_steps: int + verify_max_steps: int + + def __post_init__(self) -> None: + channels = _unique_tuple(self.analog_channels, label="focus analog_channels") + _strict_int(len(channels), label="focus analog channel count", minimum=1, maximum=64) + for channel in channels: + _strict_int(channel, label="focus analog channel", minimum=1, maximum=65_535) + if tuple(sorted(channels)) != channels: + raise ValueError("focus analog_channels must use ascending channel order") + self._validate_bounds( + self.time_range_min_s, + self.time_range_max_s, + label="focus time range", + ) + self._validate_bounds( + self.vertical_scale_min_v_per_div, + self.vertical_scale_max_v_per_div, + label="focus vertical scale", + ) + for label, value in ( + ("focus time range tolerance", self.time_range_abs_tolerance_s), + ("focus vertical scale tolerance", self.vertical_scale_abs_tolerance_v_per_div), + ("focus vertical range tolerance", self.vertical_range_abs_tolerance_v), + ("focus time position tolerance", self.time_position_abs_tolerance_s), + ("focus position tolerance", self.position_abs_tolerance), + ("focus offset tolerance", self.offset_abs_tolerance_v), + ): + if _finite(value, label=label) < 0: + raise ValueError(f"{label} must be >= 0") + for label, value, minimum in ( + ("snapshot_max_steps", self.snapshot_max_steps, 1), + ("configure_max_steps", self.configure_max_steps, 2), + ("restore_max_steps", self.restore_max_steps, 1), + ("verify_max_steps", self.verify_max_steps, 1), + ): + _strict_int(value, label=f"focus {label}", minimum=minimum, maximum=512) + + @staticmethod + def _validate_bounds(lower: object, upper: object, *, label: str) -> None: + lower_value = _finite(lower, label=f"{label} minimum") + upper_value = _finite(upper, label=f"{label} maximum") + if lower_value <= 0 or lower_value > upper_value: + raise ValueError(f"{label} bounds are invalid") + + @staticmethod + def _close(left: float, right: float, tolerance: float) -> bool: + return isclose(left, right, rel_tol=0.0, abs_tol=tolerance) + + @staticmethod + def _channel_map(state: ScopeFocusState) -> dict[int, ScopeFocusChannelState]: + return {item.channel: item for item in state.channels} + + def validate_request(self, request: ScopeFocusRequest) -> None: + if not isinstance(request, ScopeFocusRequest): + raise TypeError("focus request has an invalid type") + if not set(request.channels) <= set(self.analog_channels): + raise ValueError("focus request is outside the descriptor analog channels") + if request.time_range_s is not None and not ( + self.time_range_min_s <= request.time_range_s <= self.time_range_max_s + ): + raise ValueError("focus time range is outside the descriptor profile") + for item in request.vertical_scales: + if not ( + self.vertical_scale_min_v_per_div + <= item.scale_v_per_div + <= self.vertical_scale_max_v_per_div + ): + raise ValueError("focus vertical scale is outside the descriptor profile") + + def validate_state(self, state: ScopeFocusState) -> None: + if not isinstance(state, ScopeFocusState): + raise TypeError("focus driver returned an invalid state") + if tuple(item.channel for item in state.channels) != self.analog_channels: + raise ValueError("focus driver state does not cover the descriptor analog channels") + if not self.time_range_min_s <= state.time_range_s <= self.time_range_max_s: + raise ValueError("focus driver time range is outside the descriptor profile") + if any( + not self.vertical_scale_min_v_per_div + <= item.scale_v_per_div + <= self.vertical_scale_max_v_per_div + for item in state.channels + ): + raise ValueError("focus driver vertical scale is outside the descriptor profile") + + def restore_order_for(self, request: ScopeFocusRequest) -> tuple[ScopeFocusField, ...]: + self.validate_request(request) + return ( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ) + + def request_satisfied(self, state: ScopeFocusState, request: ScopeFocusRequest) -> bool: + self.validate_request(request) + self.validate_state(state) + channel_map = self._channel_map(state) + targets = set(request.channels) + if any(not channel_map[channel].enabled for channel in targets): + return False + if request.hide_others and any( + item.enabled for item in state.channels if item.channel not in targets + ): + return False + if request.time_range_s is not None and not self._close( + state.time_range_s, + request.time_range_s, + self.time_range_abs_tolerance_s, + ): + return False + return all( + self._close( + channel_map[item.channel].scale_v_per_div, + item.scale_v_per_div, + self.vertical_scale_abs_tolerance_v_per_div, + ) + for item in request.vertical_scales + ) + + def transition_matches( + self, + before: ScopeFocusState, + after: ScopeFocusState, + request: ScopeFocusRequest, + ) -> bool: + self.validate_request(request) + self.validate_state(before) + self.validate_state(after) + if not self.request_satisfied(after, request): + return False + if request.time_range_s is None and not self._close( + before.time_range_s, + after.time_range_s, + self.time_range_abs_tolerance_s, + ): + return False + if not self._close( + before.time_position_s, + after.time_position_s, + self.time_position_abs_tolerance_s, + ): + return False + before_map = self._channel_map(before) + after_map = self._channel_map(after) + targets = set(request.channels) + requested_scales = {item.channel: item.scale_v_per_div for item in request.vertical_scales} + for channel in self.analog_channels: + old = before_map[channel] + new = after_map[channel] + if channel not in targets and not request.hide_others and old.enabled is not new.enabled: + return False + requested_scale = requested_scales.get(channel) + if requested_scale is None and not self._close( + old.scale_v_per_div, + new.scale_v_per_div, + self.vertical_scale_abs_tolerance_v_per_div, + ): + return False + if requested_scale is None and not self._close( + old.range_v, + new.range_v, + self.vertical_range_abs_tolerance_v, + ): + return False + if not self._close(old.position, new.position, self.position_abs_tolerance): + return False + if not self._close(old.offset_v, new.offset_v, self.offset_abs_tolerance_v): + return False + return True + + def states_equivalent(self, expected: ScopeFocusState, observed: ScopeFocusState) -> bool: + self.validate_state(expected) + self.validate_state(observed) + if not self._close( + expected.time_range_s, + observed.time_range_s, + self.time_range_abs_tolerance_s, + ): + return False + if not self._close( + expected.time_position_s, + observed.time_position_s, + self.time_position_abs_tolerance_s, + ): + return False + expected_map = self._channel_map(expected) + observed_map = self._channel_map(observed) + for channel in self.analog_channels: + left = expected_map[channel] + right = observed_map[channel] + if left.enabled is not right.enabled: + return False + if not self._close( + left.range_v, + right.range_v, + self.vertical_range_abs_tolerance_v, + ): + return False + if not self._close( + left.scale_v_per_div, + right.scale_v_per_div, + self.vertical_scale_abs_tolerance_v_per_div, + ): + return False + if not self._close(left.position, right.position, self.position_abs_tolerance): + return False + if not self._close(left.offset_v, right.offset_v, self.offset_abs_tolerance_v): + return False + return True + + +@dataclass(frozen=True, slots=True) +class ScopeFocusResult: + request: ScopeFocusRequest + before: ScopeFocusState + after: ScopeFocusState + write_performed: bool + + def __post_init__(self) -> None: + if not isinstance(self.request, ScopeFocusRequest): + raise TypeError("focus result request has an invalid type") + if not isinstance(self.before, ScopeFocusState) or not isinstance( + self.after, + ScopeFocusState, + ): + raise TypeError("focus result state has an invalid type") + if tuple(item.channel for item in self.before.channels) != tuple( + item.channel for item in self.after.channels + ): + raise ValueError("focus result states use different channel sets") + if not isinstance(self.write_performed, bool): + raise TypeError("focus result write_performed must be bool") + if not self.write_performed and self.before != self.after: + raise ValueError("focus no-op result must preserve its exact state") + + +@dataclass(frozen=True, slots=True) +class ScopeFocusBaseline: + context_id: str + session_epoch: str + baseline_nonce: str + snapshot: ScopeFocusState + restore_order: tuple[ScopeFocusField, ...] + + def __post_init__(self) -> None: + _safe_token(self.context_id, label="focus baseline context_id") + _safe_token(self.session_epoch, label="focus baseline session_epoch") + _safe_token(self.baseline_nonce, label="focus baseline nonce") + if not isinstance(self.snapshot, ScopeFocusState): + raise TypeError("focus baseline snapshot has an invalid type") + order = _unique_tuple(self.restore_order, label="focus restore_order") + expected = ( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ) + if order != expected: + raise ValueError("focus restore_order must cover all fields in the fixed safe order") + + +@dataclass(frozen=True, slots=True) +class ScopeFocusRestoreResult: + status: Literal["completed", "failed", "not_attempted"] + attempted_fields: tuple[ScopeFocusField, ...] + restored_fields: tuple[ScopeFocusField, ...] + error_code: str | None = None + + def __post_init__(self) -> None: + _literal(self.status, {"completed", "failed", "not_attempted"}, label="focus restore status") + attempted = _unique_tuple(self.attempted_fields, label="attempted_fields") + restored = _unique_tuple(self.restored_fields, label="restored_fields") + if not set(attempted + restored) <= _SCOPE_FOCUS_FIELDS: + raise ValueError("focus restore fields are invalid") + _optional_safe_token(self.error_code, label="error_code") + + def validate_for(self, baseline: ScopeFocusBaseline) -> None: + if not isinstance(baseline, ScopeFocusBaseline): + raise TypeError("focus baseline has an invalid type") + _validate_prefix_and_subsequence( + expected=baseline.restore_order, + attempted=self.attempted_fields, + completed=self.restored_fields, + status=self.status, + ) + + @dataclass(frozen=True, slots=True) class ScopeAcquisitionControlSnapshot: run_state: ScopeAcquisitionRunState @@ -2880,6 +3430,8 @@ class ScopeDescriptorExtensions: fft_status_profile_v2: ScopeFftStatusProfileV2 | None = None cursor_readout_profile_v2: ScopeCursorReadoutProfileV2 | None = None average_capture_profile_v2: ScopeAverageCaptureProfileV2 | None = None + channel_display_profile_v2: ScopeChannelDisplayProfileV2 | None = None + focus_profile_v2: ScopeFocusProfileV2 | None = None def __post_init__(self) -> None: for label, value, expected in ( @@ -2925,6 +3477,16 @@ def __post_init__(self) -> None: self.average_capture_profile_v2, ScopeAverageCaptureProfileV2, ), + ( + "channel_display_profile_v2", + self.channel_display_profile_v2, + ScopeChannelDisplayProfileV2, + ), + ( + "focus_profile_v2", + self.focus_profile_v2, + ScopeFocusProfileV2, + ), ): if value is not None and not isinstance(value, expected): raise TypeError(f"{label} has an invalid type") @@ -3074,6 +3636,43 @@ def acquire_single( ) -> ScopeAcquisitionCompletion: ... +@runtime_checkable +class ScopeChannelDisplayDriverV2(InstrumentDriver, Protocol): + def get_channel_display_state_v2( + self, + channel: int, + ) -> ScopeChannelDisplayState: ... + + def configure_channel_display_v2( + self, + request: ScopeChannelDisplayRequest, + *, + baseline: ScopeChannelDisplayBaseline, + ) -> None: ... + + def restore_channel_display_v2( + self, + baseline: ScopeChannelDisplayBaseline, + ) -> ScopeChannelDisplayRestoreResult: ... + + +@runtime_checkable +class ScopeFocusDriverV2(InstrumentDriver, Protocol): + def get_focus_state_v2(self) -> ScopeFocusState: ... + + def configure_focus_v2( + self, + request: ScopeFocusRequest, + *, + baseline: ScopeFocusBaseline, + ) -> None: ... + + def restore_focus_v2( + self, + baseline: ScopeFocusBaseline, + ) -> ScopeFocusRestoreResult: ... + + @runtime_checkable class ScopeTraceTransferRecoveryDriver(InstrumentDriver, Protocol): def snapshot_trace_transfer_state( diff --git a/src/wavebench/services/operation_specs.py b/src/wavebench/services/operation_specs.py index 226e777c..0e503c25 100644 --- a/src/wavebench/services/operation_specs.py +++ b/src/wavebench/services/operation_specs.py @@ -1633,6 +1633,49 @@ def _scope_operation( risk_flags=("acquisition_state", "recovery_required"), error_check_minimum="disabled", ), + _scope_operation( + "scope.channel_display_configure_v2", + required_capabilities=("scope.channel_display_configure_v2",), + effect="write", + timeout_ms=SCOPE_PROFILE_OPERATION_TIMEOUT_MS, + changed_fields=("scope.channel_display", "scope.error_queue"), + restore_coverage="failure-cleanup-only", + verification_fields=("scope.channel_display",), + postcondition_fields=("scope.channel_display",), + cleanup_verification_fields=("scope.channel_display",), + risk_flags=("front_panel_state", "recovery_required"), + error_check_minimum="disabled", + ), + _scope_operation( + "scope.focus_configure_v2", + required_capabilities=("scope.focus_configure_v2",), + effect="write", + timeout_ms=SCOPE_PROFILE_OPERATION_TIMEOUT_MS, + changed_fields=( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + "scope.error_queue", + ), + restore_coverage="failure-cleanup-only", + verification_fields=( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ), + postcondition_fields=( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ), + cleanup_verification_fields=( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ), + risk_flags=("front_panel_state", "multi_channel_state", "recovery_required"), + error_check_minimum="disabled", + ), _scope_operation( "scope.trace_metadata", required_capabilities=("scope.trace_metadata",), diff --git a/src/wavebench/services/scope_extension_service.py b/src/wavebench/services/scope_extension_service.py index 683b0291..f096a87c 100644 --- a/src/wavebench/services/scope_extension_service.py +++ b/src/wavebench/services/scope_extension_service.py @@ -22,7 +22,19 @@ ScopeAcquisitionRunState, ScopeBaselineRestoreResult, ScopeBaselineVerification, + ScopeChannelDisplayBaseline, + ScopeChannelDisplayProfileV2, + ScopeChannelDisplayRequest, + ScopeChannelDisplayRestoreResult, + ScopeChannelDisplayResult, + ScopeChannelDisplayState, ScopeContinuousAcquisitionRequest, + ScopeFocusBaseline, + ScopeFocusProfileV2, + ScopeFocusRequest, + ScopeFocusRestoreResult, + ScopeFocusResult, + ScopeFocusState, ScopeScreenshot, ScopeScreenshotBaseline, ScopeScreenshotProfile, @@ -357,6 +369,342 @@ def acquisition_run_state( self._attach_diagnostics(exc, context, None) raise + def configure_channel_display_v2( + self, + request: ScopeChannelDisplayRequest, + *, + error_check: ErrorCheckSpec | None = None, + deadline: float | None = None, + ) -> ScopeExtensionOperationResult: + if not isinstance(request, ScopeChannelDisplayRequest): + raise DataError("channel display request has an invalid type") + spec = self._require("scope.channel_display_configure_v2") + profile = self._channel_display_profile_v2() + try: + profile.validate_request(request) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + context = self._context(spec, deadline=deadline) + try: + error_executor = self._error_executor(spec, error_check, context) + except BaseException: + context.complete() + raise + handle: ScopeBaselineHandle | None = None + baseline: ScopeChannelDisplayBaseline | None = None + before: ScopeChannelDisplayState | None = None + after: ScopeChannelDisplayState | None = None + value: ScopeChannelDisplayResult | None = None + primary: BaseException | None = None + cleanup_error: BaseException | None = None + cleanup_diagnostics: dict[str, object] = {} + write_attempted = False + try: + fields = ("scope.channel_display",) + preflight = context.make_phase_spec( + OperationPhase.PREFLIGHT, + allowed_io={"query"}, + fields={"scope.identity", *fields}, + max_steps=1 + profile.snapshot_max_steps, + ) + with context.authorize_phase(preflight) as authorization: + self._verify_identity() + before = self.driver.get_channel_display_state_v2(request.channel) + self._validate_channel_display_state(before, profile, request.channel) + if before.enabled is not request.enabled: + handle = context.create_baseline( + kind="channel_display", + fields=fields, + restore_order=fields, + ) + baseline = ScopeChannelDisplayBaseline( + context_id=handle.context_id, + session_epoch=handle.session_epoch, + baseline_nonce=handle.baseline_nonce, + snapshot=before, + restore_order=fields, + ) + context.pass_baseline_to_main(handle) + context.complete_phase_verification( + authorization, + io_kind="query", + fields={"scope.identity", *fields}, + ) + + error_executor.run(context, phase="before") + main = context.make_phase_spec( + OperationPhase.MAIN, + allowed_io={"write", "query", "query_opc"}, + fields=set(spec.changed_fields), + max_steps=profile.configure_max_steps, + ) + try: + with context.authorize_phase(main): + assert before is not None + if baseline is not None: + write_attempted = True + error_executor.mark_main_sent() + self.driver.configure_channel_display_v2( + request, + baseline=baseline, + ) + after = self.driver.get_channel_display_state_v2(request.channel) + self._validate_channel_display_state(after, profile, request.channel) + if after.enabled is not request.enabled: + raise DataError( + "channel display postcondition does not match the request" + ) + else: + after = before + value = ScopeChannelDisplayResult( + request=request, + before=before, + after=after, + write_performed=write_attempted, + ) + error_executor.run(context, phase="after") + except BaseException as exc: + primary = exc + if error_executor.wants("after") and not context.has_phase( + OperationPhase.ERROR_AFTER + ): + error_executor.omit_after( + "session_unhealthy" + if self.session_state.health is not SessionHealth.HEALTHY + else "main_operation_failed" + ) + if primary is None and handle is not None: + context.consume_baseline_after_success(handle) + elif primary is not None and handle is not None and write_attempted: + context.mark_cleanup_required() + cleanup_error, cleanup_diagnostics = self._cleanup_channel_display( + context, + handle, + baseline, + profile, + ) + context.complete() + operation_details = { + "request": _json_safe(request), + "before": _json_safe(before), + "after": _json_safe(after), + } + if primary is not None: + self._attach_diagnostics( + primary, + context, + error_executor, + cleanup_error=cleanup_error, + extra={ + "channel_display": operation_details, + "cleanup": cleanup_diagnostics, + }, + ) + raise primary + assert value is not None and after is not None and before is not None + return self._result( + context, + value=value, + error_executor=error_executor, + observed_state={ + "before": asdict(before), + "after": asdict(after), + }, + extra={ + "postcondition": { + "status": "verified", + "fields": ["scope.channel_display"], + "write_performed": value.write_performed, + }, + "cleanup": cleanup_diagnostics, + }, + ) + except BaseException as exc: + if not context.terminal: + context.complete() + if not hasattr(exc, "scope_operation_diagnostics"): + self._attach_diagnostics( + exc, + context, + error_executor, + extra={ + "channel_display": { + "request": _json_safe(request), + "before": _json_safe(before), + "after": _json_safe(after), + } + }, + ) + raise + + def configure_focus_v2( + self, + request: ScopeFocusRequest, + *, + error_check: ErrorCheckSpec | None = None, + deadline: float | None = None, + ) -> ScopeExtensionOperationResult: + if not isinstance(request, ScopeFocusRequest): + raise DataError("focus request has an invalid type") + spec = self._require("scope.focus_configure_v2") + profile = self._focus_profile_v2() + try: + profile.validate_request(request) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + context = self._context(spec, deadline=deadline) + try: + error_executor = self._error_executor(spec, error_check, context) + except BaseException: + context.complete() + raise + handle: ScopeBaselineHandle | None = None + baseline: ScopeFocusBaseline | None = None + before: ScopeFocusState | None = None + after: ScopeFocusState | None = None + value: ScopeFocusResult | None = None + primary: BaseException | None = None + cleanup_error: BaseException | None = None + cleanup_diagnostics: dict[str, object] = {} + write_attempted = False + all_fields = ( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ) + try: + restore_order = profile.restore_order_for(request) + preflight = context.make_phase_spec( + OperationPhase.PREFLIGHT, + allowed_io={"query"}, + fields={"scope.identity", *all_fields}, + max_steps=1 + profile.snapshot_max_steps, + ) + with context.authorize_phase(preflight) as authorization: + self._verify_identity() + before = self.driver.get_focus_state_v2() + self._validate_focus_state(before, profile) + if not profile.request_satisfied(before, request): + handle = context.create_baseline( + kind="focus", + fields=restore_order, + restore_order=restore_order, + ) + baseline = ScopeFocusBaseline( + context_id=handle.context_id, + session_epoch=handle.session_epoch, + baseline_nonce=handle.baseline_nonce, + snapshot=before, + restore_order=restore_order, + ) + context.pass_baseline_to_main(handle) + context.complete_phase_verification( + authorization, + io_kind="query", + fields={"scope.identity", *all_fields}, + ) + + error_executor.run(context, phase="before") + main = context.make_phase_spec( + OperationPhase.MAIN, + allowed_io={"write", "query", "query_opc"}, + fields=set(spec.changed_fields), + max_steps=profile.configure_max_steps, + ) + try: + with context.authorize_phase(main): + assert before is not None + if baseline is not None: + write_attempted = True + error_executor.mark_main_sent() + self.driver.configure_focus_v2(request, baseline=baseline) + after = self.driver.get_focus_state_v2() + self._validate_focus_state(after, profile) + if not profile.transition_matches(before, after, request): + raise DataError("focus postcondition does not match the request") + else: + after = before + value = ScopeFocusResult( + request=request, + before=before, + after=after, + write_performed=write_attempted, + ) + error_executor.run(context, phase="after") + except BaseException as exc: + primary = exc + if error_executor.wants("after") and not context.has_phase( + OperationPhase.ERROR_AFTER + ): + error_executor.omit_after( + "session_unhealthy" + if self.session_state.health is not SessionHealth.HEALTHY + else "main_operation_failed" + ) + if primary is None and handle is not None: + context.consume_baseline_after_success(handle) + elif primary is not None and handle is not None and write_attempted: + context.mark_cleanup_required() + cleanup_error, cleanup_diagnostics = self._cleanup_focus( + context, + handle, + baseline, + profile, + ) + context.complete() + operation_details = { + "request": _json_safe(request), + "before": _json_safe(before), + "after": _json_safe(after), + } + if primary is not None: + self._attach_diagnostics( + primary, + context, + error_executor, + cleanup_error=cleanup_error, + extra={ + "focus": operation_details, + "cleanup": cleanup_diagnostics, + }, + ) + raise primary + assert value is not None and after is not None and before is not None + return self._result( + context, + value=value, + error_executor=error_executor, + observed_state={ + "before": asdict(before), + "after": asdict(after), + }, + extra={ + "postcondition": { + "status": "verified", + "fields": list(all_fields), + "write_performed": value.write_performed, + }, + "cleanup": cleanup_diagnostics, + }, + ) + except BaseException as exc: + if not context.terminal: + context.complete() + if not hasattr(exc, "scope_operation_diagnostics"): + self._attach_diagnostics( + exc, + context, + error_executor, + extra={ + "focus": { + "request": _json_safe(request), + "before": _json_safe(before), + "after": _json_safe(after), + } + }, + ) + raise + def start_acquisition( self, request: ScopeContinuousAcquisitionRequest, @@ -1011,6 +1359,148 @@ def _cleanup_acquisition( "verification": _json_safe(verification), } + def _cleanup_channel_display( + self, + context: ScopeOperationContextCoordinator, + handle: ScopeBaselineHandle, + baseline: ScopeChannelDisplayBaseline | None, + profile: ScopeChannelDisplayProfileV2, + ) -> tuple[BaseException | None, dict[str, object]]: + assert baseline is not None + restore_result: ScopeChannelDisplayRestoreResult | None = None + verification: dict[str, object] | None = None + error: BaseException | None = None + try: + restore = context.make_phase_spec( + OperationPhase.FAILURE_CLEANUP, + allowed_io={"write"}, + fields=handle.fields, + max_steps=profile.restore_max_steps, + ) + with context.authorize_phase(restore): + context.begin_restore(handle) + try: + restore_result = self.driver.restore_channel_display_v2(baseline) + if not isinstance(restore_result, ScopeChannelDisplayRestoreResult): + raise TypeError( + "restore_channel_display_v2() returned an invalid result" + ) + restore_result.validate_for(baseline) + succeeded = restore_result.status == "completed" + except BaseException: + context.finish_restore(handle, succeeded=False) + raise + context.finish_restore(handle, succeeded=succeeded) + if not succeeded: + raise InstrumentError("channel display restore did not complete") + except BaseException as exc: + error = exc + + if self.session_state.health is not SessionHealth.POISONED: + try: + verify = context.make_phase_spec( + OperationPhase.CLEANUP_VERIFICATION, + allowed_io={"query"}, + fields=handle.fields, + max_steps=profile.verify_max_steps, + ) + with context.authorize_phase(verify) as authorization: + context.begin_verification(handle) + observed = self.driver.get_channel_display_state_v2( + baseline.snapshot.channel + ) + self._validate_channel_display_state( + observed, + profile, + baseline.snapshot.channel, + ) + matched = observed == baseline.snapshot + verification = { + "status": "verified" if matched else "mismatch", + "expected": _json_safe(baseline.snapshot), + "observed": _json_safe(observed), + } + context.finish_verification( + handle, + authorization, + io_kind="query", + verified_fields=handle.fields, + matched=matched, + ) + except BaseException as exc: + error = error or exc + return error, { + "restore": _json_safe(restore_result), + "verification": verification, + } + + def _cleanup_focus( + self, + context: ScopeOperationContextCoordinator, + handle: ScopeBaselineHandle, + baseline: ScopeFocusBaseline | None, + profile: ScopeFocusProfileV2, + ) -> tuple[BaseException | None, dict[str, object]]: + assert baseline is not None + restore_result: ScopeFocusRestoreResult | None = None + verification: dict[str, object] | None = None + error: BaseException | None = None + try: + restore = context.make_phase_spec( + OperationPhase.FAILURE_CLEANUP, + allowed_io={"write"}, + fields=handle.fields, + max_steps=profile.restore_max_steps, + ) + with context.authorize_phase(restore): + context.begin_restore(handle) + try: + restore_result = self.driver.restore_focus_v2(baseline) + if not isinstance(restore_result, ScopeFocusRestoreResult): + raise TypeError("restore_focus_v2() returned an invalid result") + restore_result.validate_for(baseline) + succeeded = restore_result.status == "completed" + except BaseException: + context.finish_restore(handle, succeeded=False) + raise + context.finish_restore(handle, succeeded=succeeded) + if not succeeded: + raise InstrumentError("focus restore did not complete") + except BaseException as exc: + error = exc + + if self.session_state.health is not SessionHealth.POISONED: + try: + verify = context.make_phase_spec( + OperationPhase.CLEANUP_VERIFICATION, + allowed_io={"query"}, + fields=handle.fields, + max_steps=profile.verify_max_steps, + ) + with context.authorize_phase(verify) as authorization: + context.begin_verification(handle) + observed = self.driver.get_focus_state_v2() + self._validate_focus_state(observed, profile) + matched = profile.states_equivalent(baseline.snapshot, observed) + verification = { + "status": "verified" if matched else "mismatch", + "expected": _json_safe(baseline.snapshot), + "observed": _json_safe(observed), + } + context.finish_verification( + handle, + authorization, + io_kind="query", + verified_fields=handle.fields, + matched=matched, + ) + except BaseException as exc: + error = error or exc + return error, { + "restore": _json_safe(restore_result), + "verification": verification, + } + def _cleanup_stop( self, context: ScopeOperationContextCoordinator, @@ -1131,6 +1621,20 @@ def _acquisition_profile(self) -> ScopeAcquisitionControlProfile: raise ConfigError("scope acquisition control requires a descriptor profile") return profile + def _channel_display_profile_v2(self) -> ScopeChannelDisplayProfileV2: + extensions = self.descriptor.scope_extensions + profile = extensions.channel_display_profile_v2 if extensions is not None else None + if profile is None: + raise ConfigError("scope channel display capability requires a descriptor profile") + return profile + + def _focus_profile_v2(self) -> ScopeFocusProfileV2: + extensions = self.descriptor.scope_extensions + profile = extensions.focus_profile_v2 if extensions is not None else None + if profile is None: + raise ConfigError("scope focus capability requires a descriptor profile") + return profile + def _trace_profile(self) -> ScopeTraceProfile: extensions = self.descriptor.scope_extensions profile = extensions.trace_profile if extensions is not None else None @@ -1231,6 +1735,24 @@ def _validate_acquisition_snapshot(snapshot: object) -> None: if not isinstance(snapshot, ScopeAcquisitionControlSnapshot): raise DataError("acquisition control snapshot has an invalid type") + @staticmethod + def _validate_channel_display_state( + state: object, + profile: ScopeChannelDisplayProfileV2, + channel: int, + ) -> None: + try: + profile.validate_state(state, channel=channel) # type: ignore[arg-type] + except (TypeError, ValueError) as exc: + raise DataError(f"channel display driver returned an invalid state: {exc}") from exc + + @staticmethod + def _validate_focus_state(state: object, profile: ScopeFocusProfileV2) -> None: + try: + profile.validate_state(state) # type: ignore[arg-type] + except (TypeError, ValueError) as exc: + raise DataError(f"focus driver returned an invalid state: {exc}") from exc + @staticmethod def _validate_start_result( value: object, diff --git a/src/wavebench/services/scope_service.py b/src/wavebench/services/scope_service.py index c3a66117..8c5411da 100644 --- a/src/wavebench/services/scope_service.py +++ b/src/wavebench/services/scope_service.py @@ -70,6 +70,10 @@ ScopeAverageCaptureProfileV2, ScopeAverageCaptureRequestV2, ScopeAverageCaptureResultV2, + ScopeChannelDisplayProfileV2, + ScopeChannelDisplayRequest, + ScopeFocusProfileV2, + ScopeFocusRequest, ScopeCursorReadoutProfileV2, ScopeFftStatusProfileV2, ScopeMeasurementStatisticsProfileV2, @@ -1010,6 +1014,57 @@ def acquisition_run_state( with self._scope_session() as scope: return self._scope_extension_service(scope).acquisition_run_state(deadline=deadline) + def configure_channel_display_v2( + self, + request: ScopeChannelDisplayRequest, + *, + error_check: ErrorCheckSpec | None = None, + deadline: float | None = None, + ) -> ScopeExtensionOperationResult: + if not isinstance(request, ScopeChannelDisplayRequest): + raise DataError("channel display request has an invalid type") + self._require( + "scope.channel_display_configure_v2", + "scope.channel_display_configure_v2", + ) + profile = self._channel_display_v2_profile() + if profile is None: + raise ConfigError("scope channel display capability requires a descriptor profile") + try: + profile.validate_request(request) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + with self._scope_session() as scope: + return self._scope_extension_service(scope).configure_channel_display_v2( + request, + error_check=error_check, + deadline=deadline, + ) + + def configure_focus_v2( + self, + request: ScopeFocusRequest, + *, + error_check: ErrorCheckSpec | None = None, + deadline: float | None = None, + ) -> ScopeExtensionOperationResult: + if not isinstance(request, ScopeFocusRequest): + raise DataError("focus request has an invalid type") + self._require("scope.focus_configure_v2", "scope.focus_configure_v2") + profile = self._focus_v2_profile() + if profile is None: + raise ConfigError("scope focus capability requires a descriptor profile") + try: + profile.validate_request(request) + except (TypeError, ValueError) as exc: + raise ConfigError(str(exc)) from exc + with self._scope_session() as scope: + return self._scope_extension_service(scope).configure_focus_v2( + request, + error_check=error_check, + deadline=deadline, + ) + def start_acquisition( self, request: ScopeContinuousAcquisitionRequest, @@ -1257,6 +1312,20 @@ def _v2_profile( def _snapshot_v2_profile(self) -> ScopeSnapshotProfileV2 | None: return self._v2_profile("snapshot_profile_v2", ScopeSnapshotProfileV2, "snapshot") + def _channel_display_v2_profile(self) -> ScopeChannelDisplayProfileV2 | None: + return self._v2_profile( + "channel_display_profile_v2", + ScopeChannelDisplayProfileV2, + "channel display", + ) + + def _focus_v2_profile(self) -> ScopeFocusProfileV2 | None: + return self._v2_profile( + "focus_profile_v2", + ScopeFocusProfileV2, + "focus", + ) + def _acquisition_status_v2_profile(self) -> ScopeAcquisitionStatusProfileV2 | None: descriptor = self.descriptor or resolve_instrument_descriptor( self.config.scope.driver, diff --git a/tests/test_scope_channel_display_configure_v2.py b/tests/test_scope_channel_display_configure_v2.py new file mode 100644 index 00000000..2be86b0d --- /dev/null +++ b/tests/test_scope_channel_display_configure_v2.py @@ -0,0 +1,381 @@ +from __future__ import annotations + +from contextlib import redirect_stdout +from dataclasses import asdict, replace +import io +import json +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from wavebench.cli import main +from wavebench.cli_parser import build_parser +from wavebench.errors import AccessDeniedError, ConfigError, TransportIOError +from wavebench.instruments import InstrumentDescriptor, ScopeChannelDisplayDriverV2 +from wavebench.instruments.capabilities import CAPABILITY_METHODS, validate_declared_capabilities +from wavebench.instruments.contracts import ScopeDriver +from wavebench.instruments.factory import open_instrument_driver +from wavebench.instruments.scope_extension_capabilities import ( + SCOPE_CAPABILITY_METHODS, + SCOPE_STRICT_V2_CAPABILITIES, + validate_scope_descriptor, +) +from wavebench.instruments.scope_extensions import ( + ScopeChannelDisplayBaseline, + ScopeChannelDisplayProfileV2, + ScopeChannelDisplayRequest, + ScopeChannelDisplayRestoreResult, + ScopeChannelDisplayResult, + ScopeChannelDisplayState, + ScopeDescriptorExtensions, +) +from wavebench.logging import CommandLogger +from wavebench.services.capability_explain import explain_operation +from wavebench.services.operation_specs import require_operation_spec +from wavebench.services.scope_service import ScopeService +from wavebench.transport.contracts import ReplayPolicy + + +def _profile() -> ScopeChannelDisplayProfileV2: + return ScopeChannelDisplayProfileV2( + analog_channels=(1, 2), + snapshot_max_steps=1, + configure_max_steps=2, + restore_max_steps=1, + verify_max_steps=1, + ) + + +def _descriptor( + *, + minimum: str = "0.8.24", + capabilities: tuple[str, ...] = ( + "scope.idn", + "scope.channel_display_configure_v2", + ), + extensions: bool = True, +) -> InstrumentDescriptor: + return InstrumentDescriptor( + driver_id="example.channel-display-v2", + kind="scope", + display_name="Example Scope", + manufacturer="Example", + models=("EX1",), + aliases=(), + capabilities=capabilities, + idn_patterns=("EXAMPLE,EX1",), + backends=("pyvisa",), + option_specs=(), + permissions=("instrument.io",), + factory=lambda _context: object(), + wavebench_min_version=minimum, + scope_extensions=( + ScopeDescriptorExtensions(channel_display_profile_v2=_profile()) + if extensions + else None + ), + ) + + +def test_channel_display_models_and_profile_freeze_single_field_semantics() -> None: + request = ScopeChannelDisplayRequest(channel=2, enabled=True) + before = ScopeChannelDisplayState(channel=2, enabled=False) + after = ScopeChannelDisplayState(channel=2, enabled=True) + result = ScopeChannelDisplayResult(request, before, after, write_performed=True) + baseline = ScopeChannelDisplayBaseline( + context_id="context", + session_epoch="epoch", + baseline_nonce="nonce", + snapshot=before, + restore_order=("scope.channel_display",), + ) + restored = ScopeChannelDisplayRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + assert asdict(result) == { + "request": {"channel": 2, "enabled": True}, + "before": {"channel": 2, "enabled": False}, + "after": {"channel": 2, "enabled": True}, + "write_performed": True, + } + _profile().validate_request(request) + _profile().validate_state(after, channel=2) + restored.validate_for(baseline) + + with pytest.raises(ValueError, match="integer"): + ScopeChannelDisplayRequest(channel=True, enabled=True) + with pytest.raises(TypeError, match="bool"): + ScopeChannelDisplayState(channel=1, enabled=1) # type: ignore[arg-type] + with pytest.raises(ValueError, match="ascending"): + ScopeChannelDisplayProfileV2((2, 1), 1, 2, 1, 1) + with pytest.raises(ValueError, match="outside"): + _profile().validate_request(ScopeChannelDisplayRequest(channel=3, enabled=True)) + with pytest.raises(ValueError, match="write_performed"): + ScopeChannelDisplayResult(request, before, after, write_performed=False) + + +def test_channel_display_is_an_additive_profile_gated_capability() -> None: + class Driver: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + def get_channel_display_state_v2(self, channel: int) -> ScopeChannelDisplayState: + return ScopeChannelDisplayState(channel, False) + + def configure_channel_display_v2(self, request, *, baseline) -> None: + pass + + def restore_channel_display_v2(self, baseline): + return ScopeChannelDisplayRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + assert isinstance(Driver(), ScopeChannelDisplayDriverV2) + assert "configure_channel_display_v2" not in ScopeDriver.__dict__ + expected_methods = ( + "get_channel_display_state_v2", + "configure_channel_display_v2", + "restore_channel_display_v2", + ) + assert SCOPE_CAPABILITY_METHODS["scope.channel_display_configure_v2"] == expected_methods + assert CAPABILITY_METHODS["scope.channel_display_configure_v2"] == expected_methods + assert "scope.channel_display_configure_v2" in SCOPE_STRICT_V2_CAPABILITIES + validate_declared_capabilities(_descriptor(), Driver()) + + +def test_channel_display_descriptor_requires_floor_profile_dependency_and_methods() -> None: + with pytest.raises(ConfigError, match="scope portability V2 capabilities require.*0.8.24"): + validate_scope_descriptor(_descriptor(minimum="0.8.23")) + with pytest.raises(ConfigError, match="scope_extensions.channel_display_profile_v2"): + validate_scope_descriptor(_descriptor(extensions=False)) + with pytest.raises(ConfigError, match="scope.idn"): + validate_scope_descriptor( + _descriptor(capabilities=("scope.channel_display_configure_v2",)) + ) + + class MissingMethods: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + with pytest.raises(TypeError, match="get_channel_display_state_v2"): + validate_declared_capabilities(_descriptor(), MissingMethods()) + + +def test_channel_display_operation_spec_and_explain_are_fail_closed() -> None: + spec = require_operation_spec("scope.channel_display_configure_v2") + + assert spec.instrument_kind == "scope" + assert spec.required_capabilities == ("scope.channel_display_configure_v2",) + assert spec.effect == "write" + assert spec.lease_mode == "exclusive" + assert spec.changed_fields == ("scope.channel_display", "scope.error_queue") + assert spec.restore_coverage == "failure-cleanup-only" + assert spec.verification_fields == ("scope.channel_display",) + assert spec.postcondition_fields == ("scope.channel_display",) + assert spec.cleanup_verification_fields == ("scope.channel_display",) + + descriptor = SimpleNamespace( + driver_id="example.channel-display-v2", + kind="scope", + capabilities=("scope.channel_display_configure_v2",), + ) + assert explain_operation( + "scope.channel_display_configure_v2", + descriptor=descriptor, + ).status == "supported" + assert explain_operation( + "scope.channel_display_configure_v2", + descriptor=descriptor, + access="read_only", + ).status == "access_denied" + + +def test_scope_service_rejects_access_and_unknown_channel_before_opening_session() -> None: + request = ScopeChannelDisplayRequest(channel=3, enabled=True) + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace( + driver="example.channel-display-v2", + access="read_write", + ), + ), + logger=SimpleNamespace(), + descriptor=_descriptor(), + ) + + with patch.object(service, "_open_scope") as open_scope: + with pytest.raises(ConfigError, match="outside"): + service.configure_channel_display_v2(request) + open_scope.assert_not_called() + + service.config.scope.access = "read_only" + with patch.object(service, "_open_scope") as open_scope: + with pytest.raises(AccessDeniedError, match="read_only"): + service.configure_channel_display_v2( + ScopeChannelDisplayRequest(channel=1, enabled=True) + ) + open_scope.assert_not_called() + + +def test_scope_display_cli_builds_typed_request_and_emits_versioned_result() -> None: + args = build_parser().parse_args(["scope", "display", "--channel", "2", "on"]) + assert args.command == "display" + assert args.channel == 2 + assert args.state == "on" + + calls: list[ScopeChannelDisplayRequest] = [] + scope_payload = { + "schema": "wavebench.scope.result.v1", + "result": { + "request": {"channel": 2, "enabled": True}, + "write_performed": True, + }, + "diagnostics": {"schema": "wavebench.scope.operation.v1"}, + "observed_state": None, + } + + class Result: + def as_dict(self) -> dict[str, object]: + return scope_payload + + class Service: + def configure_channel_display_v2(self, request, *, error_check=None): + assert error_check is None + calls.append(request) + return Result() + + stdout = io.StringIO() + with patch("wavebench.cli._load_service", return_value=Service()), redirect_stdout(stdout): + code = main(["scope", "display", "--channel", "2", "on"]) + + assert code == 0 + assert calls == [ScopeChannelDisplayRequest(channel=2, enabled=True)] + assert json.loads(stdout.getvalue()) == scope_payload + + stdout = io.StringIO() + with patch("wavebench.cli._load_service", return_value=Service()), redirect_stdout(stdout): + code = main(["scope", "display", "--channel", "2", "on", "--json"]) + + assert code == 0 + payload = json.loads(stdout.getvalue()) + assert payload["schema"] == "wavebench.cli.result.v1" + assert payload["result"] == scope_payload + + +class _FactoryTransport: + resource = "TCPIP::example::INSTR" + + def __init__(self) -> None: + self.queries: list[str] = [] + self.closed = 0 + + def record_event(self, _direction: str, _text: str) -> None: + pass + + def query(self, command: str, *, replay: ReplayPolicy = ReplayPolicy.NO_REPLAY) -> str: + assert replay is ReplayPolicy.NO_REPLAY + self.queries.append(command) + return "ok" + + def close(self) -> None: + self.closed += 1 + + +def _open_factory_descriptor() -> object: + return open_instrument_driver( + driver_reference="example.channel-display-v2", + expected_kind="scope", + resource="TCPIP::example::INSTR", + configured_backend="pyvisa", + timeout_ms=1_000, + opc_timeout_ms=2_000, + read_retry_attempts=1, + read_retry_delay_ms=0, + logger=CommandLogger(), + ) + + +def test_channel_display_factory_latch_and_missing_method_are_zero_io( + monkeypatch: pytest.MonkeyPatch, +) -> None: + inner = _FactoryTransport() + errors: list[TransportIOError] = [] + + class Driver: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + def get_channel_display_state_v2(self, channel: int) -> ScopeChannelDisplayState: + return ScopeChannelDisplayState(channel, False) + + def configure_channel_display_v2(self, request, *, baseline) -> None: + pass + + def restore_channel_display_v2(self, baseline): + return ScopeChannelDisplayRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + def factory(context): + transport = context.open_transport() + with pytest.raises(TransportIOError) as raised: + transport.query("*IDN?") + errors.append(raised.value) + return Driver() + + descriptor = replace(_descriptor(), factory=factory) + monkeypatch.setattr( + "wavebench.instruments.factory.resolve_instrument_descriptor", + lambda _reference, expected_kind: descriptor, + ) + monkeypatch.setattr("wavebench.instruments.factory._open_transport", lambda **_kwargs: inner) + + opened = _open_factory_descriptor() + + assert [error.reason_code for error in errors] == ["factory_construction_pending"] + assert errors[0].attempts == 0 + assert inner.queries == [] + assert opened.transport is not None + assert opened.transport.query("*IDN?") == "ok" + assert inner.queries == ["*IDN?"] + + inner = _FactoryTransport() + + class MissingMethods: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + missing = replace( + _descriptor(), + factory=lambda context: (context.open_transport(), MissingMethods())[1], + ) + monkeypatch.setattr( + "wavebench.instruments.factory.resolve_instrument_descriptor", + lambda _reference, expected_kind: missing, + ) + monkeypatch.setattr("wavebench.instruments.factory._open_transport", lambda **_kwargs: inner) + + with pytest.raises(ConfigError, match="get_channel_display_state_v2"): + _open_factory_descriptor() + assert inner.queries == [] + assert inner.closed == 1 diff --git a/tests/test_scope_extension_registry.py b/tests/test_scope_extension_registry.py index b706d3d7..c810302d 100644 --- a/tests/test_scope_extension_registry.py +++ b/tests/test_scope_extension_registry.py @@ -187,8 +187,10 @@ def test_scope_descriptor_extension_is_append_only_for_positional_compatibility( "measurement_statistics_profile_v2", "fft_status_profile_v2", "cursor_readout_profile_v2", - "average_capture_profile_v2", - ] + "average_capture_profile_v2", + "channel_display_profile_v2", + "focus_profile_v2", + ] def test_new_old_core_plugin_capability_matrix_is_fail_closed() -> None: diff --git a/tests/test_scope_extension_service.py b/tests/test_scope_extension_service.py index dcce2655..6d3c4d83 100644 --- a/tests/test_scope_extension_service.py +++ b/tests/test_scope_extension_service.py @@ -1,5 +1,7 @@ from __future__ import annotations +from types import SimpleNamespace + import numpy as np import pytest import zlib @@ -17,8 +19,20 @@ ScopeAcquisitionRunState, ScopeAxisMetadata, ScopeBaselineRestoreResult, + ScopeChannelDisplayProfileV2, + ScopeChannelDisplayRequest, + ScopeChannelDisplayRestoreResult, + ScopeChannelDisplayResult, + ScopeChannelDisplayState, ScopeContinuousAcquisitionRequest, ScopeDescriptorExtensions, + ScopeFocusChannelState, + ScopeFocusProfileV2, + ScopeFocusRequest, + ScopeFocusRestoreResult, + ScopeFocusResult, + ScopeFocusState, + ScopeFocusVerticalScale, ScopeScreenshot, ScopeScreenshotProfile, ScopeScreenshotRequest, @@ -40,6 +54,7 @@ ExperimentalScopeExtensionService, ScopeExtensionService, ) +from wavebench.services.scope_service import ScopeService from wavebench.transport.contracts import ( BinaryQueryResult, BinaryResponseFraming, @@ -164,6 +179,15 @@ def __init__( self.trace_verify_mismatch = False self.fail_single_completion = False self.fail_stop_once = False + self.display_postcondition_mismatch = False + self.display_restore_mismatch = False + self.display_enabled = False + self.display_write_calls = 0 + self.display_restore_calls = 0 + self.focus_postcondition_mismatch = False + self.focus_restore_mismatch = False + self.focus_write_calls = 0 + self.focus_restore_calls = 0 self.terminal_stop_completion = terminal_stop_completion self.terminal_stop_mode = "single" self.restore_calls = 0 @@ -215,6 +239,45 @@ def __init__( restore_max_steps=8, verify_max_steps=8, ) + self.channel_display_profile = ScopeChannelDisplayProfileV2( + analog_channels=(1, 2), + snapshot_max_steps=1, + configure_max_steps=2, + restore_max_steps=1, + verify_max_steps=1, + ) + self.focus_profile = ScopeFocusProfileV2( + analog_channels=(1, 2, 3, 4), + time_range_min_s=1e-9, + time_range_max_s=100.0, + time_range_abs_tolerance_s=1e-12, + vertical_scale_min_v_per_div=1e-3, + vertical_scale_max_v_per_div=10.0, + vertical_scale_abs_tolerance_v_per_div=1e-6, + vertical_range_abs_tolerance_v=1e-6, + time_position_abs_tolerance_s=1e-12, + position_abs_tolerance=1e-6, + offset_abs_tolerance_v=1e-6, + snapshot_max_steps=22, + configure_max_steps=64, + restore_max_steps=22, + verify_max_steps=22, + ) + self.focus_state = ScopeFocusState( + time_range_s=0.01, + time_position_s=0.001, + channels=tuple( + ScopeFocusChannelState( + channel, + enabled=channel in {1, 3}, + range_v=float(channel * 10), + scale_v_per_div=float(channel), + position=channel / 10, + offset_v=-channel / 20, + ) + for channel in self.focus_profile.analog_channels + ), + ) self.screenshot_snapshot = ScopeScreenshotStateSnapshot( captured_fields=("scope.display_menu", "scope.display_color"), menu_state_token="MENU_ON", @@ -294,6 +357,110 @@ def get_acquisition_run_state(self): self.transport.query("RUN_STATE?") return self.run_state + def get_channel_display_state_v2(self, channel): + self.transport.query(f"DISPLAY:{channel}?") + return ScopeChannelDisplayState(channel=channel, enabled=self.display_enabled) + + def configure_channel_display_v2(self, request, *, baseline): + self.display_write_calls += 1 + self.transport.write( + f"DISPLAY:{request.channel} {'ON' if request.enabled else 'OFF'}" + ) + if not self.display_postcondition_mismatch: + self.display_enabled = request.enabled + + def restore_channel_display_v2(self, baseline): + self.display_restore_calls += 1 + self.transport.write( + f"DISPLAY:{baseline.snapshot.channel} " + f"{'ON' if baseline.snapshot.enabled else 'OFF'}" + ) + self.display_enabled = ( + not baseline.snapshot.enabled + if self.display_restore_mismatch + else baseline.snapshot.enabled + ) + return ScopeChannelDisplayRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + def get_focus_state_v2(self): + self.transport.query("FOCUS:TIMEBASE?") + self.transport.query("FOCUS:TIMEPOSITION?") + for channel in self.focus_profile.analog_channels: + self.transport.query(f"FOCUS:{channel}:DISPLAY?") + self.transport.query(f"FOCUS:{channel}:RANGE?") + self.transport.query(f"FOCUS:{channel}:SCALE?") + self.transport.query(f"FOCUS:{channel}:POSITION?") + self.transport.query(f"FOCUS:{channel}:OFFSET?") + return self.focus_state + + def configure_focus_v2(self, request, *, baseline): + self.focus_write_calls += 1 + if request.time_range_s is not None: + self.transport.write(f"FOCUS:TIMEBASE {request.time_range_s}") + scale_map = {item.channel: item.scale_v_per_div for item in request.vertical_scales} + target_channels = set(request.channels) + updated = [] + for item in self.focus_state.channels: + scale = scale_map.get(item.channel, item.scale_v_per_div) + enabled = True if item.channel in target_channels else ( + False if request.hide_others else item.enabled + ) + if item.channel in scale_map: + self.transport.write(f"FOCUS:{item.channel}:SCALE {scale}") + if item.enabled is not enabled: + self.transport.write( + f"FOCUS:{item.channel}:DISPLAY {'ON' if enabled else 'OFF'}" + ) + updated.append( + ScopeFocusChannelState( + item.channel, + enabled, + item.range_v if item.channel not in scale_map else scale * 10, + scale, + item.position, + item.offset_v, + ) + ) + if not self.focus_postcondition_mismatch: + self.focus_state = ScopeFocusState( + request.time_range_s or self.focus_state.time_range_s, + self.focus_state.time_position_s, + tuple(updated), + ) + + def restore_focus_v2(self, baseline): + self.focus_restore_calls += 1 + for field in baseline.restore_order: + self.transport.write(f"FOCUS:RESTORE:{field}") + self.focus_state = ( + ScopeFocusState( + baseline.snapshot.time_range_s, + baseline.snapshot.time_position_s, + ( + ScopeFocusChannelState( + 1, + not baseline.snapshot.channels[0].enabled, + baseline.snapshot.channels[0].range_v, + baseline.snapshot.channels[0].scale_v_per_div, + baseline.snapshot.channels[0].position, + baseline.snapshot.channels[0].offset_v, + ), + *baseline.snapshot.channels[1:], + ), + ) + if self.focus_restore_mismatch + else baseline.snapshot + ) + return ScopeFocusRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + def snapshot_acquisition_control(self): self.transport.query("RUN_STATE?") self.transport.query("TRIGGER?") @@ -472,6 +639,8 @@ def _service( "scope.screenshot_v2", "scope.acquisition_run_state", "scope.acquisition_control", + "scope.channel_display_configure_v2", + "scope.focus_configure_v2", "scope.trace_metadata", "scope.fetch_trace", ] @@ -490,11 +659,13 @@ def _service( option_specs=(), permissions=("instrument.io",), factory=lambda context: driver, - wavebench_min_version="0.8.23", + wavebench_min_version="0.8.26", scope_extensions=ScopeDescriptorExtensions( screenshot_profile=driver.screenshot_profile, acquisition_control_profile=driver.acquisition_profile, trace_profile=driver.trace_profile, + channel_display_profile_v2=driver.channel_display_profile, + focus_profile_v2=driver.focus_profile, ), ) service = ExperimentalScopeExtensionService( @@ -600,6 +771,246 @@ def test_acquisition_success_keeps_postcondition_and_failure_restores_baseline() assert raised.value.scope_operation_diagnostics["cleanup"]["verification"]["status"] == "verified" +def test_channel_display_success_keeps_target_and_matching_state_is_zero_write() -> None: + service, driver, transport, backend = _service() + + changed = service.configure_channel_display_v2( + ScopeChannelDisplayRequest(channel=1, enabled=True) + ) + + assert isinstance(changed.value, ScopeChannelDisplayResult) + assert changed.value.write_performed is True + assert changed.value.before.enabled is False + assert changed.value.after.enabled is True + assert driver.display_write_calls == 1 + assert driver.display_restore_calls == 0 + assert transport.session_state.health is SessionHealth.HEALTHY + assert backend.writes == ["DISPLAY:1 ON"] + assert [phase["phase"] for phase in changed.diagnostics["phases"]] == [ + "preflight", + "main", + ] + + service, driver, transport, backend = _service() + unchanged = service.configure_channel_display_v2( + ScopeChannelDisplayRequest(channel=1, enabled=False) + ) + + assert unchanged.value.write_performed is False + assert unchanged.value.before == unchanged.value.after + assert driver.display_write_calls == 0 + assert driver.display_restore_calls == 0 + assert backend.writes == [] + assert transport.session_state.health is SessionHealth.HEALTHY + + +def test_scope_service_routes_channel_display_through_the_public_extension_service() -> None: + internal, driver, transport, _ = _service() + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace( + driver="example.scope", + access="read_write", + check_errors=False, + ), + connection=SimpleNamespace(timeout_ms=1_000), + ), + logger=SimpleNamespace(), + session=driver, + descriptor=internal.descriptor, + transport=transport, + session_state=transport.session_state, + ) + + result = service.configure_channel_display_v2( + ScopeChannelDisplayRequest(channel=2, enabled=True) + ) + + assert isinstance(result.value, ScopeChannelDisplayResult) + assert result.value.request.channel == 2 + assert result.value.after.enabled is True + assert driver.display_write_calls == 1 + + +def test_channel_display_postcondition_failure_restores_and_verifies_baseline() -> None: + service, driver, transport, _ = _service() + driver.display_postcondition_mismatch = True + + with pytest.raises(DataError, match="postcondition") as raised: + service.configure_channel_display_v2( + ScopeChannelDisplayRequest(channel=1, enabled=True) + ) + + assert driver.display_write_calls == 1 + assert driver.display_restore_calls == 1 + assert driver.display_enabled is False + assert transport.session_state.health is SessionHealth.HEALTHY + diagnostics = raised.value.scope_operation_diagnostics + assert diagnostics["cleanup_error"] is None + assert diagnostics["cleanup"]["verification"]["status"] == "verified" + + +def test_channel_display_restore_mismatch_preserves_primary_and_poisons_session() -> None: + service, driver, transport, _ = _service() + driver.display_postcondition_mismatch = True + driver.display_restore_mismatch = True + + with pytest.raises(DataError, match="postcondition") as raised: + service.configure_channel_display_v2( + ScopeChannelDisplayRequest(channel=1, enabled=True) + ) + + assert driver.display_restore_calls == 1 + assert transport.session_state.health is SessionHealth.POISONED + diagnostics = raised.value.scope_operation_diagnostics + assert diagnostics["cleanup_error"] == "ValueError" + assert diagnostics["cleanup"]["verification"]["status"] == "mismatch" + + +def test_focus_multi_channel_success_and_matching_state_are_zero_write() -> None: + service, driver, transport, backend = _service() + request = ScopeFocusRequest( + channels=(2, 4), + time_range_s=0.02, + vertical_scales=( + ScopeFocusVerticalScale(2, 0.5), + ScopeFocusVerticalScale(4, 1.5), + ), + hide_others=True, + ) + + changed = service.configure_focus_v2(request) + + assert isinstance(changed.value, ScopeFocusResult) + assert changed.value.write_performed is True + assert changed.value.after.time_range_s == 0.02 + assert {item.channel for item in changed.value.after.channels if item.enabled} == {2, 4} + assert tuple(item.position for item in changed.value.after.channels) == ( + 0.1, + 0.2, + 0.3, + 0.4, + ) + assert tuple(item.offset_v for item in changed.value.after.channels) == ( + -0.05, + -0.1, + -0.15, + -0.2, + ) + assert driver.focus_write_calls == 1 + assert driver.focus_restore_calls == 0 + assert transport.session_state.health is SessionHealth.HEALTHY + assert backend.writes == [ + "FOCUS:TIMEBASE 0.02", + "FOCUS:1:DISPLAY OFF", + "FOCUS:2:SCALE 0.5", + "FOCUS:2:DISPLAY ON", + "FOCUS:3:DISPLAY OFF", + "FOCUS:4:SCALE 1.5", + "FOCUS:4:DISPLAY ON", + ] + + backend.writes.clear() + unchanged = service.configure_focus_v2(request) + + assert unchanged.value.write_performed is False + assert unchanged.value.before == unchanged.value.after + assert driver.focus_write_calls == 1 + assert backend.writes == [] + + +def test_scope_service_routes_focus_through_public_extension_service() -> None: + internal, driver, transport, _ = _service() + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace( + driver="example.scope", + access="read_write", + check_errors=False, + ), + connection=SimpleNamespace(timeout_ms=1_000), + ), + logger=SimpleNamespace(), + session=driver, + descriptor=internal.descriptor, + transport=transport, + session_state=transport.session_state, + ) + + result = service.configure_focus_v2(ScopeFocusRequest(channels=(1, 3))) + + assert isinstance(result.value, ScopeFocusResult) + assert result.value.request.channels == (1, 3) + assert driver.focus_write_calls == 0 + + +def test_focus_postcondition_failure_restores_complete_baseline() -> None: + service, driver, transport, _ = _service() + original = driver.focus_state + driver.focus_postcondition_mismatch = True + + with pytest.raises(DataError, match="postcondition") as raised: + service.configure_focus_v2( + ScopeFocusRequest( + channels=(2,), + time_range_s=0.02, + vertical_scales=(ScopeFocusVerticalScale(2, 0.5),), + hide_others=True, + ) + ) + + assert driver.focus_write_calls == 1 + assert driver.focus_restore_calls == 1 + assert driver.focus_state == original + assert transport.session_state.health is SessionHealth.HEALTHY + diagnostics = raised.value.scope_operation_diagnostics + assert diagnostics["cleanup_error"] is None + assert diagnostics["cleanup"]["verification"]["status"] == "verified" + + +def test_focus_protected_field_drift_restores_and_restore_mismatch_poisons() -> None: + service, driver, transport, _ = _service() + original_configure = driver.configure_focus_v2 + + def configure_with_drift(request, *, baseline): + original_configure(request, baseline=baseline) + first, *remaining = driver.focus_state.channels + driver.focus_state = ScopeFocusState( + driver.focus_state.time_range_s, + driver.focus_state.time_position_s, + (ScopeFocusChannelState( + first.channel, + first.enabled, + first.range_v, + first.scale_v_per_div, + first.position + 1, + first.offset_v, + ), *remaining), + ) + + driver.configure_focus_v2 = configure_with_drift + + with pytest.raises(DataError, match="postcondition"): + service.configure_focus_v2( + ScopeFocusRequest(channels=(1,), time_range_s=0.02) + ) + assert driver.focus_restore_calls == 1 + assert transport.session_state.health is SessionHealth.HEALTHY + + service, driver, transport, _ = _service() + driver.focus_postcondition_mismatch = True + driver.focus_restore_mismatch = True + with pytest.raises(DataError, match="postcondition") as raised: + service.configure_focus_v2( + ScopeFocusRequest(channels=(2,), hide_others=True) + ) + assert driver.focus_restore_calls == 1 + assert transport.session_state.health is SessionHealth.POISONED + diagnostics = raised.value.scope_operation_diagnostics + assert diagnostics["cleanup_error"] == "ValueError" + assert diagnostics["cleanup"]["verification"]["status"] == "mismatch" + + def test_acquisition_service_accepts_profile_gated_terminal_stop_proof() -> None: service, driver, transport, _ = _service(terminal_stop_completion=True) diff --git a/tests/test_scope_focus_configure_v2.py b/tests/test_scope_focus_configure_v2.py new file mode 100644 index 00000000..c900a29a --- /dev/null +++ b/tests/test_scope_focus_configure_v2.py @@ -0,0 +1,442 @@ +from __future__ import annotations + +from contextlib import redirect_stdout +from dataclasses import replace +import io +import json +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from wavebench.cli import main +from wavebench.cli_parser import build_parser +from wavebench.errors import AccessDeniedError, ConfigError, TransportIOError +from wavebench.instruments import InstrumentDescriptor, ScopeFocusDriverV2 +from wavebench.instruments.capabilities import CAPABILITY_METHODS, validate_declared_capabilities +from wavebench.instruments.contracts import ScopeDriver +from wavebench.instruments.factory import open_instrument_driver +from wavebench.instruments.scope_extension_capabilities import ( + SCOPE_CAPABILITY_METHODS, + SCOPE_STRICT_V2_CAPABILITIES, + validate_scope_descriptor, +) +from wavebench.instruments.scope_extensions import ( + ScopeDescriptorExtensions, + ScopeFocusBaseline, + ScopeFocusChannelState, + ScopeFocusProfileV2, + ScopeFocusRequest, + ScopeFocusRestoreResult, + ScopeFocusState, + ScopeFocusVerticalScale, +) +from wavebench.logging import CommandLogger +from wavebench.services.capability_explain import explain_operation +from wavebench.services.operation_specs import require_operation_spec +from wavebench.services.scope_service import ScopeService +from wavebench.transport.contracts import ReplayPolicy + + +def _profile(*, channels: tuple[int, ...] = (1, 2)) -> ScopeFocusProfileV2: + return ScopeFocusProfileV2( + analog_channels=channels, + time_range_min_s=1e-9, + time_range_max_s=100.0, + time_range_abs_tolerance_s=1e-12, + vertical_scale_min_v_per_div=1e-3, + vertical_scale_max_v_per_div=10.0, + vertical_scale_abs_tolerance_v_per_div=1e-6, + vertical_range_abs_tolerance_v=1e-6, + time_position_abs_tolerance_s=1e-12, + position_abs_tolerance=1e-6, + offset_abs_tolerance_v=1e-6, + snapshot_max_steps=9, + configure_max_steps=32, + restore_max_steps=9, + verify_max_steps=9, + ) + + +def _state( + *, + time_range_s: float = 0.01, + enabled: tuple[bool, ...] = (True, True), + ranges: tuple[float, ...] = (10.0, 20.0), + scales: tuple[float, ...] = (1.0, 2.0), + positions: tuple[float, ...] = (0.25, -0.5), + offsets: tuple[float, ...] = (0.1, -0.2), +) -> ScopeFocusState: + return ScopeFocusState( + time_range_s=time_range_s, + time_position_s=0.001, + channels=tuple( + ScopeFocusChannelState(channel, display, range_v, scale, position, offset) + for channel, display, range_v, scale, position, offset in zip( + range(1, len(enabled) + 1), + enabled, + ranges, + scales, + positions, + offsets, + strict=True, + ) + ), + ) + + +def _descriptor( + *, + minimum: str = "0.8.26", + capabilities: tuple[str, ...] = ("scope.idn", "scope.focus_configure_v2"), + extensions: bool = True, +) -> InstrumentDescriptor: + return InstrumentDescriptor( + driver_id="example.scope-focus-v2", + kind="scope", + display_name="Example Scope", + manufacturer="Example", + models=("EX1",), + aliases=(), + capabilities=capabilities, + idn_patterns=("EXAMPLE,EX1",), + backends=("pyvisa",), + option_specs=(), + permissions=("instrument.io",), + factory=lambda _context: object(), + wavebench_min_version=minimum, + scope_extensions=(ScopeDescriptorExtensions(focus_profile_v2=_profile()) if extensions else None), + ) + + +def test_focus_models_freeze_atomic_multi_channel_semantics() -> None: + request = ScopeFocusRequest( + channels=(1, 2), + time_range_s=0.02, + vertical_scales=( + ScopeFocusVerticalScale(1, 0.5), + ScopeFocusVerticalScale(2, 1.0), + ), + hide_others=True, + ) + before = _state() + after = _state(time_range_s=0.02, scales=(0.5, 1.0)) + profile = _profile() + baseline = ScopeFocusBaseline( + context_id="context", + session_epoch="epoch", + baseline_nonce="nonce", + snapshot=before, + restore_order=( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ), + ) + restored = ScopeFocusRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + profile.validate_request(request) + profile.validate_state(before) + assert profile.restore_order_for(request) == baseline.restore_order + assert profile.request_satisfied(after, request) + assert profile.transition_matches(before, after, request) + assert profile.states_equivalent(before, before) + restored.validate_for(baseline) + + position_drift = replace( + after, + channels=(replace(after.channels[0], position=9.0), after.channels[1]), + ) + assert not profile.transition_matches(before, position_drift, request) + + with pytest.raises(ValueError, match="ascending"): + ScopeFocusRequest(channels=(2, 1)) + with pytest.raises(ValueError, match="selected channels"): + ScopeFocusRequest( + channels=(1,), + vertical_scales=(ScopeFocusVerticalScale(2, 1.0),), + ) + with pytest.raises(ValueError, match="outside"): + profile.validate_request(ScopeFocusRequest(channels=(3,))) + with pytest.raises(ValueError, match="fixed safe order"): + ScopeFocusBaseline( + "context", + "epoch", + "nonce", + before, + ("scope.channel_display", "scope.timebase"), + ) + + +def test_focus_is_an_additive_profile_gated_capability() -> None: + class Driver: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + def get_focus_state_v2(self) -> ScopeFocusState: + return _state() + + def configure_focus_v2(self, request, *, baseline) -> None: + pass + + def restore_focus_v2(self, baseline): + return ScopeFocusRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + expected_methods = ( + "get_focus_state_v2", + "configure_focus_v2", + "restore_focus_v2", + ) + assert isinstance(Driver(), ScopeFocusDriverV2) + assert "configure_focus_v2" not in ScopeDriver.__dict__ + assert SCOPE_CAPABILITY_METHODS["scope.focus_configure_v2"] == expected_methods + assert CAPABILITY_METHODS["scope.focus_configure_v2"] == expected_methods + assert "scope.focus_configure_v2" in SCOPE_STRICT_V2_CAPABILITIES + validate_declared_capabilities(_descriptor(), Driver()) + + +def test_focus_descriptor_requires_version_profile_dependency_and_methods() -> None: + with pytest.raises(ConfigError, match="scope focus V2 capability requires.*0.8.26"): + validate_scope_descriptor(_descriptor(minimum="0.8.25")) + with pytest.raises(ConfigError, match="scope_extensions.focus_profile_v2"): + validate_scope_descriptor(_descriptor(extensions=False)) + with pytest.raises(ConfigError, match="scope.idn"): + validate_scope_descriptor(_descriptor(capabilities=("scope.focus_configure_v2",))) + + class MissingMethods: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + with pytest.raises(TypeError, match="get_focus_state_v2"): + validate_declared_capabilities(_descriptor(), MissingMethods()) + + +def test_focus_operation_spec_and_explain_are_fail_closed() -> None: + spec = require_operation_spec("scope.focus_configure_v2") + fields = ( + "scope.timebase", + "scope.channel_vertical", + "scope.channel_display", + ) + + assert spec.instrument_kind == "scope" + assert spec.required_capabilities == ("scope.focus_configure_v2",) + assert spec.effect == "write" + assert spec.lease_mode == "exclusive" + assert spec.changed_fields == (*fields, "scope.error_queue") + assert spec.restore_coverage == "failure-cleanup-only" + assert spec.verification_fields == fields + assert spec.postcondition_fields == fields + assert spec.cleanup_verification_fields == fields + + descriptor = SimpleNamespace( + driver_id="example.scope-focus-v2", + kind="scope", + capabilities=("scope.focus_configure_v2",), + ) + assert explain_operation("scope.focus_configure_v2", descriptor=descriptor).status == "supported" + assert ( + explain_operation( + "scope.focus_configure_v2", + descriptor=descriptor, + access="read_only", + ).status + == "access_denied" + ) + + +def test_scope_service_rejects_access_and_unknown_focus_channel_before_session() -> None: + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace(driver="example.scope-focus-v2", access="read_write"), + ), + logger=SimpleNamespace(), + descriptor=_descriptor(), + ) + + with patch.object(service, "_open_scope") as open_scope: + with pytest.raises(ConfigError, match="outside"): + service.configure_focus_v2(ScopeFocusRequest(channels=(3,))) + open_scope.assert_not_called() + + service.config.scope.access = "read_only" + with patch.object(service, "_open_scope") as open_scope: + with pytest.raises(AccessDeniedError, match="read_only"): + service.configure_focus_v2(ScopeFocusRequest(channels=(1,))) + open_scope.assert_not_called() + + +def test_scope_focus_cli_builds_typed_multi_channel_request() -> None: + argv = [ + "scope", + "focus", + "--channel", + "2", + "--channel", + "1", + "--time-range", + "0.02", + "--vertical-scale", + "1=0.5", + "--vertical-scale", + "2=1.0", + "--hide-others", + ] + args = build_parser().parse_args(argv) + assert args.command == "focus" + assert args.channels == [2, 1] + + calls: list[ScopeFocusRequest] = [] + scope_payload = { + "schema": "wavebench.scope.result.v1", + "result": {"write_performed": True}, + "diagnostics": {"schema": "wavebench.scope.operation.v1"}, + "observed_state": None, + } + + class Result: + def as_dict(self) -> dict[str, object]: + return scope_payload + + class Service: + def configure_focus_v2(self, request, *, error_check=None): + assert error_check is None + calls.append(request) + return Result() + + stdout = io.StringIO() + with patch("wavebench.cli._load_service", return_value=Service()), redirect_stdout(stdout): + code = main(argv) + + assert code == 0 + assert calls == [ + ScopeFocusRequest( + channels=(1, 2), + time_range_s=0.02, + vertical_scales=( + ScopeFocusVerticalScale(1, 0.5), + ScopeFocusVerticalScale(2, 1.0), + ), + hide_others=True, + ) + ] + assert json.loads(stdout.getvalue()) == scope_payload + + +class _FactoryTransport: + resource = "TCPIP::example::INSTR" + + def __init__(self) -> None: + self.queries: list[str] = [] + self.closed = 0 + + def record_event(self, _direction: str, _text: str) -> None: + pass + + def query(self, command: str, *, replay: ReplayPolicy = ReplayPolicy.NO_REPLAY) -> str: + assert replay is ReplayPolicy.NO_REPLAY + self.queries.append(command) + return "ok" + + def close(self) -> None: + self.closed += 1 + + +def _open_factory_descriptor() -> object: + return open_instrument_driver( + driver_reference="example.scope-focus-v2", + expected_kind="scope", + resource="TCPIP::example::INSTR", + configured_backend="pyvisa", + timeout_ms=1_000, + opc_timeout_ms=2_000, + read_retry_attempts=1, + read_retry_delay_ms=0, + logger=CommandLogger(), + ) + + +def test_focus_factory_latch_and_missing_method_are_zero_io( + monkeypatch: pytest.MonkeyPatch, +) -> None: + inner = _FactoryTransport() + errors: list[TransportIOError] = [] + + class Driver: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + def get_focus_state_v2(self) -> ScopeFocusState: + return _state() + + def configure_focus_v2(self, request, *, baseline) -> None: + pass + + def restore_focus_v2(self, baseline): + return ScopeFocusRestoreResult( + "completed", + baseline.restore_order, + baseline.restore_order, + ) + + def factory(context): + transport = context.open_transport() + with pytest.raises(TransportIOError) as raised: + transport.query("*IDN?") + errors.append(raised.value) + return Driver() + + descriptor = replace(_descriptor(), factory=factory) + monkeypatch.setattr( + "wavebench.instruments.factory.resolve_instrument_descriptor", + lambda _reference, expected_kind: descriptor, + ) + monkeypatch.setattr("wavebench.instruments.factory._open_transport", lambda **_kwargs: inner) + + opened = _open_factory_descriptor() + + assert [error.reason_code for error in errors] == ["factory_construction_pending"] + assert errors[0].attempts == 0 + assert inner.queries == [] + assert opened.transport is not None + + inner = _FactoryTransport() + + class MissingMethods: + def idn(self) -> str: + return "EXAMPLE,EX1" + + def close(self) -> None: + pass + + missing = replace( + _descriptor(), + factory=lambda context: (context.open_transport(), MissingMethods())[1], + ) + monkeypatch.setattr( + "wavebench.instruments.factory.resolve_instrument_descriptor", + lambda _reference, expected_kind: missing, + ) + monkeypatch.setattr("wavebench.instruments.factory._open_transport", lambda **_kwargs: inner) + + with pytest.raises(ConfigError, match="get_focus_state_v2"): + _open_factory_descriptor() + assert inner.queries == [] + assert inner.closed == 1 diff --git a/tests/test_scope_portability_m0.py b/tests/test_scope_portability_m0.py index 42002c68..a6664b58 100644 --- a/tests/test_scope_portability_m0.py +++ b/tests/test_scope_portability_m0.py @@ -54,6 +54,7 @@ "scope.measurement_statistics_v2", "scope.fft_status_v2", "scope.cursor_readout_v2", + "scope.channel_display_configure_v2", } )