Converting Legacy CSV Instrument Logs to HL7 ORU^R01 Messages

Problem Statement

A legacy chemistry or hematology analyzer that only knows how to drop a comma-separated log file into a network share cannot talk to a modern Laboratory Information Management System, which expects structured HL7 v2.x result messages. The gap looks trivial — split a row on commas, glue in some pipes — but a naive script double-posts results on retries, silently corrupts patient records when an unescaped | or ^ lands inside a name field, and leaves no defensible record of how a numeric value became an OBX-5. This page builds the deterministic converter that turns one flat CSV row into a valid ORU^R01 message: it acquires the file atomically, validates every field at the boundary, assembles the segments with correct HL7 escaping, and writes an immutable audit record for each message it emits.

Prerequisites

• Python 3.11+ — the converter uses asyncio, the walrus operator, and datetime.UTC.
• Pydantic v2 (pydantic>=2.6) for the typed row contract, and aiofiles (aiofiles>=23.2) for non-blocking file reads during acquisition.
• A stable instrument export contract. You must know the exact column order, encoding, and date format the analyzer emits. This converter is downstream of the serial and FTP polling architecture that captures the raw file and of the async batch processing workers that drain the durable queue — it never reads a socket itself.
• A resolved code map. Each instrument test code must already crosswalk to a LOINC code and a UCUM unit; see how to map LOINC codes to LIMS test panels for the mapping table this stage reads.
• Regulatory baseline: CLIA §493.1291 (test report integrity), 21 CFR Part 11.10(e) (immutable audit trail), and HIPAA §164.312(b) audit controls.

Field-level segment placement follows the canonical HL7 v2 Segment Mapping reference; this page implements one concrete instance of that mapping for a legacy CSV source, as one task within the CSV to HL7 Transformation stage.

Step 1: Acquire the File Atomically and Deduplicate

The single most common corruption source is reading a CSV while the analyzer is still writing it. Move each file from a staging directory into a processing directory with an atomic os.replace() only after confirming the file descriptor is closed, and derive a SHA-256 content hash so a file that is re-dropped by a retrying instrument resolves to the same identity and is never converted twice.

import asyncio
import hashlib
import logging
import os
from datetime import datetime, UTC
from pathlib import Path
from typing import AsyncGenerator, Any

import aiofiles

logger = logging.getLogger("csv_hl7_ingest")


async def atomic_file_poller(
    staging_dir: Path,
    processing_dir: Path,
    seen: set[str],
    poll_interval: float = 2.0,
    chunk_size: int = 8192,
) -> AsyncGenerator[dict[str, Any], None]:
    """Non-blocking poller: atomic move + SHA-256 content dedup."""
    while True:
        for csv_path in staging_dir.glob("*.csv"):
            try:
                digest = hashlib.sha256()
                async with aiofiles.open(csv_path, mode="rb") as f:
                    while chunk := await f.read(chunk_size):
                        digest.update(chunk)
                payload_hash = digest.hexdigest()

                if payload_hash in seen:
                    logger.info("Duplicate export ignored: %s", payload_hash[:12])
                    csv_path.unlink(missing_ok=True)
                    continue

                dest = processing_dir / f"{csv_path.stem}_{payload_hash[:8]}.csv"
                os.replace(csv_path, dest)  # atomic; fails if still write-locked
                seen.add(payload_hash)
                yield {
                    "path": dest,
                    "hash": payload_hash,
                    "acquired_at": datetime.now(UTC).isoformat(),
                }
            except (PermissionError, BlockingIOError):
                logger.debug("File locked by analyzer, deferring: %s", csv_path)
            except OSError as exc:
                logger.error("Acquisition failure for %s: %s", csv_path, exc)

        await asyncio.sleep(poll_interval)

In production the seen set is a Redis-backed cache with a 72-hour TTL so deduplication survives a process restart; the in-memory set shown here keeps the example runnable.

Step 2: Validate the Row at the Boundary with Pydantic

Legacy exports carry truncated MRNs, locale-specific decimal commas, empty result cells, and stray delimiters. A typed InstrumentRow rejects those at the edge instead of letting them reach segment assembly, where they would produce a syntactically valid but clinically wrong message. A row that fails validation is quarantined with the raw bytes attached — never dropped, never guessed at — matching the schema validation and error handling contract for this pipeline.

from pydantic import BaseModel, Field, ValidationError, field_validator


class InstrumentRow(BaseModel):
    """One validated result row from a legacy analyzer CSV export."""

    sample_id: str = Field(min_length=6, max_length=20)
    patient_mrn: str = Field(pattern=r"^\d{8,12}$")
    test_code: str = Field(min_length=2)
    loinc_code: str = Field(pattern=r"^\d{2,6}-\d$")
    result_value: float
    units: str = Field(min_length=1)
    reference_low: float
    reference_high: float
    run_timestamp: str = Field(pattern=r"^\d{14}$")  # YYYYMMDDHHMMSS

    @field_validator("units")
    @classmethod
    def _no_delimiters_in_units(cls, v: str) -> str:
        if any(c in v for c in "|^~\\&"):
            raise ValueError(f"units must not contain HL7 delimiters: {v!r}")
        return v


def validate_row(raw: dict[str, str]) -> InstrumentRow | None:
    try:
        return InstrumentRow(**raw)
    except ValidationError as exc:
        logger.warning("Row rejected to quarantine: %s", exc.json())
        return None

Step 3: Escape Delimiters, Then Assemble the ORU^R01

