feat: 6 new MCP analysis tools (Workstream C)

Expose RFlect's existing deterministic analysis math as MCP tools so an agent
can drive the full companion-plotter workflow. All reuse tested functions in
plot_antenna (calculations / uwb_analysis / analysis_engine / file_utils);
none reimplement math. Each returns a structured dict and never raises.

New tools (MCP count 35 -> 41):
- compare_antennas (comparison_tools): cross-measurement overlay — per-freq
  peak gain/TRP, deltas vs a reference, best-per-freq, best-overall, optional
  CSV. Closes the long-standing cross-file comparison request.
- analyze_s11 (vna_tools): return loss / impedance bandwidth / VSWR /
  resonance from an S11 sweep.
- analyze_group_delay (vna_tools): group delay + in-band flatness from a
  transmission-phase sweep (phase-only input).
- estimate_link_budget (propagation_tools): max range + link margin from
  measured gains via Friis / log-distance / ITU-indoor, with optional
  Rayleigh/Rician fade margin and de-rated reliable range.
- analyze_mimo_diversity (mimo_tools): ECC -> Vaughan-Andersen diversity gain
  + 2x2 capacity + isolation rating + optional capacity-vs-SNR curve.
- generate_active_cal (calibration_tools): scriptable active chamber cal-file
  generation (also records into cal-drift history).

12 new tests cover happy paths + error/validation paths for every tool.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: RFingAdam

rflect-mcp/server.py

@@ -33,6 +33,11 @@
 from tools.cal_drift_tools import register_cal_drift_tools
 from tools.orchestration import register_orchestration_tools
 from tools.iperf_angle_tools import register_iperf_angle_tools
+from tools.comparison_tools import register_comparison_tools
+from tools.vna_tools import register_vna_tools
+from tools.propagation_tools import register_propagation_tools
+from tools.mimo_tools import register_mimo_tools
+from tools.calibration_tools import register_calibration_tools
 
 # Create MCP server
 mcp = FastMCP("rflect")
@@ -46,6 +51,11 @@
 register_cal_drift_tools(mcp)
 register_orchestration_tools(mcp)
 register_iperf_angle_tools(mcp)
+register_comparison_tools(mcp)
+register_vna_tools(mcp)
+register_propagation_tools(mcp)
+register_mimo_tools(mcp)
+register_calibration_tools(mcp)
 
 
 @mcp.resource("rflect://status")

rflect-mcp/tools/calibration_tools.py

