End-to-End Lab Automation Pipeline Assembly

Assembling a lab automation pipeline is the act of composing six independently correct stages — session acquisition, transport, dispatch, capture, injection, and durable sink — into one deterministic control path that starts, drains, and stops as a unit. This guide, part of the Instrument Control Architecture & Taxonomy section, covers the assembly seams: startup ordering, inter-stage backpressure, and graceful shutdown, rather than the internals of any single stage. Get the seams wrong and each stage passes its own tests while the rig deadlocks, drops samples on shutdown, or silently corrupts provenance under load.

What Breaks Without Explicit Assembly

Every failure that survives stage-level testing lives in the wiring between stages. A digitizer opens before its VISA session is negotiated and the first sweep reads garbage. A capture stage that outruns a stalled LIMS sink fills memory until the acquisition process is OOM-killed mid-run, losing the in-flight buffer. A Ctrl-C during a plate read tears the transport down while the injector still holds twelve validated-but-unwritten records, so the run ends with a gap no operator can reconstruct. None of these are bugs in a stage — they are bugs in the lifecycle contract between stages, and they only appear once the stages run together against real hardware under real load. Explicit assembly makes that contract a first-class object: a Pipeline orchestrator that owns startup order, the bounded queues that carry backpressure, and the drain-then-stop sequence that guarantees no acknowledged sample is lost.

Prerequisites and Deployment Scope

The patterns here assume Python 3.11+ (for asyncio.TaskGroup and ExceptionGroup), pyvisa 1.14+ over a NI-VISA or pyvisa-py backend for SCPI instruments, pyserial 3.5 for RS-232/RS-485 controllers, and pydantic 2.x for the typed records that cross stage boundaries. The durable sink is any transactional store — PostgreSQL through psycopg 3.x, a LIMS REST endpoint, or a local SQLite write-ahead-log buffer for the disconnected case. The orchestrator is transport-agnostic on purpose: it composes stage objects that expose a uniform lifecycle, so a rig mixing a GPIB source meter, a USB-TMC oscilloscope, and a serial temperature controller assembles from the same skeleton. Every stage boundary is a typed queue, never a shared mutable buffer, which is what makes backpressure and shutdown tractable.

Stage Responsibilities and Boundaries

Assembly starts by fixing what each stage owns and — more importantly — what it must never reach across a boundary to touch. The table is the contract every stage object implements; the individual technique guides linked in each row cover the internals.

The load-bearing rule across the whole table: failure isolation is defined at the boundary that produces the failure, not at the top of the pipeline. A transport that fails to open must not be allowed to abort a four-instrument rig; a validation reject storm must be absorbed at the capture boundary so it cannot flood the injector. Each stage translates its own faults — categorized through Error Code Categorization — into one of three outcomes: retry locally, divert to a side channel, or raise a fatal that the orchestrator catches during a controlled stop.

Pipeline Topology and Backpressure Path

The stages form a linear flow, but the control signals are bidirectional: data moves left to right through bounded queues, while backpressure and health move right to left. When the sink slows, its input queue fills, which blocks the injector’s put, which fills the capture queue, which blocks dispatch — the stall propagates upstream one bounded buffer at a time until the acquisition loop itself pauses. That is the design goal: a bounded queue converts “unbounded memory growth” into “the source stops pulling,” which is the only safe response to a slow sink.

End-to-end topology: data flows left to right through bounded queues; backpressure (amber) and health/liveness (green) flow upstream. Capture diverts rejects to quarantine and injection diverts unwritable records to a dead-letter path, so neither fault propagates into the sink.

Startup Ordering, Drain, and Shutdown

Assembly order is not cosmetic — it is a dependency topological sort. Stages come up sink-first, source-last: the durable sink and its queue must be ready to accept before the injector runs, the injector before capture, and acquisition dead last, because acquisition is the only stage that pulls new work into the system. Bringing the source up first would let samples arrive before anything downstream can accept them, forcing an immediate stall or drop. Shutdown reverses this exactly: stop the source first so no new work enters, then drain — let the in-flight samples flow through capture, injection, and the sink until every queue empties and every acknowledged record is committed — and only then tear down transports and release sessions. The distinction between stop (halt intake) and drain (flush what is already inside) is what separates a clean shutdown from a lossy one.

Timing determinism follows from the same ordering. The end-to-end latency of one sample is bounded by the sum of per-stage service times plus the time it spends queued, so the drain deadline must exceed the worst-case in-flight depth times the slowest stage:

$$T_{\text{drain}}\ge \sum _{i=1}^N{q}_i\cdot t_{\text{stage},i}+t_{\text{sink,commit}}$$

