Controlling serial lab instruments from Python with pyserial

To control a serial lab instrument from Python, open the port with pyserial and an explicit timeout, send each command as bytes with the instrument's line terminator, and read exactly one reply with read_until(), treating a reply that lacks the terminator as a timeout. Wrap that exchange in a small class that logs every command and reply, retries only commands that are safe to repeat, and owns the port from one thread. The same code drives RS-232 instruments through USB adapters and USB instruments that appear as virtual COM ports.

Key points

• pyserial 3.5 (November 2020) is still the current release on PyPI; the API in this guide has been stable for years.
• Always set timeout. readline() and read_until() return early, with a partial line, when it expires.
• Find ports by USB vendor ID, product ID and serial number rather than by COM5 or /dev/ttyUSB0.
• Retry queries, never actions: after a timeout on "dispense", ask the instrument what it did.
• Test without hardware: a fake link object for unit tests, loop:// for framing, a pseudo-terminal for end-to-end runs.

The examples use a fictional syringe pump, the "PX-1", with an ASCII protocol: commands and replies end in CR LF, STATUS? returns a line such as OK IDLE FLOW=0.250, FLOW 1.500 sets the flow rate, and errors come back as ERR <code> <text>. Swap in your instrument's commands and terminator. The code needs Python 3.10 or later and was run against pyserial 3.5.

How do I find the right serial port?

Install pyserial with pip install pyserial, then list ports from the shell with python -m serial.tools.list_ports -v, or from code:

from serial.tools import list_ports

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

Each entry is a ListPortInfo with device, description and hwid, plus vid, pid, serial_number, manufacturer and product for USB devices. The documentation warns that enumeration can be incomplete on some systems; the extended USB fields are available on Linux, macOS and Windows. Match on them:

def find_port(vid: int, pid: int, serial_number: str | None = None) -> str:
    matches = [
        p for p in list_ports.comports()
        if p.vid == vid and p.pid == pid
        and (serial_number is None or p.serial_number == serial_number)
    ]
    if len(matches) != 1:
        raise RuntimeError(f"expected one matching port, found {[p.device for p in matches]}")
    return matches[0].device

Typical device names:

Linux: permission denied

Serial devices belong to a group. On Debian and Ubuntu it is dialout, described in the Debian wiki as "full and direct access to serial ports". Add yourself with sudo usermod -aG dialout $USER and log in again. On other distributions, ls -l /dev/ttyACM0 shows which group owns the device.

macOS: /dev/cu.* or /dev/tty.*?

macOS creates two device files per port. Apple's serial device guide says "the callout device should almost always be used" and reserves the dial-in /dev/tty.* device for monitoring incoming calls. Use /dev/cu.*; pyserial's comports() lists the callout devices on macOS.

Windows: COM numbers above 9

COM1 to COM9 are reserved device names. For higher numbers the Win32 API needs the device namespace prefix, as in \\.\COM56 (Microsoft). pyserial adds the prefix itself, so pass "COM12". COM numbers are assigned by Windows and differ between PCs, which is another reason to match on VID and PID.

USB virtual COM ports

Many instruments now implement the USB CDC ACM class instead of shipping an adapter cable. On Linux they appear as ttyACM devices handled by the cdc-acm driver. If the instrument's firmware talks USB directly, the baud rate you set may be ignored; if a USB chip inside the instrument bridges to a UART, it still matters. Use the value in the manual either way. A USB-to-RS-232 adapter must match the instrument's baud rate, parity and stop bits exactly.

Opening the port: baud rate, framing and timeouts

import serial

port = serial.Serial("/dev/ttyACM0", baudrate=115200, timeout=1.0, write_timeout=1.0)

The defaults are 8 data bits, no parity, 1 stop bit (8N1). The timeout argument controls read() and everything built on it: None waits forever, 0 returns immediately, and a number waits up to that many seconds. write_timeout makes write() raise SerialTimeoutException instead of blocking when the device stops accepting data. On POSIX systems, exclusive=True refuses to open a port that is already open in exclusive mode, so if all your scripts use it, two of them cannot interleave commands.

serial.serial_for_url() takes the same arguments and also accepts URLs: loop:// (a loopback for tests), socket://host:port (a serial-to-Ethernet converter) and spy://<port>?file=wire.log (a hex dump of all traffic). Using it everywhere costs nothing and lets the same driver run against hardware, a network bridge or a test loop.

Line endings and the request-response pattern

Many lab instruments speak in lines: one command line, one reply line. The terminator is CR (\r), LF (\n) or CR LF, and the manual will say which. Send bytes, not strings, and read until the terminator:

port.reset_input_buffer()
port.write(b"STATUS?\r\n")
line = port.read_until(b"\r\n")
if not line.endswith(b"\r\n"):
    raise TimeoutError(f"incomplete reply: {line!r}")
