Skip to navigation
StanzaCore ConceptsRoutines

Routines

Routines are Python functions that orchestrate your device measurements. Think of them as recipes - they take a device, perform a sequence of operations, and return results. The @routine decorator handles parameter management, logging, and result storage automatically.

Your First Routine

Let’s write a simple routine to sweep a gate and measure current:

import numpy as np
from stanza.routines import routine
@routine
def pinchoff_sweep(ctx, gate, v_start, v_stop, n_points, contact, session=None):
    """Sweep a gate to find the pinchoff voltage."""
    device = ctx.resources.device
    # Generate voltage points
    voltages = np.linspace(v_start, v_stop, n_points)
    # Sweep and measure (automatically logged if session provided)
    v_data, i_data = device.sweep_1d(
        gate,
        voltages.tolist(),
        contact,
        session=session
    )
    # Find pinchoff (where current drops below threshold)
    threshold = 1e-12  # 1 pA
    pinchoff_idx = np.where(np.abs(i_data) < threshold)[0]
    pinchoff_voltage = v_data[pinchoff_idx[0]] if len(pinchoff_idx) > 0 else None
    return {
        "voltages": v_data,
        "currents": i_data,
        "pinchoff_voltage": pinchoff_voltage
    }

Run it:

from stanza.routines import RoutineRunner
from stanza.models import DeviceConfig
config = DeviceConfig.from_yaml("device.yaml")
runner = RoutineRunner(configs=[config])
result = runner.run(
    "pinchoff_sweep",
    gate="LEFT_BARRIER",
    v_start=-2.0,
    v_stop=0.0,
    n_points=100,
    contact="DRAIN"
)
print(f"Pinchoff voltage: {result['pinchoff_voltage']:.3f} V")

Understanding the Context

The ctx parameter gives you access to everything your routine needs:

Accessing Your Device

@routine
def check_voltages(ctx):
    """Read current voltage on all gates."""
    device = ctx.resources.device
    voltages = {}
    for gate in device.gates:
        voltages[gate] = device.check(gate)
    return {"voltages": voltages}

Using Previous Results

Routines can access results from earlier routines:

@routine
def find_pinchoff(ctx, gate, contact):
    """Find pinchoff voltage."""
    device = ctx.resources.device
    voltages = np.linspace(-2.0, 0.0, 100)
    v_data, i_data = device.sweep_1d(gate, voltages.tolist(), contact)
    # Find where current drops below 1 pA
    threshold_idx = np.where(np.abs(i_data) < 1e-12)[0]
    pinchoff = v_data[threshold_idx[0]] if len(threshold_idx) > 0 else None
    return {"gate": gate, "pinchoff_voltage": pinchoff}
@routine
def set_to_pinchoff(ctx, gate):
    """Set gate to its pinchoff voltage."""
    device = ctx.resources.device
    # Get result from find_pinchoff routine
    pinchoff_result = ctx.results.get("find_pinchoff")
    if pinchoff_result and pinchoff_result["pinchoff_voltage"]:
        voltage = pinchoff_result["pinchoff_voltage"]
        device.jump({gate: voltage})
        return {"status": "success", "voltage": voltage}
    else:
        return {"status": "failed", "reason": "no pinchoff found"}

Run them in sequence:

runner.run("find_pinchoff", gate="LEFT_BARRIER", contact="DRAIN")
runner.run("set_to_pinchoff", gate="LEFT_BARRIER")

Pre-configuring Parameters

Instead of passing parameters every time, configure them in your device YAML:

routines:
  - name: pinchoff_sweep
    parameters:
      gate: LEFT_BARRIER
      v_start: -2.0
      v_stop: 0.0
      n_points: 100
      contact: DRAIN

Now run without arguments:

# Uses parameters from YAML
runner.run("pinchoff_sweep")
# Override specific parameters
runner.run("pinchoff_sweep", n_points=200)

Example: Building a 2D Charge Stability Diagram