where q_i is the occupancy of queue i at the moment intake stops and t_stage,i is that stage’s per-item service time. If T_drain is set shorter than this bound, drain terminates with records still buffered and the shutdown becomes lossy — so the orchestrator must either wait for the queues to empty or explicitly flush the residue to the durable buffer before exiting.

Orchestrator Skeleton: Composing Typed Stages

The orchestrator below composes stage objects behind one Stage protocol, wires them with bounded asyncio.Queue boundaries, and implements the sink-first start, forward drain, and reverse stop. Each stage is a pure consumer/producer over its queues; the orchestrator owns lifecycle and error propagation. This is the wiring made concrete in Wiring VISA, Serial, Async Capture, and LIMS into One Pipeline; the surrounding project layout is covered in Structuring a New Lab Automation Project from Scratch.

Copy
from __future__ import annotations

import asyncio
import logging
from typing import Protocol, runtime_checkable

log = logging.getLogger("pipeline")

SENTINEL = object()  # in-band signal that a stage's upstream has drained


class PipelineFatal(Exception):
    """A stage fault the orchestrator cannot recover; triggers a clean stop."""


@runtime_checkable
class Stage(Protocol):
    """Uniform lifecycle contract every pipeline stage implements."""

    name: str

    async def open(self) -> None:
        """Acquire resources (sessions, sockets, files). Must be idempotent."""

    async def run(self, inbox: asyncio.Queue, outbox: asyncio.Queue | None) -> None:
        """Consume from inbox, produce to outbox until a SENTINEL arrives."""

    async def close(self) -> None:
        """Release resources. Must tolerate being called after a failed open."""


class Pipeline:
    """Composes ordered stages with bounded queues and explicit lifecycle.

    Stages are listed source-first for readability; startup runs them in
    reverse (sink-first) so every consumer is ready before its producer.
    """

    def __init__(self, stages: list[Stage], *, queue_depth: int = 256) -> None:
        if len(stages) < 2:
            raise ValueError("Pipeline needs at least a source and a sink")
        self._stages = stages
        # One bounded queue between each adjacent pair; depth bounds memory
        # and is the mechanism by which a slow sink applies backpressure.
        self._queues = [asyncio.Queue(maxsize=queue_depth)
                        for _ in range(len(stages) - 1)]
        self._opened: list[Stage] = []

    async def start(self) -> None:
        """Open stages sink-first; roll back cleanly on partial startup."""
        for stage in reversed(self._stages):
            try:
                await stage.open()
                self._opened.append(stage)
            except Exception as exc:  # noqa: BLE001 - convert to controlled rollback
                log.error("open failed for %s: %s", stage.name, exc)
                await self._rollback()
                raise PipelineFatal(f"startup aborted at {stage.name}") from exc

    async def _rollback(self) -> None:
        """Close whatever opened, newest-first; no partial rig stays live."""
        for stage in reversed(self._opened):
            try:
                await stage.close()
            except Exception:  # noqa: BLE001 - best-effort teardown
                log.exception("rollback close failed for %s", stage.name)
        self._opened.clear()

    async def run_until_drained(self) -> None:
        """Run every stage concurrently; a fatal in any stage stops all."""
        try:
            async with asyncio.TaskGroup() as tg:
                for i, stage in enumerate(self._stages):
                    inbox = self._queues[i - 1] if i > 0 else asyncio.Queue()
                    outbox = self._queues[i] if i < len(self._queues) else None
                    tg.create_task(self._supervise(stage, inbox, outbox))
        except* PipelineFatal as eg:
            for exc in eg.exceptions:
                log.error("stage fatal: %s", exc)
            raise

    async def _supervise(self, stage: Stage, inbox: asyncio.Queue,
                         outbox: asyncio.Queue | None) -> None:
        """Run one stage; on exit forward a SENTINEL so drain propagates."""
        try:
            await stage.run(inbox, outbox)
        finally:
            if outbox is not None:
                await outbox.put(SENTINEL)  # let downstream finish its backlog

    async def stop(self) -> None:
        """Reverse teardown: source already halted, now release everything."""
        await self._rollback()

The design decisions that make this composable: stages never reference each other, only their inbox/outbox, so any stage can be swapped or unit-tested against two plain queues. The SENTINEL is an in-band drain signal — when the source stops producing and exits, its finally pushes a sentinel that each stage forwards after emptying its backlog, so drain flows forward to the sink without a separate control channel. asyncio.TaskGroup (the except* clause) propagates the first PipelineFatal while cancelling siblings, converting one stage’s unrecoverable fault into a coordinated stop instead of a half-running rig. Startup rollback guarantees the partial-startup invariant: if the third of six stages fails to open, the two already opened are closed newest-first and start raises, so the process never lingers with a live transport and no sink.

