Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions packages/embeddedci-mcp/CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,12 @@
# Changelog — `embeddedci-mcp`

## 2.13.3

- New tools `calibrate_dac` and `calibrate_adc_gain`: the DMM-guided calibration (firmware 3.9.4+).
`calibrate` takes `clear_part` (offset, dac or adc); `calibration` reports the DAC lines and
ADC gains.
- Requires embeddedci 2.13.3.

## 2.13.2

- `calibrate` and `calibration` describe the ADC offset calibration: from firmware 3.9.3 it corrects
Expand Down
8 changes: 5 additions & 3 deletions packages/embeddedci-mcp/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -290,8 +290,10 @@ the pod's second UART, so `uart_open` keeps the console. Needs gateware v48 (`gp
| `dac_output` | Route a DAC output and drive a calibrated DC voltage on it. | **`path`**, `volts` |
| `current_out` | Hold a current on the 4-20 mA output (J9), in amps, or read its range. | `current` |
| `adc_read` | One calibrated reading: front SMA, DAC loopbacks or the 4-20 mA input (J8). | `source=ext` |
| `calibration` | The stored calibration of the 4-20 mA input and the fit it uses. | none |
| `calibrate` | Measure and store the 4-20 mA input's offset, or clear it. | `source=current_in`, `clear=false` |
| `calibration` | The stored calibration: the ADC offset (measured on the 4-20 mA input) and the fit it uses, plus the DMM lines and gains (firmware 3.9.4+). | none |
| `calibrate` | Measure and store the ADC offset on the 4-20 mA input (J8 open), or clear the calibration (all of it, or one part). | `source=current_in`, `clear=false`, `clear_part` |
| `calibrate_dac` | Store this pod's line for a DAC output from two DMM readings of set points driven with `dac_output` (firmware 3.9.4+). | **`path`**, **`code1`**, **`volts1`**, **`code2`**, **`volts2`** |
| `calibrate_adc_gain` | Measure an ADC source's gain through a loopback against a DMM-calibrated output (firmware 3.9.4+). | **`source`** |

### Capture and decode