@routine
def charge_stability(ctx, gate_x, gate_y, v_x_start, v_x_stop, n_x,
                      v_y_start, v_y_stop, n_y, contact, session=None):
    """Measure a 2D charge stability diagram."""
    device = ctx.resources.device
    # Create voltage arrays
    v_x = np.linspace(v_x_start, v_x_stop, n_x)
    v_y = np.linspace(v_y_start, v_y_stop, n_y)
    # 2D sweep with automatic logging
    v_data, signal = device.sweep_2d(
        gate_x, v_x.tolist(),
        gate_y, v_y.tolist(),
        contact,
        session=session
    )
    # Analyze: find charge transitions
    grad_x = np.gradient(signal, axis=0)
    grad_y = np.gradient(signal, axis=1)
    edges = np.sqrt(grad_x**2 + grad_y**2)
    return {
        "voltages_x": v_data[0],
        "voltages_y": v_data[1],
        "signal": signal,
        "edges": edges
    }

Configure in YAML:

routines:
  - name: charge_stability
    parameters:
      gate_x: LEFT_PLUNGER
      gate_y: RIGHT_PLUNGER
      v_x_start: -2.0
      v_x_stop: -0.5
      n_x: 150
      v_y_start: -2.0
      v_y_stop: -0.5
      n_y: 150
      contact: DRAIN

Run it:

result = runner.run("charge_stability")
# Plot the diagram
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 8))
plt.pcolormesh(
    result["voltages_x"],
    result["voltages_y"],
    result["signal"].T,
    cmap="RdBu_r"
)
plt.xlabel("Left Plunger (V)")
plt.ylabel("Right Plunger (V)")
plt.colorbar(label="Signal")
plt.title("Charge Stability Diagram")
plt.show()

Example: Iterative Tuning with Feedback

Routines can iterate until they find what they’re looking for:

@routine
def optimize_coupling(ctx, barrier_gate, plunger_gates, target_current,
                       max_iterations=20):
    """Iteratively tune barrier gate to achieve target interdot coupling."""
    device = ctx.resources.device
    v_barrier = -1.5  # Starting voltage
    step_size = 0.05  # Initial step
    for iteration in range(max_iterations):
        # Set barrier voltage
        device.jump({barrier_gate: v_barrier})
        # Measure charge diagram to estimate coupling
        v_plunger = np.linspace(-2.0, -0.5, 50)
        v_data, i_data = device.sweep_2d(
            plunger_gates[0], v_plunger.tolist(),
            plunger_gates[1], v_plunger.tolist(),
            "DRAIN"
        )
        # Estimate coupling from gradient
        gradient = np.gradient(i_data)
        coupling_strength = np.max(np.abs(gradient))
        # Check if we hit target
        if np.abs(coupling_strength - target_current) < target_current * 0.1:
            return {
                "converged": True,
                "barrier_voltage": v_barrier,
                "coupling_strength": coupling_strength,
                "iterations": iteration + 1
            }
        # Adjust barrier voltage
        if coupling_strength < target_current:
            v_barrier += step_size  # Increase coupling
        else:
            v_barrier -= step_size  # Decrease coupling
        step_size *= 0.9  # Reduce step size
    return {
        "converged": False,
        "barrier_voltage": v_barrier,
        "coupling_strength": coupling_strength,
        "iterations": max_iterations
    }

Example: Conditional Workflows

Build intelligent workflows that adapt based on measurements:

@routine
def check_leakage(ctx, threshold=1e-9):
    """Check if leakage current is acceptable."""
    device = ctx.resources.device
    current = device.measure("DRAIN")
    passed = np.abs(current) < threshold
    return {"current": current, "passed": passed, "threshold": threshold}
