HL7 v2 ADT feeds: from pipe-delimited events to an encounters table

// healthcare · hl7 v2

TL;DR

An ADT feed is a stream of events about a visit, not a table of visits. Parsing a message takes about 40 lines of Python. Most of the work is elsewhere: ACK only after the raw message is stored, key visits by facility + visit number, and rebuild each visit from all of its events so that cancellations and out-of-order delivery come out right.

The problem

The Roster API series answers who is ours. An ADT feed answers what just happened to them: admitted last night, transferred to ICU, discharged to a skilled nursing facility. Care management teams in value-based care live on this: a discharge without a follow-up visit within 7 days is a readmission waiting to happen.

Since May 2021, hospitals that take Medicare must send these notifications electronically (the Conditions of Participation change in the CMS Interoperability and Patient Access rule, CMS-9115-F). In practice they arrive as HL7 v2 ADT messages, straight from the hospital or through an HIE.

This is what one looks like (synthetic, segments on separate lines for readability; on the wire they are separated by \r):

MSH|^~\&|EPIC|MERCY_HOSP|GLB|VBC_ORG|20261030221504||ADT^A01^ADT_A01|MSG00001|P|2.5.1
EVN|A01|20261030221500
PID|1||884213^^^MERCY_HOSP^MR~1EG4TE5MK73^^^CMS^MC||Rivera^Ana^L||19520314|F|||12 Oak St^^Springfield^MA^01103
PV1|1|I|4W^412^B^MERCY_HOSP||||1234567893^Chen^David^^^^^^NPI|||MED||||7|||||V556677|||||||||||||||||||||||||202610302158

The four segments you need first:

Segment	What it holds	Fields worth knowing
`MSH`	Envelope	`MSH-4` sending facility, `MSH-9` message type (`ADT^A01`), `MSH-10` control ID, `MSH-12` version
`EVN`	Event	`EVN-2` when the event was recorded
`PID`	Patient	`PID-3` identifiers (repeating), `PID-5` name, `PID-7` birth date
`PV1`	Visit	`PV1-2` patient class, `PV1-3` location, `PV1-19` visit number, `PV1-36` discharge disposition, `PV1-44`/`PV1-45` admit/discharge time

And the trigger events that matter for an encounters table:

Trigger	Meaning
`A01`	Admit (inpatient)
`A04`	Register (ED, outpatient)
`A02`	Transfer
`A03`	Discharge
`A08`	Update patient/visit information
`A11`	Cancel admit
`A13`	Cancel discharge
`A06` / `A07`	Outpatient → inpatient / inpatient → outpatient
`A40`	Merge two patient identities

What I tried

`line.split("|")` and indexing

It almost works, which is why it’s dangerous:

MSH is off by one. MSH-1 is the | character, so in the split list MSH-9 sits at index 8 while PV1-19 sits at index 19. Every MSH lookup was wrong by one.
Repetitions. PID-3 held two identifiers separated by ~. Taking the first component of the whole field gave me the MRN by luck, and the Medicare Beneficiary Identifier never.
Escapes. A last name of Smith&Jones arrives as Smith\T\Jones, because & is a delimiter.
Segment terminators. The spec says \r. Files that went through an SFTP hop or a Windows box arrive with \r\n or \n.

One row per message

I inserted every A01 as an encounter and updated it on A03. Then:

An A11 (cancel admit) arrived for a patient who was registered to the wrong bed. The encounter stayed: a phantom admission in the care management worklist.
An A03 arrived before its A01 (two interface engines, two queues). The update hit nothing, then the insert created an encounter that was never discharged.
An A13 (cancel discharge) reopened a visit, and the next A03 carried the real discharge time. My update-in-place logic had already sent a follow-up task for the first one.

Messages are events. Treating them as rows loses the order and the cancellations.

What actually worked

1. A small parser that respects the structure

# hl7v2.py
import re
from datetime import datetime

ESCAPES = {"F": "|", "S": "^", "T": "&", "R": "~", "E": "\\"}
TS_FORMATS = {4: "%Y", 6: "%Y%m", 8: "%Y%m%d", 10: "%Y%m%d%H", 12: "%Y%m%d%H%M", 14: "%Y%m%d%H%M%S"}


def unescape(value: str) -> str:
    return re.sub(r"\\([FSTRE])\\", lambda m: ESCAPES[m.group(1)], value)