Edge Cases in Multi-Device Assembly

• Partial-startup rollback. A four-instrument rig where instrument three’s session negotiation times out must not leave instruments one and two holding open GPIB locks. The sink-first open with newest-first rollback in start handles this: every successfully opened stage is closed before the exception surfaces, and no queue has yet received a sample, so there is nothing to drain. The operator sees a clean PipelineFatal naming the failed stage, not a wedged rig requiring a power cycle.

• One instrument down in a multi-device rig. When acquisition fans out across several devices, a single device’s transport failure must degrade the rig, not halt it. Model each device as its own source-side sub-stage feeding a shared dispatch queue; a failed device raises a non-fatal that quarantines that device and emits a health event, while the pipeline keeps servicing the survivors. Only a sink or injector fault — a stage with no redundancy — escalates to PipelineFatal. This is where the health arrow matters: the orchestrator must surface “3 of 4 devices live” rather than reporting a false all-clear.

• A stage that outruns the sink. A digitizer streaming at 2 MS/s into a LIMS that commits at 200 records/s will, without bounds, exhaust memory in seconds. The bounded queue is the entire defense: once the injector→sink queue is full, await outbox.put(...) blocks the injector, which blocks capture, which blocks dispatch, until the source’s own put blocks and acquisition naturally throttles to the sink’s sustainable rate. If the source cannot be throttled (a free-running instrument that will drop samples in its own hardware FIFO), spill the overflow to the local durable buffer and reconcile later rather than letting the queue grow unbounded — the sink outage row below covers the mechanism.

Fault Categorization Across the Assembly

Each row maps a fault to the boundary that owns it. The categorization discipline itself — deciding transient-retry versus divert versus fatal — reuses the taxonomy in Error Code Categorization; assembly’s contribution is deciding at which boundary the categorized fault is absorbed so it cannot cascade.

Integrating the Assembly with Each Stage Guide

The orchestrator is deliberately hollow — its value is the lifecycle, and every stage’s substance lives in its own guide. Session acquisition composes over a VISA Resource Manager for multi-vendor handle management; transport comes up through PySerial Configuration & Tuning for serial devices and a Protocol Abstraction Layer that lets legacy and modern instruments present one command interface. Dispatch is an Async Command Queuing System whose bounded queue is the backpressure boundary this guide relies on. Capture chains Binary & ASCII Format Parsing into Checksum & CRC Validation, and the injector applies Metadata Injection Workflows before the sink. Wrap the whole path in Threshold Tuning & Alerting so queue depth, drain time, and per-boundary reject rates are watched as first-class pipeline health signals rather than discovered post-mortem.

Implementation Checklist

• Startup opens stages sink-first and a forced open failure at any stage rolls back every earlier stage, leaving zero live sessions and raising a named PipelineFatal.
• Every stage boundary is a bounded asyncio.Queue; a deliberately stalled sink is verified to apply backpressure all the way to the source rather than growing memory.
• Shutdown distinguishes stop (halt the source) from drain (flush in-flight records), and a drain leaves zero acknowledged-but-uncommitted records in any queue.
• The drain deadline is set from measured worst-case queue occupancy times slowest-stage service time, and a timeout flushes residual records to the durable buffer instead of dropping them.
• A single device failing to open in a multi-device rig degrades to “N-1 live” with a health event, while only sink/injector faults escalate to a full stop.
• Capture diverts validation rejects to quarantine and injection diverts unwritable records to a dead-letter path, verified so neither fault reaches the sink.
• A simulated LIMS outage flips the sink to the local durable buffer, keeps upstream unblocked, and drains the buffer with bounded backoff on recovery with no lost records.
• Health/liveness for every stage is surfaced to Threshold Tuning & Alerting with alerts on queue saturation, drain overrun, and per-boundary reject rate.

Related

• Wiring VISA, Serial, Async Capture, and LIMS into One Pipeline — the concrete stage implementations behind this orchestrator.
• Structuring a New Lab Automation Project from Scratch — the repository layout and packaging that hosts the assembly.
• Async Command Queuing Systems — the dispatch stage and the queue that carries backpressure.
• Metadata Injection Workflows — the enrichment stage upstream of the durable sink.
• Protocol Abstraction Layers — the uniform command interface that lets mixed hardware share one transport contract.

← Back to Instrument Control Architecture & Taxonomy