Plugs

Plugs Overview (long-lived resources, per-slot status)
  Keysight DAQ        | active
  EtherCAT Manager    | initializing
  Rigol Power Supply  | idle

You can create plugs to manage persistent resources like test equipment or hardware interfaces.

TofuPilot Framework handles plug initialization, lifecycle management, and automatically passes plug instances to your phase functions.

plugs: 
  - name: EtherCAT Manager
    python: hardware.ethercat:EtherCATManager

main:
  - name: Test Motor
    python: phases.motor:test_motor

class EtherCATManager:
    def __init__(self):
        # Initialize connection
        pass

    def send_command(self, cmd):
        # Send command to hardware
        pass

def test_motor(ethercat_manager):
    ethercat_manager.send_command("START")

Plugs are standard Python classes.

Name

You can create a new plug by adding it to the plugs list.

plugs:
  - name: EtherCAT Manager

TofuPilot will automatically trim whitespace and enforce a 100 character limit.

Description

You can add an optional description to explain what the plug is.

plugs:
  - name: EtherCAT Manager
    description: "Manages EtherCAT communication with motors"

TofuPilot will enforce a 50,000 character limit on descriptions.

Key

You can define a key for referencing this plug in your phase function parameters.

plugs:
  - name: EtherCAT Manager
    description: "Manages EtherCAT communication with motors"
    key: ethercat_manager

If not specified, TofuPilot auto-generates a key from the plug name. Keys must be valid Python identifiers: start with a letter or underscore, followed by letters, numbers, or underscores.

Python

You can define plugs as standard Python classes.

class PowerSupply:
    def __init__(self):
        self.address = "192.168.1.100"

    def __del__(self):
        self.write("OUTP OFF")

    def set_voltage(self, volts):
        self.write(f"VOLT {volts}")

    def write(self, command):
        pass  # Actual hardware communication here

And reference the Python class in YAML:

plugs:
  - name: Power Supply
    description: "Programmable power supply"
    python: instruments.power_supply:PowerSupply

TofuPilot will automatically initialize the plug at setup and ensure its destruction at teardown.

A class from an installed package, such as a vendor's driver listed in the procedure's dependencies, is referenced the same way (python: acme_drivers.psu:PSU). When no file of that name exists in the procedure folder, the module is imported from the procedure's venv; tofupilot validate reports it until the venv shows the package (installed, or editable from a uv workspace member or pip install -e).

With

You can pass a with mapping to the plug. Each key becomes a keyword argument on the plug class __init__, so you can keep instrument addresses and settings in the procedure instead of hard-coding them in Python.

plugs:
  - name: Power Supply
    python: instruments.power_supply:PowerSupply
    with: 
      address: "192.168.1.100"

class PowerSupply:
    def __init__(self, address): 
        self.address = address

Values must be JSON-serializable. A key the __init__ does not take, or a required parameter left unset, is reported when the procedure loads; at setup, Python would raise a TypeError.

Values can hold {config.X}, {secrets.X} and {run.*} markers, resolved when the run starts, so one procedure can reach a different instrument on each station:

config:
  PSU_ADDRESS:
    required: true

plugs:
  - name: Power Supply
    python: instruments.power_supply:PowerSupply
    with:
      address: "{config.PSU_ADDRESS}"

A plug can also set environment variables on its process with env:, except a station plug.

Multiple instances

Register the same class more than once, each with its own key and with. Each entry is an independent instance, injected into phases by its key — useful for several instruments of the same model.

name: Power Supply Test

plugs:
  - name: Rigol A
    key: rigol_a
    python: instruments.rigol:RigolDP832
    with:
      address: "192.168.1.100"
  - name: Rigol B
    key: rigol_b
    python: instruments.rigol:RigolDP832
    with:
      address: "192.168.1.101"

main:
  - name: Test Both
    python: phases.test_both

def test_both(rigol_a, rigol_b):
    rigol_a.set_voltage(3.3)
    rigol_b.set_voltage(5.0)

Init timeout

The plug's __init__ has 30 seconds to open the instrument. An instrument that warms up, calibrates or loads firmware when it opens needs more: set init_timeout, a duration up to 1h. Runs and tofupilot plug wait the same.

plugs:
  - key: spectrum_analyzer
    name: Spectrum Analyzer
    python: plugs/analyzer.py:Analyzer
    init_timeout: 2m

