From 952cd47fbc8503e9b3ee7ee5807770a2ae690634 Mon Sep 17 00:00:00 2001 From: Nept Epslion <11226778+Nept-Epslion@user.noreply.gitee.com> Date: Tue, 28 Jul 2026 16:30:29 +0800 Subject: [PATCH 1/2] Add explicit scope display and focus controls --- README.md | 8 ++ src/wavebench/cli.py | 21 +++++ src/wavebench/cli_output.py | 14 +++ src/wavebench/cli_parser.py | 32 +++++++ src/wavebench/drivers/ds1104.py | 38 +++++++- src/wavebench/drivers/rtm2032.py | 36 ++++++++ src/wavebench/instruments/builtin.py | 4 + src/wavebench/instruments/capabilities.py | 2 + src/wavebench/instruments/contracts.py | 18 ++++ src/wavebench/plugins/builtin.py | 4 + src/wavebench/services/scope_service.py | 59 +++++++++++++ tests/test_cli.py | 97 ++++++++++++++++++++ tests/test_ds1104.py | 48 ++++++++++ tests/test_instrument_models.py | 6 +- tests/test_scope_snapshot.py | 102 ++++++++++++++++++++++ tests/test_waveform.py | 24 +++++ 16 files changed, 509 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 6e5d01c..f0e7a96 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ WaveBench 主包长期预装 RTM2000/RTM2032、DS1104Z/DS1000Z、DG4000/DG4202 - LAN VISA 连接 - `scope idn`、`scope errors`；声明相应 capability 的驱动还支持只读 `scope status`、`scope acquisition-status`、`scope history-timestamps` 与 `scope measurement-statistics` - 显式 `scope auto` / `scope autoscale` +- 显式 `scope display --channel N on|off` 与 `scope focus --channel N`，可审计地调整通道显示、时基窗口和垂直档位；不会控制信号源或电源 - `scope fetch` 与 `scope capture`；默认先只读确认输入为高阻，50 Ω 需显式 `--allow-50ohm` - 声明 `scope.capture_average` 的驱动可执行受控平均采集；公共结果要求逐项恢复并返回恢复前后配置证据 - 声明 `scope.digital_status` 的驱动可读取既有 MSO 数字通道状态；该能力不读取数字波形，也不隐式配置阈值、显示或传输格式 @@ -326,6 +327,13 @@ python -m wavebench scope capture --config wavebench.toml --channel 1 --label sm python -m wavebench scope capture --config wavebench.toml --channel 1 --label smoke_with_screen --points def --no-csv --screenshot ``` +显式调整示波器显示： + +```powershell +python -m wavebench scope display --config wavebench.toml --channel 2 off +python -m wavebench scope focus --config wavebench.toml --channel 1 --time-range 0.01 --vertical-scale 0.2 +``` + DS1104Z 配置示例： ```toml diff --git a/src/wavebench/cli.py b/src/wavebench/cli.py index 6468f0f..7b4f0ef 100644 --- a/src/wavebench/cli.py +++ b/src/wavebench/cli.py @@ -55,6 +55,7 @@ _print_scope_digital_status, _print_scope_digital_waveform, _print_scope_measurement_statistics, + _print_scope_mutation_manifest, _print_scope_cursor_readout, _print_scope_derived_waveform_metadata, _print_scope_fft_status, @@ -725,6 +726,26 @@ def main(argv: list[str] | None = None) -> int: service.autoscale() print("AUToscale completed") return 0 + if args.command == "display": + result = service.set_channel_display( + channel=args.channel, + enabled=args.state.lower() == "on", + ) + _print_scope_mutation_manifest(result) + return 0 + if args.command == "focus": + if args.time_range is not None and args.time_range <= 0: + raise ConfigError("--time-range must be > 0") + if args.vertical_scale is not None and args.vertical_scale <= 0: + raise ConfigError("--vertical-scale must be > 0") + result = service.focus_channel( + channel=args.channel, + time_range_s=args.time_range, + vertical_scale_v_per_div=args.vertical_scale, + hide_other_channels=args.hide_other_channels, + ) + _print_scope_mutation_manifest(result) + return 0 if args.command == "fetch": channel = args.channel or service.config.scope.default_channel service.require_high_impedance(channel, allow_50ohm=args.allow_50ohm) diff --git a/src/wavebench/cli_output.py b/src/wavebench/cli_output.py index 403a6c2..df61528 100644 --- a/src/wavebench/cli_output.py +++ b/src/wavebench/cli_output.py @@ -455,6 +455,20 @@ def number(value: float | None) -> str: print(f"cursor.x_ratio={number(readout.x_ratio)}") print(f"cursor.y_ratio={number(readout.y_ratio)}") + +def _print_scope_mutation_manifest(manifest: dict[str, Any]) -> None: + print(f"operation={manifest['operation']}") + print(f"mutates_instrument={str(bool(manifest['mutates_instrument'])).lower()}") + print(f"raw_scpi={str(bool(manifest['raw_scpi'])).lower()}") + print(f"channel={manifest['channel']}") + for key in ("display", "time_range_s", "vertical_scale_v_per_div", "hide_other_channels"): + if key in manifest and manifest[key] is not None: + value = manifest[key] + if isinstance(value, bool): + value = str(value).lower() + print(f"{key}={value}") + print("affected_settings=" + ",".join(str(item) for item in manifest["affected_settings"])) + def _print_dmm_function_status(function: str) -> None: print(f"功能 / Function: {function}") diff --git a/src/wavebench/cli_parser.py b/src/wavebench/cli_parser.py index 76faac4..9e52801 100644 --- a/src/wavebench/cli_parser.py +++ b/src/wavebench/cli_parser.py @@ -747,6 +747,38 @@ def build_parser() -> argparse.ArgumentParser: autoscale = scope_sub.add_parser("autoscale", help="Alias of scope auto") add_runtime_options(autoscale) + display = scope_sub.add_parser( + "display", + help="Explicitly turn one analog channel display on or off", + ) + display.add_argument("--channel", type=int, required=True) + display.add_argument("state", choices=["on", "off", "ON", "OFF"]) + add_runtime_options(display) + + focus = scope_sub.add_parser( + "focus", + help="Explicitly focus the scope display/acquisition window on one channel", + ) + focus.add_argument("--channel", type=int, required=True) + focus.add_argument( + "--time-range", + type=float, + default=None, + help="Set total horizontal acquisition/display window in seconds", + ) + focus.add_argument( + "--vertical-scale", + type=float, + default=None, + help="Set selected channel vertical scale in V/div", + ) + focus.add_argument( + "--hide-other-channels", + action="store_true", + help="Turn CH1-CH4 displays off except the selected channel", + ) + add_runtime_options(focus) + fetch = scope_sub.add_parser("fetch", help="Fetch waveform data without creating full package") fetch.add_argument("--channel", type=int, default=None) fetch.add_argument("--points", default=None, help="Override waveform points: def, max, or dmax") diff --git a/src/wavebench/drivers/ds1104.py b/src/wavebench/drivers/ds1104.py index dfdf9a3..89aaac3 100644 --- a/src/wavebench/drivers/ds1104.py +++ b/src/wavebench/drivers/ds1104.py @@ -118,7 +118,43 @@ def assert_no_errors(self) -> None: def autoscale(self, wait_opc: bool = True, check_errors: bool = True) -> None: self.transport.write(":AUToscale") if wait_opc: - self.transport.query_opc() + self._record_telemetry( + "stage=autoscale wait_opc=skipped reason=ds1000z_socket_opc_response_can_desync" + ) + if check_errors: + self.assert_no_errors() + + def set_channel_display( + self, + channel: int, + enabled: bool, + *, + check_errors: bool = True, + ) -> None: + self._validate_channel(channel) + self.transport.write(f":CHANnel{channel}:DISPlay {'ON' if enabled else 'OFF'}") + if check_errors: + self.assert_no_errors() + + def focus_channel( + self, + channel: int, + *, + time_range_s: float | None = None, + vertical_scale_v_per_div: float | None = None, + hide_other_channels: bool = False, + check_errors: bool = True, + ) -> None: + self._validate_channel(channel) + if hide_other_channels: + for other in range(1, 5): + if other != channel: + self.transport.write(f":CHANnel{other}:DISPlay OFF") + self.transport.write(f":CHANnel{channel}:DISPlay ON") + if time_range_s is not None: + self.set_time_range(time_range_s) + if vertical_scale_v_per_div is not None: + self.set_vertical_scale(channel, vertical_scale_v_per_div) if check_errors: self.assert_no_errors() diff --git a/src/wavebench/drivers/rtm2032.py b/src/wavebench/drivers/rtm2032.py index 94db489..a16e864 100644 --- a/src/wavebench/drivers/rtm2032.py +++ b/src/wavebench/drivers/rtm2032.py @@ -62,6 +62,42 @@ def autoscale(self, wait_opc: bool = True, check_errors: bool = True) -> None: if check_errors: self.assert_no_errors() + def set_channel_display( + self, + channel: int, + enabled: bool, + *, + check_errors: bool = True, + ) -> None: + if channel < 1: + raise DataError("channel must be >= 1") + self.transport.write(f"CHAN{channel}:STAT {'ON' if enabled else 'OFF'}") + if check_errors: + self.assert_no_errors() + + def focus_channel( + self, + channel: int, + *, + time_range_s: float | None = None, + vertical_scale_v_per_div: float | None = None, + hide_other_channels: bool = False, + check_errors: bool = True, + ) -> None: + if channel < 1: + raise DataError("channel must be >= 1") + if hide_other_channels: + for other in range(1, 3): + if other != channel: + self.transport.write(f"CHAN{other}:STAT OFF") + self.transport.write(f"CHAN{channel}:STAT ON") + if time_range_s is not None: + self.set_time_range(time_range_s) + if vertical_scale_v_per_div is not None: + self.set_vertical_scale(channel, vertical_scale_v_per_div) + if check_errors: + self.assert_no_errors() + def set_time_range(self, time_range_s: float) -> None: if time_range_s <= 0: raise DataError("time range must be > 0") diff --git a/src/wavebench/instruments/builtin.py b/src/wavebench/instruments/builtin.py index bdf936e..5900363 100644 --- a/src/wavebench/instruments/builtin.py +++ b/src/wavebench/instruments/builtin.py @@ -61,6 +61,8 @@ def _open_dm3000(context): "scope.idn", "scope.errors", "scope.autoscale", + "scope.channel_display", + "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", @@ -88,6 +90,8 @@ def _open_dm3000(context): "scope.idn", "scope.errors", "scope.autoscale", + "scope.channel_display", + "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", diff --git a/src/wavebench/instruments/capabilities.py b/src/wavebench/instruments/capabilities.py index fddb9f9..afc04c5 100644 --- a/src/wavebench/instruments/capabilities.py +++ b/src/wavebench/instruments/capabilities.py @@ -11,6 +11,8 @@ "scope.idn": ("idn",), "scope.errors": ("errors",), "scope.autoscale": ("autoscale",), + "scope.channel_display": ("set_channel_display",), + "scope.focus_channel": ("focus_channel",), "scope.fetch_waveform": ("fetch_waveform",), "scope.capture_waveform": ("capture_waveform",), "scope.capture_waveforms": ("capture_waveforms",), diff --git a/src/wavebench/instruments/contracts.py b/src/wavebench/instruments/contracts.py index 66c31d0..f890b38 100644 --- a/src/wavebench/instruments/contracts.py +++ b/src/wavebench/instruments/contracts.py @@ -58,6 +58,24 @@ def channel_coupling(self, channel: int) -> str: ... def autoscale(self, wait_opc: bool = True, check_errors: bool = True) -> None: ... + def set_channel_display( + self, + channel: int, + enabled: bool, + *, + check_errors: bool = True, + ) -> None: ... + + def focus_channel( + self, + channel: int, + *, + time_range_s: float | None = None, + vertical_scale_v_per_div: float | None = None, + hide_other_channels: bool = False, + check_errors: bool = True, + ) -> None: ... + def fetch_waveform( self, channel: int, diff --git a/src/wavebench/plugins/builtin.py b/src/wavebench/plugins/builtin.py index 25f6f8f..ded3c97 100644 --- a/src/wavebench/plugins/builtin.py +++ b/src/wavebench/plugins/builtin.py @@ -13,6 +13,8 @@ "scope.idn", "scope.errors", "scope.autoscale", + "scope.channel_display", + "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", @@ -33,6 +35,8 @@ "scope.idn", "scope.errors", "scope.autoscale", + "scope.channel_display", + "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", diff --git a/src/wavebench/services/scope_service.py b/src/wavebench/services/scope_service.py index 9a2f822..76b852b 100644 --- a/src/wavebench/services/scope_service.py +++ b/src/wavebench/services/scope_service.py @@ -328,6 +328,65 @@ def autoscale(self) -> None: check_errors=self.config.autoscale.check_errors, ) + def set_channel_display(self, channel: int, enabled: bool) -> dict[str, Any]: + required = ["scope.channel_display"] + if self.config.scope.check_errors: + required.append("scope.errors") + self._require("scope.channel_display", *required) + with self._scope_session() as scope: + scope.set_channel_display( + channel, + enabled, + check_errors=self.config.scope.check_errors, + ) + return { + "operation": "scope.channel_display", + "mutates_instrument": True, + "raw_scpi": False, + "channel": channel, + "display": "on" if enabled else "off", + "affected_settings": [f"CH{channel}.display"], + } + + def focus_channel( + self, + *, + channel: int, + time_range_s: float | None = None, + vertical_scale_v_per_div: float | None = None, + hide_other_channels: bool = False, + ) -> dict[str, Any]: + required = ["scope.focus_channel"] + if self.config.scope.check_errors: + required.append("scope.errors") + self._require("scope.focus_channel", *required) + with self._scope_session() as scope: + scope.focus_channel( + channel, + time_range_s=time_range_s, + vertical_scale_v_per_div=vertical_scale_v_per_div, + hide_other_channels=hide_other_channels, + check_errors=self.config.scope.check_errors, + ) + affected = [f"CH{channel}.display"] + if time_range_s is not None: + affected.append("timebase.range") + if vertical_scale_v_per_div is not None: + affected.append(f"CH{channel}.vertical_scale") + affected.append(f"CH{channel}.offset") + if hide_other_channels: + affected.extend(f"CH{other}.display" for other in range(1, 5) if other != channel) + return { + "operation": "scope.focus_channel", + "mutates_instrument": True, + "raw_scpi": False, + "channel": channel, + "time_range_s": time_range_s, + "vertical_scale_v_per_div": vertical_scale_v_per_div, + "hide_other_channels": hide_other_channels, + "affected_settings": affected, + } + def fetch_waveform(self, channel: int) -> WaveformData: if self.config.waveform.format.lower() != "real": raise ConfigError("MVP-1 only supports waveform.format = 'real'") diff --git a/tests/test_cli.py b/tests/test_cli.py index 9729133..c5956aa 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -74,6 +74,103 @@ def test_capture_accepts_repeated_channels(self): self.assertEqual(args.command, "capture") self.assertEqual(args.channel, [1, 2]) + def test_scope_display_accepts_channel_and_state(self): + args = build_parser().parse_args(["scope", "display", "--channel", "2", "off"]) + self.assertEqual(args.domain, "scope") + self.assertEqual(args.command, "display") + self.assertEqual(args.channel, 2) + self.assertEqual(args.state, "off") + + def test_scope_focus_accepts_display_adjustment_options(self): + args = build_parser().parse_args([ + "scope", + "focus", + "--channel", + "1", + "--time-range", + "0.01", + "--vertical-scale", + "0.2", + "--hide-other-channels", + ]) + self.assertEqual(args.domain, "scope") + self.assertEqual(args.command, "focus") + self.assertEqual(args.channel, 1) + self.assertEqual(args.time_range, 0.01) + self.assertEqual(args.vertical_scale, 0.2) + self.assertTrue(args.hide_other_channels) + + def test_scope_display_prints_mutation_manifest(self): + service = Mock() + service.set_channel_display.return_value = { + "operation": "scope.channel_display", + "mutates_instrument": True, + "raw_scpi": False, + "channel": 2, + "display": "off", + "affected_settings": ["CH2.display"], + } + stdout = io.StringIO() + + with patch("wavebench.cli._load_service", return_value=service), redirect_stdout(stdout): + code = main(["scope", "display", "--channel", "2", "off"]) + + self.assertEqual(code, 0) + service.set_channel_display.assert_called_once_with(channel=2, enabled=False) + output = stdout.getvalue() + self.assertIn("operation=scope.channel_display\n", output) + self.assertIn("mutates_instrument=true\n", output) + self.assertIn("raw_scpi=false\n", output) + self.assertIn("affected_settings=CH2.display\n", output) + + def test_scope_focus_prints_mutation_manifest(self): + service = Mock() + service.focus_channel.return_value = { + "operation": "scope.focus_channel", + "mutates_instrument": True, + "raw_scpi": False, + "channel": 1, + "time_range_s": 0.01, + "vertical_scale_v_per_div": 0.2, + "hide_other_channels": True, + "affected_settings": [ + "CH1.display", + "timebase.range", + "CH1.vertical_scale", + "CH1.offset", + "CH2.display", + "CH3.display", + "CH4.display", + ], + } + stdout = io.StringIO() + + with patch("wavebench.cli._load_service", return_value=service), redirect_stdout(stdout): + code = main([ + "scope", + "focus", + "--channel", + "1", + "--time-range", + "0.01", + "--vertical-scale", + "0.2", + "--hide-other-channels", + ]) + + self.assertEqual(code, 0) + service.focus_channel.assert_called_once_with( + channel=1, + time_range_s=0.01, + vertical_scale_v_per_div=0.2, + hide_other_channels=True, + ) + output = stdout.getvalue() + self.assertIn("operation=scope.focus_channel\n", output) + self.assertIn("time_range_s=0.01\n", output) + self.assertIn("vertical_scale_v_per_div=0.2\n", output) + self.assertIn("hide_other_channels=true\n", output) + def test_power_status_accepts_channel(self): args = build_parser().parse_args(["power", "status", "--channel", "1"]) self.assertEqual(args.domain, "power") diff --git a/tests/test_ds1104.py b/tests/test_ds1104.py index d9b46fb..57992ff 100644 --- a/tests/test_ds1104.py +++ b/tests/test_ds1104.py @@ -232,6 +232,54 @@ def test_capture_translates_total_time_range_to_12_divisions(): assert "*OPC?" in transport.queries +def test_autoscale_skips_opc_query_to_avoid_ds1000z_socket_desync(): + transport = FakeTransport(responses={":SYSTem:ERRor?": '0,"No error"'}) + scope = DS1104Scope(transport=transport) + + scope.autoscale(wait_opc=True, check_errors=True) + + assert transport.writes == [":AUToscale"] + assert transport.queries == [":SYSTem:ERRor?"] + assert any( + direction == "telemetry" and "wait_opc=skipped" in text + for direction, text in transport.events + ) + + +def test_set_channel_display_writes_rigol_display_command(): + transport = FakeTransport() + scope = DS1104Scope(transport=transport) + + scope.set_channel_display(2, False, check_errors=False) + + assert transport.writes == [":CHANnel2:DISPlay OFF"] + + +def test_focus_channel_can_hide_others_and_adjust_selected_channel(): + transport = FakeTransport() + scope = DS1104Scope(transport=transport) + + scope.focus_channel( + 2, + time_range_s=0.012, + vertical_scale_v_per_div=0.2, + hide_other_channels=True, + check_errors=False, + ) + + assert transport.writes == [ + ":CHANnel1:DISPlay OFF", + ":CHANnel3:DISPlay OFF", + ":CHANnel4:DISPlay OFF", + ":CHANnel2:DISPlay ON", + ":TIMebase:MODE MAIN", + ":TIMebase:MAIN:SCALe 0.001", + ":CHANnel2:DISPlay ON", + ":CHANnel2:SCALe 0.2", + ":CHANnel2:OFFSet 0", + ] + + def test_multichannel_capture_uses_one_single_and_one_opc(): current_channel = {"value": 1} diff --git a/tests/test_instrument_models.py b/tests/test_instrument_models.py index 3500783..88430a0 100644 --- a/tests/test_instrument_models.py +++ b/tests/test_instrument_models.py @@ -516,9 +516,9 @@ def __getattr__(self, name): class _Scope(_DynamicDriver): - idn = close = errors = channel_coupling = autoscale = fetch_waveform = capture_waveform = ( - screenshot_png - ) = lambda *args, **kwargs: None + idn = close = errors = channel_coupling = autoscale = set_channel_display = ( + focus_channel + ) = fetch_waveform = capture_waveform = screenshot_png = lambda *args, **kwargs: None class _Source(_DynamicDriver): diff --git a/tests/test_scope_snapshot.py b/tests/test_scope_snapshot.py index cab99ca..9b05ba8 100644 --- a/tests/test_scope_snapshot.py +++ b/tests/test_scope_snapshot.py @@ -65,6 +65,108 @@ def test_scope_service_status_rejects_missing_capability_before_opening(): open_scope.assert_not_called() +def test_scope_service_channel_display_returns_mutation_manifest(): + calls = [] + + def set_channel_display(channel, enabled, *, check_errors=True): + calls.append((channel, enabled, check_errors)) + + driver = SimpleNamespace(set_channel_display=set_channel_display) + descriptor = SimpleNamespace( + driver_id="example.scope", + capabilities=("scope.channel_display",), + ) + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace(driver="example.scope", check_errors=False) + ), + logger=SimpleNamespace(), + session=driver, + descriptor=descriptor, + ) + + manifest = service.set_channel_display(channel=3, enabled=False) + + assert calls == [(3, False, False)] + assert manifest == { + "operation": "scope.channel_display", + "mutates_instrument": True, + "raw_scpi": False, + "channel": 3, + "display": "off", + "affected_settings": ["CH3.display"], + } + + +def test_scope_service_focus_returns_auditable_mutation_manifest(): + calls = [] + + def focus_channel( + channel, + *, + time_range_s=None, + vertical_scale_v_per_div=None, + hide_other_channels=False, + check_errors=True, + ): + calls.append( + { + "channel": channel, + "time_range_s": time_range_s, + "vertical_scale_v_per_div": vertical_scale_v_per_div, + "hide_other_channels": hide_other_channels, + "check_errors": check_errors, + } + ) + + driver = SimpleNamespace(focus_channel=focus_channel) + descriptor = SimpleNamespace( + driver_id="example.scope", + capabilities=("scope.focus_channel", "scope.errors"), + ) + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace(driver="example.scope", check_errors=True) + ), + logger=SimpleNamespace(), + session=driver, + descriptor=descriptor, + ) + + manifest = service.focus_channel( + channel=2, + time_range_s=0.01, + vertical_scale_v_per_div=0.2, + hide_other_channels=True, + ) + + assert calls == [ + { + "channel": 2, + "time_range_s": 0.01, + "vertical_scale_v_per_div": 0.2, + "hide_other_channels": True, + "check_errors": True, + } + ] + assert manifest["operation"] == "scope.focus_channel" + assert manifest["mutates_instrument"] is True + assert manifest["raw_scpi"] is False + assert manifest["channel"] == 2 + assert manifest["time_range_s"] == 0.01 + assert manifest["vertical_scale_v_per_div"] == 0.2 + assert manifest["hide_other_channels"] is True + assert manifest["affected_settings"] == [ + "CH2.display", + "timebase.range", + "CH2.vertical_scale", + "CH2.offset", + "CH1.display", + "CH3.display", + "CH4.display", + ] + + def test_scope_status_cli_uses_default_channel_and_prints_stable_fields(): service = SimpleNamespace( config=SimpleNamespace(scope=SimpleNamespace(default_channel=2)), diff --git a/tests/test_waveform.py b/tests/test_waveform.py index 7fac4fd..533a941 100644 --- a/tests/test_waveform.py +++ b/tests/test_waveform.py @@ -88,6 +88,30 @@ def test_scope_screenshot_png_rejects_non_png(self): with self.assertRaises(DataError): scope.screenshot_png() + def test_scope_focus_hides_only_rtm2032_analog_channels(self): + transport = FakeTransport({}) + scope = RTM2032Scope(transport=transport) + + scope.focus_channel( + 2, + time_range_s=0.012, + vertical_scale_v_per_div=0.2, + hide_other_channels=True, + check_errors=False, + ) + + self.assertEqual( + transport.writes, + [ + "CHAN1:STAT OFF", + "CHAN2:STAT ON", + "TIMebase:RANGe 0.012", + "CHAN2:STAT ON", + "CHAN2:SCAL 0.2", + "CHAN2:POS 0", + ], + ) + if __name__ == "__main__": unittest.main() From de78947b48d7257906612b1cf60d27e30a6bebb9 Mon Sep 17 00:00:00 2001 From: Nept Epslion <11226778+Nept-Epslion@user.noreply.gitee.com> Date: Wed, 29 Jul 2026 23:20:22 +0800 Subject: [PATCH 2/2] Narrow scope display controls to shared surface --- src/wavebench/drivers/ds1104.py | 38 +--------- src/wavebench/drivers/rtm2032.py | 36 --------- src/wavebench/instruments/builtin.py | 4 - src/wavebench/instruments/contracts.py | 4 +- src/wavebench/plugins/builtin.py | 4 - src/wavebench/services/scope_service.py | 48 ++++++++++-- tests/test_ds1104.py | 48 ------------ tests/test_scope_snapshot.py | 99 ++++++++++++++++++++++++- tests/test_waveform.py | 24 ------ 9 files changed, 141 insertions(+), 164 deletions(-) diff --git a/src/wavebench/drivers/ds1104.py b/src/wavebench/drivers/ds1104.py index 89aaac3..dfdf9a3 100644 --- a/src/wavebench/drivers/ds1104.py +++ b/src/wavebench/drivers/ds1104.py @@ -118,43 +118,7 @@ def assert_no_errors(self) -> None: def autoscale(self, wait_opc: bool = True, check_errors: bool = True) -> None: self.transport.write(":AUToscale") if wait_opc: - self._record_telemetry( - "stage=autoscale wait_opc=skipped reason=ds1000z_socket_opc_response_can_desync" - ) - if check_errors: - self.assert_no_errors() - - def set_channel_display( - self, - channel: int, - enabled: bool, - *, - check_errors: bool = True, - ) -> None: - self._validate_channel(channel) - self.transport.write(f":CHANnel{channel}:DISPlay {'ON' if enabled else 'OFF'}") - if check_errors: - self.assert_no_errors() - - def focus_channel( - self, - channel: int, - *, - time_range_s: float | None = None, - vertical_scale_v_per_div: float | None = None, - hide_other_channels: bool = False, - check_errors: bool = True, - ) -> None: - self._validate_channel(channel) - if hide_other_channels: - for other in range(1, 5): - if other != channel: - self.transport.write(f":CHANnel{other}:DISPlay OFF") - self.transport.write(f":CHANnel{channel}:DISPlay ON") - if time_range_s is not None: - self.set_time_range(time_range_s) - if vertical_scale_v_per_div is not None: - self.set_vertical_scale(channel, vertical_scale_v_per_div) + self.transport.query_opc() if check_errors: self.assert_no_errors() diff --git a/src/wavebench/drivers/rtm2032.py b/src/wavebench/drivers/rtm2032.py index a16e864..94db489 100644 --- a/src/wavebench/drivers/rtm2032.py +++ b/src/wavebench/drivers/rtm2032.py @@ -62,42 +62,6 @@ def autoscale(self, wait_opc: bool = True, check_errors: bool = True) -> None: if check_errors: self.assert_no_errors() - def set_channel_display( - self, - channel: int, - enabled: bool, - *, - check_errors: bool = True, - ) -> None: - if channel < 1: - raise DataError("channel must be >= 1") - self.transport.write(f"CHAN{channel}:STAT {'ON' if enabled else 'OFF'}") - if check_errors: - self.assert_no_errors() - - def focus_channel( - self, - channel: int, - *, - time_range_s: float | None = None, - vertical_scale_v_per_div: float | None = None, - hide_other_channels: bool = False, - check_errors: bool = True, - ) -> None: - if channel < 1: - raise DataError("channel must be >= 1") - if hide_other_channels: - for other in range(1, 3): - if other != channel: - self.transport.write(f"CHAN{other}:STAT OFF") - self.transport.write(f"CHAN{channel}:STAT ON") - if time_range_s is not None: - self.set_time_range(time_range_s) - if vertical_scale_v_per_div is not None: - self.set_vertical_scale(channel, vertical_scale_v_per_div) - if check_errors: - self.assert_no_errors() - def set_time_range(self, time_range_s: float) -> None: if time_range_s <= 0: raise DataError("time range must be > 0") diff --git a/src/wavebench/instruments/builtin.py b/src/wavebench/instruments/builtin.py index 5900363..bdf936e 100644 --- a/src/wavebench/instruments/builtin.py +++ b/src/wavebench/instruments/builtin.py @@ -61,8 +61,6 @@ def _open_dm3000(context): "scope.idn", "scope.errors", "scope.autoscale", - "scope.channel_display", - "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", @@ -90,8 +88,6 @@ def _open_dm3000(context): "scope.idn", "scope.errors", "scope.autoscale", - "scope.channel_display", - "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", diff --git a/src/wavebench/instruments/contracts.py b/src/wavebench/instruments/contracts.py index f890b38..e39c51f 100644 --- a/src/wavebench/instruments/contracts.py +++ b/src/wavebench/instruments/contracts.py @@ -64,7 +64,7 @@ def set_channel_display( enabled: bool, *, check_errors: bool = True, - ) -> None: ... + ) -> dict[str, Any] | None: ... def focus_channel( self, @@ -74,7 +74,7 @@ def focus_channel( vertical_scale_v_per_div: float | None = None, hide_other_channels: bool = False, check_errors: bool = True, - ) -> None: ... + ) -> dict[str, Any] | None: ... def fetch_waveform( self, diff --git a/src/wavebench/plugins/builtin.py b/src/wavebench/plugins/builtin.py index ded3c97..25f6f8f 100644 --- a/src/wavebench/plugins/builtin.py +++ b/src/wavebench/plugins/builtin.py @@ -13,8 +13,6 @@ "scope.idn", "scope.errors", "scope.autoscale", - "scope.channel_display", - "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", @@ -35,8 +33,6 @@ "scope.idn", "scope.errors", "scope.autoscale", - "scope.channel_display", - "scope.focus_channel", "scope.fetch_waveform", "scope.capture_waveform", "scope.capture_waveforms", diff --git a/src/wavebench/services/scope_service.py b/src/wavebench/services/scope_service.py index 76b852b..6766edf 100644 --- a/src/wavebench/services/scope_service.py +++ b/src/wavebench/services/scope_service.py @@ -2,6 +2,7 @@ import csv import json +import math import os import traceback from collections.abc import Iterator @@ -116,6 +117,33 @@ class MultiCaptureResult: screenshot_path: Path | None commands_log_path: Path | None + +def _validate_scope_channel(channel: int) -> None: + if isinstance(channel, bool) or not isinstance(channel, int) or channel < 1: + raise ConfigError("scope channel must be a positive integer") + + +def _validate_optional_positive_finite(value: float | None, name: str) -> None: + if value is None: + return + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ConfigError(f"{name} must be a finite number > 0") + if not math.isfinite(float(value)) or float(value) <= 0: + raise ConfigError(f"{name} must be a finite number > 0") + + +def _scope_mutation_manifest( + result: object, + default: dict[str, Any], +) -> dict[str, Any]: + if not isinstance(result, dict): + return default + merged = dict(default) + merged.update(result) + merged.setdefault("affected_settings", default["affected_settings"]) + return merged + + @dataclass class ScopeService: config: WaveBenchConfig @@ -329,17 +357,18 @@ def autoscale(self) -> None: ) def set_channel_display(self, channel: int, enabled: bool) -> dict[str, Any]: + _validate_scope_channel(channel) required = ["scope.channel_display"] if self.config.scope.check_errors: required.append("scope.errors") self._require("scope.channel_display", *required) with self._scope_session() as scope: - scope.set_channel_display( + result = scope.set_channel_display( channel, enabled, check_errors=self.config.scope.check_errors, ) - return { + default = { "operation": "scope.channel_display", "mutates_instrument": True, "raw_scpi": False, @@ -347,6 +376,7 @@ def set_channel_display(self, channel: int, enabled: bool) -> dict[str, Any]: "display": "on" if enabled else "off", "affected_settings": [f"CH{channel}.display"], } + return _scope_mutation_manifest(result, default) def focus_channel( self, @@ -356,12 +386,18 @@ def focus_channel( vertical_scale_v_per_div: float | None = None, hide_other_channels: bool = False, ) -> dict[str, Any]: + _validate_scope_channel(channel) + _validate_optional_positive_finite(time_range_s, "time_range_s") + _validate_optional_positive_finite( + vertical_scale_v_per_div, + "vertical_scale_v_per_div", + ) required = ["scope.focus_channel"] if self.config.scope.check_errors: required.append("scope.errors") self._require("scope.focus_channel", *required) with self._scope_session() as scope: - scope.focus_channel( + result = scope.focus_channel( channel, time_range_s=time_range_s, vertical_scale_v_per_div=vertical_scale_v_per_div, @@ -373,10 +409,9 @@ def focus_channel( affected.append("timebase.range") if vertical_scale_v_per_div is not None: affected.append(f"CH{channel}.vertical_scale") - affected.append(f"CH{channel}.offset") if hide_other_channels: - affected.extend(f"CH{other}.display" for other in range(1, 5) if other != channel) - return { + affected.append("other_channels.display") + default = { "operation": "scope.focus_channel", "mutates_instrument": True, "raw_scpi": False, @@ -386,6 +421,7 @@ def focus_channel( "hide_other_channels": hide_other_channels, "affected_settings": affected, } + return _scope_mutation_manifest(result, default) def fetch_waveform(self, channel: int) -> WaveformData: if self.config.waveform.format.lower() != "real": diff --git a/tests/test_ds1104.py b/tests/test_ds1104.py index 57992ff..d9b46fb 100644 --- a/tests/test_ds1104.py +++ b/tests/test_ds1104.py @@ -232,54 +232,6 @@ def test_capture_translates_total_time_range_to_12_divisions(): assert "*OPC?" in transport.queries -def test_autoscale_skips_opc_query_to_avoid_ds1000z_socket_desync(): - transport = FakeTransport(responses={":SYSTem:ERRor?": '0,"No error"'}) - scope = DS1104Scope(transport=transport) - - scope.autoscale(wait_opc=True, check_errors=True) - - assert transport.writes == [":AUToscale"] - assert transport.queries == [":SYSTem:ERRor?"] - assert any( - direction == "telemetry" and "wait_opc=skipped" in text - for direction, text in transport.events - ) - - -def test_set_channel_display_writes_rigol_display_command(): - transport = FakeTransport() - scope = DS1104Scope(transport=transport) - - scope.set_channel_display(2, False, check_errors=False) - - assert transport.writes == [":CHANnel2:DISPlay OFF"] - - -def test_focus_channel_can_hide_others_and_adjust_selected_channel(): - transport = FakeTransport() - scope = DS1104Scope(transport=transport) - - scope.focus_channel( - 2, - time_range_s=0.012, - vertical_scale_v_per_div=0.2, - hide_other_channels=True, - check_errors=False, - ) - - assert transport.writes == [ - ":CHANnel1:DISPlay OFF", - ":CHANnel3:DISPlay OFF", - ":CHANnel4:DISPlay OFF", - ":CHANnel2:DISPlay ON", - ":TIMebase:MODE MAIN", - ":TIMebase:MAIN:SCALe 0.001", - ":CHANnel2:DISPlay ON", - ":CHANnel2:SCALe 0.2", - ":CHANnel2:OFFSet 0", - ] - - def test_multichannel_capture_uses_one_single_and_one_opc(): current_channel = {"value": 1} diff --git a/tests/test_scope_snapshot.py b/tests/test_scope_snapshot.py index 9b05ba8..dd738ee 100644 --- a/tests/test_scope_snapshot.py +++ b/tests/test_scope_snapshot.py @@ -1,4 +1,5 @@ import io +import math from contextlib import redirect_stdout from types import SimpleNamespace from unittest.mock import patch @@ -70,6 +71,15 @@ def test_scope_service_channel_display_returns_mutation_manifest(): def set_channel_display(channel, enabled, *, check_errors=True): calls.append((channel, enabled, check_errors)) + return { + "operation": "scope.channel_display", + "mutates_instrument": True, + "raw_scpi": False, + "channel": channel, + "display": "off", + "affected_settings": ["CH3.display"], + "driver_manifest": True, + } driver = SimpleNamespace(set_channel_display=set_channel_display) descriptor = SimpleNamespace( @@ -95,6 +105,7 @@ def set_channel_display(channel, enabled, *, check_errors=True): "channel": 3, "display": "off", "affected_settings": ["CH3.display"], + "driver_manifest": True, } @@ -118,6 +129,22 @@ def focus_channel( "check_errors": check_errors, } ) + return { + "operation": "scope.focus_channel", + "mutates_instrument": True, + "raw_scpi": False, + "channel": channel, + "time_range_s": time_range_s, + "vertical_scale_v_per_div": vertical_scale_v_per_div, + "hide_other_channels": hide_other_channels, + "affected_settings": [ + "CH2.display", + "timebase.range", + "CH2.vertical_scale", + "CH1.display", + ], + "driver_manifest": True, + } driver = SimpleNamespace(focus_channel=focus_channel) descriptor = SimpleNamespace( @@ -160,11 +187,77 @@ def focus_channel( "CH2.display", "timebase.range", "CH2.vertical_scale", - "CH2.offset", "CH1.display", - "CH3.display", - "CH4.display", ] + assert manifest["driver_manifest"] is True + + +@pytest.mark.parametrize("bad_channel", [0, -1, True, 1.5, "1"]) +def test_scope_service_display_rejects_invalid_channel_before_opening(bad_channel): + descriptor = SimpleNamespace( + driver_id="example.scope", + capabilities=("scope.channel_display",), + ) + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace(driver="example.scope", check_errors=False) + ), + logger=SimpleNamespace(), + descriptor=descriptor, + ) + + with patch.object(service, "_open_scope") as open_scope: + with pytest.raises(ConfigError, match="positive integer"): + service.set_channel_display(channel=bad_channel, enabled=True) + + open_scope.assert_not_called() + + +@pytest.mark.parametrize("bad_value", [math.nan, math.inf, -math.inf, 0.0, -1.0]) +def test_scope_service_focus_rejects_invalid_time_range_before_opening(bad_value): + descriptor = SimpleNamespace( + driver_id="example.scope", + capabilities=("scope.focus_channel",), + ) + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace(driver="example.scope", check_errors=False) + ), + logger=SimpleNamespace(), + descriptor=descriptor, + ) + + with patch.object(service, "_open_scope") as open_scope: + with pytest.raises(ConfigError, match="finite number > 0"): + service.focus_channel(channel=1, time_range_s=bad_value) + + open_scope.assert_not_called() + + +@pytest.mark.parametrize("bad_value", [math.nan, math.inf, -math.inf, 0.0, -1.0]) +def test_scope_service_focus_rejects_invalid_vertical_scale_before_driver_call(bad_value): + calls = [] + + def focus_channel(*args, **kwargs): + calls.append((args, kwargs)) + + descriptor = SimpleNamespace( + driver_id="example.scope", + capabilities=("scope.focus_channel",), + ) + service = ScopeService( + config=SimpleNamespace( + scope=SimpleNamespace(driver="example.scope", check_errors=False) + ), + logger=SimpleNamespace(), + session=SimpleNamespace(focus_channel=focus_channel), + descriptor=descriptor, + ) + + with pytest.raises(ConfigError, match="finite number > 0"): + service.focus_channel(channel=1, vertical_scale_v_per_div=bad_value) + + assert calls == [] def test_scope_status_cli_uses_default_channel_and_prints_stable_fields(): diff --git a/tests/test_waveform.py b/tests/test_waveform.py index 533a941..7fac4fd 100644 --- a/tests/test_waveform.py +++ b/tests/test_waveform.py @@ -88,30 +88,6 @@ def test_scope_screenshot_png_rejects_non_png(self): with self.assertRaises(DataError): scope.screenshot_png() - def test_scope_focus_hides_only_rtm2032_analog_channels(self): - transport = FakeTransport({}) - scope = RTM2032Scope(transport=transport) - - scope.focus_channel( - 2, - time_range_s=0.012, - vertical_scale_v_per_div=0.2, - hide_other_channels=True, - check_errors=False, - ) - - self.assertEqual( - transport.writes, - [ - "CHAN1:STAT OFF", - "CHAN2:STAT ON", - "TIMebase:RANGe 0.012", - "CHAN2:STAT ON", - "CHAN2:SCAL 0.2", - "CHAN2:POS 0", - ], - ) - if __name__ == "__main__": unittest.main()