fix(parsers): honour HL7 v2 MSH delimiters, repetitions and escape sequences (#382)

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

Author: kevincostner17

src/freshdata/parsers/hl7v2.py

@@ -5,14 +5,25 @@
 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.
 """
 
 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()))