Expand Down Expand Up @@ -372,7 +374,7 @@ resistors and analog paths) and `benchpod://help` (the server instructions).
`disable_i2c_sensor` and `disable_gps`; a pull conflict to `set_pull(las=[7], enabled=false)`.
- **Capabilities.** `connect` and `status` report what the pod can do as flags, so an agent can
check before it calls a tool: for example `la_pins`, `capture_trigger`, `power_profile`,
`nrst_pin`, `can`, `calibrate`, `current_out`, `pod_current` and `analog` (false on the
`nrst_pin`, `can`, `calibrate`, `cal_dmm`, `current_out`, `pod_current` and `analog` (false on the
digital-only board), plus the pod's policy and cloud settings (`dac_limits`, `flash_kb`,
`ota_sig`, `sig_policy`, `lan_policy`, `tunnel_max_tier`, `lease_state`, `cloud_ca`,
`cloud_proxy`).
Expand Down
4 changes: 2 additions & 2 deletions packages/embeddedci-mcp/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "embeddedci-mcp"
version = "2.13.2"
version = "2.13.3"
description = "MCP server that lets AI agents drive an EmbeddedCI BenchPod hardware-in-the-loop bench"
readme = "README.md"
requires-python = ">=3.10"
Expand All @@ -26,7 +26,7 @@ classifiers = [
dependencies = [
# cloud: the embeddedci:<device> destination; discovery: connect("discover") over mDNS;
# analysis: dominant frequency in capture summaries.
"embeddedci[cloud,discovery,analysis]>=2.13.2,<3",
"embeddedci[cloud,discovery,analysis]>=2.13.3,<3",
"mcp>=1.12,<2",
"anyio>=4",
"pydantic>=2.7",
Expand Down
4 changes: 2 additions & 2 deletions packages/embeddedci-mcp/server.json
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
"name": "io.github.embeddedci-com/embeddedci-mcp",
"title": "EmbeddedCI BenchPod",
"description": "Drive a BenchPod hardware-in-the-loop bench: power, flash, UART, I2C, CAN, analog and logic.",
"version": "2.13.2",
"version": "2.13.3",
"websiteUrl": "https://www.embeddedci.com/docs/benchpod-mcp",
"repository": {
"url": "https://github.com/embeddedci-com/embeddedci-python",
Expand All @@ -15,7 +15,7 @@
"registryType": "pypi",
"registryBaseUrl": "https://pypi.org",
"identifier": "embeddedci-mcp",
"version": "2.13.2",
"version": "2.13.3",
"runtimeHint": "uvx",
"transport": {
"type": "stdio"
Expand Down
17 changes: 17 additions & 0 deletions packages/embeddedci-mcp/src/embeddedci_mcp/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -418,6 +418,13 @@ class AdcReadResult(BaseModel):
"Loop current in amps on the 4-20 mA input (current_in only): 0.004 to 0.020 for a live loop."))


class DacLineResult(BaseModel):
a: float = Field(description="Volts at 16-bit code 0 (volts = a + b * code16).")
b: float = Field(description="Volts per 16-bit code.")
offset: float = Field(description="Offset against the built-in line in volts (at 0 V on 12v, at code 0 on the others).")
gain_ppm: int = Field(description="Gain against the built-in line, ppm.")


class CalibrationResult(BaseModel):
source: str
calibrated: bool = Field(description="True when the pod has its own calibration stored.")
Expand All @@ -427,6 +434,16 @@ class CalibrationResult(BaseModel):
count: Optional[int] = Field(None, description="Mean raw count of the calibration measurement (only right after calibrating).")
span: Optional[int] = Field(None, description="Its peak-to-peak spread in counts.")
samples: Optional[int] = Field(None, description="Samples averaged.")
dac: Dict[str, Optional[DacLineResult]] = Field(default_factory=dict, description=(
"This pod's DMM line per DAC output (3v3, 5v, 12v); null = the built-in line. Empty before firmware 3.9.4."))
adc_gain_ppm: Dict[str, Optional[int]] = Field(default_factory=dict, description=(
"ADC gain correction per source (cal1, cal2, ext) in ppm; null = the built-in gain."))
dmm: bool = Field(False, description="True when any DMM-guided part is stored.")
zero_count: Optional[float] = Field(None, description=(
"The raw zero the offset is derived from: the mean ADC count with J8 open, unwrapped (firmware 3.9.4+)."))
measured: Optional[Dict[str, float]] = Field(None, description=(
"After calibrate_adc_gain: v1, v2 (what the DAC put out), r1, r2 (what the ADC read before the gain) "
"and residual (offset left after it), in volts."))


class AdcCaptureResult(BaseModel):
Expand Down
56 changes: 51 additions & 5 deletions packages/embeddedci-mcp/src/embeddedci_mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,10 @@
from embeddedci.benchpod import (
AdcSource,
AnalogPath,
AdcGainSource,
CalibrateSource,
CalibrationPart,
DacCalPath,
CanMode,
DacOutputPath,
DacPath,
Expand Down Expand Up @@ -1346,15 +1349,20 @@ async def adc_read(source: AdcSource = "ext") -> m.AdcReadResult:


def _calibration_result(c: Any) -> m.CalibrationResult:
dac = {k: (None if v is None else m.DacLineResult(a=v.a, b=v.b, offset=v.offset, gain_ppm=v.gain_ppm))
for k, v in (c.dac or {}).items()}
return m.CalibrationResult(source=c.source, calibrated=c.calibrated, offset=c.offset, a=c.a, b=c.b,
count=c.count, span=c.span, samples=c.samples)
count=c.count, span=c.span, samples=c.samples, dac=dac,
adc_gain_ppm=dict(c.adc_gain_ppm or {}), dmm=c.dmm, measured=c.measured,
zero_count=c.zero_count)


@mcp.tool(annotations=_ann("Read the pod's calibration", read_only=True))
async def calibration() -> m.CalibrationResult:
"""The ADC offset calibration the pod has stored (measured on current_in, terminal J8), and the fit it uses.
"""The calibration the pod has stored: the ADC offset (measured on current_in, terminal J8) and the fit it uses,
and on firmware 3.9.4+ the DMM-guided parts (dac: a line per DAC output, adc_gain_ppm: a gain per ADC source).

calibrated=false means the pod was never calibrated: analog readings can be off by up to about 50 mV.
calibrated=false means the pod has no ADC offset: analog readings can be off by up to about 50 mV.
"""
return _calibration_result(await _call(lambda: SESSION.require().calibration()))

Expand All @@ -1363,6 +1371,8 @@ async def calibration() -> m.CalibrationResult:
async def calibrate(
source: Annotated[CalibrateSource, Field(description="What to calibrate. Only current_in, the 4-20 mA input.")] = "current_in",
clear: Annotated[bool, Field(description="True removes the stored calibration instead of measuring one.")] = False,
clear_part: Annotated[Optional[CalibrationPart], Field(description=(
"With clear=true, remove only this part: offset, dac (the DMM lines) or adc (the ADC gains). Firmware 3.9.4+."))] = None,
) -> m.CalibrationResult:
"""Calibrate the pod's ADC offset on its 4-20 mA input (current_in, terminal J8) and store it on the pod.

Expand All @@ -1371,13 +1381,49 @@ async def calibrate(
has it taken out, and on firmware 3.9.3+ every other ADC source and capture too (they share the front end).
Once per pod is enough: it survives reboots and firmware updates.
The pod refuses, and keeps what it had, when something is driving J8 (more than 150 mV; 50 mV before 3.9.3).
clear=true removes the stored calibration.
clear=true removes the stored calibration (all of it, or clear_part only).
"""
if clear:
return _calibration_result(await _call(lambda: SESSION.require().clear_calibration()))
return _calibration_result(await _call(lambda: SESSION.require().clear_calibration(clear_part)))
return _calibration_result(await _call(lambda: SESSION.require().calibrate(source)))


@mcp.tool(annotations=_ann("Store a DAC output's DMM calibration", destructive=True, idempotent=True))
async def calibrate_dac(
path: Annotated[DacCalPath, Field(description="The DAC output: 3v3, 5v or 12v.")],
code1: Annotated[int, Field(ge=0, le=65535, description="The 16-bit code of the first set point: code16 from dac_output.")],
volts1: Annotated[float, Field(description="What the DMM read at the first set point, in volts.")],
code2: Annotated[int, Field(ge=0, le=65535, description="The 16-bit code of the second set point.")],
volts2: Annotated[float, Field(description="What the DMM read at the second set point, in volts.")],
) -> m.CalibrationResult:
"""Store this pod's own line for a DAC output from two DMM readings (firmware 3.9.4+, cap cal_dmm).

The flow, with the user holding a bench DMM: ask them to disconnect the DUT from the output and put the DMM
(DC volts) on that output's SMA. Drive each set point with dac_output(path, volts) and keep its code16
(3v3: 0.5 and 3.0 V, 5v: 1.0 and 4.0 V, 12v: -5 and +5 V), ask the user what the DMM reads, then call this
with both points and switch the output off (dac_output path off). From then on dac_output, generate and
the DAC output limits use this line. The pod refuses, and keeps what it had, when the codes are too close,
a reading is far from what the output should give (DMM on the wrong SMA) or the gain is more than 2 % off.
"""
return _calibration_result(await _call(
lambda: SESSION.require().calibrate_dac(path, [(code1, volts1), (code2, volts2)])))


@mcp.tool(annotations=_ann("Measure an ADC gain through a loopback", destructive=True, idempotent=True))
async def calibrate_adc_gain(
source: Annotated[AdcGainSource, Field(description=(
"cal1 (and current_in) through the 5 V output, cal2 through the ±12 V output, ext (the front SMA, "
"captures and the control loop) through an SMA cable from the ±12 V output to the ADC input."))],
) -> m.CalibrationResult:
"""Measure and store an ADC source's gain against a DMM-calibrated DAC output (firmware 3.9.4+).

Needs the ADC offset (calibrate) and the driving output's DMM line (calibrate_dac on 5v for cal1, 12v for
cal2 and ext). For ext, ask the user to connect the SMA cable first. The pod drives two set points itself
and switches the outputs off afterwards; measured says what it saw. Nothing to read on a DMM.
"""
return _calibration_result(await _call(lambda: SESSION.require().calibrate_adc_gain(source)))


@mcp.tool(annotations=_ann("Capture an ADC waveform"))
async def capture_adc(
samples: Samples = 4096,
Expand Down
33 changes: 30 additions & 3 deletions packages/embeddedci-mcp/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,13 @@ def __init__(self) -> None:
#: Capabilities the pod reports; drop entries to test a pod without a feature.
self.caps = ["signal", "la", "uart", "dac", "dac_replay", "dac_cotrig",
"la_pins", "gpio_read", "capture_trigger", "power_profile", "calibrate",
"current_out"]
"cal_dmm", "current_out"]
#: The pod's stored `current_in` calibration (the `calibrate` command).
self.current_in_calibrated = False
self.current_in_offset_uv = 0
#: DMM parts: the DAC lines by path and the ADC gains by source (ppm).
self.dac_lines: Dict[str, Any] = {}
self.adc_gains: Dict[str, int] = {}
# -- LA pin ownership (the firmware's table): function, gpio mode and commanded level.
self.function: Dict[int, str] = {la: "none" for la in range(1, 15)}
self.mode: Dict[int, Any] = {la: None for la in range(1, 15)}
Expand Down Expand Up @@ -285,19 +288,43 @@ def _cmd_adc_read(self, req):
return reply

def _cmd_calibrate(self, req):
if req.get("clear"):
measured = None
clear = req.get("clear")
if clear in ("dac", "adc", "offset"):
if clear == "dac":
self.dac_lines = {}
elif clear == "adc":
self.adc_gains = {}
else:
self.current_in_offset_uv, self.current_in_calibrated = 0, False
elif clear:
self.current_in_offset_uv = 0
self.current_in_calibrated = False
self.dac_lines, self.adc_gains = {}, {}
elif "dac" in req:
if self.error:
raise FirmwareError(self.error, cmd="calibrate")
self.dac_lines[req["dac"]] = {"a_uv": -609, "b_pv": 76249841, "offset_uv": -3127, "gain_ppm": -996,
"_req": dict(req)}
elif "adc" in req:
self.adc_gains[req["adc"]] = 1850
measured = {"v1_uv": 995900, "v2_uv": 3992900, "r1_uv": 994100, "r2_uv": 3985500, "residual_uv": 1200}
elif "source" in req:
if self.error:
raise FirmwareError(self.error, cmd="calibrate")
self.current_in_offset_uv = 4356
self.current_in_calibrated = True
reply = {"source": "current_in", "calibrated": self.current_in_calibrated,
"offset_mv": round(self.current_in_offset_uv / 1000), "offset_uv": self.current_in_offset_uv,
"a_uv": 65832397 - self.current_in_offset_uv, "b_nv": -1004471}
"a_uv": 65832397 - self.current_in_offset_uv, "b_nv": -1004471,
"dac": {p: ({k: v for k, v in self.dac_lines[p].items() if k != "_req"} if p in self.dac_lines else None)
for p in ("3v3", "5v", "12v")},
"adc_gain_ppm": {s: self.adc_gains.get(s) for s in ("cal1", "cal2", "ext")},
"dmm": bool(self.dac_lines or self.adc_gains)}
if "source" in req:
reply.update(count=65535, span=6, samples=512)
if measured:
reply["measured"] = measured
return reply

def _cmd_generate(self, req):
Expand Down
20 changes: 19 additions & 1 deletion packages/embeddedci-mcp/tests/test_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
"i2c_sensor_regs", "i2c_sensor_capture", "i2c_sensor_types",
"enable_gps", "set_gps", "disable_gps", "gps_status",
"set_pull", "pull_status", "usb_network", "lan_tls",
"analog_path", "dac_output", "current_out", "adc_read", "calibration", "calibrate",
"analog_path", "dac_output", "current_out", "adc_read", "calibration", "calibrate", "calibrate_dac", "calibrate_adc_gain",
"capture_adc", "capture_la", "capture_correlated", "decode_la", "la_timing",
"generate", "dac_stop", "replay", "list_waveforms", "replay_waveform",
"save_capture_as_recording",
Expand Down Expand Up @@ -478,6 +478,24 @@ def test_calibrate_tools(connected):
assert cleared["calibrated"] is False and cleared["offset"] == 0.0


def test_dmm_calibration_tools(connected):
done = call("calibrate_dac", path="5v", code1=13069, volts1=0.9959, code2=52374, volts2=3.9929)
assert connected.dac_lines["5v"]["_req"] == {"cmd": "calibrate", "dac": "5v", "code1": 13069, "uv1": 995900,
"code2": 52374, "uv2": 3992900}
assert done["dac"]["5v"]["gain_ppm"] == -996 and done["dac"]["3v3"] is None and done["dmm"] is True
assert done["dac"]["5v"]["offset"] == pytest.approx(-3.127e-3)
gain = call("calibrate_adc_gain", source="cal1")
assert gain["adc_gain_ppm"]["cal1"] == 1850 and gain["measured"]["residual"] == pytest.approx(0.0012)
cleared = call("calibrate", clear=True, clear_part="dac")
assert cleared["dac"]["5v"] is None and cleared["adc_gain_ppm"]["cal1"] == 1850


def test_dmm_calibration_needs_newer_firmware(connected):
without_caps(connected, "cal_dmm")
with pytest.raises(Exception, match="cal_dmm"):
call("calibrate_adc_gain", source="cal1")


def test_calibrate_refusal_is_a_tool_error(connected):
connected.error = "calibrate: current_in reads 996 mV, it must be within +/-50 mV of zero. Something is driving J8. Disconnect it and try again"
with pytest.raises(Exception, match="996 mV"):
Expand Down
Loading