@@ -0,0 +1,104 @@
+"""
+Active chamber calibration MCP tool for RFlect.
+
+Scriptable wrapper over plot_antenna.file_utils.generate_active_cal_file so the
+chamber active-calibration workflow can be driven by an MCP agent. The
+underlying routine also records the run into the cal-drift history, so this
+pairs with the cal_drift_* tools. No LLM, no network. Returns a structured
+dict and never raises; failures populate `warnings`.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, List
+
+from plot_antenna.file_utils import generate_active_cal_file
+
+
+def register_calibration_tools(mcp):
+    """Register active-calibration tools with the MCP server."""
+
+    @mcp.tool()
+    def generate_active_cal(
+        power_measurement_file: str,
+        gain_standard_file: str,
+        hpol_file: str,
+        vpol_file: str,
+        freq_list: List[float],
+        cable_loss: float = 0.0,
+    ) -> Dict[str, Any]:
+        """
+        Generate an active chamber calibration file from reference-antenna data.
+
+        Computes the per-frequency path-loss calibration (P + G − HPol and
+        P + G − VPol) from a power-measurement file, a gain-standard file, and the
+        reference antenna's HPol/VPol scans, writing a TRP cal file + summary. The
+        run is automatically recorded into the cal-drift history (see the
+        cal_drift_* tools to compare against prior calibrations).
+
+        Note: the gain-standard reference is rotated 90 deg between polarizations,
+        so HPol/VPol angle headers legitimately differ — this routine accounts for
+        that (it does not require matching angle grids).
+
+        Args:
+            power_measurement_file: Path to the power-measurement file.
+            gain_standard_file: Path to the gain-standard (reference antenna) file.
+            hpol_file: Path to the reference HPol data file.
+            vpol_file: Path to the reference VPol data file.
+            freq_list: Frequencies (MHz) to include in the calibration.
+            cable_loss: Cable loss (dB); reserved for future use.
+
+        Returns:
+            Dict: output_path, summary_path, rows_written, rows_missing. Never
+            raises; missing files / generation errors populate `warnings`.
+        """
+        result: Dict[str, Any] = {
+            "output_path": None,
+            "summary_path": None,
+            "rows_written": None,
+            "rows_missing": None,
+            "warnings": [],
+        }
+
+        for label, path in (
+            ("power_measurement_file", power_measurement_file),
+            ("gain_standard_file", gain_standard_file),
+            ("hpol_file", hpol_file),
+            ("vpol_file", vpol_file),
+        ):
+            if not path or not os.path.isfile(path):
+                result["warnings"].append(f"file_not_found: {label}={path!r}")
+        if result["warnings"]:
+            return result
+        if not freq_list:
+            result["warnings"].append("empty_freq_list")
+            return result
+
+        try:
+            cal = generate_active_cal_file(
+                power_measurement_file=power_measurement_file,
+                gain_standard_file=gain_standard_file,
+                hpol_file=hpol_file,
+                vpol_file=vpol_file,
+                cable_loss=cable_loss,
+                freq_list=list(freq_list),
+            )
+        except Exception as exc:
+            result["warnings"].append(f"generate_active_cal_failed: {exc}")
+            return result
+
+        if isinstance(cal, dict):
+            result.update(
+                {
+                    "output_path": cal.get("output_path"),
+                    "summary_path": cal.get("summary_path"),
+                    "rows_written": cal.get("rows_written"),
+                    "rows_missing": cal.get("rows_missing"),
+                }
+            )
+            if cal.get("rows_missing"):
+                result["warnings"].append(
+                    f"{cal['rows_missing']} frequencies had no data and were skipped"
+                )
+        return result

rflect-mcp/tools/comparison_tools.py

@@ -0,0 +1,166 @@
+"""
+Cross-measurement comparison MCP tool for RFlect.
+
+Overlays N loaded measurements and reports a per-frequency comparison of peak
+gain / TRP with deltas versus a reference and the best performer per frequency.
+Reuses the deterministic AntennaAnalyzer; no LLM, no network. Returns a
+structured dict and never raises; failures populate `warnings`.
+
+Closes the long-standing request for a cross-file overlay comparison tool.
+"""
+
+from __future__ import annotations
+
+import csv
+import os
+from typing import Any, Dict, List, Optional
+
+from plot_antenna.analysis_engine import AntennaAnalyzer
+
+from .import_tools import get_loaded_measurements
+
+
+def _peak_metric(stats: Dict[str, Any], scan_type: str):
+    """Return (value, unit) of the headline 'peak' metric for a scan type."""
+    if scan_type == "active":
+        return stats.get("max_power_dBm"), "dBm"
+    return stats.get("max_gain_dBi"), "dBi"
+
+
+def register_comparison_tools(mcp):
+    """Register cross-measurement comparison tools with the MCP server."""
+
+    @mcp.tool()
+    def compare_antennas(
+        measurement_names: Optional[List[str]] = None,
+        reference: Optional[str] = None,
+        out_csv: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """
+        Compare loaded antenna measurements head-to-head across frequency.
+
+        For every measurement, reports the peak metric (gain dBi for passive,
+        TRP/peak power dBm for active) at each common frequency, the delta versus
+        a reference measurement, and the best performer per frequency. Import the
+        measurements first (import_antenna_file / import_passive_pair /
+        import_antenna_folder), then call this to overlay them.
+
+        Args:
+            measurement_names: Names to compare (default: all loaded). Must share
+                a scan type for the metric to be apples-to-apples.
+            reference: Measurement to compute deltas against (default: the first).
+            out_csv: Optional path to also write the comparison table as CSV.
+
+        Returns:
+            Dict: metric, unit, measurements, frequencies, rows (per-frequency:
+            {freq_mhz, values:{name:peak}, best:{name,value}, deltas_vs_ref}),
+            best_overall (name with the highest mean peak), out_csv. Never raises;
+            failures populate `warnings`.
+        """
+        result: Dict[str, Any] = {
+            "metric": None,
+            "unit": None,
+            "measurements": [],
+            "frequencies": [],
+            "rows": [],
+            "best_overall": None,
+            "out_csv": None,
+            "warnings": [],
+        }
+
+        loaded = get_loaded_measurements()
+        if not loaded:
+            result["warnings"].append("no_data_loaded")
+            return result
+
+        names = measurement_names or list(loaded.keys())
+        names = [n for n in names if n in loaded]
+        missing = [n for n in (measurement_names or []) if n not in loaded]
+        for n in missing:
+            result["warnings"].append(f"measurement_not_found: {n}")
+        if len(names) < 2:
+            result["warnings"].append("need_at_least_2_measurements_to_compare")
+            return result
+
+        scan_types = {loaded[n].scan_type for n in names}
+        if len(scan_types) > 1:
+            result["warnings"].append(
+                f"mixed_scan_types: {sorted(scan_types)} — peak metric not comparable"
+            )
+        scan_type = loaded[names[0]].scan_type
+        result["metric"] = "peak_power" if scan_type == "active" else "peak_gain"
+        result["unit"] = "dBm" if scan_type == "active" else "dBi"
+        result["measurements"] = names
+
+        ref = reference if reference in names else names[0]
+        if reference and reference not in names:
+            result["warnings"].append(f"reference_not_in_set: {reference}; using {ref}")
+
+        # Build analyzers + the union of frequencies (rounded to MHz).
+        analyzers: Dict[str, AntennaAnalyzer] = {}
+        for n in names:
+            m = loaded[n]
+            try:
+                analyzers[n] = AntennaAnalyzer(
+                    measurement_data=m.data, scan_type=m.scan_type, frequencies=m.frequencies
+                )
+            except Exception as exc:
+                result["warnings"].append(f"analyzer_failed:{n}: {exc}")
+
+        all_freqs = sorted({round(float(fq), 1) for n in names for fq in loaded[n].frequencies})
+        result["frequencies"] = all_freqs
+
+        sums: Dict[str, float] = {n: 0.0 for n in names}
+        counts: Dict[str, int] = {n: 0 for n in names}
+
+        for fq in all_freqs:
+            values: Dict[str, float] = {}
+            for n in names:
+                az = analyzers.get(n)
+                if az is None:
+                    continue
+                try:
+                    stats = az.get_gain_statistics(frequency=fq)
+                    val, _ = _peak_metric(stats, scan_type)
+                    if val is not None:
+                        values[n] = float(val)
+                        sums[n] += float(val)
+                        counts[n] += 1
+                except Exception as exc:
+                    result["warnings"].append(f"stats_failed:{n}@{fq}: {exc}")
+            if not values:
+                continue
+            best_name = max(values, key=values.get)
+            ref_val = values.get(ref)
+            deltas = (
+                {k: round(v - ref_val, 2) for k, v in values.items()} if ref_val is not None else {}
+            )
+            result["rows"].append(
+                {
+                    "freq_mhz": fq,
+                    "values": {k: round(v, 2) for k, v in values.items()},
+                    "best": {"name": best_name, "value": round(values[best_name], 2)},
+                    "deltas_vs_ref": deltas,
+                }
+            )
+
+        means = {n: (sums[n] / counts[n]) for n in names if counts[n] > 0}
+        if means:
+            result["best_overall"] = max(means, key=means.get)
+            result["mean_peak_by_measurement"] = {k: round(v, 2) for k, v in means.items()}
+
+        if out_csv:
+            try:
+                with open(out_csv, "w", newline="", encoding="utf-8") as fh:
+                    w = csv.writer(fh)
+                    w.writerow(["freq_mhz"] + names + ["best", "best_value"])
+                    for row in result["rows"]:
+                        vals = [row["values"].get(n, "") for n in names]
+                        w.writerow(
+                            [row["freq_mhz"]] + vals + [row["best"]["name"], row["best"]["value"]]
+                        )
+                result["out_csv"] = out_csv
+            except (OSError, UnicodeError) as exc:
+                result["warnings"].append(f"csv_write_failed: {exc}")
+
+        return result