Serial & FTP Polling Architectures: Deterministic Acquisition Off Legacy Analyzer Interfaces

Serial and FTP polling is the acquisition subsystem of the instrument ingestion pipeline — the layer that pulls raw output off analyzers that expose only an RS-232 port, a vendor middleware socket, or a passive SFTP drop directory. Its single job is narrow and unforgiving: capture every unit of instrument output exactly once, timestamp it, persist the untouched bytes, and hand a durable reference downstream — without ever letting transport-layer volatility leak into result processing. This subsystem is part of Instrument Data Ingestion & HL7/CSV Pipelines, and it owns the leftmost contract in that flow: the boundary where uncontrolled instrument telemetry becomes an attributable, replayable clinical artifact.

Polling — rather than push — is the deterministic default here for a concrete reason. Many hematology, chemistry, and coagulation platforms were never designed for high-availability integration; they emit a serial frame or write a CSV file and assume a listener is present. Polling puts the laboratory in control of timing, retry, and backpressure, and it degrades gracefully: a poller that misses a cycle simply catches up on the next one, whereas a dropped push is gone. Every design decision below follows from treating acquisition as a regulated control surface, not as file-copy glue.

Context & Pipeline Position

Within the end-to-end pipeline, this subsystem sits upstream of everything. It precedes the CSV to HL7 transformation stage that converts vendor-native formats into canonical ORU^R01 messages, and it feeds the schema validation and error handling gate that decides whether a result may cross into the patient record. Throughput smoothing between acquisition and those stages is delegated to the async batch processing workers, so a burst of morning-draw files never stalls the pollers or overruns an analyzer’s internal buffer.

The acquisition contract is deliberately minimal. Pollers observe interfaces on fixed schedules, and for each discrete unit of output they perform one thing before any parsing: write the raw payload to durable storage with a receipt timestamp and a content hash. This “capture-then-process” ordering is the forensic anchor of the whole pipeline — the original bytes are the record of record, and they must survive a crash in any later stage. Only after that persistence succeeds does the poller emit an ingest.received audit event and enqueue a reference for downstream work.

Stage Boundaries

The acquisition subsystem exposes exactly one ingress surface and one egress surface, each governed by an explicit contract. Making these contracts precise is what keeps failure semantics local: a serial parity error or an FTPS listing timeout is resolved (or quarantined) inside acquisition and never surfaces as a malformed HL7 message three stages later.

Ingress — instrument interfaces. What enters is uninterpreted bytes over one of three transports:

Egress — durable ingest handoff. What leaves is never raw bytes on the wire. It is a persisted artifact plus a typed reference message: the storage key of the untouched payload, its SHA-256 content hash, the receipt timestamp, the source instrument identifier, and the poll correlation ID. Downstream stages consume only this reference; they can always re-fetch the original bytes for replay.

Failure semantics at the boundary. The boundary distinguishes two outcomes. A transient failure (connection reset, unstable file size, checksum mismatch on a serial frame) triggers bounded exponential backoff and retry within the same poll worker — no artifact is produced. A permanent failure (a file that never stabilizes, a frame whose checksum fails after all retries, an undecodable payload) produces a quarantine artifact: the raw bytes are still persisted, but they are tagged with an error code and routed to a dead-letter path rather than the normal ingest queue. Nothing is ever silently discarded, because a discarded specimen result is a reportable data-integrity failure.

Schema & Protocol Specification

Serial acquisition speaks the ASTM E1381/E1394 stack. E1381 defines the low-level framing and the send/receive handshake; E1394 defines the record grammar carried inside those frames. A poller must understand both to know when a logical message is complete.

The record-type ordering is a sequence contract: a well-formed E1394 message is H → (P → (O → R*)*)* → L. A poller that assembles frames into a logical message must not release that message downstream until it has seen the terminating L record and every intervening checksum has verified.

FTPS and SFTP acquisition have no record grammar at this layer — the unit of output is a whole file — but they impose their own contract: a file is only a candidate once it has stopped changing. Instruments frequently create the file, then stream content into it before closing the handle, so a naïve “list and fetch” reads a truncated export. The specification the poller enforces is a stabilization window: identical size and mtime across two probes separated by a fixed interval, plus a non-zero length, before the file is eligible for acquisition.

Implementation Patterns

