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/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..e39c51f 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, + ) -> dict[str, Any] | 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, + ) -> dict[str, Any] | None: ... + def fetch_waveform( self, channel: int, diff --git a/src/wavebench/services/scope_service.py b/src/wavebench/services/scope_service.py index 9a2f822..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 @@ -328,6 +356,73 @@ def autoscale(self) -> None: check_errors=self.config.autoscale.check_errors, ) + 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: + result = scope.set_channel_display( + channel, + enabled, + check_errors=self.config.scope.check_errors, + ) + default = { + "operation": "scope.channel_display", + "mutates_instrument": True, + "raw_scpi": False, + "channel": channel, + "display": "on" if enabled else "off", + "affected_settings": [f"CH{channel}.display"], + } + return _scope_mutation_manifest(result, default) + + 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]: + _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: + result = 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") + if hide_other_channels: + affected.append("other_channels.display") + default = { + "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, + } + return _scope_mutation_manifest(result, default) + 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_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..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 @@ -65,6 +66,200 @@ 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)) + 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( + 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"], + "driver_manifest": True, + } + + +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, + } + ) + 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( + 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", + "CH1.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(): service = SimpleNamespace( config=SimpleNamespace(scope=SimpleNamespace(default_channel=2)),