Building a pyserial Plug for Serial Devices

pip install pyserial

from serial.tools import list_ports

for p in list_ports.comports():
    print(p.device, p.vid, p.pid, p.serial_number, p.description)

/dev/ttyUSB0 4292 60000 A50285BI USB Serial
COM3 1027 24577 FT5ZQ7X8 USB Serial Port (COM3)

import re
import time

import serial
from serial.tools import list_ports
from openhtf.core import base_plugs


class DutUart(base_plugs.BasePlug):
    """ASCII command/response console on the DUT's debug UART."""

    def __init__(self, port: str, baudrate: int = 115200, adapter_serial: str | None = None):
        if adapter_serial:
            port = self._port_for_adapter(adapter_serial)
        self._ser = serial.Serial(port, baudrate, timeout=1)   # read timeout, seconds
        self._ser.reset_input_buffer()
        self.logger.info("Opened %s @ %d", port, baudrate)

    @staticmethod
    def _port_for_adapter(adapter_serial: str) -> str:
        for p in list_ports.comports():
            if p.serial_number == adapter_serial:
                return p.device
        raise RuntimeError(f"No serial adapter with serial {adapter_serial!r}")

    def command(self, cmd: str, timeout_s: float = 2.0) -> str:
        """Send `cmd`, return everything up to the next prompt."""
        self._ser.reset_input_buffer()
        self._ser.write((cmd + "\r\n").encode())
        deadline = time.monotonic() + timeout_s
        lines = []
        while time.monotonic() < deadline:
            line = self._ser.readline().decode(errors="replace").rstrip("\r\n")
            if line == ">":                 # DUT prompt: response complete
                break
            if line and line != cmd:        # drop echo
                lines.append(line)
        else:
            raise TimeoutError(f"No prompt after {cmd!r}")
        response = "\n".join(lines)
        self.logger.debug("%s -> %s", cmd, response)
        return response

    def read_until_pattern(self, pattern: str, timeout_s: float = 30.0) -> str:
        """Collect output until a regex matches — e.g. a boot banner."""
        regex = re.compile(pattern)
        deadline = time.monotonic() + timeout_s
        buf = ""
        while time.monotonic() < deadline:
            chunk = self._ser.read(self._ser.in_waiting or 1).decode(errors="replace")
            buf += chunk
            if regex.search(buf):
                return buf
        raise TimeoutError(f"Pattern {pattern!r} not seen within {timeout_s}s")

    def tearDown(self) -> None:
        self._ser.close()

import openhtf as htf
from openhtf.util import configuration

from dut_uart import DutUart

CONF = configuration.CONF
UART_PORT = CONF.declare("uart_port", default_value="/dev/ttyUSB0")
UART_BAUD = CONF.declare("uart_baud", default_value=115200)
Uart = configuration.bind_init_args(DutUart, UART_PORT, UART_BAUD)


@htf.plug(uart=Uart)
def wait_for_boot(test, uart):
    banner = uart.read_until_pattern(r"login:|>\s*$", timeout_s=20)
    test.attach("boot_log", banner.encode(), mimetype="text/plain")


@htf.plug(uart=Uart)
@htf.measures(
    htf.Measurement("fw_version").matches_regex(r"^2\.1\.\d+$"),
    htf.Measurement("mcu_serial").matches_regex(r"^[0-9A-F]{16}$"),
    htf.Measurement("vbat_mv").in_range(3600, 4200),
)
def read_identity(test, uart):
    test.measurements.fw_version = uart.command("version")
    test.measurements.mcu_serial = uart.command("serial")
    test.measurements.vbat_mv = int(uart.command("adc vbat"))


@htf.plug(uart=Uart)
def set_dut_id_from_device(test, uart):
    # Use the MCU's own serial as the DUT ID when there is no barcode.
    test.test_record.dut_id = uart.command("serial")


def main():
    test = htf.Test(wait_for_boot, set_dut_id_from_device, read_identity)
    test.execute()      # DUT ID set by set_dut_id_from_device

if __name__ == "__main__":
    main()