feat(sp8): outbound 837P serializer — full rebuild + round-trip tests

- backend/src/cyclone/parsers/serialize_837.py — full-rebuild 837P serializer.
  Emits envelope (ISA/GS/ST/SE/GE/IEA + BHT) + submitter/receiver/billing
  provider/subscriber/payer hierarchy + editable segments (CLM/REF*G1/HI)
  + per-service-line LX/SV1/DTP*472/REF*6R — all from canonical
  ClaimOutput fields.

  Pivoted from spec §3.1 hybrid to full rebuild because ClaimOutput.raw_segments
  only captures post-CLM segments (CLM, REF*G1, HI, LX, SV1 pairs) — not the
  envelope or hierarchies. A pass-through approach cannot regenerate those
  without expanding raw_segments in parse_837.py (out of scope for this SP).

- backend/tests/test_serialize_837.py — 36 tests covering envelope shape,
  hierarchy segments, claim-level builders, service-line builders, edited-field
  propagation, round-trip, custom sender/receiver IDs, and resubmit helper.

- backend/tests/test_prodfiles_smoke.py::test_claims_prodfile_round_trip —
  every file in docs/prodfiles/claims/ (113 files) round-trips through
  serialize_837 → parse_837_text with deep-equal ClaimOutput (modulo
  validation, which is recomputed by the parser).

- docs/superpowers/plans/2026-06-20-cyclone-serialize-837.md — full plan
  with amendment note documenting the Approach A pivot.

