Building a Python FTP Watcher for Hematology Analyzers

Problem Statement

A hematology analyzer — a Sysmex XN, a Beckman Coulter DxH, an Abbott CELL-DYN — rarely offers a modern integration surface. It runs an embedded FTP server and drops a CSV of CBC results into a directory on a fixed export cadence, assuming a listener will pick them up. A naive watcher that lists the directory and downloads whatever it finds will eventually read a file the instrument is still writing, ingest a truncated result, download the same export twice after a network blip, or exhaust the analyzer’s tiny embedded connection pool and stall its export queue. This page builds the watcher that does the opposite: it polls the drop directory without blocking, proves each file is quiescent before it touches it, acquires it atomically with a content hash so a re-dropped file is never processed twice, and writes an immutable audit record for every acquisition — then hands a durable reference downstream and gets out of the way.

Prerequisites

• Python 3.11+ — the watcher uses asyncio, the walrus operator, and datetime.UTC.
• aioftp (aioftp>=0.21) for non-blocking FTP polling, and Pydantic v2 (pydantic>=2.6) for a typed acquisition receipt.
• A known export contract. You must know the analyzer’s export directory, file-naming pattern, encoding, and roughly how long a single CSV takes to write. This watcher owns only acquisition; it is the leftmost stage of the serial and FTP polling architecture and it never parses results itself.
• A downstream queue. Acquired files are handed to the async batch processing workers that drain a durable queue, not processed inline — a burst of morning-draw exports must never stall the poller or overrun the analyzer’s buffer.
• Regulatory baseline: CLIA §493.1105 (record retention), 21 CFR Part 11.10(e) (immutable, time-stamped audit trail), and HIPAA §164.312(b) audit controls.

Capture-then-process is the ordering that makes the whole pipeline defensible: the untouched bytes are the record of record, and they must survive a crash in any later stage before any parsing happens. Result parsing and canonicalization belong to the downstream CSV to HL7 Transformation stage — see converting legacy CSV instrument logs to HL7 ORU^R01 messages for the sibling task that consumes what this watcher produces.

Step 1: Poll the Drop Directory Without Blocking

The watcher observes the analyzer’s export directory on a fixed schedule using aioftp, so a slow instrument never blocks the event loop or other pollers. Each cycle opens one short-lived control connection, lists the directory, and yields candidate files; it never holds a connection open between cycles, because embedded analyzer FTP servers commonly cap concurrent sessions at two or four.

import asyncio
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import AsyncGenerator

import aioftp

logger = logging.getLogger("hematology_ftp_watcher")


@dataclass(frozen=True)
class FTPTarget:
    host: str
    port: int
    user: str
    password: str
    remote_dir: str
    glob: str = "*.csv"
    poll_interval: float = 20.0  # match the analyzer's export cadence


async def poll_drop_directory(
    target: FTPTarget,
) -> AsyncGenerator[tuple[aioftp.Client, str, dict], None]:
    """Yield (client, remote_path, stat) for each candidate file, one cycle at a time."""
    while True:
        try:
            async with aioftp.Client.context(
                target.host, port=target.port,
                user=target.user, password=target.password,
            ) as client:
                async for path, info in client.list(target.remote_dir, recursive=False):
                    if info.get("type") == "file" and Path(str(path)).match(target.glob):
                        yield client, str(path), info
        except (ConnectionError, aioftp.StatusCodeError) as exc:
            logger.warning("Poll cycle failed, will retry: %s", exc)
        await asyncio.sleep(target.poll_interval)

Calibrate poll_interval to the instrument’s export cadence — typically 15–30 seconds for a mid-volume hematology line. Polling faster than the analyzer writes only increases the chance of catching a half-written file; polling slower inflates result turnaround time.

Step 2: Prove the File Is Quiescent Before Touching It

The single most common corruption source is downloading a CSV while the analyzer is still appending rows to it. Because FTP gives no “file closed” event, the watcher infers quiescence: it samples the remote size and modification time twice across a settling window and only proceeds when both are unchanged and the size is non-zero. This is the same mtime-quiescence guard the acquisition contract mandates across the schema validation and error handling boundary downstream.

