Building a pymodbus Plug for PLCs and Modbus Devices
Control fixtures, chambers and power meters over Modbus TCP or RTU from an OpenHTF plug with pymodbus — connect, read holding registers into measurements, write coils to actuate a fixture, decode 32-bit floats, and handle Modbus exceptions.
Last updated · Verified with OpenHTF 1.6.1
Fixtures and environmental equipment usually expose Modbus: a PLC that clamps the DUT, a thermal chamber, a programmable load, a power meter. pymodbus speaks Modbus TCP and RTU; an OpenHTF plug turns register reads into measurements and coil writes into fixture actions.
pip install pymodbus # add pyserial for Modbus RTU over a serial lineExamples use pymodbus 3.x's synchronous client.
The plug
import struct
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException
from openhtf.core import base_plugs
class FixturePlc(base_plugs.BasePlug):
"""Test fixture PLC over Modbus TCP: clamp control, pressure and temperature."""
COIL_CLAMP = 0
COIL_POWER = 1
HR_PRESSURE_KPA = 100 # uint16, kPa
HR_TEMP_C_X10 = 101 # int16, °C × 10
HR_ENERGY_WH = 200 # float32 in two registers (big-endian word order)
def __init__(self, host: str, port: int = 502, unit: int = 1):
self._unit = unit
self._client = ModbusTcpClient(host, port=port, timeout=2)
if not self._client.connect():
raise ConnectionError(f"Cannot reach PLC at {host}:{port}")
self.logger.info("Connected to PLC %s:%d", host, port)
# --- actuation ---
def clamp(self, engaged: bool) -> None:
self._write_coil(self.COIL_CLAMP, engaged)
def dut_power(self, on: bool) -> None:
self._write_coil(self.COIL_POWER, on)
# --- readings ---
def pressure_kpa(self) -> int:
return self._read_registers(self.HR_PRESSURE_KPA, 1)[0]
def temperature_c(self) -> float:
raw = self._read_registers(self.HR_TEMP_C_X10, 1)[0]
return struct.unpack(">h", struct.pack(">H", raw))[0] / 10 # signed int16
def energy_wh(self) -> float:
hi, lo = self._read_registers(self.HR_ENERGY_WH, 2)
return struct.unpack(">f", struct.pack(">HH", hi, lo))[0]
# --- helpers ---
def _read_registers(self, address: int, count: int) -> list[int]:
rr = self._client.read_holding_registers(address, count=count, slave=self._unit)
if rr.isError():
raise ModbusException(f"read_holding_registers({address}, {count}) -> {rr}")
return rr.registers
def _write_coil(self, address: int, value: bool) -> None:
wr = self._client.write_coil(address, value, slave=self._unit)
if wr.isError():
raise ModbusException(f"write_coil({address}, {value}) -> {wr}")
self.logger.info("coil %d <- %s", address, value)
def tearDown(self) -> None:
# Release the DUT whatever happened during the test.
try:
self.dut_power(False)
self.clamp(False)
finally:
self._client.close()pymodbus 3.9 renamed the slave= keyword to device_id=. Check pymodbus.__version__ and use the one your install expects.
Use it in phases
import time
import openhtf as htf
from openhtf.util import configuration, units
from fixture_plc import FixturePlc
CONF = configuration.CONF
PLC_HOST = CONF.declare("plc_host", default_value="192.168.10.5")
Plc = configuration.bind_init_args(FixturePlc, PLC_HOST)
@htf.plug(plc=Plc)
@htf.measures(htf.Measurement("clamp_pressure").in_range(400, 600).with_units(units.KILOPASCAL))
def clamp_dut(test, plc):
plc.clamp(True)
time.sleep(0.5)
test.measurements.clamp_pressure = plc.pressure_kpa()
@htf.plug(plc=Plc)
@htf.measures(
htf.Measurement("ambient_c").in_range(18, 28).with_units(units.DEGREE_CELSIUS),
htf.Measurement("energy_wh").in_range(maximum=0.5).with_units(units.WATT_HOUR),
)
def power_and_measure(test, plc):
plc.dut_power(True)
time.sleep(5)
test.measurements.ambient_c = plc.temperature_c()
test.measurements.energy_wh = plc.energy_wh()
def main():
test = htf.Test(
htf.PhaseGroup(
main=[clamp_dut, power_and_measure],
# tearDown() also releases, but an explicit teardown phase is recorded.
teardown=[release_dut],
)
)
test.execute(lambda: "SN1234")
@htf.plug(plc=Plc)
def release_dut(test, plc):
plc.dut_power(False)
plc.clamp(False)
if __name__ == "__main__":
main()The phase group guarantees release_dut runs even when power_and_measure fails; the plug's tearDown() is a second safety net if the process dies mid-phase.
Modbus RTU
Swap the client; the rest of the plug is identical:
from pymodbus.client import ModbusSerialClient
self._client = ModbusSerialClient(port="/dev/ttyUSB1", baudrate=19200, parity="E", stopbits=1, timeout=1)Practical notes
- Addresses are zero-based in pymodbus. A device manual's "register 40101" is holding register address 100.
- Data types. Modbus only knows 16-bit registers; ints, floats and word order (big/little endian) are conventions of each device.
structmakes the conversion explicit and testable. - Check
isError()on every response; a Modbus exception response is not a Python exception. - Timeouts.
timeout=2on the client plus phasetimeout_sinPhaseOptionsbounds a dead PLC. - Shared devices. If a chamber is shared by several stations, one plug per station is fine for reads; coordinate writes outside OpenHTF.
- Environmental profiles. For a thermal cycle, combine
write_registersetpoints with a monitor samplingtemperature_c()andconsistent_end_dimension_pivot_validate(validators) to prove the chamber settled.
Related
pyserial plug
Talk to a DUT or instrument over a UART from an OpenHTF plug with pyserial — port discovery, command/response with timeouts, line framing, reading a serial number and firmware version into measurements, and capturing a boot log as an attachment.
OpenHTF vs pytest
A practical comparison of OpenHTF and pytest for testing physical products — execution model, measurements vs assertions, fixtures vs plugs, operator interaction, output records — and when to use each or both.