Acquisition is I/O-bound and fan-out-heavy — dozens of instruments, each on its own cadence — which makes asyncio the natural concurrency model. The patterns below are async-first and typed, with Pydantic v2 models pinning the egress contract so that a malformed reference can never leave this stage.

The egress reference is a Pydantic model. Making it the only object that crosses the boundary means the type system enforces the contract:

from datetime import datetime
from pydantic import BaseModel, Field, field_validator


class RawArtifactRef(BaseModel):
    """The only object the acquisition stage emits downstream."""

    storage_key: str = Field(min_length=1)
    content_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    received_at: datetime
    instrument_id: str = Field(min_length=1)
    source_transport: str  # "serial" | "ftps" | "sftp"
    byte_length: int = Field(gt=0)
    correlation_id: str

    @field_validator("source_transport")
    @classmethod
    def _known_transport(cls, v: str) -> str:
        if v not in {"serial", "ftps", "sftp"}:
            raise ValueError(f"unknown transport: {v}")
        return v

FTPS/SFTP polling centres on the stabilization check followed by an atomic claim. Probing size twice and requiring equality is what defends against partial writes:

import asyncio
from dataclasses import dataclass


@dataclass(slots=True)
class StabilityConfig:
    window_seconds: float = 2.5
    probes: int = 2
    max_attempts: int = 4


async def is_stable(stat_size, remote_path: str, cfg: StabilityConfig) -> bool:
    """True once the file size holds steady across probes and is non-zero."""
    for attempt in range(cfg.max_attempts):
        sizes: list[int] = []
        for _ in range(cfg.probes):
            sizes.append(await stat_size(remote_path))
            await asyncio.sleep(cfg.window_seconds)
        if sizes[0] > 0 and all(s == sizes[0] for s in sizes):
            return True
    return False

Capture-then-process is expressed as: fetch bytes, hash them, persist under the hash-derived key, and only then build the reference. If persistence fails, no reference is emitted and the file is left in place for the next cycle — the poll is idempotent by construction.

import hashlib
from datetime import datetime, timezone


async def acquire_file(fetch_bytes, persist, remote_path: str,
                       instrument_id: str, correlation_id: str) -> RawArtifactRef:
    payload: bytes = await fetch_bytes(remote_path)
    digest = hashlib.sha256(payload).hexdigest()
    storage_key = f"raw/{instrument_id}/{digest}"
    await persist(storage_key, payload)  # durable write BEFORE any parsing
    return RawArtifactRef(
        storage_key=storage_key,
        content_sha256=digest,
        received_at=datetime.now(timezone.utc),
        instrument_id=instrument_id,
        source_transport="ftps",
        byte_length=len(payload),
        correlation_id=correlation_id,
    )

Serial acquisition assembles E1381 frames into a logical message, verifying each checksum and honouring the link handshake. The delimiters are read from the H record rather than assumed:

def astm_checksum(frame_body: bytes) -> str:
    """Modulo-256 sum of the frame body, upper-case hex (E1381)."""
    return f"{sum(frame_body) & 0xFF:02X}"


def verify_frame(frame_body: bytes, transmitted: str) -> bool:
    return astm_checksum(frame_body) == transmitted.upper()

Fan-out across many instruments uses one supervised task per interface with bounded concurrency, so a single stalled analyzer cannot starve the others. A TaskGroup gives structured cancellation if the whole poller is shut down for maintenance:

import asyncio


async def run_pollers(interfaces: list[str], poll_one, concurrency: int = 16) -> None:
    limit = asyncio.Semaphore(concurrency)

    async def _guarded(iface: str) -> None:
        async with limit:
            await poll_one(iface)

    async with asyncio.TaskGroup() as tg:
        for iface in interfaces:
            tg.create_task(_guarded(iface))

The detailed, production build of the file-watching variant — including aioftp session handling, encoding quirks, and atomic directory moves — is worked end to end in building a Python FTP watcher for hematology analyzers.

Error Classification & Handling

Acquisition errors sort into three tiers, and the tier dictates the response. Conflating them is the classic failure that either drops results (retrying a permanent error forever) or double-reports them (quarantining a transient blip).

Transient transport failures use jittered exponential backoff to avoid thundering-herd reconnects against a shared serial-to-Ethernet bridge:

import asyncio
import random


async def with_backoff(op, *, max_attempts: int = 5, base: float = 0.5,
                       cap: float = 30.0):
    for attempt in range(max_attempts):
        try:
            return await op()
        except (ConnectionError, TimeoutError):
            if attempt == max_attempts - 1:
                raise
            delay = min(cap, base * 2 ** attempt) * (0.5 + random.random())
            await asyncio.sleep(delay)

Deduplication is the semantic-tier safeguard against double acquisition. Because the storage key is derived from the content hash, re-reading an identical file is a no-op: the persist step sees the key already exists and the poller records a dedup.skip audit event instead of enqueueing a duplicate. This is what prevents a re-listed SFTP file or a resent serial block from becoming a duplicate patient result. Acknowledgment-side deduplication and retry — the mirror-image problem at the LIMS routing boundary — is covered in handling HL7 ACK timeouts in clinical data pipelines.

Every quarantine decision emits a structured event carrying the error tier, the reason code, the storage key of the preserved raw bytes, and the correlation ID, so a data-integrity investigation can reconstruct exactly what the instrument sent and why it was held back.

Regulatory Touchpoints

Acquisition is the stage where several regulatory obligations first attach, because it is the first point at which instrument output becomes an electronic record the laboratory is accountable for.

The load-bearing detail for 21 CFR Part 11 is that the receipt timestamp is authoritative and machine-generated. Pollers derive it from a synchronized NTP source at the moment of persistence, never from the instrument’s own clock (which drifts and is often unmonitored), so the audit trail’s chronology is defensible under inspection. The broader compliance framing for the whole pipeline lives in the LIMS architecture and regulatory compliance foundations reference.

Testing & Validation

Because acquisition is where determinism is established, its tests assert invariants rather than example outputs. Property-based testing with hypothesis is the right tool: it searches for the byte sequence that breaks a parser.

The checksum verifier must round-trip for any body and reject any single-byte corruption:

from hypothesis import given, strategies as st


@given(st.binary(min_size=1, max_size=512))
def test_checksum_roundtrips(body: bytes) -> None:
    assert verify_frame(body, astm_checksum(body))


@given(st.binary(min_size=1, max_size=512), st.integers(0, 511))
def test_single_byte_corruption_detected(body: bytes, idx: int) -> None:
    good = astm_checksum(body)
    i = idx % len(body)
    corrupted = body[:i] + bytes([body[i] ^ 0x01]) + body[i + 1:]
    # A one-bit flip must change the checksum (or the frames are not equal).
    assert astm_checksum(corrupted) != good or corrupted == body

The stabilization contract is a property too: a file whose size is still growing must never be declared stable.

import asyncio
from hypothesis import given, strategies as st


@given(st.lists(st.integers(min_value=1, max_value=10_000),
                min_size=2, max_size=6).map(sorted).filter(lambda xs: xs[0] != xs[-1]))
def test_growing_file_is_never_stable(sizes: list[int]) -> None:
    it = iter(sizes)

    async def stat_size(_path: str) -> int:
        return next(it, sizes[-1])

    cfg = StabilityConfig(window_seconds=0, probes=2, max_attempts=1)
    assert asyncio.run(is_stable(stat_size, "x", cfg)) is False

Golden-file fixtures pin real-world instrument quirks. A committed corpus of captured ASTM transfers and CSV exports — one per instrument model and firmware generation — is replayed on every CI run so that a change to framing or stabilization logic that would alter acquisition of a known-good message fails loudly. Contract tests assert the egress RawArtifactRef validates and its content_sha256 matches a recomputed hash of the persisted bytes, guaranteeing the reference downstream stages trust is faithful to what was captured.

Related

• CSV to HL7 transformation — the next stage, mapping acquired vendor formats to canonical ORU^R01 messages.
• Schema validation & error handling — the tiered gate that decides whether an acquired result reaches the record.
• Async batch processing — bounded-concurrency workers that absorb acquisition bursts.
• Building a Python FTP watcher for hematology analyzers — a production build of the file-watching poller.
• Handling HL7 ACK timeouts in clinical data pipelines — the acknowledgment-side retry and dedup mirror of this stage.

Part of: Instrument Data Ingestion & HL7/CSV Pipelines

Frequently Asked Engineering Questions