Add explicit scope display and focus controls

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

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)

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}")
 

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")

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",),

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,

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'")

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")

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):