Per session convention, plan note about unrelated modifications to
parse_835.py / fixtures stashed separately.
This commit is contained in:
Tyler
2026-06-20 20:27:48 -06:00
parent b3de9c4d22
commit 561018c690
4 changed files with 2573 additions and 1 deletions
@@ -0,0 +1,528 @@
"""Serialize a :class:`ClaimOutput` to a complete X12 837P text.
Full rebuild (Approach A — see plan amendment 2026-06-20). Every segment
in the output is emitted from canonical :class:`ClaimOutput` fields;
``raw_segments`` is not consulted.
Why full rebuild instead of the spec's §3.1 hybrid:
- ``ClaimOutput.raw_segments`` only captures post-CLM segments (CLM,
REF*G1, HI, LX, SV1 pairs). The envelope (ISA/GS/ST/BHT) and
hierarchy segments (NM1, N3, N4, HL, PRV, SBR, DMG, REF*EI) are not
captured, so a pass-through approach cannot regenerate them
byte-for-byte. Expanding ``raw_segments`` would require modifying
``parse_837.py`` (out of scope for this SP).
- The 837P spec is stable; the emitter only needs canonical fields
that already live on ``ClaimOutput``. No data loss vs. the original
inbound file for any field the user can edit (charge, frequency,
prior_auth, diagnoses, service lines, dates).
The serializer emits:
- ISA / GS / ST / SE / GE / IEA — envelope (caller supplies the
control numbers via ``interchange_control_number`` /
``group_control_number`` so back-to-back resubmits get unique values)
- BHT — from ``claim.transaction_type_code`` + ``claim.transaction_date``
- Submitter block (NM1*41, PER) — ``sender_id`` / ``submitter_name`` /
``submitter_contact_*`` (caller-supplied; defaults below)
- Receiver block (NM1*40) — ``receiver_id`` / ``receiver_name``
- Billing provider hierarchy (HL*1, NM1*85, N3, N4, REF*EI) — from
``claim.billing_provider``
- Subscriber hierarchy (HL*2, SBR, NM1*IL, N3, N4, DMG) — from
``claim.subscriber``
- Payer (NM1*PR) — from ``claim.payer``
- CLM / REF*G1 / HI — from ``claim.claim`` / ``claim.diagnoses``
- Per service line: LX / SV1 / DTP*472 / REF*6R — from
``claim.service_lines``
If ``claim.service_lines`` is empty (a header-only 837), the LX/SV1
loop emits nothing; the rest still produces a valid envelope.
Round-trip is verified by ``backend/tests/test_serialize_837.py`` and
the prodfile parametrized smoke in
``backend/tests/test_prodfiles_smoke.py``.
"""
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from cyclone.parsers.models import ClaimOutput
_SEG = "~"
_ELEM = "*"
_ISA_COMPONENT_SEPARATOR = ":"
_ISA_REPETITION_SEPARATOR = "^"
_IMPLEMENTATION_GUIDE = "005010X222A1"
_FUNCTIONAL_ID_HEALTH_CARE = "HC"
class SerializeError(Exception):
"""Raised when a claim cannot be serialized."""
# ---------------------------------------------------------------------------
# Envelope helpers
# ---------------------------------------------------------------------------
def _today_yymmdd() -> str:
t = date.today()
return f"{t.year % 100:02d}{t.month:02d}{t.day:02d}"
def _today_yyyymmdd() -> str:
t = date.today()
return f"{t.year:04d}{t.month:02d}{t.day:02d}"
def _today_hhmm() -> str:
t = datetime.now()
return f"{t.hour:02d}{t.minute:02d}"
def _pad(s: str | None, width: int) -> str:
return (s or "").ljust(width)[:width]
def _build_isa(sender_id: str, receiver_id: str, interchange_control_number: str) -> str:
"""Fixed-width 106-char ISA envelope (mirrors serialize_999 layout)."""
parts = [
"ISA",
"00", _pad("", 10),
"00", _pad("", 10),
"ZZ", _pad(sender_id, 15),
"ZZ", _pad(receiver_id, 15),
_today_yymmdd(),
_today_hhmm(),
_ISA_REPETITION_SEPARATOR,
"00501",
_pad(interchange_control_number, 9),
"0",
"P",
_ISA_COMPONENT_SEPARATOR,
]
return _ELEM.join(parts) + _SEG
def _build_gs(sender_id: str, receiver_id: str, group_control_number: str) -> str:
parts = [
"GS",
_FUNCTIONAL_ID_HEALTH_CARE,
sender_id,
receiver_id,
_today_yymmdd(),
_today_hhmm(),
group_control_number,
"X",
_IMPLEMENTATION_GUIDE,
]
return _ELEM.join(parts) + _SEG
def _build_st(control_number: str) -> str:
return _ELEM.join(["ST", "837", control_number, _IMPLEMENTATION_GUIDE]) + _SEG
def _build_se(count: int, control_number: str) -> str:
return _ELEM.join(["SE", str(count), control_number]) + _SEG
def _build_bht(
*,
transaction_type_code: str | None,
reference_id: str,
transaction_date: date | None,
transaction_time: str | None,
) -> str:
"""BHT segment. BHT06 transaction type code defaults to 'CH' (chargeable)."""
bht06 = transaction_type_code or "CH"
date_str = transaction_date.strftime("%Y%m%d") if transaction_date else _today_yyyymmdd()
time_str = transaction_time or _today_hhmm()
parts = [
"BHT",
"0019", # BHT01 — hierarchical structure (originator)
"00", # BHT02 — purpose code (original)
reference_id, # BHT03 — reference identification
date_str, # BHT04 — transaction set creation date
time_str, # BHT05 — transaction set creation time
bht06, # BHT06 — transaction type code
]
return _ELEM.join(parts) + _SEG
# ---------------------------------------------------------------------------
# Hierarchy segment builders
# ---------------------------------------------------------------------------
def _build_nm1(entity_id_qualifier: str, entity_type: str, name: str,
id_code_qualifier: str | None, id_code: str | None) -> str:
"""Generic NM1 segment. entity_type is the 2nd element ('85', 'IL', 'PR', etc.)."""
parts = [
"NM1",
entity_type, # NM101 — entity identifier code
_FUNCTIONAL_ID_HEALTH_CARE if entity_type in {"85", "IL"} else "", # NM102 (1=person, 2=non-person)
"", # NM103 — name last / organization
"", # NM104 — name first
"", # NM105 — name middle
"", # NM106 — name prefix
"", # NM107 — name suffix
id_code_qualifier or "", # NM108 — id code qualifier
id_code or "", # NM109 — id code
]
# Strip the empty 2nd/3rd slots for non-person entities.
if entity_type in {"85", "PR", "40", "41"}:
# Non-person: NM102=2, name goes in NM103
parts[2] = "2"
parts[3] = name
else:
# Person: NM102=1, last in NM103, first in NM104
parts[2] = "1"
names = (name or "").rsplit(" ", 1)
parts[3] = names[0] if names else ""
parts[4] = names[1] if len(names) > 1 else ""
return _ELEM.join(parts) + _SEG
def _build_per(contact_name: str | None, contact_phone: str | None) -> str:
"""PER segment — submitter contact. Returns empty when no contact info."""
if not contact_name and not contact_phone:
return ""
parts = [
"PER",
"IC", # PER01 — contact function code (Information Contact)
contact_name or "",
"TE", # PER03 — phone qualifier
contact_phone or "",
]
return _ELEM.join(parts) + _SEG
def _build_n3(line1: str | None, line2: str | None) -> str:
if not line1:
return ""
parts = ["N3", line1]
if line2:
parts.append(line2)
return _ELEM.join(parts) + _SEG
def _build_n4(city: str | None, state: str | None, zip_code: str | None) -> str:
if not city and not state and not zip_code:
return ""
parts = ["N4", city or "", state or "", zip_code or ""]
return _ELEM.join(parts) + _SEG
def _build_dmg(dob: date | None, gender: str | None) -> str:
"""DMG segment — subscriber demographics."""
if not dob and not gender:
return ""
parts = ["DMG", "D8", dob.strftime("%Y%m%d") if dob else "", gender or ""]
return _ELEM.join(parts) + _SEG
def _build_hl(hl_id: str, parent_id: str, level_code: str, child_code: str) -> str:
"""HL segment — hierarchical level. child_code='0' when this HL has no children."""
parts = [
"HL",
hl_id,
parent_id if parent_id else "",
level_code,
child_code,
]
return _ELEM.join(parts) + _SEG
def _build_sbr(relationship_code: str | None, member_id: str | None,
payer_name: str | None) -> str:
"""SBR segment — subscriber information.
SBR01 (relationship code) defaults to ``"P"`` (Patient = self) which is
the most common case for professional claims; the parser does not store
this on the canonical Subscriber model so we cannot thread it through
without adding a model field.
"""
parts = [
"SBR",
relationship_code or "P",
"", # SBR02 — group number
"", # SBR03 — group name
"", # SBR04 — claim filing indicator code
"", # SBR05 — sequence number code
payer_name or "", # SBR06 — claim filing indicator code (CO uses MC)
"", # SBR07
"", # SBR08
member_id or "", # SBR09 — claim submitter's id
]
return _ELEM.join(parts) + _SEG
def _build_ref(qualifier: str, value: str | None) -> str:
if not value:
return ""
return _ELEM.join(["REF", qualifier, value]) + _SEG
# ---------------------------------------------------------------------------
# Claim-level segment builders
# ---------------------------------------------------------------------------
def _build_clm(claim) -> str:
"""CLM01..CLM09.
The parser reads CLM05 as a composite ``place_of_service:facility_qualifier:frequency_code``
and stores the three sub-parts back on ``ClaimHeader.place_of_service``,
``.facility_code_qualifier``, ``.frequency_code``. CLM03 + CLM04 are
non-institutional claim filing fields — emitted as empty strings.
"""
charge = f"{Decimal(claim.total_charge or 0):.2f}"
# CLM05 composite — composed from the three canonical fields.
clm05_parts: list[str] = []
if claim.place_of_service:
clm05_parts.append(claim.place_of_service)
if claim.facility_code_qualifier:
clm05_parts.append(claim.facility_code_qualifier)
if claim.frequency_code:
clm05_parts.append(claim.frequency_code)
clm05 = ":".join(clm05_parts)
parts = [
"CLM",
claim.claim_id, # CLM01
charge, # CLM02
"", # CLM03 — non-institutional claim filing indicator
"", # CLM04 — non-institutional claim filing code
clm05, # CLM05 — composite POS:qualifier:frequency_code
claim.provider_signature or "Y", # CLM06
claim.assignment or "Y", # CLM07
"", # CLM08 — benefit assignment certification
claim.release_of_info or "Y", # CLM09
]
return _ELEM.join(parts) + _SEG
def _build_ref_g1(prior_auth: str | None) -> str:
if not prior_auth:
return ""
return _ELEM.join(["REF", "G1", prior_auth]) + _SEG
def _build_hi(diagnoses) -> str:
"""One HI segment with up to 12 diagnoses (qualifier:code)."""
parts = ["HI"]
for diag in (diagnoses or [])[:12]:
qualifier = diag.qualifier or "BK"
parts.append(f"{qualifier}:{diag.code}")
return _ELEM.join(parts) + _SEG
def _build_lx(line_number: int) -> str:
return _ELEM.join(["LX", str(line_number)]) + _SEG
def _build_sv1(line) -> str:
"""SV1 segment — professional service line."""
proc = line.procedure
code = proc.code if proc else ""
mods = proc.modifiers if proc else []
composite = "HC:" + code + "".join(f":{m}" for m in (mods or [])[:4])
charge = f"{Decimal(line.charge or 0):.2f}"
units = f"{Decimal(line.units):g}" if line.units is not None else "1"
parts = [
"SV1",
composite,
charge,
line.unit_type or "UN",
units,
line.place_of_service or "",
]
return _ELEM.join(parts) + _SEG
def _build_dtp_472(service_date: date | None) -> str:
"""DTP*472 (service date). D8 qualifier with CCYYMMDD single date."""
if service_date is None:
return ""
stamp = service_date.strftime("%Y%m%d")
return _ELEM.join(["DTP", "472", "D8", stamp]) + _SEG
# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------
def _build_submitter_block(sender_id: str, submitter_name: str | None,
contact_name: str | None,
contact_phone: str | None) -> list[str]:
out = [
_build_nm1("41", "41", submitter_name or sender_id, "46", sender_id),
]
per = _build_per(contact_name, contact_phone)
if per:
out.append(per)
return out
def _build_receiver_block(receiver_id: str, receiver_name: str | None) -> list[str]:
return [
_build_nm1("40", "40", receiver_name or receiver_id, "46", receiver_id),
]
def _build_billing_provider_block(provider) -> list[str]:
"""HL*1 → NM1*85 → N3 → N4 → REF*EI."""
out = [
_build_hl("1", "", "20", "1"), # HL*1 — information source, 1 child (subscriber)
_build_nm1("85", "85", provider.name, "XX", provider.npi),
]
addr = provider.address
if addr:
n3 = _build_n3(addr.line1, addr.line2)
n4 = _build_n4(addr.city, addr.state, addr.zip)
if n3:
out.append(n3)
if n4:
out.append(n4)
ref_ei = _build_ref("EI", provider.tax_id)
if ref_ei:
out.append(ref_ei)
return out
def _build_subscriber_block(subscriber, payer_name: str | None) -> list[str]:
"""HL*2 → SBR → NM1*IL → N3 → N4 → DMG. Subscriber has no children."""
out = [
_build_hl("2", "1", "22", "0"), # HL*2 — subscriber, 0 children
_build_sbr("18", subscriber.member_id, payer_name),
_build_nm1(
"IL", "IL",
f"{subscriber.last_name} {subscriber.first_name}".strip(),
"MI",
subscriber.member_id,
),
]
addr = subscriber.address
if addr:
n3 = _build_n3(addr.line1, addr.line2)
n4 = _build_n4(addr.city, addr.state, addr.zip)
if n3:
out.append(n3)
if n4:
out.append(n4)
dmg = _build_dmg(subscriber.dob, subscriber.gender)
if dmg:
out.append(dmg)
return out
def _build_payer_block(payer) -> list[str]:
return [
_build_nm1("PR", "PR", payer.name, "PI", payer.id),
]
def _build_service_lines_block(service_lines) -> list[str]:
"""Per line: LX / SV1 / DTP*472 / REF*6R."""
out: list[str] = []
for idx, line in enumerate(service_lines or [], start=1):
out.append(_build_lx(idx))
out.append(_build_sv1(line))
dtp = _build_dtp_472(line.service_date)
if dtp:
out.append(dtp)
ref_6r = _build_ref("6R", getattr(line, "provider_reference", None))
if ref_6r:
out.append(ref_6r)
return out
def serialize_837(
claim: ClaimOutput,
*,
sender_id: str = "CYCLONE",
receiver_id: str = "RECEIVER",
submitter_name: str | None = None,
submitter_contact_name: str | None = None,
submitter_contact_phone: str | None = None,
receiver_name: str | None = None,
interchange_control_number: str = "000000001",
group_control_number: str = "1",
) -> str:
"""Serialize a single claim to a complete 837P file (full rebuild).
Sender/receiver info is required for a valid envelope but isn't
captured on the parsed ``ClaimOutput`` — it's deployment-specific
(one billing office ↔ one trading partner). Defaults are provided
(``"CYCLONE"`` / ``"RECEIVER"``) but real deployments should pass
the configured values.
Editable fields (CLM, REF*G1, HI, service-line SV1, DTP*472) are
emitted from the canonical ``ClaimOutput`` fields, so post-parse
edits propagate to the output.
Returns the complete X12 document as a string (segments
``~``-terminated).
"""
segments: list[str] = [
_build_isa(sender_id, receiver_id, interchange_control_number),
_build_gs(sender_id, receiver_id, group_control_number),
_build_st(claim.control_number or "0001"),
_build_bht(
transaction_type_code=claim.transaction_type_code,
reference_id=claim.claim_id,
transaction_date=claim.transaction_date,
transaction_time=None,
),
]
segments.extend(_build_submitter_block(
sender_id, submitter_name, submitter_contact_name, submitter_contact_phone,
))
segments.extend(_build_receiver_block(receiver_id, receiver_name))
segments.extend(_build_billing_provider_block(claim.billing_provider))
segments.extend(_build_subscriber_block(claim.subscriber, claim.payer.name))
segments.extend(_build_payer_block(claim.payer))
# Claim-level editable segments.
segments.append(_build_clm(claim.claim))
ref_g1 = _build_ref_g1(claim.claim.prior_auth)
if ref_g1:
segments.append(ref_g1)
if claim.diagnoses:
segments.append(_build_hi(claim.diagnoses))
# Service lines (LX / SV1 / DTP*472 / REF*6R).
segments.extend(_build_service_lines_block(claim.service_lines))
# SE segment count includes ST (line 3, 1-based) through SE itself
# — i.e. the entire ST..SE block inclusive.
seg_count = len(segments) - 2 + 1 # exclude ISA, GS; +1 for SE
segments.append(_build_se(seg_count, claim.control_number or "0001"))
segments.append(f"GE*1*{group_control_number}{_SEG}")
segments.append(f"IEA*1*{interchange_control_number}{_SEG}")
return "".join(segments)
def serialize_837_for_resubmit(
claim: ClaimOutput,
*,
interchange_index: int,
) -> str:
"""Like :func:`serialize_837` but assigns deterministic-but-unique
interchange + group control numbers for a bundle position.
Interchange number = ``f"{interchange_index:09d}"``.
Group number = ``str(interchange_index)``.
"""
return serialize_837(
claim,
interchange_control_number=f"{interchange_index:09d}",
group_control_number=str(interchange_index),
)