reply = line[:-2].decode("ascii")

The pyserial docs are blunt about readline(): specify a timeout or it can block forever, and "if the \n is missing in the return value, it returned on timeout". read_until(expected, size) works the same way but takes any terminator. Watch for three protocol habits: instruments that echo each command before replying (read and discard the echo), multi-line replies (read until a final OK or ERR line), and unsolicited messages such as "door open" (route them to a handler instead of treating them as replies).

A reusable instrument class

This class wraps one PX-1. It was tested with pytest, a pseudo-terminal device and loop://.

"""px1.py: driver for the fictional PX-1 syringe pump (line-based ASCII protocol)."""
from __future__ import annotations

import logging
import threading
import time

import serial

log = logging.getLogger("px1")

EOL = b"\r\n"


class PumpError(Exception):
    """The pump answered with an ERR reply."""


class PumpTimeout(Exception):
    """No complete reply line arrived before the timeout."""


class Pump:
    def __init__(self, link, retries: int = 2):
        self._link = link              # an open serial.Serial or anything with the same methods
        self._retries = retries
        self._lock = threading.Lock()  # one command/reply exchange at a time

    @classmethod
    def open(cls, url: str, baudrate: int = 115200, timeout: float = 1.0, **kwargs) -> "Pump":
        # serial_for_url accepts COM5, /dev/ttyACM0, loop://, socket://host:port, spy://...
        link = serial.serial_for_url(url, baudrate=baudrate, timeout=timeout, write_timeout=timeout)
        return cls(link, **kwargs)

    def close(self) -> None:
        self._link.close()

    def __enter__(self) -> "Pump":
        return self

    def __exit__(self, *exc) -> None:
        self.close()

    def query(self, command: str) -> str:
        """Send one command and return the reply line without its terminator."""
        # Only queries (ending in '?') are safe to repeat blindly.
        attempts = 1 + (self._retries if command.endswith("?") else 0)
        with self._lock:
            for attempt in range(1, attempts + 1):
                try:
                    return self._exchange(command)
                except PumpTimeout:
                    log.warning("timeout on %r (attempt %d of %d)", command, attempt, attempts)
                    self._resync()
        raise PumpTimeout(f"no reply to {command!r}")

    def _exchange(self, command: str) -> str:
        self._link.reset_input_buffer()          # drop stale bytes from an earlier reply
        raw = command.encode("ascii") + EOL
        log.debug("TX %r", raw)
        self._link.write(raw)
        line = self._link.read_until(EOL)        # returns early, possibly partial, on timeout
        log.debug("RX %r", line)
        if not line.endswith(EOL):
            raise PumpTimeout(command)
        reply = line[: -len(EOL)].decode("ascii", errors="replace")
        if reply.startswith("ERR"):
            raise PumpError(f"{command!r} -> {reply!r}")
        return reply

    def _resync(self, quiet: float = 0.2) -> None:
        """Let a late reply finish arriving, then throw it away."""
        time.sleep(quiet)
        self._link.reset_input_buffer()

    # Methods for the fictional PX-1 command set
    def status(self) -> str:
        return self.query("STATUS?")

    def set_flow(self, ml_per_min: float) -> None:
        self.query(f"FLOW {ml_per_min:.3f}")

Usage:

with Pump.open("/dev/ttyACM0") as pump:
    pump.set_flow(0.25)
    print(pump.status())   # OK IDLE FLOW=0.250

Design choices worth copying:

• The port is injected. Pump(link) accepts any object with write, read_until, reset_input_buffer and close, so tests need no hardware.
• One lock per port. A status-polling thread and a command thread cannot interleave bytes.
• The input buffer is cleared before every command, so a late reply to a previous command is not read as the answer to this one.
• Errors are exceptions, and the instrument's own error text is kept in the message.

Logging every command and reply

When a run fails at 02:14, the wire log is the only witness. Configure logging once, with millisecond timestamps:

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s.%(msecs)03d %(levelname)s %(name)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)

That produces lines such as 2026-10-13T18:34:47.876 DEBUG px1 TX b'FLOW 0.250\r\n'. For a byte-level trace, open the port as Pump.open("spy:///dev/ttyACM0?file=wire.log"), which writes a hex dump of every byte sent and received.

Timeouts, partial lines, retries and resynchronisation

The rule that saves samples: after a timeout on a command that moves liquid or hardware, do not resend it. You do not know whether the instrument received it. Query the status, then decide.

Threads and asyncio

A serial port is a single conversation, so give it one owner. With the lock in the class above, several threads can share a Pump. In an asyncio program, the simplest option is to keep the blocking driver and run calls in a worker thread, which works on every OS:

reply = await asyncio.to_thread(pump.status)

