When a measurement fails in OpenHTF, the default behavior is to mark the phase as failed and continue running subsequent phases. In production, you need deliberate control: when to stop early, when to retry, and how to preserve failure data in TofuPilot for analytics.
Prerequisites
- TofuPilot Python client installed:
pip install tofupilot - OpenHTF installed:
pip install openhtf - Basic familiarity with OpenHTF phases and measurements
Phase Failure Behavior by Default
OpenHTF phases return a PhaseResult. When a measurement fails, the phase is marked FAIL but execution continues unless you explicitly stop it.
import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import TofuPilot@htf.measures( htf.Measurement('rail_3v3_voltage').in_range(3.2, 3.4).with_units(units.VOLT), htf.Measurement('rail_5v_voltage').in_range(4.85, 5.15).with_units(units.VOLT),)def check_power_rails(test): test.measurements.rail_3v3_voltage = 3.1 # fails: below minimum test.measurements.rail_5v_voltage = 5.02 # passes@htf.measures( htf.Measurement('uart_echo').equals('OK'),)def check_uart(test): # This phase still runs even though check_power_rails failed test.measurements.uart_echo = 'OK'def main(): test = htf.Test(check_power_rails, check_uart) with TofuPilot(test): test.execute(test_start=lambda: 'SN-001')Both phases run. The run is uploaded to TofuPilot as FAIL with the specific failing measurement recorded.
Stop on First Failure
For hardware tests where a failed power rail makes downstream tests meaningless, stop early using PhaseResult.STOP.
import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import TofuPilot@htf.measures( htf.Measurement('rail_3v3_voltage').in_range(3.2, 3.4).with_units(units.VOLT),)def check_power_rails(test): test.measurements.rail_3v3_voltage = read_adc_channel(0) if not test.measurements.rail_3v3_voltage.is_pass: test.logger.error('Power rail out of spec, aborting sequence') return htf.PhaseResult.STOP@htf.measures( htf.Measurement('uart_echo').equals('OK'),)def check_uart(test): response = send_uart_command('PING') test.measurements.uart_echo = responsedef main(): test = htf.Test(check_power_rails, check_uart) with TofuPilot(test): test.execute(test_start=lambda: 'SN-001')When check_power_rails returns PhaseResult.STOP, check_uart is skipped. TofuPilot records the run as FAIL with check_uart phases marked as not executed.
Retry a Phase on Failure
Retries are useful for transient failures: communication timeouts, settling voltages, or flaky contacts. Use PhaseResult.REPEAT with a counter to avoid infinite loops.
import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import TofuPilotMAX_RETRIES = 3@htf.measures( htf.Measurement('i2c_device_present').equals(True),)def check_i2c_device(test, retries=[0]): found = probe_i2c_address(0x48) test.measurements.i2c_device_present = found if not found and retries[0] < MAX_RETRIES: retries[0] += 1 test.logger.warning('I2C device not found, retry %d/%d', retries[0], MAX_RETRIES) return htf.PhaseResult.REPEAT retries[0] = 0 # reset for future runs@htf.measures( htf.Measurement('i2c_temperature').in_range(-10, 85).with_units(units.DEGREE_CELSIUS),)def read_i2c_temperature(test): temp = read_temperature_sensor(0x48) test.measurements.i2c_temperature = tempdef main(): test = htf.Test(check_i2c_device, read_i2c_temperature) with TofuPilot(test): test.execute(test_start=lambda: 'SN-002')Using a mutable default argument (retries=[0]) persists the counter across REPEAT calls within the same phase execution. Reset it before returning to avoid stale state on the next DUT.
Retry with Backoff Using a Plug
For production lines with many fixtures, a plug keeps retry logic reusable and keeps phase code clean.
import timeimport openhtf as htffrom openhtf.plugs import BasePlugclass CommPlug(BasePlug): def setUp(self): import serial self._port = serial.Serial('/dev/ttyUSB0', 115200, timeout=1) def send_with_retry(self, command, expected, attempts=3, delay=0.5): response = '' for attempt in range(attempts): self._port.write(f'{command}'.encode()) response = self._port.readline().decode().strip() if response == expected: return response, attempt + 1 time.sleep(delay) return response, attempts def tearDown(self): if self._port: self._port.close()import openhtf as htffrom tofupilot.openhtf import TofuPilotfrom plugs.comms_plug import CommPlug@htf.plug(comm=CommPlug)@htf.measures( htf.Measurement('firmware_version').matches_regex(r'v\d+\.\d+\.\d+'), htf.Measurement('version_retry_count').in_range(minimum=1),)def check_firmware_version(test, comm): response, attempts = comm.send_with_retry('VERSION', expected='v2.1.0', attempts=3) test.measurements.firmware_version = response test.measurements.version_retry_count = attemptsdef main(): test = htf.Test(check_firmware_version) with TofuPilot(test): test.execute(test_start=lambda: 'SN-003')Recording version_retry_count as a measurement lets TofuPilot surface retry frequency across your production fleet. A spike in retries on a specific station points to a fixture problem before it becomes a yield problem.
Clean Teardown After Failures
Plugs with a tearDown method are always called by OpenHTF, even when a phase fails or the test stops early. Use this to release hardware resources reliably.
import openhtf as htffrom openhtf.plugs import BasePlugclass FixturePlug(BasePlug): def setUp(self): self._power_on = False self._relay_closed = False def power_on_dut(self): enable_power_rail() self._power_on = True def close_test_relay(self): close_relay(1) self._relay_closed = True def tearDown(self): # Always runs, even on STOP or exception if self._relay_closed: open_relay(1) if self._power_on: disable_power_rail()import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import TofuPilotfrom plugs.fixture_plug import FixturePlug@htf.plug(fixture=FixturePlug)@htf.measures( htf.Measurement('leakage_current').in_range(maximum=50).with_units(units.AMPERE),)def measure_leakage(test, fixture): fixture.power_on_dut() fixture.close_test_relay() current_ua = read_current_meter() * 1e6 test.measurements.leakage_current = current_ua if not test.measurements.leakage_current.is_pass: test.logger.error('Leakage %.1f uA exceeds 50 uA limit', current_ua) return htf.PhaseResult.STOP # fixture.tearDown() called automatically by OpenHTFdef main(): test = htf.Test(measure_leakage) with TofuPilot(test): test.execute(test_start=lambda: 'SN-004')Failure Handling Strategy Comparison
| Strategy | Use when | PhaseResult | Downstream phases |
|---|---|---|---|
| Continue on fail | Failures are independent, collect all data | (default) | Run |
| Stop on first fail | Downstream tests are invalid after failure | STOP | Skipped |
| Retry transient | Failure may be a fixture fluke | REPEAT | Run after retry |
| Retry with limit | Same, but cap attempts | REPEAT + counter | Run or STOP |
Logging Failures for Analytics
TofuPilot captures every failed measurement automatically. To make failures actionable in analytics, add structured context using test.logger.
import openhtf as htffrom openhtf.util import unitsfrom tofupilot.openhtf import TofuPilot@htf.measures( htf.Measurement('oscillator_freq_hz').in_range(7_990_000, 8_010_000).with_units(units.HERTZ),)def check_oscillator(test): freq = measure_frequency(channel=2) test.measurements.oscillator_freq_hz = freq if not test.measurements.oscillator_freq_hz.is_pass: deviation_ppm = abs(freq - 8_000_000) / 8_000_000 * 1e6 test.logger.error( 'Oscillator %.0f Hz (%.1f ppm from 8 MHz)', freq, deviation_ppm, )@htf.measures( htf.Measurement('adc_offset_mv').in_range(-5, 5), htf.Measurement('adc_gain_error_pct').within_percent(1.0, 0.1),)def calibrate_adc(test): offset = measure_adc_offset() gain = measure_adc_gain() test.measurements.adc_offset_mv = offset * 1000 test.measurements.adc_gain_error_pct = gaindef main(): test = htf.Test(check_oscillator, calibrate_adc, name='PCB-Rev-C') with TofuPilot(test): test.execute(test_start=lambda: 'SN-005')Log messages appear in the TofuPilot run detail view alongside the failing measurement. Quantitative context (deviation in ppm, not just pass/fail) makes root cause analysis faster when reviewing failures across a batch.