Instrument Control

Wiring VISA, Serial, Async Capture, and LIMS into One Pipeline

This is the runnable walkthrough behind End-to-End Lab Automation Pipeline Assembly: a single main() that opens a PyVISA session and a pyserial port, dispatches through an async queue, parses and CRC-checks each frame, injects metadata, and pushes one record to a LIMS. It is part of the Instrument Control Architecture & Taxonomy section, and where the parent guide fixes the lifecycle contract, this one fills every stage with concrete code and follows one record end to end.

What This Walkthrough Assumes

The target rig is minimal on purpose: one SCPI source-measure unit reachable over a VISA Resource Manager session and one RS-232 sensor controller on a pyserial port, both feeding a single async dispatcher. Every byte a device returns is a length-prefixed frame with a trailing CRC-32; the sink is a LIMS REST endpoint that accepts one enriched record per POST and returns a commit id. Versions match the parent: Python 3.11+ for asyncio.TaskGroup, pyvisa 1.14+, pyserial 3.5, and pydantic 2.x for the typed record that crosses each boundary. The parent guide already argues why the seams matter and defines the generic Stage protocol; this guide does not restate that design — it shows the six concrete stage bodies and the one function that wires them, so the reader can paste it, point it at mock endpoints, and watch a record traverse the whole path.

The wiring rule is that adjacent stages never call each other directly; they hand off through bounded asyncio.Queue boundaries carrying one Pydantic model, so a slow LIMS applies backpressure the same way a slow parser does. Startup runs sink-first and shutdown drains source-first, exactly as the parent prescribes — the code below is the literal implementation of those two orderings.

Latency and Throughput Budget for One Cohesive Cycle

Before wiring, fix the two inequalities that decide whether the assembled pipeline stays bounded. The per-record cycle time T_cycle cannot be shorter than the sum of the individual stage service times, because the stages are serialized through queues along the critical path of one record:

The second inequality is the one that keeps memory finite. Over any window the sink must retire records at least as fast as the sources emit them, or a bounded queue on the path fills and blocks its producer:

If r_sink drops below r_ingest — a LIMS commit slows from 40 ms to 400 ms — the injector→sink queue saturates in queue_depth / (r_ingest - r_sink) seconds, and every stage upstream stalls in turn. That is the designed response, not a failure: a bounded queue converts unbounded memory growth into a throttled source. The budget tells you the drain deadline and the alert threshold to hand to Threshold Tuning & Alerting; these numbers are what you measure, not guess.

One record traversing the concretely wired VISA plus serial pipeline Two source nodes on the left, a PyVISA session and a pyserial port, each emit a raw frame into a shared async command queue. The queue forwards a raw frame to a parse-and-CRC stage that produces a validated sample. That sample flows into a metadata-injection stage that produces an enriched record carrying a sequence number and provenance. The enriched record flows into a LIMS sink node that performs a transactional POST and returns a commit id acknowledgement. A single highlighted record is traced across all stages with forward arrows. Below the band an amber backpressure arrow runs right to left from the sink back to the queue, labelled sink slower than ingest blocks upstream. A quarantine side box hangs beneath the parse and CRC stage for failed checksums. PyVISA sessionSCPI SMU · MEAS? pyserial portRS-232 · 115200 async queue ordered dispatch parse + CRC validated sample inject seq_no · provenance LIMS sink POST → commit id raw frame sample record sink slower than ingest → blocks upstream (backpressure) Quarantine (bad CRC)

One record’s path: two sources enqueue raw frames, dispatch serializes them, parse and CRC produce a validated sample, injection stamps provenance, and the LIMS sink returns a commit id. Failed checksums divert to quarantine; a slow sink pushes backpressure (amber) upstream to the queue.

Step 1 — Define the Typed Record That Crosses Every Boundary

Fix the handoff type first, because every queue in the pipeline carries exactly one shape. A Pydantic model gives each stage a validated contract and makes an accidentally-skipped stage a load-time error rather than a silent None.

from __future__ import annotations

import asyncio
import binascii
import logging
import struct
from datetime import datetime, timezone
from typing import Optional

from pydantic import BaseModel, Field

log = logging.getLogger("pipeline.wiring")


class Sample(BaseModel):
    """A parsed, CRC-verified reading straight off a transport."""
    device_id: str
    channel: int
    value: float
    raw_crc_ok: bool


class Record(Sample):
    """A Sample after metadata injection; this is what the sink commits."""
    sequence_no: int
    captured_at: str
    run_id: str
    commit_id: Optional[str] = Field(default=None)

Step 2 — Open the VISA Session and the Serial Port (Sink-First Startup)

Resource acquisition is where a rig aborts cleanly or comes up half-wired. Open the sink first so a producer never emits into a stage that cannot accept, then the sources; if any open() raises, roll back everything already opened so no VISA lock or serial handle leaks. The stubs below call the real subsystems — a full VISA Resource Manager and tuned PySerial Configuration & Tuning live in their own guides — but the lifecycle is exact.