class StabilityGate:
    def __init__(self, settle_seconds: float = 2.5, attempts: int = 3):
        self.settle_seconds = settle_seconds
        self.attempts = attempts

    async def is_quiescent(self, client: aioftp.Client, remote_path: str) -> bool:
        """True only when size and mtime hold steady across the settling window."""
        for attempt in range(1, self.attempts + 1):
            try:
                before = await client.stat(remote_path)
                await asyncio.sleep(self.settle_seconds)
                after = await client.stat(remote_path)
            except aioftp.StatusCodeError as exc:
                logger.debug("stat failed for %s: %s", remote_path, exc)
                return False

            size_before = int(before.get("size", -1))
            size_after = int(after.get("size", -1))
            if size_before == size_after and size_after > 0 \
                    and before.get("modify") == after.get("modify"):
                return True
            logger.info("File still settling: %s (attempt %d)", remote_path, attempt)
        return False

A settle_seconds of 2.5 comfortably covers the sub-second write burst of a single CBC export; widen it for analyzers that stream large end-of-shift QC batches.

Step 3: Acquire Atomically, Hash, and Deduplicate

Only a quiescent file is downloaded. It lands first in a temporary path, then moves into the quarantine directory with an atomic os.replace(), so a downstream worker never sees a partially transferred file. A SHA-256 of the bytes gives the file a stable identity: if the analyzer re-drops the same export after a failed transfer, the hash matches an entry in the seen cache and the duplicate is discarded instead of re-ingested.

import hashlib
import os
from datetime import datetime, UTC


async def acquire(
    client: aioftp.Client,
    remote_path: str,
    quarantine_dir: Path,
    seen: set[str],
) -> dict | None:
    """Download to a temp file, hash, dedup, then atomically move into quarantine."""
    quarantine_dir.mkdir(parents=True, exist_ok=True)
    name = Path(remote_path).name
    tmp_path = quarantine_dir / f".{name}.part"

    await client.download(remote_path, tmp_path, write_into=True)

    digest = hashlib.sha256(tmp_path.read_bytes()).hexdigest()
    if digest in seen:
        logger.info("Duplicate export ignored: %s", digest[:12])
        tmp_path.unlink(missing_ok=True)
        await client.remove(remote_path)  # instrument already gave us these bytes
        return None

    final_path = quarantine_dir / f"{Path(name).stem}_{digest[:8]}.csv"
    os.replace(tmp_path, final_path)  # atomic within the same filesystem
    seen.add(digest)
    await client.remove(remote_path)  # release the analyzer's spool only after success

    return {
        "path": str(final_path),
        "sha256": digest,
        "source_name": name,
        "acquired_at": datetime.now(UTC).isoformat(),
    }

Delete the remote file only after the local atomic move succeeds; deleting first turns any local failure into permanent data loss. In production the seen set is a Redis-backed cache with a TTL longer than the instrument’s retry window, so deduplication survives a process restart — an in-memory set re-ingests everything in the drop directory after a crash.

Step 4: Emit an Audit Event and Guard With a Circuit Breaker

Every acquisition writes one append-only audit record binding the source filename to the content hash and receipt time — the evidence a CAP assessor asks for that a specific analyzer file was captured at a specific instant. A circuit breaker wraps the loop so that repeated transport failures (a rebooting instrument, a full disk) trip the watcher into a cool-down instead of hammering the analyzer and flooding the log. The audit sink itself is the immutable substrate described in implementing HIPAA-compliant audit trails in LIMS.

import json
import time


class CircuitBreaker:
    def __init__(self, threshold: int = 5, cooldown_seconds: float = 300.0):
        self.threshold = threshold
        self.cooldown = cooldown_seconds
        self.failures = 0
        self._opened_at: float = 0.0

    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self._opened_at = time.monotonic()
            logger.critical("Watcher circuit breaker OPEN — pausing %ss", self.cooldown)

    def record_success(self) -> None:
        self.failures = 0

    def is_open(self) -> bool:
        if self.failures < self.threshold:
            return False
        if time.monotonic() - self._opened_at >= self.cooldown:
            self.failures = 0  # half-open: allow one probe cycle
            return False
        return True


