product: maestro audience: test-developer authority: normative
Python Runner SDK
⚠️ Before you write a single line of Python — pip packages
If your runner imports any third-party package (e.g.
numpy,pyserial,requests), you must do both of the following or the runner will fail withModuleNotFoundErroron a production station:
- Declare the dependency in
package.json:"requirements": { "pipPackages": ["requests>=2.32.0"] }- Vendor the wheels for offline install:
Commit the resulting.\scripts\vendor-wheels.ps1 -PackagePath "path\to\my-package"wheels/folder. The runner installs from it automatically.During development only you may skip vendoring by setting
ALLOW_ONLINE_PIP=truein your environment — this is already configured in the devdocker-compose.yml.Full details:
sdk/python-dependencies.md
Module setup
Place .py files in the package python_modules/ directory. The runner adds
python_modules/ to sys.path automatically. Each module IS reloaded once per test
execution — file edits take effect between executions without restarting the runner.
Module-level state (instrument handles set in a connect() step) IS preserved across
all steps within the same execution.
Parameter passing
All YAML parameters: values arrive as str keyword arguments. Convert explicitly
inside the function.
# YAML: parameters: { sensor_id: "1", gain: "2.5" }
def read_temperature(sensor_id: str, gain: str) -> dict:
temp = hardware_read(int(sensor_id), float(gain))
return {"temperature": temp}
cfg.*, exec.*, and mes.* template values ARE resolved by the engine before
the function is called. The function receives already-resolved strings as normal
keyword arguments — there IS NO special cfg or exec object in Python code.
parameters:
address: "{{cfg.DMM_VISA}}" # resolved before Python sees it
serial_number: "{{exec.serial_number}}"
def connect(address: str, serial_number: str) -> None:
print(f"Connecting to {address} for DUT {serial_number}")
Return types
| Return value | Effect |
|---|---|
dict |
Keys become output variables |
None / no return |
Success, no outputs |
Any scalar (int, float, str, …) |
Stored as output variable result |
Measurements — YAML-defined limits only
The Python runner DOES NOT support returning MeasurementPointBase objects.
Measurements and limits MUST always be defined in the YAML measurement: /
measurements: block.
def measure_voltage(channel: str) -> dict:
return {"voltage": read_channel(int(channel))}
- name: "Measure 3V3 Rail"
runner: python
runner_type: python3.11
module: "instruments"
function: "measure_voltage"
parameters:
channel: "1"
outputs:
rail_3v3: "{{voltage}}"
measurement:
name: "RAIL_3V3"
value: "{{voltage}}"
low_limit: 3.135
high_limit: 3.465
unit: "V"
⚠️ Anti-pattern — never perform limit checking inside Python:
# WRONG — do not do this def measure_voltage(channel: str) -> dict: v = read_channel(int(channel)) if v < 3.135 or v > 3.465: raise ValueError(f"Voltage {v} V out of range") # limit logic belongs in YAML return {"voltage": v}This approach bypasses the Maestro measurement engine entirely. Limits defined this way are invisible to the test report, yield statistics, Cp/Cpk analysis, and the production dashboard. Always return the raw measured value and declare the limits in the YAML
measurement:block.
Step-splitting guidance
When a single function returns more than ~10 measurements, split along logical boundaries. Each sub-step has its own named verdict in the results tree.
.NET vs Python comparison
| Capability | .NET | Python |
|---|---|---|
| Parameter types | Auto-converted to declared C# type | Always str — convert manually |
| Return measurement points | ✅ MeasurementPointBase |
❌ Use YAML measurement: |
| Return output variables | Dictionary<string, object> |
dict |
| Scalar return | result |
result |
| Async support | ✅ async Task<T> |
❌ synchronous only |
| Limits defined in | Code or YAML | YAML only |
| Module hot-reload | ❌ Requires runner restart | ✅ Once per execution |
Accordion Python API (accordionq2)
The accordionq2 package mirrors the .NET AccordionQ2.WebApiClient API with
Pythonic naming and synchronous (blocking) calls — no .Wait() or .Result needed.
For the full Python API reference see ../Accordion/05-api-python/.
Package name in pipPackages: "accordionq2" (not "accordionq2-client").
Import: from accordionq2 import AccordionQ2Client.
Minimal connect/disconnect pattern:
from accordionq2 import AccordionQ2Client
_client = None
def connect(host: str) -> None:
global _client
_client = AccordionQ2Client(host)
_client.application.reset()
def disconnect() -> None:
global _client
if _client:
_client.close()
_client = None
Emitting Artifacts
A step can attach files (images, Plotly charts, raw data) to the execution report so they are visible in the Maestro UI without the operator having to download anything.
emit_artifact(path, description, content_type=None)
| Parameter | Type | Description |
|---|---|---|
path |
str |
Absolute path to the file to attach |
description |
str |
Human-readable label shown in the report |
content_type |
str \| None |
MIME type. Inferred from extension if omitted. |
The function POSTs the file to the API under the current execution and step. Artifacts are shown inline (images and Plotly) or as download links in the test report.
Inline image example:
import os
import matplotlib.pyplot as plt
def plot_bode(frequencies: str, gains: str) -> None:
freq = [float(f) for f in frequencies.split(",")]
gain = [float(g) for g in gains.split(",")]
fig, ax = plt.subplots()
ax.semilogx(freq, gain)
ax.set(xlabel="Frequency (Hz)", ylabel="Gain (dB)", title="Bode Plot")
out = "/tmp/bode.png"
fig.savefig(out)
plt.close(fig)
emit_artifact(out, "Bode plot")
The PNG is rendered inline as a thumbnail in the step row of the test report. Clicking it opens the full-size image.
emit_plotly(fig, description)
Convenience wrapper for plotly.graph_objects.Figure objects. Serialises the figure
via fig.to_json() and emits it with content_type="application/vnd.plotly.v1+json".
The Maestro UI renders it as a fully interactive Plotly chart (zoom, hover, export).
from plotly.graph_objects import Figure, Scatter
def bode_interactive(frequencies: str, gains: str) -> None:
freq = [float(f) for f in frequencies.split(",")]
gain = [float(g) for g in gains.split(",")]
fig = Figure()
fig.add_trace(Scatter(x=freq, y=gain, mode="lines", name="Gain (dB)"))
# Limit band — filled region between two boundary traces
fig.add_trace(Scatter(x=freq, y=[-3]*len(freq), mode="lines",
line=dict(width=0), showlegend=False))
fig.add_trace(Scatter(x=freq, y=[3]*len(freq), mode="lines",
fill="tonexty", fillcolor="rgba(255,0,0,0.15)",
line=dict(width=0), name="Limit band"))
fig.update_layout(xaxis_type="log", xaxis_title="Hz", yaxis_title="dB")
emit_plotly(fig, "Bode plot (interactive)")
Limit band convention — encode limits as filled
scattertraces withfill="tozeroy"orfill="tonexty"so all Bode plots share a consistent look.
Pip dependency —
plotlymust be declared inpackage.jsonpipPackagesand vendored.matplotlibis separate and optional.
Error signalling
Raise any exception → ABORTED verdict. The full traceback IS stored in the step log.
Returning None DOES NOT fail the step — it means "success with no outputs".
Logging
import sys
print("Connecting to", address) # Information
print("[DEBUG] Raw bytes:", raw) # Debug
print("[WARNING] Retrying...", file=sys.stderr) # Warning
Supported prefixes: [INFO], [DEBUG], [WARNING] / [WARN], [ERROR] / [ERR], [TRACE].