Past the timeout the open fails (__init__ did not finish within 2m), as it does when __init__ raises: no main phase runs (each reads SKIP, Not run: and the plug's error), the teardown still runs, and the run ends ERROR (exit code 5). On a full run the scope: execution plugs open before the first phase and the scope: slot ones before the first phase of a slot (after a shared setup), so the SKIP is on every main phase, not only on those that use the plug. The plugs open together, the scope: slot plugs of every slot included (one slot after the other only with strategy: slot_first): once one fails, the others get two seconds to finish opening, then the rest are dropped unopened (plug_status skipped, stage setup), so a failure does not wait out a slow sibling's timeout. A teardown phase that takes a plug that never opened reads SKIP too (Not run: plug psu did not open: and why), unless its parameter has a default (psu=None), which it then receives; the other teardown phases run. A stop (Ctrl-C, SIGTERM) while a plug is still opening ends the open at once: the setup never starts, the teardown runs.

Scope

The scope field sets how long a plug instance lives and which phases share it. Three values, smallest lifetime to largest:

A slot corresponds to one unit under test — and to one run on the dashboard. An execution is one pass of the whole procedure over every slot; a multi-slot execution produces one dashboard run per slot, which is why this scope is named execution and not run.

plugs:
  - name: Power Supply
    python: instruments.power_supply:PowerSupply
    scope: station
    with:
      address: "192.168.1.100"

slot and execution differ on multi-slot procedures: a slot plug is instantiated once per unit under test, an execution plug once for all of them together. On single-slot procedures they behave identically.

station keeps the instrument connection open between executions. The plug __init__ runs once when the first execution needs it; every following execution reuses the same live instance, and teardown happens when the station stops or a new deployment is applied. Use it for connections that are slow to establish (VISA sessions, TCP links, serial ports) so back-to-back units don't pay the reconnection cost.

Station scope requires station mode — the station process owns the instance. In a one-shot tofupilot run, a station plug behaves as execution scope.

The held instance is tied to the deployed procedure version: updating the deployment releases every held instance, and the next execution reconnects using the new bundle — plug code changes always take effect on the first run after a deploy. Within an unchanged deployment, the engine health-checks the plug process before each reuse and respawns it if it died. A change to its resolved with: (a config value updated, a secret rotated), to the procedure's env: or to the station login respawns that plug only, on the next run.

If the connection to the instrument itself drops mid-test, handle the reconnect inside your plug class — the engine guarantees a long-lived plug process, not a live instrument session.

Concurrency

A plug instance runs one method call at a time. When several slots call the same execution or station plug, their calls queue and execute one at a time, like commands on a SCPI socket. Keep plug methods short and put long waits in phases: a method that sleeps for minutes holds every other slot's calls to that plug for as long as it runs.

Setup / Teardown

Single Slot

TofuPilot automatically manages your plug lifecycle during execution:

1. Initialize plugs by calling __init__() method
2. Execute procedure phases
3. Destroy plugs by calling __del__() method

If your procedure specifies Setup or Teardown Phases, TofuPilot initializes plugs before setup and destroys them after teardown.

TofuPilot will stop the execution immediately on plug initialization ERROR, even before executing setup phases, but will always destroy plugs to prevent resource leaks no matter the execution status.

Plug Design

Errors

You can raise exceptions in plugs like you would typically do in Python.

class RigolDP832:
    def set_voltage(self, channel, voltage):
        if voltage > self.max_voltage:
            raise ValueError(f"Voltage {voltage}V exceeds maximum {self.max_voltage}V") 
        cmd = f":SOUR{channel}:VOLT {voltage}\n"
        self.socket.send(cmd.encode())

TofuPilot will automatically capture and report errors.

Logs

You can use print statements for plug-level logging.

class RigolDP832:
    def measure_voltage(self, channel):
        cmd = f":MEAS:VOLT? CH{channel}\n"
        self.socket.send(cmd.encode())
        voltage = self.socket.recv(1024).decode().strip()
        print(f"Measured: {voltage}V") 
        return float(voltage)

TofuPilot will capture and report them.

Phase-Based Management

You can use dedicated setup and teardown phases for complex plug management to leverage TofuPilot native phase execution features like UI prompts, timeouts and more.

plugs:
  - name: Power Supply
    python: instruments.rigol:RigolDP832

setup: 
  - name: Connect Power Supply
    python: setup.connect_psu
    timeout: 30s
    ui: 
      components: 
        - key: status
          type: text
          label: "Connection Status"

teardown:
  - name: Disconnect Power Supply
    python: teardown.disconnect_psu

class RigolDP832:
    def __init__(self):
        # Minimal initialization
        self.socket = None

    def connect(self, ip, port=5555):
        # Complex connection logic called from phase
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.connect((ip, port))
        return True

    def disconnect(self):
        if self.socket:
            self.socket.close()

def connect_psu(power_supply, phase, ui, log):
    ui.status = "Connecting to power supply..."

    if power_supply.connect("192.168.1.50"):
        log.info("Power supply connected successfully")
    else:
        phase.fail("Failed to connect to power supply")

This approach gives you full phase capabilities for plug management operations.

Garbage Collection

TofuPilot guarantees cleanup by calling __del__() at plug deletion and terminating the subprocess, unlike Python's non-deterministic garbage collector.

Method Calling

Plugs are automatically passed to phase functions by matching parameter names (case-insensitive) to plug keys.

def test_motor(ethercat_manager):
    ethercat_manager.send_command("START")  # Call plug methods directly

TofuPilot runs plugs in a separate process from phases and uses a TCP server to handle method calls. When you call ethercat_manager.send_command("START"), TofuPilot serializes the method name and arguments, sends them to the plug process, executes the method, and returns the result.

Benefits:

• Process isolation prevents plug crashes from affecting phases
• Plug resources (sockets, file handles) stay in one process
• Automatic cleanup when plug process terminates

Limitations:

You can only pass serializable arguments (strings, numbers, lists, dicts) to plug methods, and arguments must be passed positionally — keyword arguments are not supported. You cannot pass or return plug instances, functions, or other non-serializable objects:

def measure_voltage(power_supply, measurements):
    # ✅ Pass serializable arguments positionally, get serializable results
    voltage = power_supply.measure_voltage(1)
    measurements.voltage = voltage

    # ❌ Cannot use keyword arguments
    # power_supply.measure_voltage(channel=1)

    # ❌ Cannot pass plug instance to another plug
    # other_device.configure(power_supply)

    # ❌ Cannot return plug instance
    # return power_supply

A plug method can be an async def. It runs to completion in the plug's process, on one event loop that keeps running for the plug's whole life: a connection one call opens is usable by the next, and a reader or keepalive task the driver starts goes on between calls. When the plug closes, the tasks the driver started (a reader, a keepalive) are cancelled, then __del__ runs while the loop still does. Keep the loop when the driver connects (self.loop = asyncio.get_running_loop() in an async def): __del__ runs on another thread, where get_running_loop() fails. From __del__, schedule the close with asyncio.run_coroutine_threadsafe(self.aclose(), self.loop) and do not wait on its result: __del__ may also run on the loop's own thread. A task __del__ scheduled gets 0.5 s to finish, then is cancelled. The phase gets the result as a plain value: call the method without await, from an async def phase too (voltage = psu.aread(1)).

In a phase, a plug is a proxy that calls methods only. Read a property or an attribute of the driver through a method (def get_nplc(self): return self.nplc): psu.nplc without a call gives a function. The proxy keeps call, deadline, _connect, _remaining and the names starting with _tp_ for itself, and never forwards a __dunder__ name. Every other name, name() and address included, reaches the driver, so hasattr(psu, name) is true for any of them.

Example

Let's create a power supply plug to control voltage output:

1. Define the plug in procedure YAML
2. Implement the plug class
3. Use it in a phase

We'll create the procedure file and the Python modules:

We'll define the plug, implement it, and use it in a phase:

name: Power Supply Test

plugs:
  - name: Power Supply
    description: Rigol DP832 programmable power supply
    python: instruments.rigol:RigolDP832

main:
  - name: Set Voltage
    python: phases.set_voltage

import socket

class RigolDP832:
    def __init__(self):
        self.ip = "192.168.1.50"
        self.port = 5555

        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.connect((self.ip, self.port))
        print(f"Connected to power supply at {self.ip}")

    def __del__(self):
        if hasattr(self, 'socket'):
            self.socket.close()
        print("Power supply disconnected")

    def set_voltage(self, channel, voltage):
        cmd = f":SOUR{channel}:VOLT {voltage}\n"
        self.socket.send(cmd.encode())

def set_voltage(power_supply, measurements, log):
    power_supply.set_voltage(1, 5.0)

    measurements.voltage_set = 5.0
    log.info("Set channel 1 to 5.0V")

TofuPilot creates the plug instance when the slot starts, automatically passes it to your phase function by matching the parameter name power_supply to the plug key, and destroys it when the slot completes.