@routine
def conditional_tuneup(ctx):
    """Only proceed with tuneup if leakage is acceptable."""
    # Check if leakage test passed
    leakage_result = ctx.results.get("check_leakage")
    if not leakage_result or not leakage_result["passed"]:
        return {
            "status": "aborted",
            "reason": f"Leakage too high: {leakage_result['current']:.2e} A"
        }
    # Proceed with characterization
    device = ctx.resources.device
    voltages = np.linspace(-2.0, 0.0, 100)
    v_data, i_data = device.sweep_1d("LEFT_BARRIER", voltages.tolist(), "DRAIN")
    return {
        "status": "complete",
        "voltages": v_data,
        "currents": i_data
    }

Run the workflow:

# First check leakage
runner.run("check_leakage", threshold=1e-9)
# Then run conditional tuneup (only proceeds if leakage OK)
result = runner.run("conditional_tuneup")
if result["status"] == "aborted":
    print(f"Tuneup aborted: {result['reason']}")
else:
    print("Tuneup completed successfully")

Organizing Complex Tune-ups

Use nested routines in your YAML to organize multi-step processes:

routines:
  - name: full_characterization
    parameters:
      threshold: 1e-9
    routines:
      - name: check_leakage
        parameters:
          threshold: 1e-9
      - name: find_all_pinchoffs
        parameters:
          gates: [LEFT_BARRIER, CENTER_BARRIER, RIGHT_BARRIER]
          v_start: -2.0
          v_stop: 0.0
          n_points: 100
          contact: DRAIN
      - name: charge_stability
        parameters:
          gate_x: LEFT_PLUNGER
          gate_y: RIGHT_PLUNGER

Run the entire workflow:

# Run all routines under "full_characterization"
results = runner.run_all(parent_routine="full_characterization")
# Access individual results
leakage = results["check_leakage"]
pinchoffs = results["find_all_pinchoffs"]
charge_diagram = results["charge_stability"]

Automatic Data Logging

When you provide a logger to RoutineRunner, data is automatically logged:

from stanza.logger import DataLogger
logger = DataLogger(
    name="logger",
    routine_name="characterization",
    base_dir="./data",
    formats=["hdf5", "jsonl"],
    compression="gzip"
)
runner = RoutineRunner(configs=[config], logger=logger)
# This automatically logs all sweep data
runner.run("charge_stability")
# Data saved to: ./data/characterization/<session_id>/

Your routine receives a session parameter automatically:

@routine
def custom_measurement(ctx, session=None):
    """Routine with custom logging."""
    device = ctx.resources.device
    # Perform measurement
    data = do_something_custom(device)
    # Manually log specific data
    if session:
        session.log_measurement(
            "custom_data",
            data={
                "measured_value": data["value"],
                "timestamp": datetime.now()
            },
            metadata={"algorithm": "custom_v2"}
        )
    return data

Error Handling

Handle errors gracefully to prevent tune-up interruption:

from stanza.exceptions import DeviceError, InstrumentError
@routine
def safe_sweep(ctx, gate, v_start, v_stop, n_points, contact):
    """Sweep with robust error handling."""
    device = ctx.resources.device
    try:
        voltages = np.linspace(v_start, v_stop, n_points)
        v_data, i_data = device.sweep_1d(gate, voltages.tolist(), contact)
        return {
            "status": "success",
            "voltages": v_data,
            "currents": i_data
        }
    except InstrumentError as e:
        # Instrument communication failed
        return {
            "status": "instrument_error",
            "error": str(e),
            "gate": gate
        }
    except DeviceError as e:
        # Device-level error (e.g., voltage out of bounds)
        return {
            "status": "device_error",
            "error": str(e),
            "gate": gate
        }
    finally:
        # Always return to safe voltage
        device.jump({gate: 0.0})

Complete Example: Automated Tuneup Workflow

Here’s a complete example combining multiple routines:

import numpy as np
from stanza.routines import routine, RoutineRunner
from stanza.models import DeviceConfig
from stanza.logger import DataLogger
@routine
def initialize_device(ctx, safe_voltages=None):
    """Initialize device to safe voltages."""
    device = ctx.resources.device
    if safe_voltages is None:
        safe_voltages = {gate: 0.0 for gate in device.gates}
    device.jump(safe_voltages)
    return {"status": "initialized", "voltages": safe_voltages}
