diff --git a/docs/hl7v2-deidentification.md b/docs/hl7v2-deidentification.md index 7fc445988..e7ecef102 100644 --- a/docs/hl7v2-deidentification.md +++ b/docs/hl7v2-deidentification.md @@ -70,3 +70,7 @@ redacted = redact_hl7v2(message_text, field_map=field_map) For offline tests, pass a `deidentifier` callable. The callable should return either a string or an object with `deidentified_text`. + +For downstream clinical NLP or review, use the [HL7 v2 narrative +extractor](./interop/hl7v2-narrative-extraction.md) to render de-identified +flat or sectioned text with final-text offsets back to segment fields. diff --git a/docs/interop/hl7v2-narrative-extraction.md b/docs/interop/hl7v2-narrative-extraction.md new file mode 100644 index 000000000..e2bb99032 --- /dev/null +++ b/docs/interop/hl7v2-narrative-extraction.md @@ -0,0 +1,121 @@ +# HL7 v2 Narrative Extraction + +OpenMed can turn common ADT, ORU, and ORM pipe messages into readable, +de-identified text for clinical NLP or human review. The result includes exact +character ranges that point back to the originating segment occurrence and +field position. + +The extractor builds on the existing [HL7 v2 parser and structured +redactor](../hl7v2-deidentification.md). It does not implement a second parser +or attempt full HL7 conformance validation. + +## Quick Start + +```python +from openmed.interop.hl7v2_narrative import extract_hl7v2_narrative + +result = extract_hl7v2_narrative("synthetic_oru.hl7") + +print(result.text) +for span in result.spans_for("OBX", 5): + print(span.source.path, result.text_for(span)) +``` + +`result.text` is the de-identified narrative. A source path such as +`OBX[2]-5` means field 5 of the second OBX segment. `segment_index` on the +source record is the zero-based position of that segment in the complete +message. + +Provenance records contain only offsets, fixed labels, and HL7 coordinates. +They do not retain the original field value. + +## Narrative Modes + +Flat mode is the default. It emits compact sentence-like text suitable for +downstream NLP: + +```python +flat = extract_hl7v2_narrative(message, mode="flat") +``` + +Sectioned mode emits stable Markdown headings and one field per line for +review surfaces: + +```python +sectioned = extract_hl7v2_narrative(message, mode="sectioned") +print(sectioned.text) +``` + +Typical sections are `Message`, `Patient`, `Encounter`, `Orders`, +`Observations`, and `Notes`. Empty sections are omitted. Both modes preserve +message order within a section and return section-level offsets in +`result.sections`. + +## Privacy Pipeline + +Narrative extraction has two local privacy layers: + +1. `openmed.interop.hl7v2.redact_hl7v2` handles configured structured fields, + including patient identifiers, names, dates, addresses, and phone numbers. +2. The complete rendered narrative passes once through + `openmed.core.pii.deidentify` with `method="mask"` by default. This covers + free text and any other rendered value before it is returned. + +The extractor remaps field and section ranges after the second pass, so all +offsets index the final de-identified text even when placeholders change its +length. + +Use `deidentify_kwargs` to configure the final text pipeline. For deterministic +offline tests, pass a callable that returns a string or an object or mapping +with `deidentified_text`: + +```python +result = extract_hl7v2_narrative( + message, + deidentify_kwargs={"policy": "hipaa_safe_harbor"}, +) +``` + +`date_shift_days`, `lang`, `locale`, `seed`, and an optional structured +`field_map` are forwarded to the existing HL7 redaction layer. The default +30-day shift is stable across runs; dates in the complete narrative still pass +through the final privacy pipeline. + +## Provenance Lookup + +```python +offset = result.text.index("7.1") +for span in result.provenance_at(offset): + print(span.source.segment) # OBX + print(span.source.field_position) # 5 + print(span.source.path) # OBX[2]-5 +``` + +`result.spans` and its explicit alias `result.field_mappings` are tuples of +`HL7V2FieldSpan`. Each span provides: + +- `start` and `end`: offsets in the final narrative, with an exclusive end; +- `section` and `label`: the fixed narrative context; +- `source.segment`: the three-character segment name; +- `source.segment_index`: the zero-based message position; +- `source.segment_occurrence`: the one-based occurrence for that segment name; +- `source.field_position`: the one-based HL7 field number. + +## Supported Rendering Scope + +The renderer covers the common context needed from ADT, ORU, and ORM flows: + +| Segment | Rendered fields | +| --- | --- | +| `MSH` | Message type and HL7 version | +| `EVN` | Event type and recorded time | +| `PID` | Patient ID, name, date of birth, and administrative sex | +| `PV1` | Patient class, location, attending provider, admit/discharge time | +| `ORC` | Order control and placer/filler order identifiers | +| `OBR` | Requested test and observation time | +| `OBX` | Observation, result, units, reference range, flags, status, time | +| `NTE` | Note text | + +Unknown segments are ignored by the narrative renderer. They remain available +to the underlying parser and can still be covered by a custom structured field +map. Mapping the message into FHIR resources is outside this helper's scope. diff --git a/mkdocs.yml b/mkdocs.yml index 4c88f4c67..265f9ebaf 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -67,6 +67,7 @@ nav: - Prefect Batch De-identification: prefect-integration.md - spaCy Pipeline Component: spacy-component.md - HL7 v2 De-identification: hl7v2-deidentification.md + - HL7 v2 Narrative Extraction: interop/hl7v2-narrative-extraction.md - Compliance Posture: compliance.md - Breach Notification Runbook: compliance/breach-notification-runbook.md - Breach Report Template: compliance/templates/breach-report-template.md diff --git a/openmed/interop/hl7v2_narrative.py b/openmed/interop/hl7v2_narrative.py new file mode 100644 index 000000000..e371ffe6f --- /dev/null +++ b/openmed/interop/hl7v2_narrative.py @@ -0,0 +1,746 @@ +"""De-identified narrative extraction for common HL7 v2 messages. + +The extractor builds on :mod:`openmed.interop.hl7v2` rather than parsing pipe +messages independently. Structured identifiers are redacted first, the +rendered narrative is passed through the text de-identification pipeline, and +the final visible value spans retain segment/field provenance. +""" + +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from difflib import SequenceMatcher +from pathlib import Path +from typing import Any, Literal + +from openmed.interop.hl7v2 import ( + FieldKey, + HL7Message, + HL7Segment, + HL7V2Encoding, + TextDeidentifier, + parse_hl7v2, + redact_hl7v2, +) + +NarrativeMode = Literal["flat", "sectioned"] + +_SECTION_ORDER = ( + "Message", + "Patient", + "Encounter", + "Orders", + "Observations", + "Notes", +) + +_SEX_LABELS = { + "A": "Ambiguous", + "F": "Female", + "M": "Male", + "N": "Not applicable", + "O": "Other", + "U": "Unknown", +} +_PATIENT_CLASS_LABELS = { + "E": "Emergency", + "I": "Inpatient", + "O": "Outpatient", + "P": "Preadmit", + "R": "Recurring patient", +} +_RESULT_STATUS_LABELS = { + "C": "Corrected", + "F": "Final", + "I": "Specimen in lab", + "P": "Preliminary", + "R": "Entered, not verified", + "S": "Partial", + "X": "Cannot obtain", +} + + +@dataclass(frozen=True) +class HL7V2FieldSource: + """One source field in an HL7 v2 message. + + ``segment_index`` is zero-based in message order. ``segment_occurrence`` + is one-based among segments with the same three-character name. + """ + + segment: str + segment_index: int + segment_occurrence: int + field_position: int + + @property + def path(self) -> str: + """Return a compact, stable source path such as ``OBX[2]-5``.""" + + return f"{self.segment}[{self.segment_occurrence}]-{self.field_position}" + + +@dataclass(frozen=True) +class HL7V2FieldSpan: + """A final narrative character range mapped to one source field.""" + + start: int + end: int + source: HL7V2FieldSource + section: str + label: str + + @property + def segment(self) -> str: + """Return the source segment name.""" + + return self.source.segment + + @property + def field_position(self) -> int: + """Return the one-based source field position.""" + + return self.source.field_position + + +@dataclass(frozen=True) +class HL7V2NarrativeSection: + """One logical section and its range in the final narrative.""" + + name: str + start: int + end: int + + +@dataclass(frozen=True) +class HL7V2Narrative: + """De-identified narrative text plus source-field provenance.""" + + text: str + mode: NarrativeMode + message_type: str | None + spans: tuple[HL7V2FieldSpan, ...] + sections: tuple[HL7V2NarrativeSection, ...] + + @property + def field_mappings(self) -> tuple[HL7V2FieldSpan, ...]: + """Return the field mappings (an explicit alias for ``spans``).""" + + return self.spans + + def provenance_at(self, offset: int) -> tuple[HL7V2FieldSpan, ...]: + """Return every source-field span covering a narrative offset.""" + + if offset < 0 or offset >= len(self.text): + return () + return tuple(span for span in self.spans if span.start <= offset < span.end) + + def spans_for( + self, + segment: str, + field_position: int, + *, + segment_occurrence: int | None = None, + ) -> tuple[HL7V2FieldSpan, ...]: + """Return narrative spans sourced from a segment/field coordinate.""" + + normalized_segment = segment.upper() + return tuple( + span + for span in self.spans + if span.source.segment == normalized_segment + and span.source.field_position == field_position + and ( + segment_occurrence is None + or span.source.segment_occurrence == segment_occurrence + ) + ) + + def text_for(self, span: HL7V2FieldSpan) -> str: + """Return the final narrative text covered by ``span``.""" + + return self.text[span.start : span.end] + + +@dataclass(frozen=True) +class _NarrativeItem: + section: str + label: str + value: str + source: HL7V2FieldSource + + +@dataclass(frozen=True) +class _RenderedNarrative: + text: str + spans: tuple[HL7V2FieldSpan, ...] + sections: tuple[HL7V2NarrativeSection, ...] + + +def extract_hl7v2_narrative( + message_or_path: str | Path | HL7Message, + *, + mode: NarrativeMode = "flat", + field_map: Mapping[FieldKey | str, Any] | None = None, + deidentifier: TextDeidentifier | None = None, + deidentify_kwargs: Mapping[str, Any] | None = None, + date_shift_days: int = 30, + lang: str = "en", + locale: str | None = None, + seed: int | None = 0, +) -> HL7V2Narrative: + """Render a safe narrative from a common ADT, ORU, or ORM message. + + The existing HL7 v2 redactor first handles structured fields such as + patient identifiers and names. Its free-text hook is deliberately kept as + a no-op during that pass because the complete rendered narrative is then + sent through the supplied de-identification pipeline once. Final offsets + are projected after de-identification, so provenance always indexes the + returned text rather than the raw message or an intermediate narrative. + + Args: + message_or_path: Pipe-delimited message text, UTF-8 path, or parsed + :class:`~openmed.interop.hl7v2.HL7Message`. + mode: ``"flat"`` for sentence-like text or ``"sectioned"`` for + Markdown-style sections. + field_map: Optional structured redaction rules forwarded to + :func:`~openmed.interop.hl7v2.redact_hl7v2`. + deidentifier: Optional narrative de-identifier. Defaults to + :func:`openmed.core.pii.deidentify`. + deidentify_kwargs: Extra keyword arguments for the narrative + de-identifier. The defaults are ``method="mask"`` and ``lang``. + date_shift_days: Stable structured date shift applied before narrative + rendering. The default is 30 days; the complete narrative also + passes through the privacy pipeline. + lang: Language forwarded to structured and narrative de-identification. + locale: Optional locale for deterministic structured surrogates. + seed: Seed for stable structured surrogates. + + Returns: + A de-identified narrative with section ranges and field provenance. + + Raises: + ValueError: If ``mode`` is unsupported. + TypeError: If the de-identifier does not return text in a supported + shape. + """ + + if mode not in {"flat", "sectioned"}: + raise ValueError("mode must be 'flat' or 'sectioned'") + + message_input: str | Path + if isinstance(message_or_path, HL7Message): + message_input = message_or_path.serialize() + else: + message_input = message_or_path + + structured_safe = redact_hl7v2( + message_input, + field_map=field_map, + deidentifier=_identity_deidentifier, + date_shift_days=date_shift_days, + lang=lang, + locale=locale, + seed=seed, + ) + parsed = parse_hl7v2(structured_safe) + items = _collect_items(parsed) + rendered = _render_items(items, mode=mode) + + deidentify = deidentifier or _load_deidentifier() + kwargs = {"method": "mask", "lang": lang, **dict(deidentify_kwargs or {})} + result = deidentify(rendered.text, **kwargs) + safe_text = _deidentified_text(result) + + opcodes = SequenceMatcher( + None, + rendered.text, + safe_text, + autojunk=False, + ).get_opcodes() + spans = tuple( + _project_field_span(span, opcodes, len(safe_text)) for span in rendered.spans + ) + spans = tuple(span for span in spans if span.end > span.start) + sections = tuple( + _project_section(section, opcodes, len(safe_text)) + for section in rendered.sections + ) + + return HL7V2Narrative( + text=safe_text, + mode=mode, + message_type=_message_type(parsed), + spans=spans, + sections=sections, + ) + + +def _identity_deidentifier(text: str, **_: Any) -> str: + """Preserve HL7 free text until the complete narrative privacy pass.""" + + return text + + +def _load_deidentifier() -> TextDeidentifier: + from openmed.core.pii import deidentify + + return deidentify + + +def _deidentified_text(result: Any) -> str: + if isinstance(result, str): + return result + if isinstance(result, Mapping) and "deidentified_text" in result: + return str(result["deidentified_text"]) + if hasattr(result, "deidentified_text"): + return str(result.deidentified_text) + raise TypeError( + "narrative deidentifier must return a string, mapping, or object with " + "deidentified_text" + ) + + +def _collect_items(message: HL7Message) -> tuple[_NarrativeItem, ...]: + items: list[_NarrativeItem] = [] + occurrences: defaultdict[str, int] = defaultdict(int) + observation_number = 0 + order_number = 0 + note_number = 0 + + for segment_index, segment in enumerate(message.segments): + occurrences[segment.name] += 1 + occurrence = occurrences[segment.name] + source = _source_factory(segment, segment_index, occurrence) + + if segment.name == "MSH": + _add(items, "Message", "Message type", segment, 9, source, _message_code) + _add(items, "Message", "HL7 version", segment, 12, source) + elif segment.name == "EVN": + _add(items, "Encounter", "Event type", segment, 1, source) + _add(items, "Encounter", "Event recorded", segment, 2, source, _date) + elif segment.name == "PID": + _add(items, "Patient", "Patient ID", segment, 3, source, _identifier) + _add(items, "Patient", "Patient name", segment, 5, source, _name) + _add(items, "Patient", "Date of birth", segment, 7, source, _date) + _add(items, "Patient", "Administrative sex", segment, 8, source, _sex) + elif segment.name == "PV1": + _add( + items, + "Encounter", + "Patient class", + segment, + 2, + source, + _patient_class, + ) + _add(items, "Encounter", "Location", segment, 3, source, _components) + _add( + items, + "Encounter", + "Attending provider", + segment, + 7, + source, + _name, + ) + _add(items, "Encounter", "Admitted", segment, 44, source, _date) + _add(items, "Encounter", "Discharged", segment, 45, source, _date) + elif segment.name == "ORC": + _add(items, "Orders", "Order control", segment, 1, source) + _add(items, "Orders", "Placer order ID", segment, 2, source, _identifier) + _add(items, "Orders", "Filler order ID", segment, 3, source, _identifier) + elif segment.name == "OBR": + order_number += 1 + _add( + items, + "Orders", + f"Requested test {order_number}", + segment, + 4, + source, + _coded, + ) + _add( + items, + "Orders", + f"Observation requested {order_number}", + segment, + 7, + source, + _date, + ) + elif segment.name == "OBX": + observation_number += 1 + _add( + items, + "Observations", + f"Observation {observation_number}", + segment, + 3, + source, + _coded, + ) + _add( + items, + "Observations", + f"Result {observation_number}", + segment, + 5, + source, + _components, + ) + _add( + items, + "Observations", + f"Units {observation_number}", + segment, + 6, + source, + _coded, + ) + _add( + items, + "Observations", + f"Reference range {observation_number}", + segment, + 7, + source, + ) + _add( + items, + "Observations", + f"Abnormal flags {observation_number}", + segment, + 8, + source, + _components, + ) + _add( + items, + "Observations", + f"Result status {observation_number}", + segment, + 11, + source, + _result_status, + ) + _add( + items, + "Observations", + f"Observed at {observation_number}", + segment, + 14, + source, + _date, + ) + elif segment.name == "NTE": + note_number += 1 + _add(items, "Notes", f"Note {note_number}", segment, 3, source, _text) + + return tuple(items) + + +def _source_factory( + segment: HL7Segment, + segment_index: int, + segment_occurrence: int, +) -> Callable[[int], HL7V2FieldSource]: + def build(field_position: int) -> HL7V2FieldSource: + return HL7V2FieldSource( + segment=segment.name, + segment_index=segment_index, + segment_occurrence=segment_occurrence, + field_position=field_position, + ) + + return build + + +def _add( + items: list[_NarrativeItem], + section: str, + label: str, + segment: HL7Segment, + field_position: int, + source: Callable[[int], HL7V2FieldSource], + formatter: Callable[[str, HL7V2Encoding], str] | None = None, +) -> None: + raw_value = segment.get_field(field_position) + if raw_value is None or not raw_value.strip(): + return + value = (formatter or _text)(raw_value, segment.encoding).strip() + if not value: + return + items.append( + _NarrativeItem( + section=section, + label=label, + value=value, + source=source(field_position), + ) + ) + + +def _render_items( + items: Sequence[_NarrativeItem], + *, + mode: NarrativeMode, +) -> _RenderedNarrative: + by_section = { + section: [item for item in items if item.section == section] + for section in _SECTION_ORDER + } + parts: list[str] = [] + spans: list[HL7V2FieldSpan] = [] + sections: list[HL7V2NarrativeSection] = [] + cursor = 0 + + def append(value: str) -> None: + nonlocal cursor + parts.append(value) + cursor += len(value) + + emitted_sections = 0 + for section in _SECTION_ORDER: + section_items = by_section[section] + if not section_items: + continue + + if emitted_sections: + append("\n\n" if mode == "sectioned" else " ") + section_start = cursor + if mode == "sectioned": + append(f"## {section}\n") + + for item_index, item in enumerate(section_items): + if item_index: + append("\n" if mode == "sectioned" else " ") + append(f"{item.label}: ") + value_start = cursor + append(item.value) + value_end = cursor + if mode == "flat": + append(".") + spans.append( + HL7V2FieldSpan( + start=value_start, + end=value_end, + source=item.source, + section=item.section, + label=item.label, + ) + ) + + sections.append( + HL7V2NarrativeSection(name=section, start=section_start, end=cursor) + ) + emitted_sections += 1 + + return _RenderedNarrative( + text="".join(parts), + spans=tuple(spans), + sections=tuple(sections), + ) + + +def _message_type(message: HL7Message) -> str | None: + msh = next((segment for segment in message.segments if segment.name == "MSH"), None) + if msh is None: + return None + value = msh.get_field(9) + return value or None + + +def _text(value: str, encoding: HL7V2Encoding) -> str: + return _decode_escapes(value, encoding).strip() + + +def _components(value: str, encoding: HL7V2Encoding) -> str: + repetitions: list[str] = [] + for repetition in value.split(encoding.repetition): + components = [ + _decode_escapes(component, encoding).strip() + for component in repetition.split(encoding.component) + if component.strip() + ] + if components: + repetitions.append(" / ".join(components)) + return "; ".join(repetitions) + + +def _coded(value: str, encoding: HL7V2Encoding) -> str: + repetitions: list[str] = [] + for repetition in value.split(encoding.repetition): + components = repetition.split(encoding.component) + code = _decode_escapes(components[0], encoding).strip() if components else "" + display = ( + _decode_escapes(components[1], encoding).strip() + if len(components) > 1 + else "" + ) + if display and code and display != code: + repetitions.append(f"{display} ({code})") + elif display or code: + repetitions.append(display or code) + return "; ".join(repetitions) + + +def _message_code(value: str, encoding: HL7V2Encoding) -> str: + components = [ + _decode_escapes(component, encoding).strip() + for component in value.split(encoding.component) + if component.strip() + ] + return " ".join(components[:2]) + + +def _identifier(value: str, encoding: HL7V2Encoding) -> str: + first_repetition = value.split(encoding.repetition, 1)[0] + first_component = first_repetition.split(encoding.component, 1)[0] + return _decode_escapes(first_component, encoding).strip() + + +def _name(value: str, encoding: HL7V2Encoding) -> str: + first_repetition = value.split(encoding.repetition, 1)[0] + components = [ + _decode_escapes(component, encoding).strip() + for component in first_repetition.split(encoding.component) + ] + family = _component(components, 0) + given = _component(components, 1) + middle = _component(components, 2) + suffix = _component(components, 3) + prefix = _component(components, 4) + ordered = [prefix, given, middle, family, suffix] + return " ".join(part for part in ordered if part) + + +def _component(components: Sequence[str], index: int) -> str: + return components[index] if index < len(components) else "" + + +def _date(value: str, encoding: HL7V2Encoding) -> str: + raw = _identifier(value, encoding) + if len(raw) < 8 or not raw[:8].isdigit(): + return raw + rendered = f"{raw[0:4]}-{raw[4:6]}-{raw[6:8]}" + time = raw[8:14] + if time: + rendered += f" {time[0:2]}" + if len(time) >= 4: + rendered += f":{time[2:4]}" + if len(time) >= 6: + rendered += f":{time[4:6]}" + return rendered + + +def _sex(value: str, encoding: HL7V2Encoding) -> str: + code = _identifier(value, encoding).upper() + label = _SEX_LABELS.get(code) + return f"{label} ({code})" if label else code + + +def _patient_class(value: str, encoding: HL7V2Encoding) -> str: + code = _identifier(value, encoding).upper() + label = _PATIENT_CLASS_LABELS.get(code) + return f"{label} ({code})" if label else code + + +def _result_status(value: str, encoding: HL7V2Encoding) -> str: + code = _identifier(value, encoding).upper() + label = _RESULT_STATUS_LABELS.get(code) + return f"{label} ({code})" if label else code + + +def _decode_escapes(value: str, encoding: HL7V2Encoding) -> str: + escape = encoding.escape + replacements = { + f"{escape}F{escape}": encoding.field, + f"{escape}S{escape}": encoding.component, + f"{escape}R{escape}": encoding.repetition, + f"{escape}T{escape}": encoding.subcomponent, + f"{escape}E{escape}": encoding.escape, + f"{escape}.br{escape}": "\n", + } + rendered = value + for encoded, decoded in replacements.items(): + rendered = rendered.replace(encoded, decoded) + return rendered + + +def _project_field_span( + span: HL7V2FieldSpan, + opcodes: Sequence[tuple[str, int, int, int, int]], + target_length: int, +) -> HL7V2FieldSpan: + start, end = _project_range(span.start, span.end, opcodes, target_length) + return HL7V2FieldSpan( + start=start, + end=end, + source=span.source, + section=span.section, + label=span.label, + ) + + +def _project_section( + section: HL7V2NarrativeSection, + opcodes: Sequence[tuple[str, int, int, int, int]], + target_length: int, +) -> HL7V2NarrativeSection: + start, end = _project_range(section.start, section.end, opcodes, target_length) + return HL7V2NarrativeSection(name=section.name, start=start, end=end) + + +def _project_range( + start: int, + end: int, + opcodes: Sequence[tuple[str, int, int, int, int]], + target_length: int, +) -> tuple[int, int]: + projected_start = _project_boundary(start, opcodes, prefer_end=False) + projected_end = _project_boundary(end, opcodes, prefer_end=True) + projected_start = max(0, min(projected_start, target_length)) + projected_end = max(projected_start, min(projected_end, target_length)) + return projected_start, projected_end + + +def _project_boundary( + position: int, + opcodes: Sequence[tuple[str, int, int, int, int]], + *, + prefer_end: bool, +) -> int: + for tag, source_start, source_end, target_start, target_end in opcodes: + if tag == "insert": + if position == source_start: + return target_end if prefer_end else target_start + continue + if not source_start <= position <= source_end: + continue + if tag == "equal": + return target_start + (position - source_start) + if position == source_start: + return target_start + if position == source_end: + return target_end + if tag == "delete": + return target_start + source_width = source_end - source_start + target_width = target_end - target_start + relative = (position - source_start) / source_width + return target_start + round(relative * target_width) + return opcodes[-1][4] if opcodes else position + + +__all__ = [ + "HL7V2FieldSource", + "HL7V2FieldSpan", + "HL7V2Narrative", + "HL7V2NarrativeSection", + "NarrativeMode", + "extract_hl7v2_narrative", +] diff --git a/tests/unit/interop/test_hl7v2_narrative.py b/tests/unit/interop/test_hl7v2_narrative.py new file mode 100644 index 000000000..d101c791a --- /dev/null +++ b/tests/unit/interop/test_hl7v2_narrative.py @@ -0,0 +1,176 @@ +"""Tests for HL7 v2 de-identified narrative extraction.""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from openmed.interop.hl7v2 import parse_hl7v2 +from openmed.interop.hl7v2_narrative import extract_hl7v2_narrative + +FIXTURES = Path(__file__).parent / "fixtures" + + +def fake_deidentifier(text: str, **kwargs): + """Deterministic offline stand-in for the complete narrative pass.""" + + assert kwargs["method"] == "mask" + assert kwargs["lang"] == "en" + redacted = ( + text.replace("Jane Roe", "[PERSON]") + .replace("John Doe", "[PERSON]") + .replace("jane.roe@example.com", "[EMAIL]") + .replace("555-0199", "[PHONE]") + .replace("555-0101", "[PHONE]") + .replace("MRN67890", "[ID_NUM]") + .replace("MRN12345", "[ID_NUM]") + ) + return SimpleNamespace(deidentified_text=redacted) + + +def test_flat_oru_is_coherent_safe_and_maps_results_to_source_fields(): + result = extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_oru.hl7", + deidentifier=fake_deidentifier, + ) + + assert result.mode == "flat" + assert result.message_type == "ORU^R01" + assert result.text.startswith("Message type: ORU R01.") + assert "Observation 1: Clinical note (NOTE)." in result.text + assert ( + "Result 1: Patient [PERSON] called from [PHONE] about [ID_NUM]." in result.text + ) + assert ( + "Observation 2: Glucose (GLU). Result 2: 7.1. Units 2: mmol/L." in result.text + ) + assert "Note 1: Follow-up email [EMAIL] belongs to [PERSON]." in result.text + assert "Jane Roe" not in result.text + assert "jane.roe@example.com" not in result.text + assert "MRN67890" not in result.text + + result_span = result.spans_for("OBX", 5, segment_occurrence=1)[0] + assert result_span.source.segment_index == 3 + assert result_span.source.path == "OBX[1]-5" + assert result.text_for(result_span) == ( + "Patient [PERSON] called from [PHONE] about [ID_NUM]." + ) + + glucose_offset = result.text.index("7.1") + provenance = result.provenance_at(glucose_offset) + assert [(span.source.path, span.label) for span in provenance] == [ + ("OBX[2]-5", "Result 2") + ] + + +def test_sectioned_adt_has_stable_patient_and_encounter_sections(): + first = extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_adt.hl7", + mode="sectioned", + deidentifier=fake_deidentifier, + ) + second = extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_adt.hl7", + mode="sectioned", + deidentifier=fake_deidentifier, + ) + + assert first == second + assert first.text.startswith("## Message\n") + assert "\n\n## Patient\n" in first.text + assert "\n\n## Encounter\n" in first.text + assert "Administrative sex: Male (M)" in first.text + assert "Patient class: Inpatient (I)" in first.text + assert "DOE" not in first.text + assert "MRN12345" not in first.text + assert "1980-01-01" not in first.text + assert "Date of birth: 1980-01-31" in first.text + + assert [section.name for section in first.sections] == [ + "Message", + "Patient", + "Encounter", + ] + for section in first.sections: + assert first.text[section.start : section.end].startswith(f"## {section.name}") + + +def test_deidentifier_receives_complete_narrative_once(): + calls: list[str] = [] + + def deidentifier(text: str, **kwargs): + calls.append(text) + assert kwargs == {"method": "mask", "lang": "en"} + return text.replace("Jane Roe", "[PERSON]") + + result = extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_oru.hl7", + deidentifier=deidentifier, + ) + + assert len(calls) == 1 + assert calls[0].startswith("Message type:") + assert "Patient Jane Roe called" in calls[0] + assert "Patient Jane Roe called" not in result.text + + +def test_all_field_mappings_index_nonempty_final_text(): + result = extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_oru.hl7", + mode="sectioned", + deidentifier=fake_deidentifier, + ) + + assert result.field_mappings == result.spans + assert result.spans + for span in result.spans: + assert 0 <= span.start < span.end <= len(result.text) + assert result.text_for(span) + assert span.source.segment.isupper() + assert span.source.field_position > 0 + + +def test_parsed_message_and_mapping_result_are_supported(): + message = parse_hl7v2( + (FIXTURES / "synthetic_phi_oru.hl7").read_text(encoding="utf-8") + ) + + result = extract_hl7v2_narrative( + message, + deidentifier=lambda text, **_: {"deidentified_text": text}, + ) + + assert result.message_type == "ORU^R01" + assert result.spans_for("OBX", 3, segment_occurrence=2) + + +def test_extractor_calls_public_hl7_parser(monkeypatch): + import openmed.interop.hl7v2_narrative as narrative_module + + calls: list[str] = [] + parser = narrative_module.parse_hl7v2 + + def tracking_parser(message: str): + calls.append(message) + return parser(message) + + monkeypatch.setattr(narrative_module, "parse_hl7v2", tracking_parser) + + narrative_module.extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_adt.hl7", + deidentifier=lambda text, **_: text, + ) + + assert len(calls) == 1 + assert calls[0].startswith("MSH") + + +def test_invalid_mode_is_rejected_before_processing(): + with pytest.raises(ValueError, match="flat.*sectioned"): + extract_hl7v2_narrative( + FIXTURES / "synthetic_phi_oru.hl7", + mode="html", # type: ignore[arg-type] + deidentifier=fake_deidentifier, + )