Most hardware teams write a test sequencer at some point. It starts as a list of functions and a for loop, and a year later it has retry logic, an operator prompt, a CSV writer and a threading bug nobody wants to touch. This guide covers what a sequencer actually has to do, how to build one if you need to, and how to recognise when you have rebuilt something that already exists.
What a Sequencer Is Responsible For
| Responsibility | What it means in practice |
|---|---|
| Ordering | Steps run in a defined order, with dependencies between them |
| Limit evaluation | Pass/fail derived from a measurement and its limits, not hand-written per step |
| Result capture | The value, its units and its limits are recorded, not just a verdict |
| Unit identity | Every result is attached to a serial number |
| Resource lifecycle | Instruments open once, are shared, and close even when a step raises |
| Failure policy | Some failures stop the run; others continue for diagnostics |
| Retry | Flaky steps retry with a bounded count |
| Timeouts | A hung step fails instead of blocking the line forever |
| Operator interaction | Prompts, scans, confirmations |
| Persistence | Results survive the shift and the machine |
The first three are easy. The rest is where hand-rolled sequencers accumulate their complexity.
The Naive Version
def run_tests(serial: str): results = {} dmm = open_dmm() try: v = dmm.read_voltage() results["rail_3v3"] = (v, 3.2 <= v <= 3.4) i = dmm.read_current() results["idle_current"] = (i, i <= 0.045) finally: dmm.close() passed = all(ok for _, ok in results.values()) print(f"{serial}: {'PASS' if passed else 'FAIL'}") return passedThis is fine for a bench experiment and wrong for production in ways that are not obvious yet:
- The limits are embedded in comparisons, so they cannot be reported, reviewed, or trended.
- There is no record beyond a printed line.
- Every new step duplicates the same tuple-and-compare pattern.
- Adding retry or timeout means restructuring every step.
A Reasonable Hand-Rolled Version
If you are going to build one, this is roughly the minimum that survives contact with a production line:
sequencer/core.py68 lines
import timefrom dataclasses import dataclass, fieldfrom typing import Callable@dataclassclass Measurement: name: str value: float | str | bool | None = None lower: float | None = None upper: float | None = None unit: str = "" @property def passed(self) -> bool: if self.value is None: return False if self.lower is not None and self.value < self.lower: return False if self.upper is not None and self.value > self.upper: return False return True@dataclassclass Step: name: str fn: Callable measurements: list[Measurement] = field(default_factory=list) critical: bool = False retries: int = 0 timeout_s: float | None = Noneclass Sequencer: def __init__(self, serial: str, resources: dict): self.serial = serial self.resources = resources self.results = [] def run(self, steps: list[Step]) -> bool: overall = True for step in steps: ok = self._run_step(step) overall = overall and ok if not ok and step.critical: break return overall def _run_step(self, step: Step) -> bool: for attempt in range(step.retries + 1): started = time.time() try: step.fn(step.measurements, **self.resources) ok = all(m.passed for m in step.measurements) except Exception as exc: ok = False self.results.append({"step": step.name, "error": str(exc)}) if ok or attempt == step.retries: self.results.append({ "step": step.name, "passed": ok, "duration_s": time.time() - started, "measurements": [vars(m) for m in step.measurements], }) return ok return FalseThat is around 60 lines and it still lacks timeouts that actually interrupt a hung call, parallel execution, an operator interface, offline persistence, and anything resembling analytics. Each of those is another few hundred lines, and the threading one is genuinely hard to get right.
The Version That Already Exists
The same test, declared rather than implemented. The sequence, the limits, the retry policy and the failure behaviour are data:
procedure.yaml44 lines
name: PCBA Functional Testversion: 1.0.0unit: serial_number: default_value: "PCBA000001" part_number: default_value: "PCBA-100"plugs: - name: dmm python: plugs.dmm:Multimetermain: - name: Power Rails python: phases.power_rails timeout: 30s measurements: - name: Rail 3V3 unit: V validators: - operator: ">=" expected_value: 3.2 - operator: "<=" expected_value: 3.4 - name: Idle Current unit: A validators: - operator: "<=" expected_value: 0.045 - name: Comms Check python: phases.comms_check depends_on: [power_rails] retry: limit: 3 delay: 1s then: fail: retry measurements: - name: Link Established validators: - operator: "==" expected_value: truedef power_rails(measurements, dmm): measurements.rail_3v3 = dmm.read_voltage() measurements.idle_current = dmm.read_current()def comms_check(measurements, dmm): measurements.link_established = Trueclass Multimeter: def read_voltage(self) -> float: return 3.31 # Replace with a real instrument read def read_current(self) -> float: return 0.041Run it with tofupilot run ./procedure.yaml.
Two details in that YAML worth noting, because they are easy to get wrong:
timeout takes a duration string, such as 30s or 2m, not a bare number. A phase without one waits indefinitely.
Retry needs both a policy and a trigger. retry.limit bounds the attempts and retry.delay spaces them, but a phase only retries when something asks it to: either the phase returns PhaseResult.RETRY from Python, or then: maps an outcome to a retry, as then: fail: retry does above.
Mapping the hand-rolled concepts onto it:
| Hand-rolled | Declared |
|---|---|
Step.critical and the break | depends_on |
Step.retries loop | retry.limit plus a then: trigger |
Step.timeout_s | timeout: 30s |
Measurement.passed comparisons | validators |
resources dict passed to every step | Plugs injected by parameter name |
self.results and a CSV writer | Automatic upload |
| Threading for parallel steps | Independent phases run in parallel |
The dependency graph replaces the critical flag with something more precise. depends_on says which steps actually require an earlier one to have succeeded, so unrelated steps still run and give you diagnostics, while dependent steps skip.
When Building Your Own Is Right
- You have a genuinely unusual execution model that no framework expresses, such as a closed-loop search rather than a fixed sequence.
- You have an existing sequencer that works and the only complaint is maintenance. Reducing that maintenance is usually cheaper than migrating.
- Your constraints prohibit external dependencies, which happens in some regulated or air-gapped environments.
If your reason is "our sequence is simple", note that the version above was also simple a year ago.
What Usually Triggers the Rewrite
Teams tend to abandon hand-rolled sequencers at a predictable point, and it is rarely the sequencing:
- Someone asks for yield by measurement over the last month and the CSVs cannot answer it.
- The line needs four boards tested at once and the loop is single-threaded.
- An operator needs a screen, and now there is a frontend project.
- A station loses network and a shift of results is lost.
- Limits change and nobody can tell which results used which limits.
Those are the parts that are expensive to build and are already solved.
If You Prefer Decorators
OpenHTF is the other purpose-built Python option, with the sequence and limits expressed in code:
import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import upload@htf.measures( htf.Measurement("rail_3v3").in_range(3.2, 3.4).with_units(units.VOLT),)@htf.PhaseOptions(timeout_s=30)def test_power_rails(test): test.measurements.rail_3v3 = 3.31test = htf.Test(test_power_rails, part_number="PCBA-100")test.add_output_callbacks(upload())test.execute(test_start=lambda: input("Scan serial: "))Install with pip install "tofupilot[openhtf]", or skip the callback and run it with tofupilot run. See TofuPilot Framework vs OpenHTF.
Common Mistakes
| Mistake | Consequence | Fix |
|---|---|---|
| Limits inside comparisons | Cannot report, review or trend them | Declare limits as data |
| Results printed to a console | Nothing survives the shift | Persist every run |
| One global try/except | A failing step hides which one failed | Per-step error capture |
| Retry around the whole sequence | Passing steps re-run and waste cycle time | Retry per step |
| Retry configured with no trigger | The phase never actually retries | Add then: fail: retry or return PhaseResult.RETRY |
timeout: 30 instead of timeout: 30s | Rejected; durations are strings | Use a duration string |
| Threading added late | Race conditions in shared instrument access | Declare dependencies; share instruments deliberately |
