Live demo — data resets daily at 03:00 UTC. Nothing you enter is saved. Server UI →

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 with ModuleNotFoundError on a production station:

  1. Declare the dependency in package.json:
    "requirements": {
      "pipPackages": ["requests>=2.32.0"]
    }
    
  2. Vendor the wheels for offline install:
    .\scripts\vendor-wheels.ps1 -PackagePath "path\to\my-package"
    
    Commit the resulting wheels/ folder. The runner installs from it automatically.

During development only you may skip vendoring by setting ALLOW_ONLINE_PIP=true in your environment — this is already configured in the dev docker-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 scatter traces with fill="tozeroy" or fill="tonexty" so all Bode plots share a consistent look.

Pip dependencyplotly must be declared in package.json pipPackages and vendored. matplotlib is 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].

An unhandled error has occurred. Reload 🗙

Rejoining the server...

Rejoin failed... trying again in seconds.

Failed to rejoin.
Please retry or reload the page.

The session has been paused by the server.

Failed to resume the session.
Please retry or reload the page.