Migrate a Homegrown Python Test Script to OpenHTF

Turn an existing Python test script — a main() with prints, ifs and a CSV writer — into an OpenHTF test in five refactoring steps, keeping behaviour identical while gaining measurements with limits, plugs, a JSON record and an operator UI.

Every hardware team has one: a test.py that grew from a bring-up script. It opens the instruments at the top, runs checks in order with if value < limit: print("FAIL"), and appends a row to a CSV. It works, and nobody wants to touch it. This guide converts such a script incrementally; each step leaves a working test.

Before

test.py (before)

import csv, sys, time
import pyvisa

rm = pyvisa.ResourceManager()
dmm = rm.open_resource("USB0::0x2A8D::0x0101::MY53220001::INSTR")
psu = rm.open_resource("TCPIP0::192.168.1.50::inst0::INSTR")

sn = input("Serial number: ")
results = {"sn": sn, "pass": True}

psu.write("VOLT 12; OUTP ON")
time.sleep(1)

v = float(dmm.query("MEAS:VOLT:DC?"))
print(f"Vout = {v:.3f} V")
if not 4.95 <= v <= 5.05:
    print("FAIL: Vout"); results["pass"] = False
results["vout"] = v

i = float(dmm.query("MEAS:CURR:DC?"))
if i > 0.02:
    print("FAIL: Iq"); results["pass"] = False
results["iq"] = i

psu.write("OUTP OFF")
with open("results.csv", "a", newline="") as f:
    csv.writer(f).writerow([sn, results["pass"], v, i])
print("PASS" if results["pass"] else "FAIL")

Problems this script has and does not know it has: the PSU stays on if dmm.query raises; limits live in if statements no one else can read; the CSV has no units, no limits, no timestamps, no per-phase timing; a failed serial-number scan still writes a row.

Wrap the body in one phase

Keep everything, move it into a phase, let OpenHTF run it. Behaviour is unchanged; you gain a test record and the console banner.

test.py

import openhtf as htf
from openhtf.plugs import user_input

def legacy_body(test):
    ...  # the old code, unchanged, using module-level dmm/psu

def main():
    test = htf.Test(legacy_body)
    test.execute(test_start=user_input.prompt_for_test_start())   # replaces input()

if __name__ == "__main__":
    main()

Move instruments into plugs

The module-level dmm and psu become plugs. tearDown fixes the "PSU left on" bug for free.

instruments.py

import pyvisa
from openhtf.core import base_plugs

class Dmm(base_plugs.BasePlug):
    def __init__(self):
        self._rm = pyvisa.ResourceManager()
        self._inst = self._rm.open_resource("USB0::0x2A8D::0x0101::MY53220001::INSTR")
    def vdc(self): return float(self._inst.query("MEAS:VOLT:DC?"))
    def idc(self): return float(self._inst.query("MEAS:CURR:DC?"))
    def tearDown(self): self._rm.close()

class Psu(base_plugs.BasePlug):
    def __init__(self):
        self._rm = pyvisa.ResourceManager()
        self._inst = self._rm.open_resource("TCPIP0::192.168.1.50::inst0::INSTR")
    def set(self, volts): self._inst.write(f"VOLT {volts}; OUTP ON")
    def off(self): self._inst.write("OUTP OFF")
    def tearDown(self):
        self.off()          # always runs, even after an exception
        self._rm.close()

Addresses hard-coded here move to configuration in a later step; see the PyVISA guide.

Split into phases

One phase per logical step. Each is short, named, timed and individually reported.

test.py

import time
import openhtf as htf
from openhtf.plugs import user_input
from instruments import Dmm, Psu

@htf.plug(psu=Psu)
def power_on(test, psu):
    psu.set(12)
    time.sleep(1)

@htf.plug(dmm=Dmm)
def measure_vout(test, dmm):
    v = dmm.vdc()
    test.logger.info("Vout = %.3f V", v)
    if not 4.95 <= v <= 5.05:
        return htf.PhaseResult.FAIL_AND_CONTINUE

@htf.plug(dmm=Dmm)
def measure_iq(test, dmm):
    if dmm.idc() > 0.02:
        return htf.PhaseResult.FAIL_AND_CONTINUE

@htf.plug(psu=Psu)
def power_off(test, psu):
    psu.off()

FAIL_AND_CONTINUE reproduces the old "record the failure, keep going" behaviour.

Replace `if` limits with measurements

Now the values, limits and units enter the record. The if statements disappear; OpenHTF computes the outcome.

test.py

from openhtf.util import units

@htf.plug(dmm=Dmm)
@htf.measures(htf.Measurement("vout").in_range(4.95, 5.05).with_units(units.VOLT))
def measure_vout(test, dmm):
    test.measurements.vout = dmm.vdc()

@htf.plug(dmm=Dmm)
@htf.measures(htf.Measurement("iq").in_range(maximum=0.020).with_units(units.AMPERE))
def measure_iq(test, dmm):
    test.measurements.iq = dmm.idc()

Run with ConsoleSummary and a failure prints the value and the violated limit — see Output Callbacks.

Guarantee cleanup and replace the CSV

A phase group makes power_off run after any failure in main. The CSV writer becomes OutputToJSON (one file per unit with everything) — and, if you want the CSV for an existing consumer, a ten-line custom callback keeps it.

test.py

import csv
from openhtf.output.callbacks import console_summary, json_factory

def legacy_csv(record):
    """Keep feeding the old spreadsheet while downstream tools migrate."""
    m = {name: meas.measured_value
         for phase in record.phases for name, meas in phase.measurements.items()}
    with open("results.csv", "a", newline="") as f:
        csv.writer(f).writerow([record.dut_id, record.outcome.name, m.get("vout"), m.get("iq")])

def main():
    test = htf.Test(
        htf.PhaseGroup(
            main=[power_on, measure_vout, measure_iq],
            teardown=[power_off],
        ),
        test_name="PSU board EOL",
    )
    test.add_output_callbacks(
        json_factory.OutputToJSON("./records/{dut_id}.{start_time_millis}.json", indent=2),
        console_summary.ConsoleSummary(),
        legacy_csv,
    )
    test.execute(test_start=user_input.prompt_for_test_start())

After

Same instruments, same limits, same CSV — plus: the PSU always turns off; each phase is timed; every value has its limits and unit in a JSON record; a failed scan produces no bogus row; adding the Operator UI is three lines; sending records to a database is one callback.

What usually comes next

Move VISA addresses and limits into a YAML file per station — Configuration.
Add a checkpoint before the slow phases.
Replace time.sleep polling with a monitor where a value must be watched.
Wrap the loop in while True: with the Operator UI so the station runs unit after unit.

Migrate a Homegrown Python Test Script to OpenHTF

Before

Wrap the body in one phase

Move instruments into plugs

Split into phases

Replace `if` limits with measurements

Guarantee cleanup and replace the CSV

After

What usually comes next

Tutorial

Migrate from pytest

Production deployment

On this page