Monitors
Collect background measurements during OpenHTF phase execution with monitors. Poll instrument values at fixed intervals while your test logic runs.
Sample measurements in the background while a phase runs.
A monitor runs a sampling function in a background thread while the phase executes, polling the current every 500 milliseconds. Each sample is stored with its timestamp as a dimensioned measurement on the phase.
- phase_burn_in starts
- @monitors("current")poll_interval_ms=500
- phase logic runssampled in background
- phase_burn_in ends
Some tests need to monitor a value continuously — current draw during a burn-in, temperature during a stress test, voltage during a load cycle. OpenHTF monitors run a sampling function in a background thread at a fixed interval, recording each sample as a dimensioned measurement with a millisecond timestamp.
Syntax
Use the @monitors.monitors() decorator to attach a background monitor to a phase. The monitor function runs repeatedly in a background thread while the decorated phase executes.
import openhtf as htf
from openhtf.core import monitors
from openhtf.util import units
import time
def read_current(test):
"""Monitor function: returns the value to record."""
return 0.5 # In a real test, read from an ammeter
@monitors.monitors('current_draw', read_current, units=units.AMPERE, poll_interval_ms=500)
def burn_in_phase(test):
"""Main phase: runs while monitor samples in background."""
test.logger.info('Starting 5-second burn-in')
time.sleep(5)
return htf.PhaseResult.CONTINUE
def main():
test = htf.Test(burn_in_phase)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()The test record will contain a current_draw measurement with timestamped samples at ~500ms intervals.
Parameters
measurement_namestrName of the output measurement. Created automatically as a dimensioned measurement with a millisecond time axis.
monitor_funccallableFunction that returns the value to record. Called repeatedly by the background thread. Can accept test as an argument and use plugs.
unitsUnitDescriptorUnit of measure for the recorded values (e.g. units.AMPERE, units.VOLT).
poll_interval_msintMilliseconds between samples. Default is 1000 (1 second). Set to 0 for continuous sampling with no delay.
Monitors with plugs
The monitor function can use plugs, just like a regular phase. Plugs declared on the monitor function are automatically passed through.
import openhtf as htf
from openhtf.core import monitors
from openhtf.plugs import plug
from openhtf.util import units
import time
class Ammeter(htf.BasePlug):
def read_current(self):
return 0.42 # Read from real hardware
@plug(ammeter=Ammeter)
def read_current(test, ammeter):
return ammeter.read_current()
@monitors.monitors('current', read_current, units=units.AMPERE, poll_interval_ms=200)
def load_test(test):
test.logger.info('Running load test for 3 seconds')
time.sleep(3)
return htf.PhaseResult.CONTINUE
def main():
test = htf.Test(load_test)
test.execute(lambda: "SN1234")
if __name__ == "__main__":
main()How it works
- When the decorated phase starts, a daemon thread launches the monitor function.
- The monitor function is called at each
poll_interval_msinterval. - Each returned value is stored with a millisecond timestamp relative to the phase start.
- When the phase completes (pass or fail), the monitor thread stops.
- The resulting dimensioned measurement appears in the test record with
MILLISECONDas the time dimension.