def write_audit(receipt: dict) -> None:
    record = {
        "event": "ingest.received",
        "source_name": receipt["source_name"],
        "sha256": receipt["sha256"],
        "quarantine_path": receipt["path"],
        "status": "ACQUIRED",
        "logged_at": receipt["acquired_at"],
    }
    logger.info(json.dumps(record))  # ship to an append-only / WORM sink

Wiring the pieces together, one watcher instance per analyzer runs the gate, acquisition, audit, and breaker in a single loop and enqueues each receipt for downstream processing:

async def run_watcher(target: FTPTarget, quarantine_dir: Path, queue: asyncio.Queue) -> None:
    gate, breaker, seen = StabilityGate(), CircuitBreaker(), set()
    async for client, remote_path, _info in poll_drop_directory(target):
        if breaker.is_open():
            continue
        try:
            if not await gate.is_quiescent(client, remote_path):
                continue
            if receipt := await acquire(client, remote_path, quarantine_dir, seen):
                write_audit(receipt)
                await queue.put(receipt)
            breaker.record_success()
        except (OSError, aioftp.StatusCodeError) as exc:
            logger.error("Acquisition error for %s: %s", remote_path, exc)
            breaker.record_failure()

Verification & Testing

Confirm each guarantee before pointing the watcher at a live analyzer feed:

• Partial-write rejection. Start writing a CSV to the drop directory in chunks with a pause between them; assert StabilityGate.is_quiescent returns False while the size is still growing and True only after the final write settles.
• Atomicity. Kill the process mid-download; assert only a .part temp file exists in quarantine and no downstream worker ever observed a partial *.csv — the os.replace never ran, so the final path was never created.
• Deduplication. Drop the same bytes twice; assert the second acquisition logs a duplicate, unlinks its temp file, removes the remote copy, and yields no receipt.
• Audit binding. After acquiring one file, run sha256sum against the quarantined copy and assert it equals the sha256 in the emitted ingest.received record.

A healthy single-file cycle produces log output in this shape:

INFO File still settling: /export/xn_0412.csv (attempt 1)
INFO {"event": "ingest.received", "source_name": "xn_0412.csv", "sha256": "9f8a2c1b4e5d...", "quarantine_path": "/lims/quarantine/xn_0412_9f8a2c1b.csv", "status": "ACQUIRED", "logged_at": "2026-07-02T10:30:00.004+00:00"}

Compliance Note

This implementation satisfies CLIA §493.1105, which requires that instrument records supporting patient results be retained and remain retrievable. Persisting the untouched analyzer bytes to quarantine before any parsing preserves the original source record, and the append-only ingest.received event satisfies 21 CFR Part 11.10(e): a computer-generated, time-stamped audit record that binds each captured file to an immutable SHA-256 fingerprint and cannot be altered after the fact. Retain the quarantined originals and their audit records on WORM-compliant storage for the longer of your state and accreditor requirements — typically 7–10 years — and restrict access to the quarantine sink under the same HIPAA §164.312(b) audit controls that govern the rest of the pipeline.

Troubleshooting

Related

• Serial & FTP Polling Architectures — the acquisition subsystem’s ingress/egress contract and exactly-once file-handling semantics this watcher implements.
• Handling HL7 ACK timeouts in clinical data pipelines — the sibling acquisition concern for the push side, where the LIMS must acknowledge a transmitted message in time.
• Converting legacy CSV instrument logs to HL7 ORU^R01 messages — the transformation task that consumes the CSV files this watcher quarantines.
• Async batch processing workers — the durable queue that drains acquisition receipts so a burst of exports never stalls the poller.
• Implementing HIPAA-compliant audit trails in LIMS — the append-only audit substrate the ingest.received event is written to.

Part of: Serial & FTP Polling Architectures.