def parse(raw: str) -> dict[str, list[list[str]]]:
    segments: dict[str, list[list[str]]] = {}
    for line in re.split(r"\r\n|\r|\n", raw.strip()):
        if not line:
            continue
        fields = line.split("|")
        if fields[0] == "MSH":
            fields.insert(1, "|")  # MSH-1 is the field separator itself
        segments.setdefault(fields[0], []).append(fields)
    return segments


def get_raw(msg, seg: str, field: int) -> str:
    try:
        return msg[seg][0][field]
    except (KeyError, IndexError):
        return ""


def get(msg, seg: str, field: int, comp: int = 1, rep: int = 1) -> str:
    try:
        value = msg[seg][0][field]
        value = value.split("~")[rep - 1]
        value = value.split("^")[comp - 1]
    except (KeyError, IndexError):
        return ""
    return unescape(value)


def identifiers(msg) -> dict[str, str]:
    """PID-3 repetitions keyed by identifier type code (CX.5)."""
    out = {}
    for rep in get_raw(msg, "PID", 3).split("~"):
        comps = rep.split("^")
        if comps[0]:
            out[comps[4] if len(comps) > 4 else ""] = unescape(comps[0])
    return out


def parse_ts(value: str) -> datetime | None:
    m = re.fullmatch(r"(\d{4,14})(?:\.\d+)?([+-]\d{4})?", value or "")
    if not m:
        return None
    digits, tz = m.groups()
    fmt = TS_FORMATS[len(digits)]
    if tz:
        return datetime.strptime(digits + tz, fmt + "%z")
    return datetime.strptime(digits, fmt)

On the sample message:

>>> msg = parse(raw)
>>> get(msg, "MSH", 9), get(msg, "MSH", 9, comp=2), get(msg, "PV1", 19)
('ADT', 'A01', 'V556677')
>>> identifiers(msg)
{'MR': '884213', 'MC': '1EG4TE5MK73'}
>>> parse_ts(get(msg, "PV1", 44))
datetime.datetime(2026, 10, 30, 21, 58)

What this deliberately ignores: custom delimiters (it assumes the standard ^~\& from MSH-2), sub-components (&), and the rarer escapes like \X0D\. Read MSH-2 and MSH-18 (character set) before you assume. Many feeds are Latin-1, not UTF-8. For production-grade parsing of every edge, use a library. For a feed you control end to end, this is enough to see what is inside.

2. Receive over MLLP, ACK after storing

ADT usually travels over MLLP: a TCP connection where each message is framed by 0x0B at the start and 0x1C 0x0D at the end. The sender waits for an ACK before sending the next message. Port 2575 is the registered default.

# mllp_server.py
import asyncio
from datetime import datetime

from hl7v2 import get, get_raw, parse

SB, EB, CR = b"\x0b", b"\x1c", b"\r"


def ack(msg, code: str) -> str:
    msh = msg["MSH"][0]
    now = datetime.now().strftime("%Y%m%d%H%M%S")
    trigger = get(msg, "MSH", 9, comp=2)
    return "\r".join([
        f"MSH|^~\\&|{msh[5]}|{msh[6]}|{msh[3]}|{msh[4]}|{now}||ACK^{trigger}^ACK"
        f"|ACK{msh[10]}|P|{get_raw(msg, 'MSH', 12)}",
        f"MSA|{code}|{msh[10]}",
    ]) + "\r"


def land(raw: str, msg) -> None:
    # Append-only write keyed by (sending facility, control id). Replace with your store.
    with open("landing.hl7", "a", encoding="utf-8") as f:
        f.write(raw.replace("\r", "\n") + "\n")


async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    while True:
        try:
            await reader.readuntil(SB)
            frame = await reader.readuntil(EB + CR)
        except asyncio.IncompleteReadError:
            break
        raw = frame[:-2].decode("latin-1")
        msg = parse(raw)
        try:
            land(raw, msg)
            reply = ack(msg, "AA")
        except Exception:
            reply = ack(msg, "AE")
        writer.write(SB + reply.encode("latin-1") + EB + CR)
        await writer.drain()
    writer.close()


async def main() -> None:
    server = await asyncio.start_server(handle, "0.0.0.0", 2575)
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())

The reply for the sample:

MSH|^~\&|GLB|VBC_ORG|EPIC|MERCY_HOSP|20261031172826||ACK^A01^ACK|ACKMSG00001|P|2.5.1
MSA|AA|MSG00001

Two rules are worth more than the code:

AA means “I have it”, not “I processed it”. Store the raw message durably, then ACK. If you parse, map and load before ACKing, one bad message blocks the whole feed, because the sender keeps retrying it.
Expect duplicates. If an ACK gets lost, the sender resends. Deduplicate on (MSH-4, MSH-10) when you land the message.

3. Rebuild each visit from its events

A visit is identified by PV1-19, and only within the sending facility, so the key is (MSH-4, PV1-19). Some feeds leave PV1-19 empty. For those, fall back to MRN + admit time, and count how often you need the fallback.

Turn every landed message into an event:

# encounters.py
from hl7v2 import get, identifiers, parse, parse_ts


def to_event(raw: str) -> dict:
    msg = parse(raw)
    ids = identifiers(msg)
    return {
        "facility": get(msg, "MSH", 4),
        "visit": get(msg, "PV1", 19),
        "trigger": get(msg, "MSH", 9, comp=2),
        "event_ts": parse_ts(get(msg, "EVN", 2)) or parse_ts(get(msg, "MSH", 7)),
        "control_id": get(msg, "MSH", 10),
        "mrn": ids.get("MR"),
        "mbi": ids.get("MC"),
        "patient_class": get(msg, "PV1", 2),
        "location": get(msg, "PV1", 3),
        "admit_ts": parse_ts(get(msg, "PV1", 44)),
        "discharge_ts": parse_ts(get(msg, "PV1", 45)),
        "disposition": get(msg, "PV1", 36),
    }

Then fold all events of one visit, sorted by when they happened, not by when they arrived:

def fold(events: list[dict]) -> dict:
    """Rebuild one visit's state from all of its events, in event-time order."""
    enc = {"status": None, "admit_ts": None, "discharge_ts": None}
    for e in sorted(events, key=lambda e: (e["event_ts"], e["control_id"])):
        match e["trigger"]:
            case "A01" | "A04":
                enc.update(status="active", admit_ts=e["admit_ts"], discharge_ts=None)
            case "A03":
                enc.update(status="discharged", discharge_ts=e["discharge_ts"],
                           disposition=e["disposition"])
            case "A11":
                enc.update(status="cancelled", admit_ts=None)
            case "A13":
                enc.update(status="active", discharge_ts=None, disposition=None)
        for key in ("mrn", "mbi", "patient_class", "location"):
            if e[key]:
                enc[key] = e[key]
    return enc

Feed it the messy case from above, out of order: admit, discharge, cancel discharge, real discharge.

>>> fold([a03_second, a13, a01, a03_first])
{'status': 'discharged', 'admit_ts': datetime(2026, 10, 30, 21, 58),
 'discharge_ts': datetime(2026, 11, 3, 8, 45), 'mrn': '884213',
 'mbi': '1EG4TE5MK73', 'patient_class': 'I', 'location': '4W', 'disposition': '06'}
>>> fold([a01, a11])
{'status': 'cancelled', 'admit_ts': None, 'discharge_ts': None, ...}

Every new message for a visit triggers a re-fold of that visit, never a patch of the current row. Arrival order stops mattering. A late A11 turns the visit into cancelled the next time it is folded. It is the same idea as the versioned Group in part 3 of the roster series: keep the log, derive the state.

Things the fold leaves for the next iteration:

Time zones. Some senders include an offset in EVN-2, some don’t. Normalize them before sorting, because Python refuses to compare aware and naive datetimes.
A06/A07 change the patient class. Treat them like A08 unless observation stays matter to you, and in Medicare they usually do.
A40 merges belong to identity resolution, not to the visit fold. Apply them to the MRN crosswalk and re-fold the affected visits.

If the next stop is FHIR, the HL7 v2-to-FHIR IG maps PV1 to Encounter (PV1-2 → Encounter.class, PV1-44/45 → Encounter.period). The folded visit is what you map. Mapping each message is the wrong unit.

If the folded visits end up in Tuva, they land next to the claims-derived encounters without being matched to them. One hospital stay can appear twice (Tuva’s encounter grouper only groups claims).

Key takeaways

MSH-1 is the separator itself, so MSH field numbers are shifted by one compared with every other segment. Handle it once in the parser.
ACK after a durable write of the raw message, not after processing, and deduplicate on facility + control ID.
Key visits by sending facility + PV1-19. Rebuild each visit from all of its events in event-time order so cancellations (A11, A13) and out-of-order delivery come out right.

References

01 Nov 2026 · #healthcare #python #dataengineering

GLB_