From c64abfd7218b154b72dc6b25ff17cf36c90d8a79 Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:39:56 +0530 Subject: [PATCH] fix(parsers): honour HL7 v2 MSH delimiters, repetitions and escape sequences The HL7 v2 parser hard-coded "|" and "^", so it ignored the delimiters a message declares in MSH-1/MSH-2. It never split "~" repetitions and never decoded escape sequences. As a result PID-3 "12345~98765" came out as the patient id, "\S\" / "\T\" stayed in text values, and a message using "#" as its field separator parsed to zero messages. Each MSH segment now sets the delimiters for its message: the field separator comes from seg[3] and the component, repetition, escape and subcomponent characters from MSH-2. Any that are missing default to ^~\&. Those delimiters stay in force until the next MSH. Segment ids are detected with seg[:3], and a recognised segment must be followed by the current field separator. Single-valued columns (ids, names, codes, dates, units, status) take the first repetition. OBX-5 is a repeating field, so the observation value keeps every repetition, joined with "~". Fields are split before escapes are decoded, so an escaped delimiter such as \R\ never splits a value. \F\ \S\ \T\ \R\ \E\ are decoded using the message's own delimiters, and other escapes (\H\, \N\, \Xhh\, \.br\) are kept verbatim. Whole-field values are re-emitted with the standard ^ / & / ~ separators. Standard |^~\& messages without escapes, and without repetitions outside OBX-5, produce the same frames, metadata and warnings as before. Closes #261 --- src/freshdata/parsers/hl7v2.py | 224 +++++++++++----- tests/parsers/test_hl7v2_delimiters.py | 348 +++++++++++++++++++++++++ 2 files changed, 512 insertions(+), 60 deletions(-) create mode 100644 tests/parsers/test_hl7v2_delimiters.py diff --git a/src/freshdata/parsers/hl7v2.py b/src/freshdata/parsers/hl7v2.py index 2c44f020..0605d8c6 100644 --- a/src/freshdata/parsers/hl7v2.py +++ b/src/freshdata/parsers/hl7v2.py @@ -5,6 +5,17 @@ shaped for the healthcare domain pack. Observation code systems are mapped to their canonical URIs (LOINC ``http://loinc.org``, SNOMED ``http://snomed.info/sct``, ICD-10). +Delimiters are read from each message's MSH segment (HL7 v2 Chapter 2): MSH-1 is the +field separator and MSH-2 holds the component, repetition, escape and subcomponent +characters (``^~\\&`` when absent). They apply to every segment up to the next MSH. +Single-valued output columns take the first repetition of a field; OBX-5 (observation +value) is a repeating field and keeps every repetition. Fields are split first, then the +delimiter escape sequences ``\\F\\ \\S\\ \\T\\ \\R\\ \\E\\`` are decoded to the literal +characters. Other escapes (``\\H\\``, ``\\N\\``, ``\\X..\\``, ``\\.br\\`` ...) are left +verbatim. Whole-field values (e.g. OBX-5) are re-emitted with the standard ``^`` / ``&`` +/ ``~`` component / subcomponent / repetition separators, so output does not depend on a +message's custom delimiters. + This is a structural parser for the common segments, not a full HL7 v2 conformance engine: unrecognized segments are counted in :attr:`ParseResult.warnings`, and the OBX component layout follows the usual ORU convention. @@ -12,7 +23,7 @@ from __future__ import annotations -from typing import Any +from typing import Any, NamedTuple import pandas as pd @@ -31,21 +42,90 @@ } # PV1-2 patient class -> human label. -_PATIENT_CLASS = {"I": "inpatient", "O": "outpatient", "E": "emergency", - "P": "preadmit", "R": "recurring", "B": "obstetrics"} +_PATIENT_CLASS = { + "I": "inpatient", + "O": "outpatient", + "E": "emergency", + "P": "preadmit", + "R": "recurring", + "B": "obstetrics", +} +_KNOWN_SEGMENTS = frozenset({"PID", "PV1", "OBR", "OBX"}) + + +class _Delimiters(NamedTuple): + """The five HL7 v2 message delimiters declared by MSH-1 / MSH-2.""" + + field: str = "|" + component: str = "^" + repetition: str = "~" + escape: str = "\\" + subcomponent: str = "&" + + @classmethod + def from_msh(cls, segment: str) -> _Delimiters: + """Read MSH-1 (``segment[3]``) and MSH-2; missing characters use the defaults.""" + default = cls() + if len(segment) < 4: + return default + field_sep = segment[3] + encoding = segment[4:].split(field_sep, 1)[0] + chars = [encoding[i] if i < len(encoding) else d for i, d in enumerate(default[1:])] + return cls(field_sep, *chars) + + def decode(self, text: str) -> str: + """Decode the delimiter escape sequences; leave any other escape verbatim.""" + esc = self.escape + if esc not in text: + return text + literal = { + "F": self.field, + "S": self.component, + "T": self.subcomponent, + "R": self.repetition, + "E": esc, + } + out: list[str] = [] + i = 0 + while True: + start = text.find(esc, i) + end = text.find(esc, start + 1) if start != -1 else -1 + if end == -1: + out.append(text[i:]) + return "".join(out) + out.append(text[i:start]) + code = text[start + 1 : end] + out.append(literal.get(code, text[start : end + 1])) + i = end + 1 + + def value(self, field: str, *, repeating: bool = False) -> str: + """Decoded *field* re-emitted with the standard ``^`` / ``&`` / ``~`` separators. + + Only the first repetition is kept unless *repeating* is true, in which case every + repetition is kept and joined with ``~``. Splitting happens before escapes are + decoded, so an escaped delimiter (e.g. ``\\R\\``) never splits the value. + """ + reps = field.strip().split(self.repetition) + return "~".join( + "^".join( + "&".join(self.decode(sub) for sub in comp.split(self.subcomponent)) + for comp in rep.split(self.component) + ).strip() + for rep in (reps if repeating else reps[:1]) + ) -def _comp(field: str, n: int) -> str: - """1-based HL7 component *n* of a field (``DOE^JOHN`` -> 1='DOE').""" - if not field: - return "" - parts = field.split("^") - return parts[n - 1].strip() if 0 < n <= len(parts) else "" + def component_of(self, field: str, n: int) -> str: + """1-based component *n* of the first repetition (``DOE^JOHN`` -> 1='DOE').""" + if not field: + return "" + parts = field.strip().split(self.repetition, 1)[0].split(self.component) + return self.decode(parts[n - 1]).strip() if 0 < n <= len(parts) else "" def _field(fields: list[str], n: int) -> str: - """Field *n* of a split segment (``fields[0]`` is the segment id).""" - return fields[n].strip() if n < len(fields) else "" + """Raw field *n* of a split segment (``fields[0]`` is the segment id).""" + return fields[n] if n < len(fields) else "" class HL7v2Parser(Parser): @@ -57,8 +137,11 @@ class HL7v2Parser(Parser): def parse(self, source: Any) -> ParseResult: text = self.read_text(source) # HL7 segments are CR-separated; tolerate LF / CRLF too. - segments = [s for s in text.replace("\r\n", "\r").replace("\n", "\r").split("\r") - if s.strip()] + segments = [ + s.lstrip() + for s in text.replace("\r\n", "\r").replace("\n", "\r").split("\r") + if s.strip() + ] patients: list[dict[str, Any]] = [] encounters: list[dict[str, Any]] = [] @@ -71,64 +154,85 @@ def parse(self, source: Any) -> ParseResult: current_pid: str | None = None current_order: str | None = None message_type = "" + delim = _Delimiters() for seg in segments: - fields = seg.split("|") - seg_id = fields[0].strip() + seg_id = seg[:3] + is_msh = seg_id == "MSH" and (len(seg) == 3 or not seg[3].isalnum()) - if seg_id == "MSH": + if is_msh: + delim = _Delimiters.from_msh(seg) msg_index += 1 current_pid = None current_order = None # MSH is offset by one (MSH-1 is the field separator itself), so - # MSH-9 (message type, e.g. "ADT^A01") is fields[8]. - message_type = fields[8].strip() if len(fields) > 8 else "" + # MSH-2 is fields[1] and MSH-9 (message type, e.g. "ADT^A01") is fields[8]. + message_type = delim.value(_field(seg.split(delim.field), 8)) + continue + + d = delim + fields = seg.split(d.field) + if seg_id not in _KNOWN_SEGMENTS or (len(seg) > 3 and seg[3] != d.field): + key = fields[0].strip() + unknown[key] = unknown.get(key, 0) + 1 elif seg_id == "PID": - current_pid = _comp(_field(fields, 3), 1) or f"MSG{msg_index}" - patients.append({ - "patient_id": current_pid, - "family_name": _comp(_field(fields, 5), 1), - "given_name": _comp(_field(fields, 5), 2), - "birth_date": _field(fields, 7), - "gender": _field(fields, 8), - }) + current_pid = d.component_of(_field(fields, 3), 1) or f"MSG{msg_index}" + patients.append( + { + "patient_id": current_pid, + "family_name": d.component_of(_field(fields, 5), 1), + "given_name": d.component_of(_field(fields, 5), 2), + "birth_date": d.value(_field(fields, 7)), + "gender": d.value(_field(fields, 8)), + } + ) elif seg_id == "PV1": - encounters.append({ - "patient_id": current_pid or f"MSG{msg_index}", - "visit_number": _comp(_field(fields, 19), 1), - "class_code": _field(fields, 2), - "class": _PATIENT_CLASS.get(_field(fields, 2).upper(), _field(fields, 2)), - "location": _comp(_field(fields, 3), 1), - }) + class_code = d.value(_field(fields, 2)) + encounters.append( + { + "patient_id": current_pid or f"MSG{msg_index}", + "visit_number": d.component_of(_field(fields, 19), 1), + "class_code": class_code, + "class": _PATIENT_CLASS.get(class_code.upper(), class_code), + "location": d.component_of(_field(fields, 3), 1), + } + ) elif seg_id == "OBR": service = _field(fields, 4) - current_order = (_comp(_field(fields, 3), 1) - or _comp(_field(fields, 2), 1) or None) - orders.append({ - "patient_id": current_pid or f"MSG{msg_index}", - "order_id": current_order, - "placer_order": _comp(_field(fields, 2), 1), - "filler_order": _comp(_field(fields, 3), 1), - "service_code": _comp(service, 1), - "service_display": _comp(service, 2), - "service_system": _comp(service, 3), - "observed_at": _field(fields, 7), - }) - elif seg_id == "OBX": - system = _comp(_field(fields, 3), 3) - observations.append({ - "patient_id": current_pid or f"MSG{msg_index}", - "order_id": current_order, - "code": _comp(_field(fields, 3), 1), - "display": _comp(_field(fields, 3), 2), - "code_system": _CODE_SYSTEMS.get(system.upper(), system), - "value": _field(fields, 5), - "unit": _field(fields, 6), - "status": _field(fields, 11), - "observed_at": _field(fields, 14), - }) - else: - unknown[seg_id] = unknown.get(seg_id, 0) + 1 + current_order = ( + d.component_of(_field(fields, 3), 1) + or d.component_of(_field(fields, 2), 1) + or None + ) + orders.append( + { + "patient_id": current_pid or f"MSG{msg_index}", + "order_id": current_order, + "placer_order": d.component_of(_field(fields, 2), 1), + "filler_order": d.component_of(_field(fields, 3), 1), + "service_code": d.component_of(service, 1), + "service_display": d.component_of(service, 2), + "service_system": d.component_of(service, 3), + "observed_at": d.value(_field(fields, 7)), + } + ) + else: # OBX + code = _field(fields, 3) + system = d.component_of(code, 3) + observations.append( + { + "patient_id": current_pid or f"MSG{msg_index}", + "order_id": current_order, + "code": d.component_of(code, 1), + "display": d.component_of(code, 2), + "code_system": _CODE_SYSTEMS.get(system.upper(), system), + # OBX-5 repeats: keep every value, joined with "~". + "value": d.value(_field(fields, 5), repeating=True), + "unit": d.value(_field(fields, 6)), + "status": d.value(_field(fields, 11)), + "observed_at": d.value(_field(fields, 14)), + } + ) if unknown: listed = ", ".join(f"{k}({v})" for k, v in sorted(unknown.items())) diff --git a/tests/parsers/test_hl7v2_delimiters.py b/tests/parsers/test_hl7v2_delimiters.py new file mode 100644 index 00000000..4191bc24 --- /dev/null +++ b/tests/parsers/test_hl7v2_delimiters.py @@ -0,0 +1,348 @@ +"""HL7 v2 delimiter handling: MSH-1/MSH-2, repetitions and escape sequences (#261).""" + +from __future__ import annotations + +import pandas as pd +import pytest + +import freshdata as fd +from freshdata.parsers.hl7v2 import HL7v2Parser + + +def _parse(text: str): + return HL7v2Parser().parse(text) + + +# -- issue reproduction ------------------------------------------------------------- + +ISSUE_MSG = ( + "MSH|^~\\&|A|B|C|D|202401011200||ORU^R01|1|P|2.5\r" + "PID|1||12345~98765||DOE^JANE||19800101|F\r" + "OBX|1|ST|8867-4^HR^LN||ratio 5\\S\\3 \\T\\ note||||||F" +) + + +def test_issue_repro_first_repetition_and_escapes(): + result = fd.parse_domain(ISSUE_MSG, format="hl7v2") + assert result.frames["patient"]["patient_id"].iloc[0] == "12345" + assert result.frames["observation"]["value"].iloc[0] == "ratio 5^3 & note" + + +def test_issue_repro_hash_field_separator_honoured(): + result = fd.parse_domain(ISSUE_MSG.replace("|", "#"), format="hl7v2") + assert result.metadata["messages"] == 1 + assert result.metadata["message_type"] == "ORU^R01" + assert result.warnings == [] + assert result.frames["patient"]["patient_id"].iloc[0] == "12345" + assert result.frames["observation"]["value"].iloc[0] == "ratio 5^3 & note" + + +# -- custom delimiters -------------------------------------------------------------- + +CUSTOM = "\r".join( + [ + "MSH#$*@!#LAB#HOSP#EHR#CLINIC#20240101##ORU$R01#MSG1#P#2.5", + "PID#1##111$$$MRN*222$$$SSN##DOE$JOHN##19700101#M", + "PV1#1#I#ICU$101" + "#" * 16 + "VN1", # PV1-19 visit number + "OBR#1#PL1#FL1#24323-8$Panel$LN###20240101080000", + "OBX#1#CE#8867-4$Heart rate$LN##a$b!c*second#/min#####F", + ] +) + + +def test_custom_delimiters_from_msh_1_and_msh_2(): + result = _parse(CUSTOM) + assert result.metadata == {"messages": 1, "message_type": "ORU^R01"} + assert result.warnings == [] + + patient = result.frames["patient"].iloc[0] + assert patient["patient_id"] == "111" + assert (patient["family_name"], patient["given_name"]) == ("DOE", "JOHN") + assert (patient["birth_date"], patient["gender"]) == ("19700101", "M") + + encounter = result.frames["encounter"].iloc[0] + assert (encounter["class"], encounter["location"]) == ("inpatient", "ICU") + assert encounter["visit_number"] == "VN1" + + order = result.frames["order"].iloc[0] + assert (order["order_id"], order["service_display"]) == ("FL1", "Panel") + + obs = result.frames["observation"].iloc[0] + assert (obs["code"], obs["display"]) == ("8867-4", "Heart rate") + assert obs["code_system"] == "http://loinc.org" + # Whole-field values use the standard ^ / & / ~ separators; OBX-5 keeps repetitions. + assert obs["value"] == "a^b&c~second" + assert (obs["unit"], obs["status"]) == ("/min", "F") + + +def test_partial_msh_2_defaults_missing_encoding_characters(): + # MSH-2 only declares the component separator; ~ \ & fall back to the defaults. + msg = "MSH|$|A\rPID|||1~2||DOE$JOHN\rOBX|1|ST|X||a\\S\\b&c" + result = _parse(msg) + patient = result.frames["patient"].iloc[0] + assert (patient["patient_id"], patient["given_name"]) == ("1", "JOHN") + assert result.frames["observation"]["value"].iloc[0] == "a$b&c" + + +# -- repetitions -------------------------------------------------------------------- + + +def test_repetitions_take_first_value_for_single_valued_columns(): + msg = "\r".join( + [ + r"MSH|^~\&|A|B|C|D|20240101||ADT^A01~ADT^A04|1|P|2.5", + "PID|1||A1^^^MRN~B2^^^SSN||DOE^JOHN~ALIAS^JACK||19700101~19700102|M~F", + "PV1|1|I~O|ICU^1~ER^2", + "OBX|1|NM|1-1^X^LN~2-2^Y^SCT||5~6|mg~g|||||F~C", + ] + ) + result = _parse(msg) + assert result.metadata["message_type"] == "ADT^A01" + patient = result.frames["patient"].iloc[0] + assert patient.to_dict() == { + "patient_id": "A1", + "family_name": "DOE", + "given_name": "JOHN", + "birth_date": "19700101", + "gender": "M", + } + encounter = result.frames["encounter"].iloc[0] + assert (encounter["class_code"], encounter["class"]) == ("I", "inpatient") + assert encounter["location"] == "ICU" + obs = result.frames["observation"].iloc[0] + assert (obs["code"], obs["code_system"]) == ("1-1", "http://loinc.org") + # OBX-5 is a repeating field and keeps every repetition (see tests below). + assert (obs["value"], obs["unit"], obs["status"]) == ("5~6", "mg", "F") + + +# -- OBX-5 repetitions -------------------------------------------------------------- + + +def test_obx5_keeps_every_repetition_with_standard_delimiters(): + msg = "\r".join( + [ + r"MSH|^~\&|A", + "OBX|1|CE|C||1^One^LN~2^Two&x^SCT~ 3 |mg~g", + "OBX|2|NM|C||5~6", + ] + ) + obs = _parse(msg).frames["observation"] + assert list(obs["value"]) == ["1^One^LN~2^Two&x^SCT~3", "5~6"] + assert list(obs["unit"]) == ["mg", ""] # OBX-6 is still first repetition only + + +def test_obx5_custom_repetition_character_reemitted_as_tilde(): + msg = "MSH#$*@!#A\rOBX#1#CE#C##1$One$LN*2$Two!x$SCT*5#mg*g" + obs = _parse(msg).frames["observation"].iloc[0] + assert obs["value"] == "1^One^LN~2^Two&x^SCT~5" + assert obs["unit"] == "mg" + + +def test_escaped_repetition_decoded_after_split_not_split_on(): + std = "\r".join([r"MSH|^~\&|A", r"OBX|1|ST|C||a\R\b~c|u\R\v~w"]) + obs = _parse(std).frames["observation"].iloc[0] + assert obs["value"] == "a~b~c" + assert obs["unit"] == "u~v" # only one repetition: "\R\" did not split + custom = "MSH#$*@!#A\rOBX#1#ST#C##a@R@b*c" + # @R@ decodes to the message's repetition character, after the split on "*". + assert _parse(custom).frames["observation"]["value"].iloc[0] == "a*b~c" + + +# -- escape sequences --------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("escaped", "decoded"), + [ + (r"a\F\b", "a|b"), + (r"a\S\b", "a^b"), + (r"a\T\b", "a&b"), + (r"a\R\b", "a~b"), + (r"a\E\b", "a\\b"), + (r"\E\S\E\ ", r"\S\ "), # \E\ is decoded once, not re-scanned + (r"x\H\bold\N\y", r"x\H\bold\N\y"), # unknown escapes kept verbatim + (r"a\X0D\b", r"a\X0D\b"), + (r"a\.br\b", r"a\.br\b"), + ("trailing\\", "trailing\\"), # unterminated escape left as-is + ], +) +def test_escape_sequences_decoded_in_values(escaped, decoded): + msg = f"MSH|^~\\&|A\rPID|||ID1||{escaped}^{escaped}\rOBX|1|ST|C||{escaped}" + result = _parse(msg) + assert result.frames["observation"]["value"].iloc[0] == decoded.strip() + patient = result.frames["patient"].iloc[0] + assert patient["family_name"] == decoded.strip() + assert patient["given_name"] == decoded.strip() + + +def test_escapes_use_the_message_delimiters(): + msg = "MSH#$*@!#A\rOBX#1#ST#C##p@F@q@S@r@T@s@R@t@E@u\\S\\v" + value = _parse(msg).frames["observation"]["value"].iloc[0] + assert value == "p#q$r!s*t@u\\S\\v" + + +def test_escaped_delimiters_do_not_split_components(): + msg = "MSH|^~\\&|A\rPID|||ID\\S\\1^^^MRN||DOE\\F\\SMITH^JOHN\\R\\PAUL" + patient = _parse(msg).frames["patient"].iloc[0] + assert patient["patient_id"] == "ID^1" + assert patient["family_name"] == "DOE|SMITH" + assert patient["given_name"] == "JOHN~PAUL" + + +# -- multiple messages -------------------------------------------------------------- + + +def test_each_message_uses_its_own_delimiters(): + msg = "\r".join( + [ + r"MSH|^~\&|A|B|C|D|20240101||ADT^A01|1|P|2.5", + "PID|||P1~X||DOE^JOHN", + "OBX|1|ST|C1^One^LN||a\\S\\b", + "MSH#$*@!#A#B#C#D#20240102##ORU$R01#2#P#2.5", + "PID###P2*Y##ROE$JANE", + "OBX#1#ST#C2$Two$SCT##c@S@d", + r"MSH|^~\&|A|B|C|D|20240103||ADT^A08|3|P|2.5", + "PID|||P3||POE^JIM", + ] + ) + result = _parse(msg) + assert result.metadata == {"messages": 3, "message_type": "ADT^A08"} + assert result.warnings == [] + patients = result.frames["patient"] + assert list(patients["patient_id"]) == ["P1", "P2", "P3"] + assert list(patients["given_name"]) == ["JOHN", "JANE", "JIM"] + obs = result.frames["observation"] + assert list(obs["patient_id"]) == ["P1", "P2"] + assert list(obs["code_system"]) == ["http://loinc.org", "http://snomed.info/sct"] + assert list(obs["value"]) == ["a^b", "c$d"] + + +def test_segment_id_prefix_with_other_separator_is_unknown(): + # "PIDX|..." is not a PID segment; unknown segment names keep the old warning key. + msg = "MSH|^~\\&|A\rPIDX|||1\rZZ1|x" + result = _parse(msg) + assert result.frames["patient"].empty + assert result.warnings == ["skipped unrecognized segment types: PIDX(1), ZZ1(1)"] + + +# -- standard messages are unchanged ------------------------------------------------ + +STANDARD = "\r".join( + [ + r"MSH|^~\&|LAB|HOSP|EHR|CLINIC|20240101120000||ORU^R01^ORU_R01|MSG001|P|2.5", + "PID|1||12345^^^MRN||DOE^JOHN^Q||19700101|M", + "PV1|1|I|ICU^101^1||||||||||||||||VN9000^^^HOSP", + "OBR|1|PL9001^LAB|FL7001^LAB|24323-8^Comprehensive metabolic panel^LN|||20240101080000", + "OBX|1|NM|8867-4^Heart rate^LN||72|/min|60-100|N|||F|||20240101120000", + "OBX|2|CE|44054006^Diabetes^SCT||44054006^Diabetes^SCT||||||F", + "OBX|3|ST|1234-5^Note^I10|| free text |||||| C ", + "ZZZ|custom|segment", + r"MSH|^~\&|LAB|HOSP|EHR|CLINIC|20240102||ADT^A01|MSG002|P|2.5", + "PV1|1|E", + "PID|||67890||ROE^JANE||19800202|F", + "OBX|1|NM|789-8^RBC^LN||4.8|10*6/uL", + ] +) + +# Frames produced for STANDARD by the parser before #261 was fixed. +EXPECTED_STANDARD = { + "patient": [ + { + "patient_id": "12345", + "family_name": "DOE", + "given_name": "JOHN", + "birth_date": "19700101", + "gender": "M", + }, + { + "patient_id": "67890", + "family_name": "ROE", + "given_name": "JANE", + "birth_date": "19800202", + "gender": "F", + }, + ], + "encounter": [ + { + "patient_id": "12345", + "visit_number": "VN9000", + "class_code": "I", + "class": "inpatient", + "location": "ICU", + }, + { + "patient_id": "MSG2", + "visit_number": "", + "class_code": "E", + "class": "emergency", + "location": "", + }, + ], + "order": [ + { + "patient_id": "12345", + "order_id": "FL7001", + "placer_order": "PL9001", + "filler_order": "FL7001", + "service_code": "24323-8", + "service_display": "Comprehensive metabolic panel", + "service_system": "LN", + "observed_at": "20240101080000", + }, + ], + "observation": [ + { + "patient_id": "12345", + "order_id": "FL7001", + "code": "8867-4", + "display": "Heart rate", + "code_system": "http://loinc.org", + "value": "72", + "unit": "/min", + "status": "F", + "observed_at": "20240101120000", + }, + { + "patient_id": "12345", + "order_id": "FL7001", + "code": "44054006", + "display": "Diabetes", + "code_system": "http://snomed.info/sct", + "value": "44054006^Diabetes^SCT", + "unit": "", + "status": "F", + "observed_at": "", + }, + { + "patient_id": "12345", + "order_id": "FL7001", + "code": "1234-5", + "display": "Note", + "code_system": "ICD-10", + "value": "free text", + "unit": "", + "status": "C", + "observed_at": "", + }, + { + "patient_id": "67890", + "order_id": None, + "code": "789-8", + "display": "RBC", + "code_system": "http://loinc.org", + "value": "4.8", + "unit": "10*6/uL", + "status": "", + "observed_at": "", + }, + ], +} + + +def test_standard_message_frames_unchanged(): + result = fd.parse_domain(STANDARD, format="hl7v2") + assert result.metadata == {"messages": 2, "message_type": "ADT^A01"} + assert result.warnings == ["skipped unrecognized segment types: ZZZ(1)"] + assert set(result.frames) == set(EXPECTED_STANDARD) + for name, rows in EXPECTED_STANDARD.items(): + pd.testing.assert_frame_equal(result.frames[name], pd.DataFrame(rows))