class VisaSource:
    """Opens a PyVISA session and reads one framed reply per poll."""

    def __init__(self, resource: str, device_id: str) -> None:
        self.resource, self.device_id = resource, device_id
        self._sess = None

    async def open(self) -> None:
        import pyvisa
        rm = pyvisa.ResourceManager()
        self._sess = await asyncio.to_thread(rm.open_resource, self.resource)
        self._sess.timeout = 2000  # ms; a dead instrument must not hang startup

    async def poll(self) -> bytes:
        if self._sess is None:
            raise RuntimeError(f"{self.device_id}: poll before open")
        # Blocking VISA I/O is pushed off the loop so it never stalls dispatch.
        return await asyncio.to_thread(self._sess.read_raw)

    async def close(self) -> None:
        if self._sess is not None:
            await asyncio.to_thread(self._sess.close)
            self._sess = None


class SerialSource:
    """Opens a pyserial port and reads one length-prefixed frame per poll."""

    def __init__(self, port: str, device_id: str, baud: int = 115200) -> None:
        self.port, self.device_id, self.baud = port, device_id, baud
        self._ser = None

    async def open(self) -> None:
        import serial
        self._ser = await asyncio.to_thread(
            serial.Serial, self.port, self.baud, timeout=2.0
        )

    async def poll(self) -> bytes:
        if self._ser is None:
            raise RuntimeError(f"{self.device_id}: poll before open")
        header = await asyncio.to_thread(self._ser.read, 2)
        (length,) = struct.unpack(">H", header)
        return header + await asyncio.to_thread(self._ser.read, length)

    async def close(self) -> None:
        if self._ser is not None:
            await asyncio.to_thread(self._ser.close)
            self._ser = None

Both sources wrap their blocking libraries in asyncio.to_thread, the single most important wiring detail: a synchronous read_raw() or Serial.read() on the event loop would freeze dispatch, capture, and the sink at once.

Step 3 — Dispatch, Parse, Inject, and Sink as Queue-Coupled Coroutines

Each remaining stage is a coroutine consuming from an inbox and producing to an outbox, forwarding a sentinel on drain. Dispatch fans the two sources into one ordered stream — the concrete Async Command Queuing System that serializes per-transport traffic. Capture runs Binary & ASCII Format Parsing then Checksum & CRC Validation, diverting a bad CRC to quarantine instead of poisoning downstream. Injection applies Metadata Injection Workflows to stamp a monotonic sequence_no and provenance, and the sink commits one record per call.

SENTINEL = object()


async def dispatch(sources, out: asyncio.Queue, polls_per_source: int) -> None:
    """Poll each source in round-robin, forwarding raw (device_id, frame)."""
    for _ in range(polls_per_source):
        for src in sources:
            frame = await src.poll()
            await out.put((src.device_id, frame))  # blocks when out is full
    await out.put(SENTINEL)


def _parse_and_check(device_id: str, frame: bytes) -> Sample:
    """Strip the 2-byte length header, verify the trailing CRC-32, decode."""
    body, crc = frame[2:-4], frame[-4:]
    expected = struct.pack(">I", binascii.crc32(body) & 0xFFFFFFFF)
    ok = crc == expected
    channel, value = struct.unpack(">Hf", body[:6])
    return Sample(device_id=device_id, channel=channel, value=value, raw_crc_ok=ok)


async def capture(inp: asyncio.Queue, out: asyncio.Queue,
                  quarantine: asyncio.Queue) -> None:
    while True:
        item = await inp.get()
        if item is SENTINEL:
            await out.put(SENTINEL)
            return
        device_id, frame = item
        sample = _parse_and_check(device_id, frame)
        if not sample.raw_crc_ok:
            await quarantine.put(sample)   # diverted, never propagated
            continue
        await out.put(sample)


async def inject(inp: asyncio.Queue, out: asyncio.Queue, run_id: str) -> None:
    seq = 0
    while True:
        sample = await inp.get()
        if sample is SENTINEL:
            await out.put(SENTINEL)
            return
        seq += 1
        await out.put(Record(**sample.model_dump(), sequence_no=seq,
                             captured_at=datetime.now(timezone.utc).isoformat(),
                             run_id=run_id))


async def lims_sink(inp: asyncio.Queue, post, committed: list[Record]) -> None:
    while True:
        record = await inp.get()
        if record is SENTINEL:
            return
        record.commit_id = await post(record)   # transactional; may block
        committed.append(record)

Step 4 — Compose It All in One main()

The orchestration function opens sink-first, runs every stage concurrently under one asyncio.TaskGroup, and closes source-first on the way out. Backpressure is implicit: every await out.put(...) blocks when its bounded queue is full, so a slow post throttles the sources without a line of explicit flow control.