HL7 reserves five characters — |, ^, ~, \, and & — as delimiters. Any of them appearing inside a data field (a patient name, a unit, a comment) must be replaced with its escape sequence before the field is placed into a segment. Escape first, concatenate second: never build a segment by naive string joining, or a & in a name silently shifts every downstream field.

_HL7_ESCAPES = {
    "\\": "\\E\\",  # backslash first — it is the escape char itself
    "|": "\\F\\",
    "^": "\\S\\",
    "~": "\\R\\",
    "&": "\\T\\",
}


def escape_hl7(value: str) -> str:
    for raw, seq in _HL7_ESCAPES.items():
        value = value.replace(raw, seq)
    return value

With escaping in place, each InstrumentRow maps to a deterministic ORU^R01. The message-control ID in MSH-10 is derived from the file’s content hash and the row index, so regenerating the message from the same source produces byte-identical output — the property that lets audit replay reconcile a message years later.

def build_oru_r01(row: InstrumentRow, source_hash: str, row_index: int) -> str:
    """Assemble a deterministic HL7 v2.5 ORU^R01 from one validated row."""
    now = datetime.now(UTC).strftime("%Y%m%d%H%M%S")
    msg_id = f"{source_hash[:12]}-{row_index:04d}"  # deterministic control ID

    msh = (
        "MSH|^~\\&|ANALYZER_01|LAB_SYS|LIMS|HOSPITAL|"
        f"{now}||ORU^R01|{msg_id}|P|2.5"
    )
    pid = f"PID|1||{escape_hl7(row.patient_mrn)}^^^LAB^MR"
    obr = (
        f"OBR|1|{escape_hl7(row.sample_id)}||"
        f"{escape_hl7(row.loinc_code)}^{escape_hl7(row.test_code)}^LN|||"
        f"{row.run_timestamp}"
    )
    obx = (
        f"OBX|1|NM|{escape_hl7(row.loinc_code)}^{escape_hl7(row.test_code)}^LN||"
        f"{row.result_value}|{escape_hl7(row.units)}|"
        f"{row.reference_low}-{row.reference_high}|||F"
    )
    return "\r".join([msh, pid, obr, obx])  # segments end with CR, not LF

Step 4: Emit an Immutable Audit Record per Message

Every emitted message writes one append-only record binding the source file hash to the generated control ID. This is the evidence a CAP assessor asks for — proof that a specific analyzer file produced a specific result message — and it is the same immutable substrate described in implementing HIPAA-compliant audit trails in LIMS.

import json


def write_audit(row: InstrumentRow, source_hash: str, msg_id: str) -> dict[str, Any]:
    record = {
        "event": "csv_to_oru_transform",
        "source_file_hash": source_hash,
        "hl7_msg_id": msg_id,
        "sample_id": row.sample_id,
        "mrn_hash": hashlib.sha256(row.patient_mrn.encode()).hexdigest(),
        "loinc": row.loinc_code,
        "status": "EMITTED",
        "logged_at": datetime.now(UTC).isoformat(),
    }
    logger.info(json.dumps(record))  # ship to an append-only / WORM sink
    return record

Verification & Testing

Confirm correct behaviour before wiring the converter to a live LIMS feed:

• Determinism. Call build_oru_r01 twice with the same row, source_hash, and row_index; assert the MSH-10 field and every segment except MSH-7 (the send timestamp) are byte-identical.
• Escaping round-trip. Feed a units value containing a stray & through escape_hl7; assert the output contains \T\ and that no bare & remains in the assembled OBX.
• Boundary rejection. Build an InstrumentRow from a payload with a 4-digit MRN or an empty result_value; assert validate_row returns None and logs a quarantine record rather than raising.
• Deduplication. Run atomic_file_poller over a directory where the same bytes are dropped twice; assert the second file is unlinked and yields nothing.

A correctly assembled message parses cleanly and shows the expected segment order:

MSH|^~\&|ANALYZER_01|LAB_SYS|LIMS|HOSPITAL|20260702103000||ORU^R01|a1b2c3d4e5f6-0007|P|2.5
PID|1||000123456789^^^LAB^MR
OBR|1|SPEC00042||2951-2^NA^LN|||20260702102800
OBX|1|NM|2951-2^NA^LN||139|mmol/L|136-145|||F

Compliance Note

This implementation satisfies CLIA §493.1291, which requires that patient test reports be accurate and attributable to the correct specimen and patient. Boundary validation guarantees the sample_id-to-patient_mrn binding before a message is built, and the deterministic, hash-linked audit record satisfies 21 CFR Part 11.10(e): a computer-generated, time-stamped record of the transformation that cannot be altered after the fact and that ties each ORU^R01 back to the exact source file that produced it. Retain the audit records on WORM-compliant storage for the longer of your state and accreditor requirements — typically 7–10 years — hashing the MRN so the audit sink honours HIPAA minimum-necessary.

Troubleshooting

Related

• CSV to HL7 Transformation — the transformation stage’s ingress/egress contract and segment-map specification this converter implements.
• HL7 v2 Segment Mapping — the canonical field-to-segment reference for MSH, PID, OBR, and OBX placement.
• Validating ASTM E1394 instrument output with Python — the sibling converter for instruments that speak ASTM frames instead of flat CSV.
• Handling HL7 ACK timeouts in clinical data pipelines — what happens after this message is transmitted and the LIMS must acknowledge it.
• How to map LOINC codes to LIMS test panels — the crosswalk table that resolves each instrument test code to the LOINC used in OBR-4 and OBX-3.

Part of: CSV to HL7 Transformation.