@routine
def check_leakage(ctx, contact="DRAIN", threshold=1e-9):
    """Verify leakage is below threshold."""
    device = ctx.resources.device
    current = device.measure(contact)
    passed = np.abs(current) < threshold
    return {
        "current": current,
        "threshold": threshold,
        "passed": passed
    }
@routine
def find_pinchoff(ctx, gate, v_start, v_stop, n_points, contact,
                   threshold=1e-12, session=None):
    """Find pinchoff voltage for a gate."""
    device = ctx.resources.device
    voltages = np.linspace(v_start, v_stop, n_points)
    v_data, i_data = device.sweep_1d(gate, voltages.tolist(), contact, session=session)
    # Find pinchoff
    below_threshold = np.where(np.abs(i_data) < threshold)[0]
    pinchoff_voltage = v_data[below_threshold[0]] if len(below_threshold) > 0 else None
    # Log analysis
    if session:
        session.log_analysis(
            f"{gate}_pinchoff",
            {
                "pinchoff_voltage": pinchoff_voltage,
                "threshold": threshold,
                "gate": gate
            }
        )
    return {
        "gate": gate,
        "pinchoff_voltage": pinchoff_voltage,
        "voltages": v_data,
        "currents": i_data
    }
@routine
def measure_charge_diagram(ctx, gate_x, gate_y, v_x_start, v_x_stop, n_x,
                            v_y_start, v_y_stop, n_y, contact, session=None):
    """Measure 2D charge stability diagram."""
    device = ctx.resources.device
    v_x = np.linspace(v_x_start, v_x_stop, n_x)
    v_y = np.linspace(v_y_start, v_y_stop, n_y)
    v_data, signal = device.sweep_2d(
        gate_x, v_x.tolist(),
        gate_y, v_y.tolist(),
        contact,
        session=session
    )
    return {
        "gate_x": gate_x,
        "gate_y": gate_y,
        "voltages_x": v_data[0],
        "voltages_y": v_data[1],
        "signal": signal
    }
# Main execution
if __name__ == "__main__":
    # Setup
    config = DeviceConfig.from_yaml("device.yaml")
    logger = DataLogger(
        name="logger",
        routine_name="automated_tuneup",
        base_dir="./data",
        formats=["hdf5", "jsonl"]
    )
    runner = RoutineRunner(configs=[config], logger=logger)
    # Execute workflow
    runner.run("initialize_device")
    leakage = runner.run("check_leakage")
    if not leakage["passed"]:
        print(f"Leakage too high: {leakage['current']:.2e} A")
        exit(1)
    # Find pinchoffs for all barriers
    for gate in ["LEFT_BARRIER", "CENTER_BARRIER", "RIGHT_BARRIER"]:
        result = runner.run(
            "find_pinchoff",
            gate=gate,
            v_start=-2.0,
            v_stop=0.0,
            n_points=100,
            contact="DRAIN"
        )
        print(f"{gate} pinchoff: {result['pinchoff_voltage']:.3f} V")
    # Measure charge stability diagram
    diagram = runner.run("measure_charge_diagram")
    print(f"Data saved to: ./data/automated_tuneup/")

Tips and Best Practices

Always return dictionaries: Makes results easy to access and extend later.

Use type hints: Helps with IDE autocomplete and documentation:

from typing import Dict, List, Optional
@routine
def my_routine(ctx, gate: str, voltages: List[float]) -> Dict[str, any]:
    ...

Document your routines: Write clear docstrings explaining what the routine does and what parameters it expects.

Keep routines focused: Each routine should do one thing well. Compose complex workflows from simple routines.

Check for previous results: Always check if ctx.results.get() returns None before using results.

Use finally blocks: Ensure cleanup code runs even if errors occur:

try:
    # measurement code
finally:
    device.jump(safe_voltages)  # Always return to safe state

Test incrementally: Test each routine individually before chaining them together.

Version your code: Keep routines in git and document changes.

Next Steps