async def run_pipeline(sources, post, *, run_id: str, polls_per_source: int,
                       depth: int = 64) -> list[Record]:
    raw_q: asyncio.Queue = asyncio.Queue(maxsize=depth)
    clean_q: asyncio.Queue = asyncio.Queue(maxsize=depth)
    record_q: asyncio.Queue = asyncio.Queue(maxsize=depth)
    quarantine: asyncio.Queue = asyncio.Queue()
    committed: list[Record] = []

    opened = []
    for src in sources:                     # sources are the only openables here
        try:
            await src.open()
            opened.append(src)
        except Exception as exc:            # roll back; no half-open rig
            for s in reversed(opened):
                await s.close()
            raise RuntimeError(f"startup aborted opening {src.device_id}") from exc

    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(dispatch(sources, raw_q, polls_per_source))
            tg.create_task(capture(raw_q, clean_q, quarantine))
            tg.create_task(inject(clean_q, record_q, run_id))
            tg.create_task(lims_sink(record_q, post, committed))
    finally:
        for src in reversed(opened):        # source-first teardown after drain
            await src.close()
    log.info("committed %d records, quarantined %d", committed.count, quarantine.qsize())
    return committed

Step 5 — Dry-Run Against Mock Endpoints and Assert One Record Traverses Every Stage

Prove the wiring before any hardware is attached by swapping mock sources and a mock post, then asserting that a record leaves the sink carrying a commit_id, a sequence_no, and a CRC-verified value — evidence it passed through parse, inject, and sink in order.

import pytest


def _good_frame(channel: int, value: float) -> bytes:
    body = struct.pack(">Hf", channel, value)
    crc = struct.pack(">I", binascii.crc32(body) & 0xFFFFFFFF)
    return struct.pack(">H", len(body)) + body + crc


class MockSource:
    def __init__(self, device_id, frames):
        self.device_id, self._frames, self._i = device_id, frames, 0
    async def open(self): ...
    async def close(self): ...
    async def poll(self):
        f = self._frames[self._i % len(self._frames)]
        self._i += 1
        return f


@pytest.mark.asyncio
async def test_one_record_traverses_all_stages() -> None:
    src = MockSource("smu-1", [_good_frame(3, 1.25)])
    async def post(rec: Record) -> str:
        return f"lims-{rec.sequence_no}"
    out = await run_pipeline([src], post, run_id="R42", polls_per_source=1)
    assert len(out) == 1
    r = out[0]
    assert r.raw_crc_ok and r.value == pytest.approx(1.25)   # parse + CRC ran
    assert r.sequence_no == 1 and r.run_id == "R42"          # injection ran
    assert r.commit_id == "lims-1"                            # sink ran

On a live rig the same assertions hold, plus three hardware-side signals: the LIMS returns monotonically increasing sequence_no values (proof the injector saw records in dispatch order), quarantine.qsize() stays flat (proof CRC seeds and framing match firmware), and none of the queues’ qsize() climbs monotonically (proof r_sink ≥ r_ingest holds). A single *OPC? after a poll batch confirms the instruments physically settled before the sink acknowledged.

Failure Modes in the Assembled Wiring

A source polled before its transport is open. Calling poll() before open() — the classic effect of bringing a source up before the sink and its queue are ready — raises RuntimeError: poll before open from the guard in each source. The fix is structural, not defensive: run_pipeline opens every source before the TaskGroup starts any coroutine, so no poll can run against a None handle. If you reorder startup to launch dispatch first, this guard converts a garbage first-sweep into an immediate, named abort.

Sink slower than ingest → the queue would grow unbounded. When post slows below the aggregate poll rate, record_q fills to depth, the injector’s await out.put(...) blocks, and the stall propagates back through capture to dispatch until the sources stop polling — the r_sink ≥ r_ingest inequality made physical. Memory stays bounded at 3 × depth records. The bug is raising maxsize to mask a slow sink; the fix is to spill overflow to a local durable buffer and drain it on recovery, exactly as the parent guide’s sink-outage row prescribes.

One instrument failing to open aborts the whole rig. If VisaSource.open() times out on a powered-off SMU, the rollback loop closes every already-opened source newest-first and raises startup aborted opening smu-1, so no serial handle or VISA lock leaks. For a rig that must degrade to N-1 rather than abort, catch the per-source open failure, emit a health event, and start the TaskGroup with the survivors — but never leave a partially opened rig live, and classify the fault through Error Code Categorization to decide retry-versus-quarantine.

Shutdown losing in-flight records. Cancelling the TaskGroup on Ctrl-C instead of letting the sentinel propagate would strand validated-but-uncommitted records in clean_q and record_q. The wiring here drains by design: dispatch emits one SENTINEL when polling ends, each stage forwards it only after its inbox empties, and the sink returns only after committing its backlog — so a clean stop leaves zero acknowledged records behind. A hard kill mid-commit still needs the idempotent upsert on the LIMS side to avoid a double-write on restart.

← Back to End-to-End Lab Automation Pipeline Assembly

References