pyserial-asyncio (0.6, September 2021) gives you a StreamReader and StreamWriter for the port. Its module docstring says "Posix platforms only", although the code includes a polling fallback for Windows. pyserial-asyncio-fast, maintained under the Home Assistant libraries organisation, is a fork for macOS, Linux and BSD with eager writes.

import asyncio

import serial_asyncio


async def query(reader, writer, command: str, timeout: float = 1.0) -> str:
    writer.write(command.encode("ascii") + b"\r\n")
    await writer.drain()
    line = await asyncio.wait_for(reader.readuntil(b"\r\n"), timeout)
    return line[:-2].decode("ascii")


async def main():
    reader, writer = await serial_asyncio.open_serial_connection(url="/dev/ttyACM0", baudrate=115200)
    try:
        print(await query(reader, writer, "STATUS?"))
    finally:
        writer.close()

asyncio.run(main())

Testing without the instrument

Because the class takes any link object, a fake pump is a few lines:

import pytest

from px1 import EOL, Pump, PumpError, PumpTimeout


class FakePX1:
    """Stands in for the serial port and the pump behind it."""

    def __init__(self):
        self.rx = bytearray()     # bytes the 'pump' has sent back
        self.flow = 0.0
        self.lose_replies = 0     # simulate replies lost on the wire
        self.sent = []

    def write(self, data: bytes) -> int:
        command = data.decode("ascii").removesuffix("\r\n")
        self.sent.append(command)
        reply = self._handle(command)
        if self.lose_replies:
            self.lose_replies -= 1
        else:
            self.rx += reply.encode("ascii") + EOL
        return len(data)

    def _handle(self, command: str) -> str:
        if command == "STATUS?":
            return f"OK IDLE FLOW={self.flow:.3f}"
        if command.startswith("FLOW "):
            try:
                self.flow = float(command[5:])
            except ValueError:
                return "ERR 2 BAD_ARGUMENT"
            return "OK"
        return "ERR 1 UNKNOWN_COMMAND"

    def read_until(self, expected=b"\n", size=None) -> bytes:
        end = self.rx.find(expected)
        if end < 0:                       # behave like a timeout: return what we have
            data = bytes(self.rx)
            self.rx.clear()
            return data
        end += len(expected)
        data = bytes(self.rx[:end])
        del self.rx[:end]
        return data

    def reset_input_buffer(self) -> None:
        self.rx.clear()

    def close(self) -> None:
        pass


def test_status_and_flow():
    with Pump(FakePX1()) as pump:
        pump.set_flow(1.5)
        assert pump.status() == "OK IDLE FLOW=1.500"


def test_error_reply_raises():
    with pytest.raises(PumpError):
        Pump(FakePX1()).query("FLOW fast")


def test_query_is_retried_after_lost_reply():
    fake = FakePX1()
    fake.lose_replies = 1
    assert Pump(fake).status().startswith("OK")
    assert fake.sent == ["STATUS?", "STATUS?"]


def test_set_command_is_not_retried():
    fake = FakePX1()
    fake.lose_replies = 1
    with pytest.raises(PumpTimeout):
        Pump(fake).set_flow(2.0)
    assert fake.sent == ["FLOW 2.000"]

loop:// is useful for a different check: everything written is read back, so you can see exactly what bytes your code sends and how it behaves when a terminator never arrives.

>>> import serial
>>> s = serial.serial_for_url("loop://", timeout=0.5)
>>> s.write(b"STATUS?\r\n"); s.read_until(b"\r\n")
9
b'STATUS?\r\n'
>>> s.write(b"STAT"); s.read_until(b"\r\n")   # no terminator: returns after the timeout
4
b'STAT'

On Linux and macOS, os.openpty() gives a pseudo-terminal pair: run a fake instrument on one end in a thread and open the other end with the real Pump.open(), which exercises pyserial itself.

LabVIEW and MATLAB equivalents

LabVIEW (NI-VISA). VISA Configure Serial Port defaults to 9600 baud, 8 data bits, no parity, one stop bit, a 10 s timeout, and termination-character detection enabled with LF (0xA). The termination character ends read operations; for CR-terminated replies set it to 0xD. The setting governs reads, so make sure the strings you pass to VISA Write carry the terminator the instrument expects, then use VISA Read with a byte count larger than the longest reply.

MATLAB. serialport (R2019b and later) has a 10 s default timeout and an LF default terminator:

s = serialport("/dev/ttyACM0", 115200, "Timeout", 1);
configureTerminator(s, "CR/LF");
writeline(s, "STATUS?");
reply = readline(s)

readline returns an empty array ([]) if no data arrives within the timeout, so check the result before parsing it.

From a driver to an integration

A tested driver class is the foundation for everything above it: a status poller for a self-driving lab, a sampling script for online HPLC reaction monitoring, or a SiLA 2 server, as shown in SiLA 2, OPC UA LADS and serial. If you are building an instrument rather than scripting one, the OEM integration checklist covers the protocol decisions from the other side.