diff --git a/README.md b/README.md index aef2ac7..730333e 100644 --- a/README.md +++ b/README.md @@ -145,6 +145,9 @@ WaveBench 主包长期预装 RTM2000/RTM2032、DS1104Z/DS1000Z、DG4000/DG4202 - `run.schema`：返回 run plan schema - `run.check`：只解析并检查 `plans/*.toml` 下的 run plan，不连接仪器 - `capture.inspect`：读取 `data/raw/` 下的离线采集包摘要 + - `scope.observe`：默认只读连接配置中的示波器，返回 IDN、状态快照（若驱动支持）和高阻安全判断；支持 CH1-CH4 多通道观察；可传 `fetch_waveform=true` 读取一个或多个通道的当前波形摘要，但该模式可能改动示波器的波形传输源/模式，响应会标注 `read_only=false`。多通道 fetch 生成的 pairwise `relationships` 会标注 `same_acquisition=false`，相位等时序关系只作为建议线索 + - `scope.advise`：基于 `scope.observe` 给出每通道时基/垂直档位/display 建议；只返回建议，不应用调整 + - `doctor.config`：结构化返回配置中各仪器的只读可达性、IDN 和型号匹配检查结果 - `/mcp` 与 `/call` 的 JSON 请求体有 1 MiB 上限；路径参数按工具限制在项目内固定目录 ## 安全默认值 diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..5d759a7 --- /dev/null +++ b/TODO.md @@ -0,0 +1,48 @@ +# WaveBench AI-Agent Experience TODO + +目标：让 AI agent 能以低风险、结构化、可审计的方式观察实验台、理解已有产物、提出建议，并只通过显式 run plan 执行会改变仪器状态的动作。 + +## P0: 安全只读上下文 + +- [x] 增加 MCP 工具 `scope.observe`：读取配置中的示波器，返回 IDN、状态快照、通道高阻判断、当前波形摘要；不保存文件、不改仪器状态、不暴露 raw SCPI。 +- [x] `scope.observe` 支持 CH1-CH4 多通道观察；显式 `fetch_waveform=true` 且获取到两个以上通道时返回 pairwise relationships。 +- [x] 增加 MCP 工具 `doctor.config`：把现有 `doctor` 结果结构化返回给 agent，支持配置可达性和型号匹配判断。 +- [ ] 增加 MCP 工具 `net.discover`：结构化返回只读 LAN 仪器发现结果，默认限制网段大小和端口集合。 +- [ ] 所有 MCP 工具都标注 `read_only`、`mutates_instrument=false`、`raw_scpi=false` 等安全元数据。 + +## P1: 离线产物理解 + +- [ ] 增强 `capture.inspect`：可选返回 FFT 摘要、质量告警、推荐下一步采集参数，但不读取大数组进响应。 +- [ ] 增加 `capture.list`：列出 `data/raw` 下最近采集包，供 agent 找上下文。 +- [ ] 增加 `run.list` / `run.inspect`：列出和解释 `data/runs` 下的实验记录。 +- [ ] 给采集包生成 agent 友好的 `summary.md` 或结构化 `analysis.json`。 + +## P2: Agent 建议层 + +- [x] 增加 `scope.advise`：基于 `scope.observe` 的结果给出时基、垂直档位、触发、点数建议；只返回建议，不应用。 + - [x] 根据实测或期望频率推荐每通道 focus time-range，默认约 10 个周期。 + - [x] 根据实测 Vpp 推荐垂直档位，默认约占 5 格。 + - [x] 识别 CH1/CH2 这类大频率跨度，建议分通道/分 profile 观察，避免单时基误判形状。 +- [x] 增加 `scope.observe.expectations`：把已知闭环信号作为结构化依据，例如 CH1 1 kHz/1 Vpp/50% 方波、CH2 50 kHz/1 Vpp/500 mVdc/30% 对称三角波，并返回逐项 pass/warn/fail。 + - [x] 支持 frequency/Vpp/mean/duty/symmetry_percent 的基础断言。 + - [x] 期望断言必须显式 `fetch_waveform=true`，避免 agent 在不知情时触发波形传输状态变化。 + - [x] 返回总体 expectation status，并把 fail/warn 加入 agent hints。 +- [x] 增加多时基建议：当多个通道的频率跨度较大时，提醒 agent 不要用单个显示时基同时判断所有通道形状，应分通道或分 profile 采集。 +- [x] 多通道关系分析增加交点：对每对通道返回交点数量、采样返回点、交点时间/电压和相对斜率方向；交点过多时截断并给 warning。 +- [ ] 让 `scope.expect` 也参与多时基建议：即使当前频率估计低置信，也能利用用户给定的期望频率识别 CH1/CH2 这种 50x 频率跨度。 +- [ ] 增加 `plan.propose`：从自然语言目标或结构化目标生成保守 run plan 草案，默认写到 `plans/`，不执行。 +- [ ] 增加 `plan.explain`：解释某个 run plan 会读写哪些仪器、哪些步骤会改变输出状态、有哪些保护。 + +## P3: 可控执行边界 + +- [ ] MCP 继续保持默认只读；会改仪器状态的能力只通过显式 `run plan` 文件和人工确认入口暴露。 +- [x] 增加显式示波器显示控制 CLI：通道显示 on/off、focus 单通道、显式 autoscale；用于人类和 agent 做可审计调参。 +- [x] focus 动作只调整示波器显示/采集窗口，不改变信号源、电源或被测对象；执行后输出 mutation manifest。 +- [ ] 对所有会改状态的 plan 步骤生成 mutation manifest，便于 agent 在执行前向人类说明。 +- [ ] 增加跨进程仪器锁，避免 Windows/WSL/多个 agent 同时打开同一台仪器导致响应串线。 + +## P4: 人类与 Agent 共用体验 + +- [x] 增加 `scripts/wsl-run.ps1` 作为 Windows 到 WSL 的标准执行入口，并在 README 中记录。 +- [ ] 为 WSL/Windows 推荐环境增加 `doctor environment` 或 `env doctor`，检查 Python、PyVISA、WSL、网络可达性。 +- [ ] 增加最小可视化：采集包自动生成 waveform/FFT PNG，报告中可直接查看。 diff --git "a/doc/project/WaveBench_HTTP_MCP_\345\217\252\350\257\273\346\216\245\345\217\243.md" "b/doc/project/WaveBench_HTTP_MCP_\345\217\252\350\257\273\346\216\245\345\217\243.md" index 554ab02..1989b9e 100644 --- "a/doc/project/WaveBench_HTTP_MCP_\345\217\252\350\257\273\346\216\245\345\217\243.md" +++ "b/doc/project/WaveBench_HTTP_MCP_\345\217\252\350\257\273\346\216\245\345\217\243.md" @@ -51,16 +51,21 @@ MCP notification 请求没有 `id` 时返回空响应；普通请求返回 JSON- - `run.schema`：返回 run plan schema 文本和结构化 schema 行。 - `run.check`：参数 `{"plan": "plans/.toml"}`，只解析并检查 `plans/*.toml` 下的 run plan，不连接仪器。 - `capture.inspect`：参数 `{"path": "data/raw/"}`，读取 `data/raw/` 下的离线采集包摘要。 +- `scope.observe`：参数 `{"channels": [1, 2], "fetch_waveform": false}`，只读连接配置中的示波器，返回 IDN、状态快照（若驱动支持）和高阻安全判断。支持 CH1-CH4 多通道观察。显式传 `fetch_waveform=true` 时，会读取一个或多个通道的当前波形摘要；当至少两个通道读取成功时，返回 `relationships`，包含频率比、Vpp/均值关系、公共时间窗、归一化相关性、估计延迟、同频相位和通道间交点。该工具不保存文件，不暴露 raw SCPI，但抓波形可能改动示波器的波形传输源/模式。 + - 可选 `expectations`：例如 `{"1": {"frequency_hz": 1000, "vpp_v": 1.0, "duty_percent": 50}, "2": {"frequency_hz": 50000, "vpp_v": 1.0, "mean_v": 0.5, "symmetry_percent": 30}}`。使用 expectations 必须同时传 `fetch_waveform=true`。 +- `scope.advise`：参数同 `scope.observe`，另有 `target_cycles` 和 `target_vertical_divisions`。它基于当前观察和可选 expectations 给出 `scope focus` / `scope display` 建议、每通道推荐时基窗口和垂直档位；只返回建议，绝不应用调整。若传 `fetch_waveform=true`，同样可能改动示波器的波形传输源/模式。 +- `doctor.config`：参数 `{"timeout_ms": 1000}`，结构化返回配置中各仪器的只读可达性、IDN 和型号匹配检查结果；不执行网段发现。 ## 安全边界 - 默认只监听 `127.0.0.1`。 - 拒绝监听 `0.0.0.0`。 - `/mcp`、`/tools` 和 `/call` 强制 Bearer token。 -- 当前工具全部只读。 +- 当前工具不提供 raw SCPI，不应用显示/输出/采集建议；`scope.observe` / `scope.advise` 在 `fetch_waveform=true` 时会显式标注可能的波形传输状态影响。 - 不提供 raw SCPI。 - 不提供 power/source output on/off。 - 不提供 run 执行工具。 - `run.check` 只允许项目内 `plans/*.toml`。 - `capture.inspect` 只允许项目内 `data/raw/` 离线采集包。 +- `doctor.config` 只检查当前配置中的资源，不扫描网段。 - `/mcp` 和 `/call` 的 JSON 请求体有 1 MiB 上限。 diff --git a/src/wavebench/data/expectations.py b/src/wavebench/data/expectations.py new file mode 100644 index 0000000..0d0fe41 --- /dev/null +++ b/src/wavebench/data/expectations.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +from typing import Any + +import numpy as np + +from wavebench.instruments.models import WaveformData + + +def evaluate_waveform_expectation( + waveform: WaveformData, + expectation: dict[str, Any], +) -> dict[str, Any]: + summary = waveform.summary( + expected_frequency_hz=_optional_positive_float(expectation, "frequency_hz"), + frequency_tolerance_ratio=float(expectation.get("frequency_tolerance_ratio", 0.05)), + ) + checks: list[dict[str, Any]] = [] + _check_frequency(summary, expectation, checks) + _check_vpp(summary, expectation, checks) + _check_mean(summary, expectation, checks) + _check_duty(summary, expectation, checks) + _check_symmetry(waveform, expectation, checks) + statuses = {check["status"] for check in checks} + if "fail" in statuses: + status = "fail" + elif "warn" in statuses: + status = "warn" + else: + status = "pass" + return { + "status": status, + "channel": waveform.channel, + "label": expectation.get("label"), + "shape": expectation.get("shape"), + "checks": checks, + } + + +def expectation_summary(results: dict[int, dict[str, Any]]) -> dict[str, Any]: + statuses = {result["status"] for result in results.values()} + if "fail" in statuses: + status = "fail" + elif "warn" in statuses: + status = "warn" + else: + status = "pass" if results else "skipped" + return { + "status": status, + "channels": {str(channel): result["status"] for channel, result in sorted(results.items())}, + } + + +def estimate_triangle_symmetry_percent(waveform: WaveformData) -> float | None: + times = waveform.times_s + values = np.asarray(waveform.voltages_v, dtype=np.float64) + if times.size != values.size or values.size < 8: + return None + span = float(np.max(values) - np.min(values)) + if span <= 1e-12: + return None + centered = values - float(np.mean(values)) + diffs = np.diff(centered) + if diffs.size < 3: + return None + signs = np.sign(diffs) + for index in range(1, signs.size): + if signs[index] == 0: + signs[index] = signs[index - 1] + maxima = [ + index + for index in range(1, values.size - 1) + if signs[index - 1] > 0 and signs[index] < 0 + ] + minima = [ + index + for index in range(1, values.size - 1) + if signs[index - 1] < 0 and signs[index] > 0 + ] + fractions: list[float] = [] + for left_min, right_min in zip(minima, minima[1:]): + if right_min <= left_min: + continue + peaks = [index for index in maxima if left_min < index < right_min] + if not peaks: + continue + peak = max(peaks, key=lambda index: values[index]) + period = float(times[right_min] - times[left_min]) + if period <= 0: + continue + fractions.append(float((times[peak] - times[left_min]) / period * 100.0)) + if not fractions: + return None + return float(np.median(np.asarray(fractions, dtype=np.float64))) + + +def _check_frequency( + summary: dict[str, Any], + expectation: dict[str, Any], + checks: list[dict[str, Any]], +) -> None: + expected = _optional_positive_float(expectation, "frequency_hz") + if expected is None: + return + actual = summary.get("frequency_estimate_hz") + tolerance = float(expectation.get("frequency_tolerance_ratio", 0.05)) + low_confidence = any( + str(item).startswith("low_cycle_count") + for item in summary.get("quality_warnings", []) + ) + if not isinstance(actual, (int, float)) or actual <= 0: + checks.append(_check("frequency_hz", "warn", expected, actual, "frequency unavailable")) + return + error_ratio = abs(float(actual) - expected) / expected + if low_confidence: + checks.append( + _check( + "frequency_hz", + "warn", + expected, + float(actual), + "frequency low confidence because waveform contains too few cycles", + error_ratio=error_ratio, + tolerance_ratio=tolerance, + ) + ) + return + checks.append( + _check( + "frequency_hz", + "pass" if error_ratio <= tolerance else "fail", + expected, + float(actual), + "ok" if error_ratio <= tolerance else "frequency out of tolerance", + error_ratio=error_ratio, + tolerance_ratio=tolerance, + ) + ) + + +def _check_vpp( + summary: dict[str, Any], + expectation: dict[str, Any], + checks: list[dict[str, Any]], +) -> None: + expected = _optional_positive_float(expectation, "vpp_v") + if expected is None: + return + actual = summary.get("voltage_vpp_v") + tolerance = float(expectation.get("vpp_tolerance_ratio", 0.10)) + if not isinstance(actual, (int, float)): + checks.append(_check("vpp_v", "warn", expected, actual, "Vpp unavailable")) + return + error_ratio = abs(float(actual) - expected) / expected + checks.append( + _check( + "vpp_v", + "pass" if error_ratio <= tolerance else "fail", + expected, + float(actual), + "ok" if error_ratio <= tolerance else "Vpp out of tolerance", + error_ratio=error_ratio, + tolerance_ratio=tolerance, + ) + ) + + +def _check_mean( + summary: dict[str, Any], + expectation: dict[str, Any], + checks: list[dict[str, Any]], +) -> None: + expected = _optional_float(expectation, "mean_v") + if expected is None: + expected = _optional_float(expectation, "offset_v") + if expected is None: + return + actual = summary.get("voltage_mean_v") + tolerance = float(expectation.get("mean_tolerance_v", 0.05)) + if not isinstance(actual, (int, float)): + checks.append(_check("mean_v", "warn", expected, actual, "mean unavailable")) + return + error = abs(float(actual) - expected) + checks.append( + _check( + "mean_v", + "pass" if error <= tolerance else "fail", + expected, + float(actual), + "ok" if error <= tolerance else "mean out of tolerance", + error_abs=error, + tolerance_abs=tolerance, + ) + ) + + +def _check_duty( + summary: dict[str, Any], + expectation: dict[str, Any], + checks: list[dict[str, Any]], +) -> None: + expected = _optional_float(expectation, "duty_cycle") + if expected is None and "duty_percent" in expectation: + expected = _optional_float(expectation, "duty_percent") + if expected is not None: + expected /= 100.0 + if expected is None: + return + actual = summary.get("duty_cycle") + tolerance = float(expectation.get("duty_tolerance", 0.05)) + if not isinstance(actual, (int, float)): + checks.append(_check("duty_cycle", "warn", expected, actual, "duty unavailable")) + return + error = abs(float(actual) - expected) + checks.append( + _check( + "duty_cycle", + "pass" if error <= tolerance else "fail", + expected, + float(actual), + "ok" if error <= tolerance else "duty out of tolerance", + error_abs=error, + tolerance_abs=tolerance, + ) + ) + + +def _check_symmetry( + waveform: WaveformData, + expectation: dict[str, Any], + checks: list[dict[str, Any]], +) -> None: + expected = _optional_float(expectation, "symmetry_percent") + if expected is None: + return + actual = estimate_triangle_symmetry_percent(waveform) + tolerance = float(expectation.get("symmetry_tolerance_percent", 5.0)) + if actual is None: + checks.append(_check("symmetry_percent", "warn", expected, actual, "symmetry unavailable")) + return + error = abs(actual - expected) + checks.append( + _check( + "symmetry_percent", + "pass" if error <= tolerance else "fail", + expected, + actual, + "ok" if error <= tolerance else "symmetry out of tolerance", + error_abs=error, + tolerance_abs=tolerance, + ) + ) + + +def _check(name: str, status: str, expected: Any, actual: Any, message: str, **extra: Any) -> dict[str, Any]: + return { + "metric": name, + "status": status, + "expected": expected, + "actual": actual, + "message": message, + **extra, + } + + +def _optional_float(data: dict[str, Any], name: str) -> float | None: + if name not in data or data[name] is None: + return None + value = data[name] + if isinstance(value, bool): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _optional_positive_float(data: dict[str, Any], name: str) -> float | None: + value = _optional_float(data, name) + if value is None or value <= 0: + return None + return value diff --git a/src/wavebench/data/relationships.py b/src/wavebench/data/relationships.py new file mode 100644 index 0000000..115e764 --- /dev/null +++ b/src/wavebench/data/relationships.py @@ -0,0 +1,293 @@ +from __future__ import annotations + +from itertools import combinations +from typing import Any + +import numpy as np + +from wavebench.instruments.models import WaveformData + + +def analyze_waveform_relationships( + waveforms: dict[int, WaveformData], + *, + same_acquisition: bool = True, + max_correlation_points: int = 4096, + max_intersections: int = 64, +) -> list[dict[str, Any]]: + relationships: list[dict[str, Any]] = [] + for left_channel, right_channel in combinations(sorted(waveforms), 2): + relationships.append( + analyze_waveform_pair( + waveforms[left_channel], + waveforms[right_channel], + same_acquisition=same_acquisition, + max_correlation_points=max_correlation_points, + max_intersections=max_intersections, + ) + ) + return relationships + + +def analyze_waveform_pair( + left: WaveformData, + right: WaveformData, + *, + same_acquisition: bool = True, + max_correlation_points: int = 4096, + max_intersections: int = 64, +) -> dict[str, Any]: + left_summary = left.summary() + right_summary = right.summary() + warnings: list[str] = [] + common = _common_time_axis(left, right, max_points=max_correlation_points) + correlation = _correlation_payload(common, warnings=warnings) + intersections = _intersection_payload( + common, + warnings=warnings, + max_intersections=max_intersections, + ) + if not same_acquisition: + warnings.append("not_same_acquisition_timing_relationships_are_advisory") + left_frequency = _trusted_frequency(left_summary, warnings=warnings, label=f"CH{left.channel}") + right_frequency = _trusted_frequency(right_summary, warnings=warnings, label=f"CH{right.channel}") + frequency_ratio = None + phase_degrees = None + if left_frequency is not None and right_frequency is not None: + lower = min(left_frequency, right_frequency) + upper = max(left_frequency, right_frequency) + if lower > 0: + frequency_ratio = float(upper / lower) + if ( + same_acquisition + and + correlation.get("lag_at_max_correlation_s") is not None + and abs(left_frequency - right_frequency) / max(left_frequency, right_frequency) <= 0.01 + ): + phase_degrees = float( + (correlation["lag_at_max_correlation_s"] * left_frequency * 360.0) % 360.0 + ) + elif frequency_ratio is not None and abs(frequency_ratio - 1.0) > 0.01: + warnings.append("phase_not_meaningful_for_different_frequencies") + return { + "channels": [left.channel, right.channel], + "left_channel": left.channel, + "right_channel": right.channel, + "common_time": {**common["metadata"], "same_acquisition": same_acquisition}, + "frequency": { + "left_hz": left_frequency, + "right_hz": right_frequency, + "ratio_high_over_low": frequency_ratio, + }, + "voltage": { + "left_vpp_v": left_summary["voltage_vpp_v"], + "right_vpp_v": right_summary["voltage_vpp_v"], + "vpp_ratio_right_over_left": _safe_ratio( + right_summary["voltage_vpp_v"], + left_summary["voltage_vpp_v"], + ), + "mean_delta_right_minus_left_v": float( + right_summary["voltage_mean_v"] - left_summary["voltage_mean_v"] + ), + "rms_ratio_right_over_left": _safe_ratio( + right_summary["voltage_rms_v"], + left_summary["voltage_rms_v"], + ), + }, + "correlation": correlation, + "intersections": intersections, + "phase_degrees_at_left_frequency": phase_degrees, + "warnings": warnings, + } + + +def _common_time_axis( + left: WaveformData, + right: WaveformData, + *, + max_points: int, +) -> dict[str, Any]: + left_times = left.times_s + right_times = right.times_s + start = max(float(left_times[0]), float(right_times[0])) + stop = min(float(left_times[-1]), float(right_times[-1])) + if stop <= start: + return { + "time_s": np.array([], dtype=np.float64), + "left_v": np.array([], dtype=np.float64), + "right_v": np.array([], dtype=np.float64), + "metadata": { + "overlap": False, + "x_start_s": start, + "x_stop_s": stop, + "duration_s": 0.0, + "samples": 0, + }, + } + left_dt = left.header.x_increment + right_dt = right.header.x_increment + dt = max(value for value in (left_dt, right_dt) if value > 0) + count = int(np.floor((stop - start) / dt)) + 1 + count = max(2, min(count, max_points)) + common_times = np.linspace(start, stop, count, dtype=np.float64) + return { + "time_s": common_times, + "left_v": np.interp(common_times, left_times, left.voltages_v), + "right_v": np.interp(common_times, right_times, right.voltages_v), + "metadata": { + "overlap": True, + "x_start_s": start, + "x_stop_s": stop, + "duration_s": float(stop - start), + "samples": count, + }, + } + + +def _correlation_payload(common: dict[str, Any], *, warnings: list[str]) -> dict[str, Any]: + times = common["time_s"] + if times.size < 4: + warnings.append("insufficient_common_time_overlap") + return { + "normalized_pearson": None, + "max_cross_correlation": None, + "max_abs_cross_correlation": None, + "lag_at_max_correlation_s": None, + } + left = _normalize(common["left_v"]) + right = _normalize(common["right_v"]) + if left is None or right is None: + warnings.append("correlation_unavailable_for_flat_signal") + return { + "normalized_pearson": None, + "max_cross_correlation": None, + "max_abs_cross_correlation": None, + "lag_at_max_correlation_s": None, + } + pearson = float(np.mean(left * right)) + correlation = np.correlate(right, left, mode="full") / left.size + index = int(np.argmax(np.abs(correlation))) + lag_samples = index - (left.size - 1) + dt = float(np.median(np.diff(times))) + return { + "normalized_pearson": pearson, + "max_cross_correlation": float(correlation[index]), + "max_abs_cross_correlation": float(abs(correlation[index])), + "lag_at_max_correlation_s": float(lag_samples * dt), + } + + +def _intersection_payload( + common: dict[str, Any], + *, + warnings: list[str], + max_intersections: int, +) -> dict[str, Any]: + times = common["time_s"] + left = common["left_v"] + right = common["right_v"] + if times.size < 2: + return { + "mode": "none", + "count": 0, + "returned": 0, + "truncated": False, + "points": [], + } + diff = left - right + tolerance = max(float(np.max(np.abs(diff))) * 1e-9, 1e-12) + if bool(np.all(np.abs(diff) <= tolerance)): + warnings.append("waveforms_coincident_intersections_unbounded") + return { + "mode": "coincident", + "count": None, + "returned": 0, + "truncated": False, + "points": [], + } + points: list[dict[str, float | str]] = [] + count = 0 + last_time: float | None = None + for index in range(diff.size - 1): + d0 = float(diff[index]) + d1 = float(diff[index + 1]) + t0 = float(times[index]) + t1 = float(times[index + 1]) + if abs(d0) <= tolerance: + alpha = 0.0 + elif d0 * d1 < 0.0: + alpha = -d0 / (d1 - d0) + else: + continue + crossing_time = t0 + alpha * (t1 - t0) + if last_time is not None and abs(crossing_time - last_time) <= max(abs(t1 - t0) * 0.5, 1e-15): + continue + left_value = float(left[index] + alpha * (left[index + 1] - left[index])) + right_value = float(right[index] + alpha * (right[index + 1] - right[index])) + left_slope = _segment_slope(left, times, index) + right_slope = _segment_slope(right, times, index) + delta_slope = left_slope - right_slope + count += 1 + last_time = crossing_time + if len(points) < max_intersections: + points.append( + { + "time_s": float(crossing_time), + "voltage_v": float((left_value + right_value) / 2.0), + "left_slope_v_per_s": float(left_slope), + "right_slope_v_per_s": float(right_slope), + "delta_slope_v_per_s": float(delta_slope), + "direction": ( + "left_minus_right_rising" + if delta_slope > 0 + else "left_minus_right_falling" + if delta_slope < 0 + else "tangent_or_flat" + ), + } + ) + truncated = count > len(points) + if truncated: + warnings.append("intersections_truncated") + return { + "mode": "finite", + "count": count, + "returned": len(points), + "truncated": truncated, + "points": points, + } + + +def _segment_slope(values: np.ndarray, times: np.ndarray, index: int) -> float: + dt = float(times[index + 1] - times[index]) + if abs(dt) <= 1e-18: + return 0.0 + return float((values[index + 1] - values[index]) / dt) + + +def _normalize(values: np.ndarray) -> np.ndarray | None: + centered = values.astype(np.float64) - float(np.mean(values)) + rms = float(np.sqrt(np.mean(np.square(centered)))) + if rms <= 1e-12: + return None + return centered / rms + + +def _trusted_frequency(summary: dict[str, object], *, warnings: list[str], label: str) -> float | None: + frequency = summary.get("frequency_estimate_hz") + if not isinstance(frequency, (int, float)) or frequency <= 0: + warnings.append(f"{label}_frequency_unavailable") + return None + quality_warnings = summary.get("quality_warnings", []) + if any(str(item).startswith("low_cycle_count") for item in quality_warnings): + warnings.append(f"{label}_frequency_low_confidence") + return None + return float(frequency) + + +def _safe_ratio(numerator: object, denominator: object) -> float | None: + if not isinstance(numerator, (int, float)) or not isinstance(denominator, (int, float)): + return None + if abs(float(denominator)) <= 1e-18: + return None + return float(numerator) / float(denominator) diff --git a/src/wavebench/mcp_http.py b/src/wavebench/mcp_http.py index 1c30167..400f170 100644 --- a/src/wavebench/mcp_http.py +++ b/src/wavebench/mcp_http.py @@ -12,8 +12,11 @@ from wavebench.config import load_config from wavebench import __version__ from wavebench.data.packages import load_capture_package +from wavebench.doctor import doctor_records, has_doctor_errors from wavebench.errors import ConfigError, WaveBenchError from wavebench.logging import CommandLogger +from wavebench.services.agent_advise import scope_advise_payload +from wavebench.services.agent_observe import scope_observe_payload from wavebench.services.run_plan import load_run_plan, run_plan_schema_rows from wavebench.services.run_plan import format_run_plan_schema from wavebench.services.run_service import RunService @@ -50,12 +53,19 @@ class ToolSpec: description: str arguments: dict[str, Any] handler: Callable[[dict[str, Any], Path], dict[str, Any]] + read_only: bool = True + mutates_instrument: bool = False + raw_scpi: bool = False + instrument_state_effects: tuple[str, ...] = () def public_payload(self) -> dict[str, Any]: return { "name": self.name, "description": self.description, - "read_only": True, + "read_only": self.read_only, + "mutates_instrument": self.mutates_instrument, + "raw_scpi": self.raw_scpi, + "instrument_state_effects": list(self.instrument_state_effects), "arguments": self.arguments, } @@ -208,6 +218,125 @@ def _capture_inspect_tool(arguments: dict[str, Any], config_path: Path) -> dict[ } +def _optional_bool(arguments: dict[str, Any], name: str, default: bool) -> bool: + value = arguments.get(name, default) + if not isinstance(value, bool): + raise ConfigError(f"{name} must be a boolean / {name} 必须是布尔值") + return value + + +def _scope_observe_tool(arguments: dict[str, Any], config_path: Path) -> dict[str, Any]: + channel = arguments.get("channel") + if channel is not None and (isinstance(channel, bool) or not isinstance(channel, int)): + raise ConfigError("channel must be an integer / channel 必须是整数") + raw_channels = arguments.get("channels") + channels = None + if raw_channels is not None: + if not isinstance(raw_channels, list): + raise ConfigError("channels must be an array / channels 必须是数组") + channels = tuple(raw_channels) + if any(isinstance(item, bool) or not isinstance(item, int) for item in channels): + raise ConfigError("channels must contain integers / channels 必须包含整数") + expectations = _scope_expectations_argument(arguments.get("expectations")) + return scope_observe_payload( + config_path=_reject_sensitive_path(config_path, label="config"), + channel=channel, + channels=channels, + fetch_waveform=_optional_bool(arguments, "fetch_waveform", False), + allow_50ohm=_optional_bool(arguments, "allow_50ohm", False), + expectations=expectations, + ) + + +def _scope_advise_tool(arguments: dict[str, Any], config_path: Path) -> dict[str, Any]: + channel, channels = _scope_channel_arguments(arguments) + expectations = _scope_expectations_argument(arguments.get("expectations")) + return scope_advise_payload( + config_path=_reject_sensitive_path(config_path, label="config"), + channel=channel, + channels=channels, + fetch_waveform=_optional_bool(arguments, "fetch_waveform", False), + allow_50ohm=_optional_bool(arguments, "allow_50ohm", False), + expectations=expectations, + target_cycles=_optional_positive_number(arguments, "target_cycles", 10.0), + target_vertical_divisions=_optional_positive_number( + arguments, + "target_vertical_divisions", + 5.0, + ), + ) + + +def _scope_channel_arguments(arguments: dict[str, Any]) -> tuple[int | None, tuple[int, ...] | None]: + channel = arguments.get("channel") + if channel is not None and (isinstance(channel, bool) or not isinstance(channel, int)): + raise ConfigError("channel must be an integer / channel 必须是整数") + raw_channels = arguments.get("channels") + channels = None + if raw_channels is not None: + if not isinstance(raw_channels, list): + raise ConfigError("channels must be an array / channels 必须是数组") + channels = tuple(raw_channels) + if any(isinstance(item, bool) or not isinstance(item, int) for item in channels): + raise ConfigError("channels must contain integers / channels 必须包含整数") + return channel, channels + + +def _optional_positive_number(arguments: dict[str, Any], name: str, default: float) -> float: + value = arguments.get(name, default) + if isinstance(value, bool) or not isinstance(value, (int, float)) or value <= 0: + raise ConfigError(f"{name} must be a positive number / {name} 必须是正数") + return float(value) + + +def _scope_expectations_argument(raw: Any) -> dict[int, dict[str, Any]] | None: + if raw is None: + return None + if not isinstance(raw, dict): + raise ConfigError("expectations must be an object / expectations 必须是对象") + parsed: dict[int, dict[str, Any]] = {} + for key, value in raw.items(): + try: + channel = int(key) + except (TypeError, ValueError) as exc: + raise ConfigError("expectations keys must be channel numbers / expectations 键必须是通道号") from exc + if channel < 1: + raise ConfigError("expectations channel must be >= 1 / expectations 通道必须 >= 1") + if not isinstance(value, dict): + raise ConfigError("expectations entries must be objects / expectations 条目必须是对象") + parsed[channel] = dict(value) + return parsed + + +def _doctor_config_tool(arguments: dict[str, Any], config_path: Path) -> dict[str, Any]: + timeout_ms = arguments.get("timeout_ms") + if timeout_ms is not None and (isinstance(timeout_ms, bool) or not isinstance(timeout_ms, int) or timeout_ms <= 0): + raise ConfigError("timeout_ms must be a positive integer / timeout_ms 必须是正整数") + records = doctor_records( + load_config(_reject_sensitive_path(config_path, label="config")), + timeout_ms=timeout_ms, + include_visa=False, + ) + return { + "status": "error" if has_doctor_errors(records) else "ok", + "read_only": True, + "mutates_instrument": False, + "raw_scpi": False, + "records": [ + { + "severity": record.severity, + "target": record.target, + "driver": record.driver, + "resource": record.resource, + "idn": record.idn, + "message": record.message, + "suggestion": record.suggestion, + } + for record in records + ], + } + + READ_ONLY_TOOLS: dict[str, ToolSpec] = { "run.schema": ToolSpec( name="run.schema", @@ -240,6 +369,131 @@ def _capture_inspect_tool(arguments: dict[str, Any], config_path: Path) -> dict[ }, handler=_capture_inspect_tool, ), + "scope.observe": ToolSpec( + name="scope.observe", + description=( + "Read configured scope identity, state, and coupling safety. With fetch_waveform=true, " + "also read waveform summaries, which may change waveform-transfer source/mode/format." + ), + arguments={ + "type": "object", + "properties": { + "channel": {"type": "integer", "minimum": 1}, + "channels": { + "type": "array", + "items": {"type": "integer", "minimum": 1}, + "minItems": 1, + "uniqueItems": True, + }, + "fetch_waveform": {"type": "boolean", "default": False}, + "allow_50ohm": {"type": "boolean", "default": False}, + "expectations": { + "type": "object", + "additionalProperties": { + "type": "object", + "properties": { + "label": {"type": "string"}, + "shape": {"type": "string"}, + "frequency_hz": {"type": "number", "exclusiveMinimum": 0}, + "frequency_tolerance_ratio": {"type": "number", "minimum": 0}, + "vpp_v": {"type": "number", "exclusiveMinimum": 0}, + "vpp_tolerance_ratio": {"type": "number", "minimum": 0}, + "mean_v": {"type": "number"}, + "offset_v": {"type": "number"}, + "mean_tolerance_v": {"type": "number", "minimum": 0}, + "duty_cycle": {"type": "number", "minimum": 0, "maximum": 1}, + "duty_percent": {"type": "number", "minimum": 0, "maximum": 100}, + "duty_tolerance": {"type": "number", "minimum": 0}, + "symmetry_percent": {"type": "number", "minimum": 0, "maximum": 100}, + "symmetry_tolerance_percent": {"type": "number", "minimum": 0}, + }, + "additionalProperties": False, + }, + }, + }, + "additionalProperties": False, + }, + handler=_scope_observe_tool, + read_only=False, + mutates_instrument=True, + instrument_state_effects=( + "fetch_waveform=true may change waveform transfer source/mode/format", + "fetch_waveform=true may enable the requested channel display on some drivers", + ), + ), + "doctor.config": ToolSpec( + name="doctor.config", + description=( + "Run configured-instrument read-only doctor checks and return structured records / " + "对已配置仪器执行只读 doctor 检查并返回结构化结果" + ), + arguments={ + "type": "object", + "properties": { + "timeout_ms": {"type": "integer", "minimum": 1}, + }, + "additionalProperties": False, + }, + handler=_doctor_config_tool, + ), + "scope.advise": ToolSpec( + name="scope.advise", + description=( + "Observe the configured scope and recommend display/acquisition settings without applying " + "recommendations. With fetch_waveform=true, waveform reads may change transfer state." + ), + arguments={ + "type": "object", + "properties": { + "channel": {"type": "integer", "minimum": 1}, + "channels": { + "type": "array", + "items": {"type": "integer", "minimum": 1}, + "minItems": 1, + "uniqueItems": True, + }, + "fetch_waveform": {"type": "boolean", "default": False}, + "allow_50ohm": {"type": "boolean", "default": False}, + "target_cycles": {"type": "number", "exclusiveMinimum": 0, "default": 10}, + "target_vertical_divisions": { + "type": "number", + "exclusiveMinimum": 0, + "default": 5, + }, + "expectations": { + "type": "object", + "additionalProperties": { + "type": "object", + "properties": { + "label": {"type": "string"}, + "shape": {"type": "string"}, + "frequency_hz": {"type": "number", "exclusiveMinimum": 0}, + "frequency_tolerance_ratio": {"type": "number", "minimum": 0}, + "vpp_v": {"type": "number", "exclusiveMinimum": 0}, + "vpp_tolerance_ratio": {"type": "number", "minimum": 0}, + "mean_v": {"type": "number"}, + "offset_v": {"type": "number"}, + "mean_tolerance_v": {"type": "number", "minimum": 0}, + "duty_cycle": {"type": "number", "minimum": 0, "maximum": 1}, + "duty_percent": {"type": "number", "minimum": 0, "maximum": 100}, + "duty_tolerance": {"type": "number", "minimum": 0}, + "symmetry_percent": {"type": "number", "minimum": 0, "maximum": 100}, + "symmetry_tolerance_percent": {"type": "number", "minimum": 0}, + }, + "additionalProperties": False, + }, + }, + }, + "additionalProperties": False, + }, + handler=_scope_advise_tool, + read_only=False, + mutates_instrument=True, + instrument_state_effects=( + "fetch_waveform=true may change waveform transfer source/mode/format", + "fetch_waveform=true may enable the requested channel display on some drivers", + ), + ), } diff --git a/src/wavebench/services/agent_advise.py b/src/wavebench/services/agent_advise.py new file mode 100644 index 0000000..94a09d0 --- /dev/null +++ b/src/wavebench/services/agent_advise.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from wavebench.errors import ConfigError +from wavebench.services.agent_observe import scope_observe_payload + + +def scope_advise_payload( + *, + config_path: str | Path, + channel: int | None = None, + channels: tuple[int, ...] | None = None, + fetch_waveform: bool = False, + allow_50ohm: bool = False, + expectations: dict[int, dict[str, Any]] | None = None, + target_cycles: float = 10.0, + target_vertical_divisions: float = 5.0, +) -> dict[str, Any]: + if target_cycles <= 0: + raise ConfigError("scope.advise target_cycles must be > 0") + if target_vertical_divisions <= 0: + raise ConfigError("scope.advise target_vertical_divisions must be > 0") + observation = scope_observe_payload( + config_path=config_path, + channel=channel, + channels=channels, + fetch_waveform=fetch_waveform, + allow_50ohm=allow_50ohm, + expectations=expectations if fetch_waveform else None, + ) + recommendations = _recommendations( + observation, + expectations=expectations or {}, + target_cycles=float(target_cycles), + target_vertical_divisions=float(target_vertical_divisions), + ) + return { + "status": observation["status"], + "read_only": observation["read_only"], + "query_only": observation.get("query_only", observation["read_only"]), + "mutates_instrument": observation["mutates_instrument"], + "raw_scpi": False, + "applies_recommendations": False, + "instrument_state_effects": observation["instrument_state_effects"], + "observation": { + "channel": observation["observation"]["channel"], + "channels": observation["observation"]["channels"], + "fetch_waveform": observation["observation"]["fetch_waveform"], + }, + "recommendations": recommendations, + "agent_hints": _agent_hints(observation, recommendations), + "warnings": observation["warnings"], + } + + +def _recommendations( + observation: dict[str, Any], + *, + expectations: dict[int, dict[str, Any]], + target_cycles: float, + target_vertical_divisions: float, +) -> list[dict[str, Any]]: + recommendations: list[dict[str, Any]] = [] + channels = observation.get("channels", []) + expected_frequencies = _expected_frequencies(expectations) + channel_profiles: dict[int, dict[str, Any]] = {} + for channel_section in channels: + channel = channel_section.get("channel") + if not isinstance(channel, int): + continue + summary = _waveform_summary(channel_section) + snapshot = _scope_status_data(channel_section) + frequency_hz, source, confidence = _frequency_for_advice( + summary, + expected_frequencies.get(channel), + ) + vertical_scale = _recommended_vertical_scale( + summary, + snapshot, + target_vertical_divisions=target_vertical_divisions, + ) + time_range = ( + _recommended_time_range(frequency_hz, target_cycles=target_cycles) + if frequency_hz is not None + else None + ) + channel_profiles[channel] = { + "channel": channel, + "frequency_hz": frequency_hz, + "frequency_source": source if frequency_hz is not None else None, + "frequency_confidence": confidence, + "time_range_s": time_range, + "vertical_scale_v_per_div": vertical_scale, + } + if snapshot and snapshot.get("channel", {}).get("enabled") is False: + recommendations.append( + _command_recommendation( + "display_on", + "high", + channel, + "Channel display is off; enable it before human visual inspection.", + "display", + {"channel": channel, "state": "on"}, + ) + ) + if time_range is not None or vertical_scale is not None: + reason = _focus_reason( + summary, + frequency_hz, + source, + target_cycles=target_cycles, + ) + priority = "high" if _needs_focus(summary, channel, expected_frequencies) else "normal" + recommendations.append( + _command_recommendation( + "focus_channel", + priority, + channel, + reason, + "focus", + { + "channel": channel, + "time_range_s": time_range, + "vertical_scale_v_per_div": vertical_scale, + "frequency_confidence": confidence, + "hide_other_channels": False, + }, + ) + ) + span = _frequency_span(channel_profiles) + if span is not None and span["ratio_high_over_low"] > 10.0: + recommendations.append( + { + "id": "separate_timebase_profiles", + "priority": "high", + "action": "capture_or_observe_channels_separately", + "reason": ( + "Observed or expected channel frequencies span more than 10x; " + "do not judge every waveform shape on one timebase." + ), + "mutates_instrument_if_applied": False, + "raw_scpi": False, + "frequency_span": span, + "profiles": [ + profile + for _, profile in sorted(channel_profiles.items()) + if profile["time_range_s"] is not None + ], + } + ) + if not recommendations: + recommendations.append( + { + "id": "no_adjustment_needed", + "priority": "low", + "action": "keep_current_scope_settings", + "reason": "No obvious display or acquisition-window issue was found.", + "mutates_instrument_if_applied": False, + "raw_scpi": False, + } + ) + return recommendations + + +def _waveform_summary(channel_section: dict[str, Any]) -> dict[str, Any] | None: + waveform = channel_section.get("waveform", {}) + if waveform.get("status") != "ok": + return None + summary = waveform.get("data", {}).get("summary") + return summary if isinstance(summary, dict) else None + + +def _scope_status_data(channel_section: dict[str, Any]) -> dict[str, Any] | None: + status = channel_section.get("scope_status", {}) + data = status.get("data") + return data if isinstance(data, dict) else None + + +def _summary_frequency(summary: dict[str, Any] | None) -> float | None: + if summary is None: + return None + value = summary.get("frequency_estimate_hz") + if not isinstance(value, (int, float)) or value <= 0: + return None + return float(value) + + +def _summary_frequency_confidence(summary: dict[str, Any] | None) -> str | None: + if summary is None or _summary_frequency(summary) is None: + return None + warnings = summary.get("quality_warnings", []) + if any(str(item).startswith("low_cycle_count") for item in warnings): + return "low" + return "measured" + + +def _frequency_for_advice( + summary: dict[str, Any] | None, + expected_frequency_hz: float | None, +) -> tuple[float | None, str | None, str | None]: + measured = _summary_frequency(summary) + confidence = _summary_frequency_confidence(summary) + if expected_frequency_hz is not None and confidence == "low": + return expected_frequency_hz, "expected", "configured" + if measured is not None: + return measured, "measured", confidence + if expected_frequency_hz is not None: + return expected_frequency_hz, "expected", "configured" + return None, None, None + + +def _expected_frequencies(expectations: dict[int, dict[str, Any]]) -> dict[int, float]: + values: dict[int, float] = {} + for channel, expectation in expectations.items(): + value = expectation.get("frequency_hz") + if isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0: + values[channel] = float(value) + return values + + +def _recommended_time_range(frequency_hz: float, *, target_cycles: float) -> float: + return float(target_cycles / frequency_hz) + + +def _recommended_vertical_scale( + summary: dict[str, Any] | None, + snapshot: dict[str, Any] | None, + *, + target_vertical_divisions: float, +) -> float | None: + vpp = None if summary is None else summary.get("voltage_vpp_v") + if isinstance(vpp, (int, float)) and vpp > 0: + return float(vpp) / target_vertical_divisions + scale = None + if snapshot is not None: + scale = snapshot.get("channel", {}).get("scale_v_per_div") + if isinstance(scale, (int, float)) and scale > 0: + return float(scale) + return None + + +def _needs_focus( + summary: dict[str, Any] | None, + channel: int, + expected_frequencies: dict[int, float], +) -> bool: + if channel in expected_frequencies and summary is None: + return True + if summary is None: + return False + cycles = summary.get("estimated_cycles") + if isinstance(cycles, (int, float)) and (cycles < 5.0 or cycles > 25.0): + return True + points_per_cycle = summary.get("points_per_cycle") + if isinstance(points_per_cycle, (int, float)) and points_per_cycle < 20.0: + return True + quality = summary.get("quality_warnings", []) + return bool(quality) + + +def _focus_reason( + summary: dict[str, Any] | None, + frequency_hz: float | None, + frequency_source: str, + *, + target_cycles: float, +) -> str: + parts: list[str] = [] + if frequency_hz is not None: + confidence_note = ( + " (low-confidence estimate)" + if frequency_source == "measured" and _summary_frequency_confidence(summary) == "low" + else "" + ) + parts.append( + f"use {frequency_source} frequency {frequency_hz:.6g} Hz{confidence_note} " + f"to show about {target_cycles:.3g} cycles" + ) + if summary is not None: + cycles = summary.get("estimated_cycles") + if isinstance(cycles, (int, float)): + parts.append(f"current window contains about {cycles:.3g} cycles") + points = summary.get("points_per_cycle") + if isinstance(points, (int, float)): + parts.append(f"current sampling density is about {points:.3g} points/cycle") + return "; ".join(parts) if parts else "focus the selected channel for visual inspection" + + +def _frequency_span(profiles: dict[int, dict[str, Any]]) -> dict[str, Any] | None: + values = [ + (channel, profile["frequency_hz"]) + for channel, profile in profiles.items() + if isinstance(profile.get("frequency_hz"), (int, float)) and profile["frequency_hz"] > 0 + ] + if len(values) < 2: + return None + low_channel, low = min(values, key=lambda item: item[1]) + high_channel, high = max(values, key=lambda item: item[1]) + return { + "low_channel": low_channel, + "low_hz": low, + "high_channel": high_channel, + "high_hz": high, + "ratio_high_over_low": float(high / low), + } + + +def _command_recommendation( + recommendation_id: str, + priority: str, + channel: int, + reason: str, + command: str, + parameters: dict[str, Any], +) -> dict[str, Any]: + return { + "id": recommendation_id, + "priority": priority, + "channel": channel, + "action": f"scope.{command}", + "reason": reason, + "command": _command_text(command, parameters), + "parameters": parameters, + "mutates_instrument_if_applied": True, + "raw_scpi": False, + } + + +def _command_text(command: str, parameters: dict[str, Any]) -> str: + if command == "display": + return ( + "wavebench scope display " + f"--channel {parameters['channel']} {parameters['state']}" + ) + pieces = ["wavebench", "scope", "focus", "--channel", str(parameters["channel"])] + if parameters.get("time_range_s") is not None: + pieces.extend(["--time-range", f"{parameters['time_range_s']:.12g}"]) + if parameters.get("vertical_scale_v_per_div") is not None: + pieces.extend(["--vertical-scale", f"{parameters['vertical_scale_v_per_div']:.12g}"]) + if parameters.get("hide_other_channels"): + pieces.append("--hide-other-channels") + return " ".join(pieces) + + +def _agent_hints( + observation: dict[str, Any], + recommendations: list[dict[str, Any]], +) -> list[str]: + hints = list(observation.get("agent_hints", [])) + if any(item["id"] == "separate_timebase_profiles" for item in recommendations): + hints.append("advise: run focus/observe per channel when frequencies differ greatly") + if observation.get("mutates_instrument"): + hints.append("advise: waveform fetch was used only to compute advice; recommendations were not applied") + else: + hints.append("advise: recommendations were computed without applying instrument changes") + return hints diff --git a/src/wavebench/services/agent_observe.py b/src/wavebench/services/agent_observe.py new file mode 100644 index 0000000..2c8723d --- /dev/null +++ b/src/wavebench/services/agent_observe.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +from dataclasses import asdict +from pathlib import Path +from typing import Any + +from wavebench.config import load_config +from wavebench.data.expectations import evaluate_waveform_expectation, expectation_summary +from wavebench.data.relationships import analyze_waveform_relationships +from wavebench.errors import ConfigError, WaveBenchError +from wavebench.instruments.models import WaveformData +from wavebench.logging import CommandLogger +from wavebench.services.scope_service import ScopeService + + +def scope_observe_payload( + *, + config_path: str | Path, + channel: int | None = None, + channels: tuple[int, ...] | None = None, + fetch_waveform: bool = False, + allow_50ohm: bool = False, + expectations: dict[int, dict[str, Any]] | None = None, +) -> dict[str, Any]: + config = load_config(config_path) + observed_channels = _scope_channels( + channel=channel, + channels=channels, + default_channel=config.scope.default_channel, + ) + normalized_expectations = _normalize_expectations(expectations) + if normalized_expectations and not fetch_waveform: + raise ConfigError("scope.observe expectations require fetch_waveform=true") + service = ScopeService(config=config, logger=CommandLogger()) + sections: dict[str, Any] = {} + warnings: list[str] = [] + fetched_waveforms: dict[int, WaveformData] = {} + expectation_results: dict[int, dict[str, Any]] = {} + + sections["identity"] = _attempt(lambda: {"idn": service.idn()}, warnings=warnings, name="identity") + channel_sections = [ + _observe_channel( + service, + observed_channel, + fetch_waveform=fetch_waveform, + allow_50ohm=allow_50ohm, + warnings=warnings, + fetched_waveforms=fetched_waveforms, + expectations=normalized_expectations, + expectation_results=expectation_results, + ) + for observed_channel in observed_channels + ] + first_channel = channel_sections[0] + sections["scope_status"] = first_channel["scope_status"] + sections["coupling"] = first_channel["coupling"] + sections["waveform"] = first_channel["waveform"] + + return { + "status": "ok" if not warnings else "partial", + "read_only": not fetch_waveform, + "query_only": not fetch_waveform, + "mutates_instrument": fetch_waveform, + "raw_scpi": False, + "instrument_state_effects": _instrument_state_effects(fetch_waveform), + "config": { + "path": str(config.source_path), + "scope_driver": config.scope.driver, + "resource": config.connection.resource, + "backend": config.connection.backend, + "default_channel": config.scope.default_channel, + "waveform_points": config.waveform.points, + }, + "observation": { + "instrument": "scope", + "channel": observed_channels[0], + "channels": list(observed_channels), + "fetch_waveform": fetch_waveform, + "allow_50ohm": allow_50ohm, + }, + **sections, + "channels": channel_sections, + "relationships": ( + analyze_waveform_relationships(fetched_waveforms, same_acquisition=False) + if len(fetched_waveforms) >= 2 + else [] + ), + "expectations": expectation_summary(expectation_results), + "warnings": warnings, + "agent_hints": _agent_hints( + sections, + warnings, + channel_sections=channel_sections, + fetched_waveforms=fetched_waveforms, + expectation_results=expectation_results, + ), + } + + +def _scope_channels( + *, + channel: int | None, + channels: tuple[int, ...] | None, + default_channel: int, +) -> tuple[int, ...]: + if channel is not None and channels is not None: + raise ConfigError("scope.observe accepts either channel or channels, not both") + candidates = channels if channels is not None else (default_channel if channel is None else channel,) + if not candidates: + raise ConfigError("scope.observe channels must not be empty") + for candidate in candidates: + if isinstance(candidate, bool) or not isinstance(candidate, int) or candidate < 1: + raise ConfigError("scope.observe channel must be a positive integer") + if len(set(candidates)) != len(candidates): + raise ConfigError("scope.observe channels must be unique") + return candidates + + +def _normalize_expectations( + expectations: dict[int, dict[str, Any]] | None, +) -> dict[int, dict[str, Any]]: + if expectations is None: + return {} + normalized: dict[int, dict[str, Any]] = {} + for channel, expectation in expectations.items(): + if isinstance(channel, bool) or not isinstance(channel, int) or channel < 1: + raise ConfigError("scope.observe expectation channel must be a positive integer") + if not isinstance(expectation, dict): + raise ConfigError("scope.observe expectation entries must be objects") + normalized[channel] = dict(expectation) + return normalized + + +def _instrument_state_effects(fetch_waveform: bool) -> list[str]: + if not fetch_waveform: + return [] + return [ + "waveform transfer source/mode/format may be changed", + "some drivers may enable the requested analog channel display before fetching", + ] + + +def _observe_channel( + service: ScopeService, + channel: int, + *, + fetch_waveform: bool, + allow_50ohm: bool, + warnings: list[str], + fetched_waveforms: dict[int, WaveformData], + expectations: dict[int, dict[str, Any]], + expectation_results: dict[int, dict[str, Any]], +) -> dict[str, Any]: + section = { + "channel": channel, + "scope_status": _attempt( + lambda: asdict(service.status(channel=channel)), + warnings=warnings, + name=f"ch{channel}_scope_status", + ), + "coupling": _attempt( + lambda: _coupling_payload(service, channel, allow_50ohm=allow_50ohm), + warnings=warnings, + name=f"ch{channel}_coupling", + ), + } + if fetch_waveform: + section["waveform"] = _attempt( + lambda: _waveform_payload( + service, + channel, + allow_50ohm=allow_50ohm, + fetched_waveforms=fetched_waveforms, + ), + warnings=warnings, + name=f"ch{channel}_waveform", + ) + else: + section["waveform"] = { + "status": "skipped", + "reason": "fetch_waveform=false", + } + if channel in expectations and channel in fetched_waveforms: + result = evaluate_waveform_expectation( + fetched_waveforms[channel], + expectations[channel], + ) + expectation_results[channel] = result + section["expectation"] = { + "status": "ok", + "data": result, + } + elif channel in expectations: + section["expectation"] = { + "status": "unavailable", + "reason": "waveform unavailable", + } + else: + section["expectation"] = { + "status": "skipped", + "reason": "no expectation for channel", + } + return section + + +def _attempt(call, *, warnings: list[str], name: str) -> dict[str, Any]: + try: + return {"status": "ok", "data": call()} + except WaveBenchError as exc: + warnings.append(f"{name}_unavailable: {exc}") + return { + "status": "unavailable", + "error": {"type": type(exc).__name__, "message": str(exc)}, + } + except Exception as exc: + warnings.append(f"{name}_unavailable: {type(exc).__name__}: {exc}") + return { + "status": "unavailable", + "error": {"type": type(exc).__name__, "message": str(exc)}, + } + + +def _coupling_payload( + service: ScopeService, + channel: int, + *, + allow_50ohm: bool, +) -> dict[str, Any]: + coupling = service.require_high_impedance(channel, allow_50ohm=allow_50ohm) + return { + "channel": channel, + "coupling": coupling, + "accepted_for_capture": True, + } + + +def _waveform_payload( + service: ScopeService, + channel: int, + *, + allow_50ohm: bool, + fetched_waveforms: dict[int, WaveformData], +) -> dict[str, Any]: + service.require_high_impedance(channel, allow_50ohm=allow_50ohm) + waveform = service.fetch_waveform(channel=channel) + fetched_waveforms[channel] = waveform + return { + "channel": channel, + "summary": waveform.summary( + expected_frequency_hz=service.config.waveform.expected_frequency_hz, + frequency_tolerance_ratio=service.config.waveform.frequency_tolerance_ratio, + ), + "raw_samples_included": False, + } + + +def _agent_hints( + sections: dict[str, Any], + warnings: list[str], + *, + channel_sections: list[dict[str, Any]], + fetched_waveforms: dict[int, WaveformData], + expectation_results: dict[int, dict[str, Any]], +) -> list[str]: + hints: list[str] = [] + for channel_section in channel_sections: + waveform = channel_section.get("waveform", {}) + if waveform.get("status") != "ok": + continue + channel = channel_section.get("channel") + summary = waveform.get("data", {}).get("summary", {}) + for warning in summary.get("quality_warnings", []) or []: + hints.append(f"CH{channel}_waveform_quality_warning: {warning}") + cycles = summary.get("estimated_cycles") + if isinstance(cycles, (int, float)) and cycles < 5: + hints.append(f"CH{channel}: consider capturing a wider time window for robust periodic analysis") + if len(fetched_waveforms) >= 2: + summaries = [waveform.summary() for waveform in fetched_waveforms.values()] + frequencies = [ + summary.get("frequency_estimate_hz") + for summary in summaries + if isinstance(summary.get("frequency_estimate_hz"), (int, float)) + and not any(str(item).startswith("low_cycle_count") for item in summary.get("quality_warnings", [])) + ] + if len(frequencies) >= 2 and min(frequencies) > 0 and max(frequencies) / min(frequencies) > 10: + hints.append( + "multi_channel_frequency_span_large: use separate time windows/profiles before judging waveform shape across channels" + ) + expected_frequencies = [ + expectation.get("checks", []) + for expectation in expectation_results.values() + ] + frequency_values: list[float] = [] + for checks in expected_frequencies: + for check in checks: + if check.get("metric") == "frequency_hz" and isinstance(check.get("expected"), (int, float)): + frequency_values.append(float(check["expected"])) + if len(frequency_values) >= 2 and min(frequency_values) > 0 and max(frequency_values) / min(frequency_values) > 10: + hints.append( + "expected_multi_channel_frequency_span_large: expectation frequencies span more than 10x; use separate acquisition windows for shape judgments" + ) + for channel, result in sorted(expectation_results.items()): + if result["status"] in {"warn", "fail"}: + hints.append(f"CH{channel}_expectation_{result['status']}: inspect expectation checks") + if sections.get("scope_status", {}).get("status") == "unavailable": + hints.append("driver lacks scope.snapshot or the status query failed; use identity/waveform sections cautiously") + if sections.get("coupling", {}).get("status") == "unavailable": + hints.append("do not run capture until input coupling safety is confirmed") + if warnings: + hints.append("treat this observation as partial and avoid state-changing actions") + return hints diff --git a/tests/test_agent_advise.py b/tests/test_agent_advise.py new file mode 100644 index 0000000..cd9ae39 --- /dev/null +++ b/tests/test_agent_advise.py @@ -0,0 +1,193 @@ +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +import pytest + +from wavebench.errors import ConfigError +from wavebench.services.agent_advise import scope_advise_payload + + +def _write_config(root: Path) -> Path: + path = root / "wavebench.toml" + path.write_text( + """ +[connection] +resource = "TCPIP::scope::INSTR" + +[scope] +driver = "ds1104" +default_channel = 1 +""", + encoding="utf-8", + ) + return path + + +class _NoWaveformFakeScopeService: + def __init__(self, *, config, logger): + self.config = config + + def idn(self): + return "RIGOL TECHNOLOGIES,DS1104Z Plus,123,1.0" + + def require_high_impedance(self, channel, *, allow_50ohm=False): + return "DC" + + +def _observation(*, fetch_waveform: bool = True) -> dict: + return { + "status": "ok", + "read_only": not fetch_waveform, + "query_only": not fetch_waveform, + "mutates_instrument": fetch_waveform, + "raw_scpi": False, + "instrument_state_effects": ["waveform transfer source/mode/format may be changed"] + if fetch_waveform + else [], + "observation": { + "channel": 1, + "channels": [1, 2], + "fetch_waveform": fetch_waveform, + }, + "channels": [ + { + "channel": 1, + "scope_status": { + "status": "ok", + "data": { + "channel": {"enabled": True, "scale_v_per_div": 1.0}, + }, + }, + "waveform": { + "status": "ok", + "data": { + "summary": { + "frequency_estimate_hz": 1000.0, + "estimated_cycles": 2.4, + "points_per_cycle": 500.0, + "voltage_vpp_v": 1.0, + "quality_warnings": ["low_cycle_count: 2.4"], + } + }, + }, + }, + { + "channel": 2, + "scope_status": { + "status": "ok", + "data": { + "channel": {"enabled": True, "scale_v_per_div": 1.0}, + }, + }, + "waveform": { + "status": "ok", + "data": { + "summary": { + "frequency_estimate_hz": 50000.0, + "estimated_cycles": 120.0, + "points_per_cycle": 10.0, + "voltage_vpp_v": 1.0, + "quality_warnings": [], + } + }, + }, + }, + ], + "relationships": [], + "warnings": [], + "agent_hints": [], + } + + +def test_scope_advise_recommends_per_channel_focus_and_separate_timebases(): + with TemporaryDirectory() as tmp: + config = Path(tmp) / "wavebench.toml" + config.write_text("[scope]\n", encoding="utf-8") + with patch( + "wavebench.services.agent_advise.scope_observe_payload", + return_value=_observation(), + ): + payload = scope_advise_payload( + config_path=config, + channels=(1, 2), + fetch_waveform=True, + ) + + assert payload["read_only"] is False + assert payload["query_only"] is False + assert payload["mutates_instrument"] is True + assert payload["applies_recommendations"] is False + focus = [item for item in payload["recommendations"] if item["id"] == "focus_channel"] + assert [item["channel"] for item in focus] == [1, 2] + assert focus[0]["parameters"]["time_range_s"] == pytest.approx(0.01) + assert focus[0]["parameters"]["frequency_confidence"] == "low" + assert focus[0]["parameters"]["vertical_scale_v_per_div"] == pytest.approx(0.2) + assert focus[1]["parameters"]["time_range_s"] == pytest.approx(0.0002) + span = payload["recommendations"][-1] + assert span["id"] == "separate_timebase_profiles" + assert span["frequency_span"]["ratio_high_over_low"] == pytest.approx(50.0) + assert "not applied" in payload["agent_hints"][-1] + + +def test_scope_advise_can_use_expected_frequency_without_waveform_fetch(): + observation = _observation(fetch_waveform=False) + for channel in observation["channels"]: + channel["waveform"] = {"status": "skipped", "reason": "fetch_waveform=false"} + with TemporaryDirectory() as tmp: + config = Path(tmp) / "wavebench.toml" + config.write_text("[scope]\n", encoding="utf-8") + with patch( + "wavebench.services.agent_advise.scope_observe_payload", + return_value=observation, + ) as observe: + payload = scope_advise_payload( + config_path=config, + channels=(1, 2), + fetch_waveform=False, + expectations={ + 1: {"frequency_hz": 1000.0, "vpp_v": 1.0}, + 2: {"frequency_hz": 50000.0, "vpp_v": 1.0}, + }, + ) + + observe.assert_called_once() + assert observe.call_args.kwargs["expectations"] is None + assert payload["mutates_instrument"] is False + assert payload["query_only"] is True + focus = [item for item in payload["recommendations"] if item["id"] == "focus_channel"] + assert focus[0]["parameters"]["time_range_s"] == pytest.approx(0.01) + assert focus[0]["parameters"]["frequency_confidence"] == "configured" + assert focus[1]["parameters"]["time_range_s"] == pytest.approx(0.0002) + assert payload["recommendations"][-1]["id"] == "separate_timebase_profiles" + + +def test_scope_advise_expected_frequency_without_fetch_uses_real_observe_path(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with patch( + "wavebench.services.agent_observe.ScopeService", + _NoWaveformFakeScopeService, + ): + payload = scope_advise_payload( + config_path=config, + channels=(1, 2), + fetch_waveform=False, + expectations={ + 1: {"frequency_hz": 1000.0}, + 2: {"frequency_hz": 50000.0}, + }, + ) + + assert payload["mutates_instrument"] is False + assert payload["query_only"] is True + focus = [item for item in payload["recommendations"] if item["id"] == "focus_channel"] + assert [item["channel"] for item in focus] == [1, 2] + assert focus[0]["parameters"]["time_range_s"] == pytest.approx(0.01) + assert focus[1]["parameters"]["time_range_s"] == pytest.approx(0.0002) + assert payload["recommendations"][-1]["id"] == "separate_timebase_profiles" + + +def test_scope_advise_rejects_invalid_targets(): + with pytest.raises(ConfigError, match="target_cycles"): + scope_advise_payload(config_path="wavebench.toml", target_cycles=0) diff --git a/tests/test_agent_observe.py b/tests/test_agent_observe.py new file mode 100644 index 0000000..f91acfb --- /dev/null +++ b/tests/test_agent_observe.py @@ -0,0 +1,191 @@ +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +import numpy as np +import pytest + +from wavebench.errors import ConfigError +from wavebench.instruments.models import ( + ScopeAnalogChannelSnapshot, + ScopeEdgeTriggerSnapshot, + ScopeHealthSnapshot, + ScopeIdentitySnapshot, + ScopeProbeSnapshot, + ScopeSnapshot, + ScopeTimebaseSnapshot, + ScopeWaveformMetadataSnapshot, + WaveformData, + WaveformHeader, +) +from wavebench.services.agent_observe import scope_observe_payload + + +def _write_config(root: Path) -> Path: + path = root / "wavebench.toml" + path.write_text( + """ +[connection] +resource = "TCPIP::scope::INSTR" + +[scope] +driver = "ds1104" +default_channel = 1 + +[waveform] +points = "def" +""", + encoding="utf-8", + ) + return path + + +def _snapshot(channel: int) -> ScopeSnapshot: + return ScopeSnapshot( + identity=ScopeIdentitySnapshot("RIGOL", "DS1104Z", "123", "1.0", ()), + health=ScopeHealthSnapshot(0, 0, 0, 1, 1, 1_000_000.0, False, False), + channel=ScopeAnalogChannelSnapshot( + channel, + True, + "DC", + 8.0, + 1.0, + 0.0, + 0.0, + None, + "NORM", + 0.0, + "", + False, + False, + "SAMPLE", + ), + timebase=ScopeTimebaseSnapshot(0.001, 12, 0.0, 0.0012, 50.0, 0.0001, False), + probe=ScopeProbeSnapshot(channel, 10.0, None, None, 1_000_000.0, "P10", "PASSIVE"), + waveform=ScopeWaveformMetadataSnapshot( + channel, + -0.0005, + 0.0005, + 1000, + 1, + 1e-6, + -0.0005, + 0.001, + 0.0, + 8, + ), + trigger=ScopeEdgeTriggerSnapshot("EDGE", channel, "AUTO", "POS", "DC", 0.0, "AUTO", "OFF", 1e-6), + ) + + +class _FakeScopeService: + def __init__(self, *, config, logger): + self.config = config + + def idn(self): + return "RIGOL TECHNOLOGIES,DS1104Z Plus,123,1.0" + + def status(self, channel): + return _snapshot(channel) + + def require_high_impedance(self, channel, *, allow_50ohm=False): + return "DC" + + def fetch_waveform(self, channel): + return WaveformData( + channel=channel, + header=WaveformHeader(x_start=0.0, x_stop=0.002, points=5), + voltages_v=np.array([0.0, 1.0, 0.0, -1.0, 0.0]), + ) + + +def test_scope_observe_payload_returns_structured_read_only_context(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with patch("wavebench.services.agent_observe.ScopeService", _FakeScopeService): + payload = scope_observe_payload(config_path=config, channel=2, fetch_waveform=True) + + assert payload["status"] == "ok" + assert payload["read_only"] is False + assert payload["query_only"] is False + assert payload["mutates_instrument"] is True + assert payload["raw_scpi"] is False + assert payload["observation"]["channel"] == 2 + assert payload["observation"]["channels"] == [2] + assert payload["identity"]["data"]["idn"].startswith("RIGOL") + assert payload["scope_status"]["data"]["channel"]["channel"] == 2 + assert payload["coupling"]["data"]["accepted_for_capture"] is True + assert payload["waveform"]["data"]["summary"]["samples"] == 5 + assert payload["waveform"]["data"]["raw_samples_included"] is False + assert payload["channels"][0]["channel"] == 2 + assert payload["instrument_state_effects"] + + +def test_scope_observe_can_skip_waveform_fetch(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with patch("wavebench.services.agent_observe.ScopeService", _FakeScopeService): + payload = scope_observe_payload(config_path=config, fetch_waveform=False) + + assert payload["waveform"]["status"] == "skipped" + assert payload["read_only"] is True + assert payload["query_only"] is True + assert payload["mutates_instrument"] is False + assert payload["instrument_state_effects"] == [] + + +def test_scope_observe_payload_supports_multiple_channels(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with patch("wavebench.services.agent_observe.ScopeService", _FakeScopeService): + payload = scope_observe_payload(config_path=config, channels=(1, 2), fetch_waveform=True) + + assert payload["observation"]["channel"] == 1 + assert payload["observation"]["channels"] == [1, 2] + assert [item["channel"] for item in payload["channels"]] == [1, 2] + assert payload["channels"][1]["waveform"]["data"]["summary"]["channel"] == 2 + assert payload["relationships"][0]["channels"] == [1, 2] + assert payload["relationships"][0]["common_time"]["overlap"] is True + assert payload["relationships"][0]["common_time"]["same_acquisition"] is False + assert payload["relationships"][0]["phase_degrees_at_left_frequency"] is None + + +def test_scope_observe_payload_evaluates_channel_expectations(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with patch("wavebench.services.agent_observe.ScopeService", _FakeScopeService): + payload = scope_observe_payload( + config_path=config, + channel=1, + fetch_waveform=True, + expectations={1: {"vpp_v": 2.0, "vpp_tolerance_ratio": 0.01}}, + ) + + assert payload["expectations"]["status"] == "pass" + assert payload["channels"][0]["expectation"]["data"]["checks"][0]["metric"] == "vpp_v" + + +def test_scope_observe_expectations_require_explicit_waveform_fetch(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with pytest.raises(ConfigError, match="fetch_waveform=true"): + scope_observe_payload( + config_path=config, + channel=1, + fetch_waveform=False, + expectations={1: {"vpp_v": 2.0}}, + ) + + +def test_scope_observe_rejects_invalid_channel(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with pytest.raises(ConfigError, match="positive integer"): + scope_observe_payload(config_path=config, channel=0) + + +def test_scope_observe_rejects_ambiguous_channel_arguments(): + with TemporaryDirectory() as tmp: + config = _write_config(Path(tmp)) + with pytest.raises(ConfigError, match="either channel or channels"): + scope_observe_payload(config_path=config, channel=1, channels=(2,)) diff --git a/tests/test_mcp_http.py b/tests/test_mcp_http.py index 4353832..7aaf0f0 100644 --- a/tests/test_mcp_http.py +++ b/tests/test_mcp_http.py @@ -5,7 +5,9 @@ import urllib.error import urllib.request from pathlib import Path +from types import SimpleNamespace from tempfile import TemporaryDirectory +from unittest.mock import patch from wavebench import __version__ from wavebench.cli import build_parser @@ -162,10 +164,24 @@ def test_tools_lists_only_read_only_mvp_tools(self): self.assertEqual(status, 200) names = {tool["name"] for tool in payload["tools"]} - self.assertEqual(names, {"run.schema", "run.check", "capture.inspect"}) + self.assertEqual( + names, + { + "run.schema", + "run.check", + "capture.inspect", + "scope.observe", + "doctor.config", + "scope.advise", + }, + ) self.assertFalse(any("raw" in name.lower() for name in names)) self.assertFalse(any("output" in name.lower() for name in names)) self.assertFalse(any(name.lower().endswith((".on", ".off")) for name in names)) + by_name = {tool["name"]: tool for tool in payload["tools"]} + self.assertFalse(by_name["scope.observe"]["read_only"]) + self.assertTrue(by_name["scope.observe"]["mutates_instrument"]) + self.assertTrue(by_name["scope.observe"]["instrument_state_effects"]) def test_call_run_schema_succeeds(self): with TemporaryDirectory() as tmp: @@ -335,7 +351,17 @@ def test_mcp_jsonrpc_tools_list_and_call(self): ) self.assertEqual(status, 200) names = {tool["name"] for tool in listed["result"]["tools"]} - self.assertEqual(names, {"run.schema", "run.check", "capture.inspect"}) + self.assertEqual( + names, + { + "run.schema", + "run.check", + "capture.inspect", + "scope.observe", + "doctor.config", + "scope.advise", + }, + ) status, called = self._request( server, @@ -353,6 +379,134 @@ def test_mcp_jsonrpc_tools_list_and_call(self): self.assertEqual(called["result"]["structuredContent"]["status"], "ok") self.assertEqual(called["result"]["content"][0]["type"], "text") + def test_call_scope_observe_succeeds_with_structured_read_only_payload(self): + with TemporaryDirectory() as tmp: + root = Path(tmp) + config = self._write_config(root) + server = self._start_server(config) + + with patch( + "wavebench.mcp_http.scope_observe_payload", + return_value={ + "status": "ok", + "read_only": False, + "mutates_instrument": True, + "raw_scpi": False, + "observation": {"channel": 2, "channels": [2, 3]}, + }, + ) as observe: + status, payload = self._request( + server, + "POST", + "/call", + token="test-token", + body={ + "tool": "scope.observe", + "arguments": { + "channels": [2, 3], + "fetch_waveform": True, + "expectations": {"2": {"frequency_hz": 1000}}, + }, + }, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["result"]["status"], "ok") + self.assertFalse(payload["result"]["read_only"]) + self.assertTrue(payload["result"]["mutates_instrument"]) + observe.assert_called_once() + self.assertIsNone(observe.call_args.kwargs["channel"]) + self.assertEqual(observe.call_args.kwargs["channels"], (2, 3)) + self.assertTrue(observe.call_args.kwargs["fetch_waveform"]) + self.assertEqual(observe.call_args.kwargs["expectations"], {2: {"frequency_hz": 1000}}) + + def test_call_scope_observe_rejects_non_integer_channel(self): + with TemporaryDirectory() as tmp: + server = self._start_server(self._write_config(Path(tmp))) + + with self.assertRaises(urllib.error.HTTPError) as caught: + self._request( + server, + "POST", + "/call", + token="test-token", + body={ + "tool": "scope.observe", + "arguments": {"channel": "1"}, + }, + ) + + self.assertEqual(caught.exception.code, 400) + + def test_call_scope_advise_succeeds_without_applying_recommendations(self): + with TemporaryDirectory() as tmp: + root = Path(tmp) + config = self._write_config(root) + server = self._start_server(config) + + with patch( + "wavebench.mcp_http.scope_advise_payload", + return_value={ + "status": "ok", + "read_only": True, + "mutates_instrument": False, + "raw_scpi": False, + "applies_recommendations": False, + "recommendations": [{"id": "focus_channel"}], + }, + ) as advise: + status, payload = self._request( + server, + "POST", + "/call", + token="test-token", + body={ + "tool": "scope.advise", + "arguments": { + "channels": [1, 2], + "fetch_waveform": False, + "target_cycles": 8, + "expectations": {"1": {"frequency_hz": 1000}}, + }, + }, + ) + + self.assertEqual(status, 200) + self.assertFalse(payload["result"]["applies_recommendations"]) + advise.assert_called_once() + self.assertEqual(advise.call_args.kwargs["channels"], (1, 2)) + self.assertEqual(advise.call_args.kwargs["target_cycles"], 8.0) + self.assertEqual(advise.call_args.kwargs["expectations"], {1: {"frequency_hz": 1000}}) + + def test_call_doctor_config_returns_structured_records(self): + with TemporaryDirectory() as tmp: + root = Path(tmp) + config = self._write_config(root) + server = self._start_server(config) + record = SimpleNamespace( + severity="ok", + target="scope", + driver="ds1104", + resource="TCPIP::scope::INSTR", + idn="RIGOL,DS1104Z,123,1.0", + message="reachable", + suggestion="", + ) + + with patch("wavebench.mcp_http.doctor_records", return_value=[record]): + status, payload = self._request( + server, + "POST", + "/call", + token="test-token", + body={"tool": "doctor.config", "arguments": {"timeout_ms": 1000}}, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["result"]["status"], "ok") + self.assertEqual(payload["result"]["records"][0]["target"], "scope") + self.assertFalse(payload["result"]["mutates_instrument"]) + def test_mcp_jsonrpc_requires_token(self): with TemporaryDirectory() as tmp: server = self._start_server(self._write_config(Path(tmp))) diff --git a/tests/test_waveform_expectations.py b/tests/test_waveform_expectations.py new file mode 100644 index 0000000..7a94c66 --- /dev/null +++ b/tests/test_waveform_expectations.py @@ -0,0 +1,101 @@ +import numpy as np + +from wavebench.data.expectations import ( + estimate_triangle_symmetry_percent, + evaluate_waveform_expectation, + expectation_summary, +) +from wavebench.instruments.models import WaveformData, WaveformHeader + + +def _waveform(channel: int, times: np.ndarray, values: np.ndarray) -> WaveformData: + return WaveformData( + channel=channel, + header=WaveformHeader(x_start=float(times[0]), x_stop=float(times[-1]), points=int(times.size)), + voltages_v=values, + ) + + +def test_square_wave_expectation_passes_frequency_vpp_mean_and_duty(): + times = np.linspace(0.0, 0.009999, 10_000) + values = np.where((times * 1000.0) % 1.0 < 0.5, 0.5, -0.5) + waveform = _waveform(1, times, values) + + result = evaluate_waveform_expectation( + waveform, + { + "label": "1k square", + "shape": "square", + "frequency_hz": 1000, + "frequency_tolerance_ratio": 0.02, + "vpp_v": 1.0, + "vpp_tolerance_ratio": 0.05, + "mean_v": 0.0, + "mean_tolerance_v": 0.02, + "duty_percent": 50, + "duty_tolerance": 0.02, + }, + ) + + assert result["status"] == "pass" + assert {check["metric"] for check in result["checks"]} == { + "frequency_hz", + "vpp_v", + "mean_v", + "duty_cycle", + } + + +def test_triangle_symmetry_expectation_passes_for_asymmetric_ramp(): + times = np.linspace(0.0, 0.0002, 5000) + period = 20e-6 + symmetry = 30.0 + phase = (times % period) / period + values = np.where( + phase < symmetry / 100.0, + -0.5 + phase / (symmetry / 100.0), + 0.5 - (phase - symmetry / 100.0) / (1.0 - symmetry / 100.0), + ) + values += 0.5 + waveform = _waveform(2, times, values) + + measured = estimate_triangle_symmetry_percent(waveform) + result = evaluate_waveform_expectation( + waveform, + { + "label": "50k triangle", + "shape": "triangle", + "frequency_hz": 50_000, + "vpp_v": 1.0, + "mean_v": 0.5, + "symmetry_percent": 30, + "symmetry_tolerance_percent": 3, + }, + ) + + assert measured is not None + assert abs(measured - 30.0) < 3.0 + assert result["status"] == "pass" + + +def test_expectation_warns_instead_of_failing_low_confidence_frequency(): + times = np.linspace(0.0, 0.0005, 200) + values = np.sin(2 * np.pi * 1000 * times) + waveform = _waveform(1, times, values) + + result = evaluate_waveform_expectation(waveform, {"frequency_hz": 1000}) + + assert result["status"] == "warn" + assert result["checks"][0]["status"] == "warn" + assert "low confidence" in result["checks"][0]["message"] + + +def test_expectation_summary_rolls_up_channel_statuses(): + summary = expectation_summary( + { + 1: {"status": "pass"}, + 2: {"status": "warn"}, + } + ) + + assert summary == {"status": "warn", "channels": {"1": "pass", "2": "warn"}} diff --git a/tests/test_waveform_relationships.py b/tests/test_waveform_relationships.py new file mode 100644 index 0000000..21a0ac6 --- /dev/null +++ b/tests/test_waveform_relationships.py @@ -0,0 +1,111 @@ +import numpy as np + +from wavebench.data.relationships import analyze_waveform_pair, analyze_waveform_relationships +from wavebench.instruments.models import WaveformData, WaveformHeader + + +def _waveform(channel: int, values: np.ndarray, *, stop: float = 0.009) -> WaveformData: + return WaveformData( + channel=channel, + header=WaveformHeader(x_start=0.0, x_stop=stop, points=int(values.size)), + voltages_v=values, + ) + + +def test_waveform_pair_reports_frequency_voltage_and_phase_for_related_signals(): + t = np.linspace(0.0, 0.009, 1000) + left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1])) + right = _waveform(2, 0.5 * np.sin(2 * np.pi * 1000 * (t - 0.00025)) + 0.2, stop=float(t[-1])) + + relationship = analyze_waveform_pair(left, right) + + assert relationship["channels"] == [1, 2] + assert relationship["common_time"]["overlap"] is True + assert relationship["frequency"]["ratio_high_over_low"] == 1.0 + assert 0.45 < relationship["voltage"]["vpp_ratio_right_over_left"] < 0.55 + assert 0.19 < relationship["voltage"]["mean_delta_right_minus_left_v"] < 0.21 + assert relationship["correlation"]["max_abs_cross_correlation"] > 0.9 + assert relationship["intersections"]["mode"] == "finite" + assert relationship["intersections"]["count"] > 0 + assert relationship["phase_degrees_at_left_frequency"] is not None + + +def test_waveform_pair_suppresses_phase_when_not_same_acquisition(): + t = np.linspace(0.0, 0.009, 1000) + left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1])) + right = _waveform(2, np.sin(2 * np.pi * 1000 * (t - 0.00025)), stop=float(t[-1])) + + relationship = analyze_waveform_pair(left, right, same_acquisition=False) + + assert relationship["common_time"]["same_acquisition"] is False + assert relationship["phase_degrees_at_left_frequency"] is None + assert "not_same_acquisition_timing_relationships_are_advisory" in relationship["warnings"] + + +def test_waveform_relationships_report_all_pairs_for_four_channels(): + t = np.linspace(0.0, 0.004, 500) + waveforms = { + channel: _waveform(channel, np.sin(2 * np.pi * 1000 * t + channel), stop=float(t[-1])) + for channel in range(1, 5) + } + + relationships = analyze_waveform_relationships(waveforms) + + assert len(relationships) == 6 + assert relationships[0]["channels"] == [1, 2] + assert relationships[-1]["channels"] == [3, 4] + + +def test_waveform_pair_warns_when_frequency_confidence_is_low(): + t = np.linspace(0.0, 0.0005, 100) + left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1])) + right = _waveform(2, np.sin(2 * np.pi * 2000 * t), stop=float(t[-1])) + + relationship = analyze_waveform_pair(left, right) + + assert relationship["frequency"]["left_hz"] is None + assert any("frequency_low_confidence" in warning for warning in relationship["warnings"]) + + +def test_waveform_pair_reports_intersection_points(): + t = np.linspace(0.0, 1.0, 1001) + left = _waveform(1, t - 0.25, stop=float(t[-1])) + right = _waveform(2, np.zeros_like(t), stop=float(t[-1])) + + relationship = analyze_waveform_pair(left, right) + + intersections = relationship["intersections"] + assert intersections["mode"] == "finite" + assert intersections["count"] == 1 + assert intersections["returned"] == 1 + assert intersections["truncated"] is False + assert intersections["points"][0]["time_s"] == 0.25 + assert intersections["points"][0]["voltage_v"] == 0.0 + assert intersections["points"][0]["direction"] == "left_minus_right_rising" + + +def test_waveform_pair_can_truncate_many_intersections(): + t = np.linspace(0.0, 0.01, 2000) + left = _waveform(1, np.sin(2 * np.pi * 1000 * t), stop=float(t[-1])) + right = _waveform(2, np.zeros_like(t), stop=float(t[-1])) + + relationship = analyze_waveform_pair(left, right, max_intersections=3) + + assert relationship["intersections"]["count"] > 3 + assert relationship["intersections"]["returned"] == 3 + assert relationship["intersections"]["truncated"] is True + assert "intersections_truncated" in relationship["warnings"] + + +def test_waveform_pair_marks_coincident_waveforms_as_unbounded_intersections(): + t = np.linspace(0.0, 0.001, 100) + values = np.sin(2 * np.pi * 1000 * t) + + relationship = analyze_waveform_pair( + _waveform(1, values, stop=float(t[-1])), + _waveform(2, values, stop=float(t[-1])), + ) + + assert relationship["intersections"]["mode"] == "coincident" + assert relationship["intersections"]["count"] is None + assert "waveforms_coincident_intersections_unbounded" in relationship["warnings"]