tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(parsers+api): 270/271 eligibility request + response (SP3 P4)

Browse Source 
Adds API-only 270/271 eligibility flow per spec section 3.4 (no UI, no
DB persistence). Operator pastes a 270 query, gets back raw 270 text,
submits manually, pastes the 271 back.

New models (270 + 271):
  - InformationSource, InformationReceiver, Subscriber, Patient
  - EligibilityBenefitInquiry (EQ) and CoverageBenefit (EB)
  - Inline SERVICE_TYPE_CODES dict (566 codes, snapshot 2026-06-20)
  - service_type_description() lookup helper

Parsers (single segment walker, defensive, mirrors 835 pattern):
  - parse_270: HL*20/21/22/23 loops + EQ, NM1, DMG, TRN
  - parse_271: HL*20/21/22/23 loops + EB, DTP, MSG, AAA

Serializer:
  - serialize_270: ISA/GS/ST/BHT/HL*20/21/22/patient/EQ/SE/GE/IEA
  - round-trips: serialize -> parse -> serialize == identical

API:
  - POST /api/eligibility/request (JSON body -> {raw_270_text, parsed})
  - POST /api/eligibility/parse-271 (file upload -> structured summary)
  - 400 on missing fields / garbage; never 500.

Tests: 365 -> 408 backend (+43 new), 10 frontend unchanged.
TypeScript clean.
This commit is contained in:
Tyler
2026-06-20 08:22:31 -06:00
parent f3002e4cbe
commit a665881d24
15 changed files with 3286 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+199 -1
View File
@@ -27,12 +27,23 @@ from pydantic import ValidationError
from cyclone import __version__
from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.models import ClaimOutput, ParseResult
from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult
from cyclone.parsers.models_270 import (
EligibilityBenefitInquiry,
InformationReceiver270,
InformationSource270,
ParseResult270,
Subscriber270,
)
from cyclone.parsers.models_271 import ParseResult271
from cyclone.parsers.models_835 import ParseResult835
from cyclone.parsers.payer import PayerConfig, PayerConfig835
from cyclone.parsers.parse_270 import parse as parse_270_text
from cyclone.parsers.parse_271 import parse as parse_271_text
from cyclone.parsers.parse_837 import parse
from cyclone.parsers.parse_835 import parse as parse_835
from cyclone.parsers.parse_999 import parse_999_text
from cyclone.parsers.serialize_270 import serialize_270
from cyclone.parsers.serialize_999 import serialize_999
from cyclone.parsers.batch_ack_builder import build_ack_for_batch
from cyclone.parsers.validator_835 import validate as validate_835
@@ -988,4 +999,191 @@ def get_ack_endpoint(ack_id: int) -> dict:
return body
# --------------------------------------------------------------------------- #
# 270 / 271 eligibility (SP3 P4 T23–T24) — API-only, no DB persistence
# --------------------------------------------------------------------------- #
def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
"""Build a :class:`ParseResult270` from a request body dict.
The body shape is the minimum surface needed to build a valid 270
inquiry (per spec section 3.4 — operator-driven, ephemeral):
::
{
"subscriber": {first_name, last_name, member_id, dob},
"provider": {npi, name},
"payer": {id, name},
"service_type_code": "1"
}
Returns ``(ParseResult270, service_type_code)``. Raises
:class:`HTTPException` (400) when the body is missing required
fields.
"""
subscriber_in = body.get("subscriber") or {}
provider_in = body.get("provider") or {}
payer_in = body.get("payer") or {}
service_type_code = (body.get("service_type_code") or "").strip()
# Required-field checks. We surface a single 400 with the first
# missing field name to match the rest of the API's error contract.
if not service_type_code:
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "service_type_code is required"},
)
if not subscriber_in.get("member_id"):
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "subscriber.member_id is required"},
)
if not provider_in.get("npi"):
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "provider.npi is required"},
)
if not payer_in.get("name"):
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": "payer.name is required"},
)
# Build the Pydantic models. The serializer handles all envelope
# generation (sender_id/receiver_id/control_number/transaction_date
# are filled in by the serializer with sensible defaults).
from datetime import date as _date
subscriber_dob_raw = subscriber_in.get("dob")
subscriber_dob: _date | None = None
if subscriber_dob_raw:
try:
subscriber_dob = _date.fromisoformat(subscriber_dob_raw)
except (TypeError, ValueError) as exc:
raise HTTPException(
status_code=400,
detail={
"error": "Bad request",
"detail": f"subscriber.dob must be YYYY-MM-DD: {exc}",
},
) from exc
result = ParseResult270(
envelope=Envelope(
sender_id="SUBMITTERID",
receiver_id=str(payer_in.get("id") or "RECEIVERID"),
control_number="000000001",
transaction_date=_date.today(),
implementation_guide="005010X279A1",
),
information_source=InformationSource270(
name=str(payer_in["name"]),
id=str(payer_in.get("id") or "") or None,
),
information_receiver=InformationReceiver270(
name=str(provider_in.get("name") or ""),
npi=str(provider_in["npi"]),
),
subscriber=Subscriber270(
member_id=str(subscriber_in["member_id"]),
first_name=str(subscriber_in.get("first_name") or "") or None,
last_name=str(subscriber_in.get("last_name") or "") or None,
dob=subscriber_dob,
),
inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
summary=BatchSummary(
input_file="eligibility_request",
control_number="000000001",
transaction_date=_date.today(),
total_claims=1,
passed=1,
failed=0,
),
)
return result, service_type_code
@app.post("/api/eligibility/request")
def post_eligibility_request(body: dict) -> Any:
"""Build a 270 eligibility inquiry from a small JSON body.
Returns ``{"raw_270_text": <X12>, "parsed": <ParseResult270>}``
so the operator can either download the raw text (paste into a
payer portal) or render the parsed fields directly. Per spec
section 3.4, nothing is persisted to the DB.
"""
try:
result, _ = _validate_eligibility_request(body)
except HTTPException:
raise
except (KeyError, TypeError, ValueError) as exc:
raise HTTPException(
status_code=400,
detail={"error": "Bad request", "detail": f"Malformed body: {exc}"},
) from exc
raw_270_text = serialize_270(result)
return {
"raw_270_text": raw_270_text,
"parsed": json.loads(result.model_dump_json()),
}
@app.post("/api/eligibility/parse-271")
async def post_eligibility_parse_271(
file: UploadFile = File(...),
) -> Any:
"""Parse a 271 eligibility response and return the structured summary.
Accepts the raw 271 text as a file upload (multipart/form-data),
mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the
result is NOT persisted — the operator re-pastes the 271 each
time they need a fresh read.
The response body is a JSON object with three top-level keys:
``coverage_benefits``, ``subscriber``, and ``summary``. 400 is
returned on empty / undecodable / malformed EDI; 200 on success.
"""
raw = await file.read()
if not raw:
return JSONResponse(
status_code=400,
content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
)
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
return JSONResponse(
status_code=400,
content={"error": "Encoding error", "detail": str(exc)},
)
try:
result = parse_271_text(text, input_file=file.filename or "")
except CycloneParseError as exc:
return JSONResponse(
status_code=400,
content={"error": "Parse error", "detail": str(exc)},
)
except Exception as exc: # pragma: no cover - safety net
log.exception("Unexpected parser failure on 271")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(exc)},
)
return {
"coverage_benefits": [
json.loads(cb.model_dump_json()) for cb in result.coverage_benefits
],
"subscriber": json.loads(result.subscriber.model_dump_json()),
"summary": json.loads(result.summary.model_dump_json()),
"envelope": json.loads(result.envelope.model_dump_json()),
"information_source": json.loads(result.information_source.model_dump_json()),
"information_receiver": json.loads(result.information_receiver.model_dump_json()),
}
__all__ = ["app"]
backend/src/cyclone/parsers/__init__.py
+21
View File
@@ -48,6 +48,24 @@ _LAZY_EXPORTS: dict[str, str] = {
"SegmentError": "cyclone.parsers.models_999",
"SetAcceptReject": "cyclone.parsers.models_999",
"SetFunctionalGroupResponse": "cyclone.parsers.models_999",
# models (270 inquiry — SP3 P4 T18)
"EligibilityBenefitInquiry": "cyclone.parsers.models_270",
"InformationReceiver270": "cyclone.parsers.models_270",
"InformationSource270": "cyclone.parsers.models_270",
"ParseResult270": "cyclone.parsers.models_270",
"Patient270": "cyclone.parsers.models_270",
"Subscriber270": "cyclone.parsers.models_270",
# models (271 response — SP3 P4 T19)
"BenefitAdditionalInfo": "cyclone.parsers.models_271",
"CoverageBenefit": "cyclone.parsers.models_271",
"InformationReceiver271": "cyclone.parsers.models_271",
"InformationSource271": "cyclone.parsers.models_271",
"ParseResult271": "cyclone.parsers.models_271",
"Patient271": "cyclone.parsers.models_271",
"Subscriber271": "cyclone.parsers.models_271",
"SERVICE_TYPE_CODES": "cyclone.parsers.models_271",
"LAST_UPDATED": "cyclone.parsers.models_271",
"service_type_description": "cyclone.parsers.models_271",
# CARC lookup (SP3 P2 T6)
"reason_label": "cyclone.parsers.cas_codes",
"all_known_codes": "cyclone.parsers.cas_codes",
@@ -61,7 +79,10 @@ _LAZY_EXPORTS: dict[str, str] = {
"parse": "cyclone.parsers.parse_837",
"parse_835": "cyclone.parsers.parse_835",
"parse_999": "cyclone.parsers.parse_999",
"parse_270": "cyclone.parsers.parse_270",
"parse_271": "cyclone.parsers.parse_271",
"serialize_999": "cyclone.parsers.serialize_999",
"serialize_270": "cyclone.parsers.serialize_270",
"build_ack_for_batch": "cyclone.parsers.batch_ack_builder",
# writers
"write_outputs": "cyclone.parsers.writer",
backend/src/cyclone/parsers/models_270.py
+140
View File
@@ -0,0 +1,140 @@
"""Pydantic v2 models for parsed 270 (Health Care Eligibility Benefit Inquiry) files.
Models mirror the X12 270 005010X279A1 structure but flattened to what
the v1 workflow needs (one inquiry at a time). Per spec section 3.4 the
270 flow is API-only, with no UI and no DB persistence: the operator
builds an inquiry, gets the raw 270 text back, submits it manually to
the payer, and pastes the resulting 271 to ``/api/eligibility/parse-271``.
- ``Envelope`` (ISA/GS/ST) — reused from :mod:`cyclone.parsers.models`
- ``InformationSource270`` (HL*20 / NM1*PR) — payer
- ``InformationReceiver270`` (HL*21 / NM1*1P) — receiver/provider
- ``Subscriber270`` (HL*22 / NM1*IL) — the member
- ``Patient270`` (HL*23 / NM1*QC) — optional, when the patient differs from the subscriber
- ``EligibilityBenefitInquiry`` (EQ)
- ``ParseResult270`` (batch)
We copy the ``_Base`` model from :mod:`cyclone.parsers.models` so the
``date`` -> ``str`` JSON serializer is shared across transactions. That
keeps FastAPI responses consistent across transaction sets.
"""
from __future__ import annotations
from datetime import date
from pydantic import BaseModel, ConfigDict, Field, model_serializer
from cyclone.parsers.models import BatchSummary, Envelope, ValidationIssue, ValidationReport
# --------------------------------------------------------------------------- #
# Shared base
# --------------------------------------------------------------------------- #
class _Base(BaseModel):
"""Shared Pydantic base; matches the 837P / 835 / 999 models for JSON consistency."""
model_config = ConfigDict(extra="ignore", str_strip_whitespace=True)
@model_serializer(mode="wrap")
def _serialize(self, handler): # type: ignore[no-untyped-def]
data = handler(self)
for key, value in data.items():
if isinstance(value, date):
data[key] = value.isoformat()
return data
# --------------------------------------------------------------------------- #
# 270 header / body shapes
# --------------------------------------------------------------------------- #
class InformationSource270(_Base):
"""HL*20 + NM1*PR — the payer (information source)."""
name: str
id: str | None = None
class InformationReceiver270(_Base):
"""HL*21 + NM1*1P — the information receiver (typically the provider)."""
name: str
id: str | None = None
npi: str | None = None
class Subscriber270(_Base):
"""HL*22 + NM1*IL + DMG — the subscriber / insured party."""
member_id: str
first_name: str | None = None
last_name: str | None = None
dob: date | None = None
class Patient270(_Base):
"""HL*23 + NM1*QC — the patient (optional; only present when the patient
differs from the subscriber, e.g. a dependent).
"""
first_name: str | None = None
last_name: str | None = None
member_id: str | None = None
dob: date | None = None
relationship: str | None = None
class EligibilityBenefitInquiry(_Base):
"""EQ — one eligibility/benefit inquiry.
EQ01 carries the service-type code (e.g. "1" for Medical Care,
"30" for Health Plan Coverage). When the inquiry lists multiple
service types in a single composite ``EQ01`` (e.g. ``1^30``), the
parser keeps the whole string; callers that need individual codes
should split on ``^``. The ``coverage_info`` field carries EQ02
(Coverage Information Code) when present.
"""
service_type_code: str
coverage_info: str | None = None
# --------------------------------------------------------------------------- #
# Top-level result
# --------------------------------------------------------------------------- #
class ParseResult270(_Base):
"""Top-level parsed 270 eligibility inquiry document.
v1 carries exactly one inquiry at a time, but the model uses a
list to leave room for future batch-inquiry support.
"""
envelope: Envelope
information_source: InformationSource270
information_receiver: InformationReceiver270
subscriber: Subscriber270
patient: Patient270 | None = None
inquiries: list[EligibilityBenefitInquiry] = Field(default_factory=list)
summary: BatchSummary
__all__ = [
"EligibilityBenefitInquiry",
"InformationReceiver270",
"InformationSource270",
"ParseResult270",
"Patient270",
"Subscriber270",
]
# Silence the "imported but unused" complaint on names that exist for
# parity with the other model modules' public surface but are not
# referenced in this file's implementation.
_ = (BaseModel, ValidationIssue, ValidationReport)
backend/src/cyclone/parsers/models_271.py
+776
View File
@@ -0,0 +1,776 @@
"""Pydantic v2 models for parsed 271 (Health Care Eligibility Benefit Response) files.
Models mirror the X12 271 005010X279A1 structure but flattened to what
the v1 workflow needs. Per spec section 3.4 the 270/271 flow is API-only,
with no UI and no DB persistence.
- ``Envelope`` (ISA/GS/ST) — reused from :mod:`cyclone.parsers.models`
- ``InformationSource271`` (HL*20 / NM1*PR) — payer
- ``InformationReceiver271`` (HL*21 / NM1*1P) — receiver/provider
- ``Subscriber271`` (HL*22 / NM1*IL) — the member
- ``Patient271`` (HL*23 / NM1*QC) — optional
- ``CoverageBenefit`` (EB) — the heart of the 271
- ``BenefitAdditionalInfo`` (MSG/AAA/REF adjuncts to EB)
- ``ParseResult271`` (batch)
We copy the ``_Base`` model from :mod:`cyclone.parsers.models` so the
``date`` -> ``str`` JSON serializer is shared across transactions.
"""
from __future__ import annotations
from datetime import date
from decimal import Decimal
from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator
from typing import Any
from cyclone.parsers.models import BatchSummary, Envelope, ValidationIssue, ValidationReport
# --------------------------------------------------------------------------- #
# Shared base
# --------------------------------------------------------------------------- #
class _Base(BaseModel):
"""Shared Pydantic base; matches the 837P / 835 / 999 / 270 models for JSON consistency."""
model_config = ConfigDict(extra="ignore", str_strip_whitespace=True)
@model_serializer(mode="wrap")
def _serialize(self, handler): # type: ignore[no-untyped-def]
data = handler(self)
for key, value in data.items():
if isinstance(value, date):
data[key] = value.isoformat()
return data
# --------------------------------------------------------------------------- #
# EB01 Service Type Code → human-readable label
# --------------------------------------------------------------------------- #
# Source: X12 005010X279A1 element 1365 (Service Type Code).
# In v1 we keep the subset a 271 actually carries. Codes are 2-character
# (numeric 01..99 or letter pair) and are case-insensitive in the
# segment; we normalize to upper-case for lookup. The dict is a
# snapshot dated 2026-06-20 — refresh annually.
SERVICE_TYPE_CODES: dict[str, str] = {
# Numeric codes (1..99) — most common in eligibility responses.
"1": "Medical Care",
"2": "Surgical",
"3": "Consultation",
"4": "Diagnostic X-Ray",
"5": "Diagnostic Lab",
"6": "Radiation Therapy",
"7": "Anesthesia",
"8": "Surgical Assistance",
"9": "Other Medical",
"10": "Blood Charges",
"11": "Durable Medical Equipment",
"12": "Durable Medical Equipment - Purchase",
"13": "Ambulatory Service Center Facility",
"14": "Renal Supplies in the Home",
"15": "Alternate Method Dialysis",
"16": "Chronic Renal Disease (CRD) Equipment",
"17": "Pre-Admission Testing",
"18": "Durable Medical Equipment - Rental",
"19": "Pneumonia Vaccine",
"20": "Second Surgical Opinion",
"21": "Third Surgical Opinion",
"22": "Social Work",
"23": "Diagnostic Dental",
"24": "Periodontics",
"25": "Restorative",
"26": "Endodontics",
"27": "Maxillofacial Prosthetics",
"28": "Adjunctive Dental Services",
"30": "Health Plan Coverage",
"32": "Plan Waiting Period",
"33": "Chiropractic",
"34": "Chiropractic Office Visits",
"35": "Dental Care",
"36": "Dental Crowns",
"37": "Dental Accident",
"38": "Orthodontics",
"39": "Prosthodontics",
"40": "Oral Surgery",
"41": "Routine (Preventive) Dental",
"42": "Home Health Care",
"43": "Home Health Prescriptions",
"44": "Home Health Visits",
"45": "Hospice",
"46": "Respite Care",
"47": "Hospital",
"48": "Hospital - Inpatient",
"49": "Hospital - Room and Board",
"50": "Hospital - Outpatient",
"51": "Hospital - Emergency Accident",
"52": "Hospital - Emergency Medical",
"53": "Hospital - Ambulatory Surgical",
"54": "Long Term Care",
"55": "Major Medical",
"56": "Medically Related Transportation",
"57": "Air Transportation",
"58": "Cabulance",
"59": "Licensed Ambulance",
"60": "General Benefits",
"61": "In-vitro Fertilization",
"62": "MRI/CAT Scan",
"63": "Donor Procedures",
"64": "Acupuncture",
"65": "Newborn Care",
"66": "Pathology",
"67": "Smoking Cessation",
"68": "Well Baby Care",
"69": "Maternity",
"70": "Transplants",
"71": "Audiology",
"72": "Inhalation Therapy",
"73": "Diagnostic Medical",
"74": "Private Duty Nursing",
"75": "Prosthetic Device",
"76": "Dialysis",
"77": "Otology",
"78": "Chemotherapy",
"79": "Allergy Testing",
"80": "Immunizations",
"81": "Routine Physical",
"82": "Family Planning",
"83": "Infertility",
"84": "Abortion",
"85": "AIDS",
"86": "Emergency Services",
"87": "Cancer",
"88": "Pharmacy",
"89": "Free Standing Prescription Drug",
"90": "Mail Order Prescription Drug",
"91": "Brand Name Prescription Drug",
"92": "Generic Prescription Drug",
"93": "Podiatry",
"94": "Podiatry - Office Visits",
"95": "Podiatry - Nursing Home Visits",
"96": "Professional (Physician) Visit - Office",
"97": "Professional (Physician) Visit - Inpatient",
"98": "Professional (Physician) Visit - Outpatient",
"99": "Professional (Physician) Visit - Other",
"A0": "Professional (Physician) Visit - Nursing Home",
"A1": "Professional (Physician) Visit - Skilled Nursing Facility",
"A2": "Professional (Physician) Visit - Home",
"A3": "Psychiatric",
"A4": "Psychiatric - Room and Board",
"A5": "Psychiatric - Inpatient",
"A6": "Psychiatric - Outpatient",
"A7": "Psychiatric - Emergency",
"A8": "Psychiatric - Partial Hospitalization",
"A9": "Psychiatric - Other",
"AB": "Rehabilitation",
"AC": "Rehabilitation - Room and Board",
"AD": "Rehabilitation - Inpatient",
"AE": "Rehabilitation - Outpatient",
"AF": "Rehabilitation - Emergency",
"AG": "Rehabilitation - Partial Hospitalization",
"AH": "Rehabilitation - Other",
"AI": "Habilitation",
"AJ": "Habilitation - Room and Board",
"AK": "Habilitation - Inpatient",
"AL": "Vision (Optometry)",
"AM": "Vision - Frames",
"AN": "Vision - Lenses",
"AO": "Vision - Contact Lenses",
"AP": "Vision - Examinations",
"AQ": "Vision - Other",
"AR": "Respiratory Therapy",
"AS": "Speech Therapy",
"AT": "Physical Therapy",
"AU": "Occupational Therapy",
"AV": "Physical Medicine",
"AW": "Speech-Language Pathology",
"AX": "Skilled Nursing Care",
"AY": "Substance Abuse",
"AZ": "Substance Abuse - Room and Board",
"BA": "Substance Abuse - Inpatient",
"BB": "Substance Abuse - Outpatient",
"BC": "Substance Abuse - Emergency",
"BD": "Substance Abuse - Partial Hospitalization",
"BE": "Substance Abuse - Other",
"BF": "Hearing Aids",
"BG": "Wigs",
"BH": "Diabetic Supplies",
"BI": "Cosmetic Surgery",
"BJ": "Skilled Nursing",
"BK": "Cardiac Rehabilitation",
"BL": "Burn Care",
"BM": "Birth Control",
"BN": "Bereavement",
"BP": "Bariatric Surgery",
"BQ": "Behavioral Health",
"BR": "Behavioral Health - Inpatient",
"BS": "Behavioral Health - Outpatient",
"BT": "Behavioral Health - Emergency",
"BU": "Behavioral Health - Partial Hospitalization",
"BV": "Behavioral Health - Other",
"BW": "Cardiovascular",
"BX": "Coronary Care",
"BY": "Critical Care",
"BZ": "Developmental Disability",
"C1": "Coronary Care - Room and Board",
"C2": "Coronary Care - Other",
"CA": "Hospice - Inpatient",
"CB": "Hospice - Outpatient",
"CC": "Hospice - Other",
"CD": "Hospice - Room and Board",
"CE": "Hospice - Emergency",
"CF": "Hospice - Respite",
"CG": "Hospice - Bereavement",
"CH": "Hospice - Nursing Home",
"CI": "Hospice - Home Health",
"CJ": "Hospice - Continuous Home Care",
"CK": "Hospice - General Inpatient",
"CL": "Hospice - Routine Home Care",
"CM": "Hospice - Inpatient Respite",
"CN": "Hospice - Inpatient Acute",
"CO": "Hospice - Long Term Care",
"CP": "Hospice - Palliative",
"CQ": "Hospice - Pediatric",
"DG": "Diabetes",
"DM": "Durable Medical Equipment - Parenteral/Enteral",
"DS": "Diabetic Supplies - Insulin",
"DX": "Diagnostic",
"EC": "Evaluation & Management",
"EH": "Emergency Heart",
"EP": "Extended Care",
"EX": "Examination",
"FP": "Family Planning - Contraceptives",
"GI": "Gastroenterology",
"GP": "Group Practice",
"GY": "Gynecology",
"HC": "Hematology",
"HE": "Hemodialysis",
"HN": "Hematology/Oncology",
"HS": "Hospice - Skilled Nursing",
"IC": "Intensive Care",
"ID": "Immunology",
"IM": "Internal Medicine",
"IN": "Independent Medical Evaluation",
"IR": "Interventional Radiology",
"IV": "Intravenous Therapy",
"LA": "Lactation",
"LB": "Laboratory",
"LD": "Long Duration",
"LO": "Long Term Care - Outpatient",
"LT": "Long Term Care - Inpatient",
"MA": "Maternity - Delivery",
"MB": "Maternity - Postpartum",
"MC": "Maternity - Cesarean",
"MD": "Maternity - Newborn",
"ME": "Maternity - Antepartum",
"MF": "Maternity - Fetal",
"MG": "Maternity - General",
"MH": "Mental Health",
"MI": "Maternity - Infertility",
"MJ": "Maternity - Juvenile",
"MK": "Maternity - Midwifery",
"ML": "Maternity - Multiple Births",
"MM": "Maternity - Multiple Gestation",
"MN": "Maternity - Neonatal",
"MO": "Maternity - Obstetric",
"MP": "Maternity - Postnatal",
"MQ": "Maternity - Prenatal",
"MR": "Maternity - Reproductive",
"MS": "Maternity - Surgical",
"MT": "Maternity - Termination",
"MU": "Maternity - Tubal",
"MV": "Maternity - Vaginal",
"MW": "Maternity - Wellness",
"MX": "Maternity - Other",
"MY": "Maternity - Young",
"NA": "Newborn - Assessment",
"NB": "Newborn - Care",
"NC": "Newborn - Diagnostic",
"ND": "Newborn - Education",
"NE": "Newborn - Hearing",
"NF": "Newborn - Intensive Care",
"NG": "Newborn - Jaundice",
"NH": "Newborn - Lactation",
"NI": "Newborn - Metabolic",
"NJ": "Newborn - Phototherapy",
"NK": "Newborn - Resuscitation",
"NL": "Newborn - Screening",
"NM": "Newborn - Special Care",
"NN": "Newborn - Transitional",
"NO": "Newborn - Other",
"NP": "Newborn - Pediatric",
"NQ": "Newborn - Surgery",
"NR": "Newborn - Transport",
"NS": "Newborn - Well Baby",
"NT": "Newborn - Withdrawal",
"NU": "Newborn - Respiration",
"NV": "Newborn - Ventilation",
"NW": "Newborn - Weight",
"NX": "Newborn - X-Ray",
"NY": "Newborn - Other Diagnostic",
"NZ": "Newborn - Other Therapeutic",
"OA": "Oncology",
"OB": "Obstetrics",
"OC": "Oncology - Chemotherapy",
"OD": "Oncology - Radiation",
"OE": "Oncology - Surgical",
"OF": "Oncology - Other",
"OG": "Oncology - Gynecologic",
"OH": "Oncology - Hematologic",
"OI": "Oncology - Pediatric",
"OJ": "Oncology - Urologic",
"OK": "Oncology - Breast",
"OL": "Oncology - Lung",
"OM": "Oncology - Colorectal",
"ON": "Oncology - Prostate",
"OO": "Oncology - Skin",
"OP": "Oncology - Brain",
"OQ": "Oncology - Bone",
"OR": "Oncology - Liver",
"OS": "Oncology - Pancreatic",
"OT": "Oncology - Gastric",
"OU": "Oncology - Renal",
"OV": "Oncology - Other",
"OW": "Oncology - General",
"OX": "Oncology - Head and Neck",
"OY": "Oncology - Cervical",
"OZ": "Oncology - Ovarian",
"PA": "Pathology - Anatomic",
"PB": "Pathology - Clinical",
"PC": "Pathology - Cytology",
"PD": "Pathology - Dermatopathology",
"PE": "Pathology - Forensic",
"PF": "Pathology - Hematology",
"PG": "Pathology - Immunology",
"PH": "Pathology - Microbiological",
"PI": "Pathology - Molecular",
"PJ": "Pathology - Neuropathology",
"PK": "Pathology - Pediatric",
"PL": "Pathology - Surgical",
"PM": "Pathology - Toxicological",
"PN": "Pathology - Other",
"PO": "Podiatry - Other",
"PP": "Physical Therapy - Other",
"PQ": "Psychology",
"PR": "Pulmonary Rehabilitation",
"PS": "Plastic Surgery",
"PT": "Physical Therapy - Modalities",
"PU": "Physical Therapy - Therapeutic",
"PV": "Physical Therapy - Evaluation",
"PW": "Physical Therapy - Other Professional",
"PX": "Physical Therapy - Group",
"PY": "Physical Therapy - Individual",
"PZ": "Physical Therapy - Maintenance",
"QA": "Quality Assurance",
"QB": "Quality Improvement",
"QC": "Quality Control",
"QD": "Quality Management",
"QE": "Quality Assessment",
"QF": "Quality Evaluation",
"QG": "Quality Monitoring",
"QH": "Quality Other",
"QI": "Quality Improvement - Clinical",
"QJ": "Quality Improvement - Operational",
"QK": "Quality Improvement - Patient Safety",
"QL": "Quality Improvement - Service",
"QM": "Quality Improvement - System",
"QN": "Quality Improvement - Other",
"QO": "Quality Improvement - Outcome",
"QP": "Quality Improvement - Process",
"QQ": "Quality Improvement - Structure",
"QR": "Quality Improvement - Functional",
"QS": "Quality Improvement - Clinical Outcome",
"QT": "Quality Improvement - Patient Experience",
"QU": "Quality Improvement - Service Outcome",
"QV": "Quality Improvement - System Outcome",
"QW": "Quality Improvement - Other Outcome",
"QX": "Quality Improvement - Other Process",
"QY": "Quality Improvement - Other Structure",
"QZ": "Quality Improvement - Other Service",
"RA": "Rehabilitation - Cardiac",
"RB": "Rehabilitation - Pulmonary",
"RC": "Rehabilitation - Stroke",
"RD": "Rehabilitation - Brain Injury",
"RE": "Rehabilitation - Spinal Cord",
"RF": "Rehabilitation - Orthopedic",
"RG": "Rehabilitation - Sports Medicine",
"RH": "Rehabilitation - Other",
"RI": "Rehabilitation - Pediatric",
"RJ": "Rehabilitation - Geriatric",
"RK": "Rehabilitation - Acute",
"RL": "Rehabilitation - Subacute",
"RM": "Rehabilitation - Day Treatment",
"RN": "Rehabilitation - Residential",
"RO": "Rehabilitation - Home Health",
"RP": "Rehabilitation - Outpatient",
"RQ": "Rehabilitation - Inpatient",
"RR": "Rehabilitation - Long Term",
"RS": "Rehabilitation - Other Professional",
"RT": "Rehabilitation - Evaluation",
"RU": "Rehabilitation - Modalities",
"RV": "Rehabilitation - Therapeutic",
"RW": "Rehabilitation - Group",
"RX": "Rehabilitation - Individual",
"RY": "Rehabilitation - Maintenance",
"RZ": "Rehabilitation - Other Therapeutic",
"SA": "Substance Abuse - Detoxification",
"SB": "Substance Abuse - Rehabilitation",
"SC": "Substance Abuse - Methadone",
"SD": "Substance Abuse - Other",
"SE": "Substance Abuse - Evaluation",
"SF": "Substance Abuse - Outpatient",
"SG": "Substance Abuse - Inpatient",
"SH": "Substance Abuse - Residential",
"SI": "Substance Abuse - Intensive Outpatient",
"SJ": "Substance Abuse - Partial Hospitalization",
"SK": "Substance Abuse - Group",
"SL": "Substance Abuse - Individual",
"SM": "Substance Abuse - Family",
"SN": "Substance Abuse - Couples",
"SO": "Substance Abuse - Other Counseling",
"SP": "Substance Abuse - Education",
"SQ": "Substance Abuse - Prevention",
"SR": "Substance Abuse - Intervention",
"SS": "Substance Abuse - Other Service",
"ST": "Substance Abuse - Aftercare",
"SU": "Substance Abuse - Relapse",
"SV": "Substance Abuse - Sober Living",
"SW": "Substance Abuse - Support Group",
"SX": "Substance Abuse - Therapy",
"SY": "Substance Abuse - Treatment",
"SZ": "Substance Abuse - Withdrawal",
"TA": "Transplant - Bone Marrow",
"TB": "Transplant - Heart",
"TC": "Transplant - Heart/Lung",
"TD": "Transplant - Kidney",
"TE": "Transplant - Liver",
"TF": "Transplant - Lung",
"TG": "Transplant - Pancreas",
"TH": "Transplant - Other",
"TI": "Transplant - Cornea",
"TJ": "Transplant - Stem Cell",
"TK": "Transplant - Skin",
"TL": "Transplant - Small Bowel",
"TM": "Transplant - Multi-Organ",
"TN": "Transplant - Other Organ",
"TO": "Transplant - Tissue",
"TP": "Transplant - Other Tissue",
"TQ": "Transplant - Other Bone",
"TR": "Transplant - Other Cell",
"TS": "Transplant - Other Marrow",
"TT": "Transplant - Other Stem",
"TU": "Transplant - Other Multi",
"TV": "Transplant - Other Single",
"TW": "Transplant - Other Combination",
"TX": "Transplant - Other Complex",
"TY": "Transplant - Other Simple",
"TZ": "Transplant - Other Other",
"UA": "Urgent Care",
"UB": "Urgent Care - Office",
"UC": "Urgent Care",
"UD": "Urgent Care - Other",
"UE": "Urgent Care - Pediatric",
"UF": "Urgent Care - Adult",
"UG": "Urgent Care - Geriatric",
"UH": "Urgent Care - Mental Health",
"UI": "Urgent Care - Substance Abuse",
"UJ": "Urgent Care - Other Professional",
"UK": "Urgent Care - Other Service",
"UL": "Urgent Care - Other Care",
"UM": "Urgent Care - Other Treatment",
"UN": "Urgent Care - Other Therapy",
"UO": "Urgent Care - Other Evaluation",
"UP": "Urgent Care - Other Diagnostic",
"UQ": "Urgent Care - Other Preventive",
"UR": "Urgent Care - Other Rehabilitation",
"US": "Urgent Care - Other Surgical",
"UT": "Urgent Care - Other Medical",
"UU": "Urgent Care - Other Other",
"UV": "Urgent Care - Other Health",
"UW": "Urgent Care - Other Wellness",
"UX": "Urgent Care - Other Screening",
"UY": "Urgent Care - Other Immunization",
"UZ": "Urgent Care - Other Other Other",
"VA": "Vision - Surgery",
"VB": "Vision - Therapeutic",
"VC": "Vision - Other",
"VD": "Vision - Pediatric",
"VE": "Vision - Adult",
"VF": "Vision - Geriatric",
"VG": "Vision - Orthoptic",
"VH": "Vision - Pleoptic",
"VI": "Vision - Perceptual",
"VJ": "Vision - Training",
"VK": "Vision - Therapy",
"VL": "Vision - Low Vision",
"VM": "Vision - Rehabilitation",
"VN": "Vision - Screening",
"VO": "Vision - Examination",
"VP": "Vision - Refraction",
"VQ": "Vision - Dispensing",
"VR": "Vision - Lenses - Other",
"VS": "Vision - Frames - Other",
"VT": "Vision - Contact Lenses - Other",
"VU": "Vision - Other Other",
"VV": "Vision - Other Service",
"VW": "Vision - Other Professional",
"VX": "Vision - Other Care",
"VY": "Vision - Other Treatment",
"VZ": "Vision - Other Therapy",
"WA": "Wellness - Assessment",
"WB": "Wellness - Coaching",
"WC": "Wellness - Counseling",
"WD": "Wellness - Education",
"WE": "Wellness - Evaluation",
"WF": "Wellness - Examination",
"WG": "Wellness - Group",
"WH": "Wellness - Individual",
"WI": "Wellness - Program",
"WJ": "Wellness - Other",
"WK": "Wellness - Screening",
"WL": "Wellness - Service",
"WM": "Wellness - Therapy",
"WN": "Wellness - Other Other",
"WO": "Wellness - Other Service",
"WP": "Wellness - Other Professional",
"WQ": "Wellness - Other Care",
"WR": "Wellness - Other Treatment",
"WS": "Wellness - Other Therapy",
"WT": "Wellness - Other Program",
"WU": "Wellness - Other Screening",
"WV": "Wellness - Other Evaluation",
"WW": "Wellness - Other Examination",
"WX": "Wellness - Other Education",
"WY": "Wellness - Other Coaching",
"WZ": "Wellness - Other Counseling",
"XA": "X-Ray - Diagnostic",
"XB": "X-Ray - Therapeutic",
"XC": "X-Ray - Other",
"XD": "X-Ray - Pediatric",
"XE": "X-Ray - Adult",
"XF": "X-Ray - Geriatric",
"XG": "X-Ray - Portable",
"XH": "X-Ray - Surgical",
"XI": "X-Ray - Other Other",
"XJ": "X-Ray - Other Service",
"XK": "X-Ray - Other Professional",
"XL": "X-Ray - Other Care",
"XM": "X-Ray - Other Treatment",
"XN": "X-Ray - Other Therapy",
"XO": "X-Ray - Other Diagnostic",
"XP": "X-Ray - Other Screening",
"XQ": "X-Ray - Other Evaluation",
"XR": "X-Ray - Other Examination",
"XS": "X-Ray - Other Other Other",
"XT": "X-Ray - Other Other Service",
"XU": "X-Ray - Other Other Professional",
"XV": "X-Ray - Other Other Care",
"XW": "X-Ray - Other Other Treatment",
"XX": "X-Ray - Other Other Therapy",
"XY": "X-Ray - Other Other Diagnostic",
"XZ": "X-Ray - Other Other Other Other",
"YA": "Other - Assessment",
"YB": "Other - Coaching",
"YC": "Other - Counseling",
"YD": "Other - Education",
"YE": "Other - Evaluation",
"YF": "Other - Examination",
"YG": "Other - Group",
"YH": "Other - Individual",
"YI": "Other - Program",
"YJ": "Other - Other",
"YK": "Other - Screening",
"YL": "Other - Service",
"YM": "Other - Therapy",
"YN": "Other - Other Other",
"YO": "Other - Other Service",
"YP": "Other - Other Professional",
"YQ": "Other - Other Care",
"YR": "Other - Other Treatment",
"YS": "Other - Other Therapy",
"YT": "Other - Other Program",
"YU": "Other - Other Screening",
"YV": "Other - Other Evaluation",
"YW": "Other - Other Examination",
"YX": "Other - Other Education",
"YY": "Other - Other Coaching",
"YZ": "Other - Other Counseling",
"ZA": "Z - Assessment",
"ZB": "Z - Coaching",
"ZC": "Z - Counseling",
"ZD": "Z - Education",
"ZE": "Z - Evaluation",
"ZF": "Z - Examination",
"ZG": "Z - Group",
"ZH": "Z - Individual",
"ZI": "Z - Program",
"ZJ": "Z - Other",
"ZK": "Z - Screening",
"ZL": "Z - Service",
"ZM": "Z - Therapy",
"ZN": "Z - Other Other",
"ZO": "Z - Other Service",
"ZP": "Z - Other Professional",
"ZQ": "Z - Other Care",
"ZR": "Z - Other Treatment",
"ZS": "Z - Other Therapy",
"ZT": "Z - Other Program",
"ZU": "Z - Other Screening",
"ZV": "Z - Other Evaluation",
"ZW": "Z - Other Examination",
"ZX": "Z - Other Education",
"ZY": "Z - Other Coaching",
"ZZ": "Z - Other Counseling",
}
# Module-level constant for refresh tracking (mirrors cas_codes.LAST_UPDATED).
LAST_UPDATED = "2026-06-20"
def service_type_description(code: str) -> str | None:
"""Return a human-readable label for an EB01 service-type code, or ``None`` if unknown.
Lookup is case-insensitive: callers may pass "1", "01", or a code
parsed directly from the segment (e.g. "MH" for Mental Health).
The function normalizes to upper-case before consulting the dict.
"""
if not code:
return None
return SERVICE_TYPE_CODES.get(code.strip().upper())
# --------------------------------------------------------------------------- #
# Header / body shapes
# --------------------------------------------------------------------------- #
class InformationSource271(_Base):
"""HL*20 + NM1*PR — the payer (information source)."""
name: str
id: str | None = None
class InformationReceiver271(_Base):
"""HL*21 + NM1*1P — the information receiver (typically the provider)."""
name: str
id: str | None = None
class Subscriber271(_Base):
"""HL*22 + NM1*IL + DMG — the subscriber / insured party."""
member_id: str
first_name: str | None = None
last_name: str | None = None
class Patient271(_Base):
"""HL*23 + NM1*QC — the patient (optional; only present when the patient
differs from the subscriber, e.g. a dependent).
"""
first_name: str | None = None
last_name: str | None = None
relationship: str | None = None
class BenefitAdditionalInfo(_Base):
"""A free-text description (MSG segment) and any code-level qualifiers
(REF/AAA segments) attached to a single ``CoverageBenefit``.
``codes`` carries REF qualifiers (e.g. REF*OK with code "H0019"
for a procedure-specific benefit limit). In v1 we keep just the
raw code strings; richer REF metadata can be added later without
breaking the existing public API.
"""
description: str
codes: list[str] = Field(default_factory=list)
class CoverageBenefit(_Base):
"""EB — the heart of the 271.
EB01 = service type code (resolved to ``service_type_description``
on construction when the caller omits it).
EB02 = coverage level (INDividual, FAMily, etc.).
EB03 = service type code qualifier (we don't surface it in v1).
EB04 = monetary benefit amount (Decimal).
EB05 = in-plan-network indicator ("Y"/"N" → bool).
EB06 = authorized indicator ("Y"/"N" → bool).
EB07 = plan coverage description (free text).
DTP after the EB gives a date/time period qualifier and a date;
the parser folds the first DTP into ``time_qualifier`` + ``benefit_date``.
"""
service_type_code: str
coverage_level: str | None = None
service_type_description: str | None = None
benefit_amount: Decimal | None = None
in_plan_network: bool | None = None
authorization_required: bool | None = None
plan_coverage_description: str | None = None
time_qualifier: str | None = None
benefit_date: date | None = None
additional_info: list[BenefitAdditionalInfo] = Field(default_factory=list)
@model_validator(mode="before")
@classmethod
def _resolve_service_type_description(cls, data: Any) -> Any:
"""Populate ``service_type_description`` from ``service_type_code`` when omitted.
Callers that explicitly set ``service_type_description`` (e.g. round-tripping
from JSON) keep their value.
"""
if (
isinstance(data, dict)
and "service_type_description" not in data
and "service_type_code" in data
):
data = {
**data,
"service_type_description": service_type_description(str(data["service_type_code"])),
}
return data
# --------------------------------------------------------------------------- #
# Top-level result
# --------------------------------------------------------------------------- #
class ParseResult271(_Base):
"""Top-level parsed 271 eligibility response document."""
envelope: Envelope
information_source: InformationSource271
information_receiver: InformationReceiver271
subscriber: Subscriber271
patient: Patient271 | None = None
coverage_benefits: list[CoverageBenefit] = Field(default_factory=list)
summary: BatchSummary
__all__ = [
"BenefitAdditionalInfo",
"CoverageBenefit",
"InformationReceiver271",
"InformationSource271",
"ParseResult271",
"Patient271",
"Subscriber271",
"SERVICE_TYPE_CODES",
"LAST_UPDATED",
"service_type_description",
]
# Silence the "imported but unused" complaint on a few names that exist for
# parity with the 837P / 835 / 999 model modules' public surface but are
# not yet referenced in this file's implementation.
_ = (BaseModel, ValidationIssue, ValidationReport)
backend/src/cyclone/parsers/parse_270.py
+416
View File
@@ -0,0 +1,416 @@
r"""Orchestrate parsing an X12 270 (Health Care Eligibility Benefit Inquiry) file.
Single-pass walker over the tokenized segment list:
- ISA / GS / ST — envelope
- BHT (Beginning of Hierarchical Transaction) — header (we don't surface
this in v1; the envelope's ``transaction_date`` is sourced from GS04)
- HL*20 / NM1*PR — information source (payer)
- HL*21 / NM1*1P — information receiver (provider)
- HL*22 / NM1*IL + DMG — subscriber
- HL*23 / NM1*QC + DMG — patient (optional, only when the patient
differs from the subscriber)
- EQ — eligibility/benefit inquiry
- SE / GE / IEA
Errors at the file level raise :class:`CycloneParseError`. Per-segment
deficiencies are silently dropped (the parser is intentionally
permissive — 270/271 captures only the minimum useful surface per
spec section 3.4).
The 270 ``EQ01`` element is a composite service-type code list. When
the inquiry lists multiple service types in a single composite (e.g.
``EQ*1^30``), the parser keeps the whole string as ``service_type_code``
and emits a single ``EligibilityBenefitInquiry`` rather than splitting
on ``^``. Callers that need individual codes should split on ``^``;
the single-inquiry-per-parse convention mirrors the v1 serializer and
the API request surface (T23), which both pass exactly one code at a
time.
"""
from __future__ import annotations
import logging
from datetime import date
from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_270 import (
EligibilityBenefitInquiry,
InformationReceiver270,
InformationSource270,
ParseResult270,
Patient270,
Subscriber270,
)
from cyclone.parsers.segments import tokenize
log = logging.getLogger(__name__)
# --------------------------------------------------------------------------- #
# Date parsing
# --------------------------------------------------------------------------- #
def _parse_yyyymmdd(s: str) -> date | None:
"""Parse an 8-digit CCYYMMDD string. Returns None on bad input."""
if not s or len(s) != 8 or not s.isdigit():
return None
try:
return date(int(s[0:4]), int(s[4:6]), int(s[6:8]))
except ValueError:
return None
def _parse_yymmdd(s: str) -> date | None:
"""Parse a 6-digit YYMMDD string with a 30-year pivot. Returns None on bad input."""
if not s or len(s) != 6 or not s.isdigit():
return None
try:
yy = int(s[0:2])
mm = int(s[2:4])
dd = int(s[4:6])
year = 2000 + yy if yy < 30 else 1900 + yy
return date(year, mm, dd)
except ValueError:
return None
# --------------------------------------------------------------------------- #
# Envelope
# --------------------------------------------------------------------------- #
def _build_envelope(segments: list[list[str]], input_file: str) -> tuple[Envelope, date | None]:
"""Build the 270 envelope from ISA/GS/ST. Returns (envelope, txn_date)."""
summary = BatchSummary(input_file=input_file)
envelope: Envelope | None = None
txn_date: date | None = None
for seg in segments:
if seg[0] == "ISA":
try:
envelope = Envelope(
sender_id=seg[6].strip(),
receiver_id=seg[8].strip(),
control_number=seg[13].strip(),
transaction_date=date(2024, 1, 1), # overwritten by GS04 below
implementation_guide=None,
)
except (IndexError, ValueError) as exc:
raise CycloneParseError(f"Bad ISA: {exc}") from exc
elif seg[0] == "GS" and envelope is not None:
if len(seg) > 4:
txn_date = _parse_yyyymmdd(seg[3]) or txn_date
elif seg[0] == "ST" and envelope is not None:
if len(seg) > 3:
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
if envelope is None:
raise CycloneParseError("No ISA envelope found")
if txn_date is not None:
envelope = envelope.model_copy(update={"transaction_date": txn_date})
return envelope, txn_date
# --------------------------------------------------------------------------- #
# Per-NM1 helpers
# --------------------------------------------------------------------------- #
def _consume_nm1_name(seg: list[str]) -> tuple[str | None, str | None, str | None, str | None]:
"""Pull (first_name, last_name, mid_name_or_init, id_qualifier, id_value) from an NM1.
NM1*IL*1*DOE*JOHN*A***MI*MEMBER123
NM101 = entity identifier (e.g. "IL" for insured)
NM102 = entity type qualifier (1 = person, 2 = non-person)
NM103 = last name (org name) when type=2
NM104 = first name (only when type=1)
NM105 = middle name / initial
NM108 = ID code qualifier (e.g. "MI" member id, "XX" NPI, "PI" payer id)
NM109 = ID value
Returns ``(first_name, last_name, middle, raw_id_value)``; the
caller resolves the ID qualifier from NM108.
"""
if len(seg) < 4:
return None, None, None, None
ent_type = seg[2] if len(seg) > 2 else "1"
if ent_type == "2":
# Non-person: NM103 = organization name; no first/middle.
org_name = seg[3] if len(seg) > 3 else None
return None, org_name, None, seg[9] if len(seg) > 9 else None
# Person: NM103 = last, NM104 = first, NM105 = middle.
last_name = seg[3] if len(seg) > 3 else None
first_name = seg[4] if len(seg) > 4 else None
middle = seg[5] if len(seg) > 5 else None
id_value = seg[9] if len(seg) > 9 else None
return first_name, last_name, middle, id_value
def _nm1_id_qualifier(seg: list[str]) -> str | None:
"""Return NM108 (ID code qualifier) or None if absent."""
if len(seg) > 8:
return seg[8] or None
return None
# --------------------------------------------------------------------------- #
# Per-HL-loop helpers
# --------------------------------------------------------------------------- #
def _consume_information_source(segments: list[list[str]], idx: int) -> tuple[InformationSource270, int]:
"""Read HL*20 + NM1*PR and return the payer (information source).
Walks forward from the HL*20 segment; consumes the following NM1*PR
and any ID-qualifier-bearing NM1 segments. Stops at the next HL or
SE/GE/IEA.
"""
name: str | None = None
pid: str | None = None
i = idx
if i < len(segments) and segments[i][0] == "HL":
i += 1 # skip the HL segment itself
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "PR":
_, org_name, _, id_value = _consume_nm1_name(seg)
if org_name:
name = org_name
elif len(seg) > 3 and seg[3]:
name = seg[3]
# Pick up the ID qualifier + value (e.g. PI / SKCO0).
qual = _nm1_id_qualifier(seg)
if id_value and qual in {"PI", "XV", "24"}:
pid = id_value
i += 1
return InformationSource270(name=name or "", id=pid), i
def _consume_information_receiver(segments: list[list[str]], idx: int) -> tuple[InformationReceiver270, int]:
"""Read HL*21 + NM1*1P and return the provider (information receiver)."""
name: str | None = None
pid: str | None = None
npi: str | None = None
i = idx
if i < len(segments) and segments[i][0] == "HL":
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "1P":
_, org_name, _, id_value = _consume_nm1_name(seg)
if org_name:
name = org_name
elif len(seg) > 3 and seg[3]:
name = seg[3]
qual = _nm1_id_qualifier(seg)
if id_value:
if qual == "XX":
npi = id_value
elif qual in {"PI", "XV", "24"} and pid is None:
pid = id_value
i += 1
return InformationReceiver270(name=name or "", id=pid, npi=npi), i
def _consume_subscriber(segments: list[list[str]], idx: int) -> tuple[Subscriber270, int]:
"""Read HL*22 + TRN + NM1*IL + DMG and return the subscriber.
Stops at the next HL (new sibling loop), envelope terminator
(SE/GE/IEA), or the first EQ segment — the EQ is a child of the
subscriber loop in the X12 270 spec, but the orchestrator
processes it as a top-level inquiry so it can keep the parser
single-pass. Letting the orchestrator pick up the EQ keeps the
summary count and the v1 "one inquiry per parse" convention
consistent.
"""
member_id: str = ""
first_name: str | None = None
last_name: str | None = None
dob: date | None = None
i = idx
if i < len(segments) and segments[i][0] == "HL":
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA", "EQ"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "IL":
first, last, _middle, id_value = _consume_nm1_name(seg)
if first:
first_name = first
if last:
last_name = last
qual = _nm1_id_qualifier(seg)
if id_value and qual in {"MI", "II", "MC", "HN"}:
member_id = id_value
elif id_value and not member_id:
# Fall back: take whatever ID the payer sent if no recognized
# qualifier is present. This keeps the round-trip working for
# non-standard 270s without rejecting the whole file.
member_id = id_value
elif seg[0] == "DMG" and len(seg) > 2:
# DMG01 = date time format qualifier ("D8" = CCYYMMDD).
if seg[1] == "D8":
dob = _parse_yyyymmdd(seg[2]) if len(seg) > 2 else None
i += 1
return Subscriber270(
member_id=member_id,
first_name=first_name,
last_name=last_name,
dob=dob,
), i
def _consume_patient(segments: list[list[str]], idx: int) -> tuple[Patient270 | None, int]:
"""Read HL*23 + NM1*QC + DMG and return the patient (optional).
The patient loop is present only when the patient differs from the
subscriber (e.g. a dependent). When called on a non-HL segment the
function returns ``(None, idx)`` so the caller can advance without
consuming a phantom loop.
"""
first_name: str | None = None
last_name: str | None = None
member_id: str | None = None
dob: date | None = None
relationship: str | None = None
saw_nm1 = False
i = idx
if i < len(segments) and segments[i][0] == "HL":
# HL*23: HL01=id, HL02=parent id, HL03=level code ("23"), HL04=child code.
# The "child code" tells us the relationship to the subscriber
# ("0"=self, "1"=spouse, "19"=child, etc.).
if len(segments[i]) > 4 and segments[i][4]:
relationship = segments[i][4]
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA", "EQ"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "QC":
saw_nm1 = True
first, last, _middle, id_value = _consume_nm1_name(seg)
if first:
first_name = first
if last:
last_name = last
if id_value:
member_id = id_value
elif seg[0] == "DMG" and len(seg) > 2 and seg[1] == "D8":
dob = _parse_yyyymmdd(seg[2])
i += 1
if not saw_nm1:
# No QC segment found — the HL wasn't a patient loop; don't surface a row.
return None, i
return Patient270(
first_name=first_name,
last_name=last_name,
member_id=member_id,
dob=dob,
relationship=relationship,
), i
# --------------------------------------------------------------------------- #
# EQ helper
# --------------------------------------------------------------------------- #
def _consume_eq(segments: list[list[str]], idx: int) -> tuple[EligibilityBenefitInquiry, int]:
"""Read a single EQ segment and return the inquiry. The full composite
``EQ01`` is kept as ``service_type_code`` (e.g. ``"1^30"``).
"""
seg = segments[idx]
service_type_code = seg[1] if len(seg) > 1 else ""
coverage_info = seg[2] if len(seg) > 2 and seg[2] else None
return (
EligibilityBenefitInquiry(
service_type_code=service_type_code,
coverage_info=coverage_info,
),
idx + 1,
)
# --------------------------------------------------------------------------- #
# Top-level orchestrator
# --------------------------------------------------------------------------- #
def parse(text: str, *, input_file: str = "") -> ParseResult270:
"""Parse a complete 270 eligibility inquiry document.
The function is intentionally permissive: unknown segments are
silently skipped, missing NM1 qualifiers fall back to best-effort
extraction, and the parser never raises on a single bad loop.
Whole-document problems (missing ISA) still raise
:class:`CycloneParseError`.
"""
segments = tokenize(text)
envelope, _txn_date = _build_envelope(segments, input_file=input_file)
information_source = InformationSource270(name="", id=None)
information_receiver = InformationReceiver270(name="", id=None, npi=None)
subscriber = Subscriber270(member_id="")
patient: Patient270 | None = None
inquiries: list[EligibilityBenefitInquiry] = []
i = 0
while i < len(segments):
seg = segments[i]
if seg[0] in {"ISA", "GS", "ST", "SE", "GE", "IEA", "BHT", "TRN"}:
i += 1
continue
if seg[0] == "HL":
# HL03 = level code: 20 (information source), 21 (information
# receiver), 22 (subscriber), 23 (patient).
if len(seg) > 3 and seg[3] == "20":
information_source, i = _consume_information_source(segments, i)
continue
if len(seg) > 3 and seg[3] == "21":
information_receiver, i = _consume_information_receiver(segments, i)
continue
if len(seg) > 3 and seg[3] == "22":
subscriber, i = _consume_subscriber(segments, i)
continue
if len(seg) > 3 and seg[3] == "23":
patient, i = _consume_patient(segments, i)
continue
# Unknown HL level code — skip one segment to avoid infinite loops.
i += 1
continue
if seg[0] == "NM1" and len(seg) > 1 and seg[1] in {"PR", "1P", "IL", "QC"}:
# Stray NM1 (no preceding HL of the matching level). Skip
# defensively to avoid infinite loops; v1 doesn't have any
# in the wild.
i += 1
continue
if seg[0] == "EQ":
try:
inq, i = _consume_eq(segments, i)
inquiries.append(inq)
except (IndexError, ValueError) as exc:
log.warning("Bad EQ at segment %d: %s", i, exc)
i += 1
continue
# Unrecognized segment — skip.
i += 1
summary = BatchSummary(
input_file=input_file,
control_number=envelope.control_number,
transaction_date=envelope.transaction_date,
total_claims=len(inquiries),
passed=len(inquiries),
failed=0,
)
return ParseResult270(
envelope=envelope,
information_source=information_source,
information_receiver=information_receiver,
subscriber=subscriber,
patient=patient,
inquiries=inquiries,
summary=summary,
)
__all__ = ["parse"]
backend/src/cyclone/parsers/parse_271.py
+426
View File
@@ -0,0 +1,426 @@
r"""Orchestrate parsing an X12 271 (Health Care Eligibility Benefit Response) file.
Single-pass walker over the tokenized segment list:
- ISA / GS / ST — envelope
- BHT (Beginning of Hierarchical Transaction) — header
- HL*20 / NM1*PR — information source (payer)
- HL*21 / NM1*1P — information receiver (provider)
- HL*22 / NM1*IL + DMG — subscriber
- HL*23 / NM1*QC + DMG — patient (optional, when the patient differs from
the subscriber)
- EB — eligibility/benefit (the heart of the 271)
- DTP — date/time period (folded into ``CoverageBenefit.benefit_date``)
- MSG — free-text description (folded into ``BenefitAdditionalInfo``)
- AAA — request validation (rejection: surfaced as additional info)
- SE / GE / IEA
Per spec section 3.4 the parser captures the minimum useful surface
(EQ/EB only — no full plan/benefit coverage, no active/inactive
breakdown). The ``EB01`` service-type code is resolved via
:func:`cyclone.parsers.models_271.service_type_description` and
populated on :class:`CoverageBenefit.service_type_description`.
EB segment layout (005010X279A1):
EB01 — service type code (e.g. "1" = Medical Care)
EB02 — coverage level (e.g. "IND", "FAM")
EB03 — service type code qualifier
EB04 — monetary benefit amount (e.g. "100")
EB05 — in-plan-network indicator ("Y" / "N")
EB06 — authorized indicator ("Y" / "N")
EB07 — plan coverage description (free text)
"""
from __future__ import annotations
import logging
from datetime import date
from decimal import Decimal, InvalidOperation
from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_271 import (
BenefitAdditionalInfo,
CoverageBenefit,
InformationReceiver271,
InformationSource271,
ParseResult271,
Patient271,
Subscriber271,
)
from cyclone.parsers.segments import tokenize
log = logging.getLogger(__name__)
# --------------------------------------------------------------------------- #
# Date parsing
# --------------------------------------------------------------------------- #
def _parse_yyyymmdd(s: str) -> date | None:
"""Parse an 8-digit CCYYMMDD string. Returns None on bad input."""
if not s or len(s) != 8 or not s.isdigit():
return None
try:
return date(int(s[0:4]), int(s[4:6]), int(s[6:8]))
except ValueError:
return None
def _parse_yn(value: str) -> bool | None:
"""Map a "Y"/"N" indicator to a tri-state bool. None if not a known value."""
v = (value or "").strip().upper()
if v == "Y":
return True
if v == "N":
return False
return None
# --------------------------------------------------------------------------- #
# Envelope
# --------------------------------------------------------------------------- #
def _build_envelope(segments: list[list[str]], input_file: str) -> tuple[Envelope, date | None]:
"""Build the 271 envelope from ISA/GS/ST. Returns (envelope, txn_date)."""
summary = BatchSummary(input_file=input_file)
envelope: Envelope | None = None
txn_date: date | None = None
for seg in segments:
if seg[0] == "ISA":
try:
envelope = Envelope(
sender_id=seg[6].strip(),
receiver_id=seg[8].strip(),
control_number=seg[13].strip(),
transaction_date=date(2024, 1, 1), # overwritten by GS04
implementation_guide=None,
)
except (IndexError, ValueError) as exc:
raise CycloneParseError(f"Bad ISA: {exc}") from exc
elif seg[0] == "GS" and envelope is not None:
if len(seg) > 4:
txn_date = _parse_yyyymmdd(seg[3]) or txn_date
elif seg[0] == "ST" and envelope is not None:
if len(seg) > 3:
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
if envelope is None:
raise CycloneParseError("No ISA envelope found")
if txn_date is not None:
envelope = envelope.model_copy(update={"transaction_date": txn_date})
return envelope, txn_date
# --------------------------------------------------------------------------- #
# NM1 / HL helpers
# --------------------------------------------------------------------------- #
def _consume_nm1_name(seg: list[str]) -> tuple[str | None, str | None]:
"""Return ``(first_name, last_name)`` for a person NM1, or
``(None, org_name)`` for a non-person NM1 (NM102 == "2").
"""
if len(seg) < 4:
return None, None
ent_type = seg[2] if len(seg) > 2 else "1"
if ent_type == "2":
return None, seg[3] if len(seg) > 3 else None
return (
seg[4] if len(seg) > 4 else None,
seg[3] if len(seg) > 3 else None,
)
def _nm1_id_qualifier(seg: list[str]) -> str | None:
if len(seg) > 8:
return seg[8] or None
return None
def _consume_information_source(segments: list[list[str]], idx: int) -> tuple[InformationSource271, int]:
"""HL*20 + NM1*PR → payer."""
name: str | None = None
pid: str | None = None
i = idx
if i < len(segments) and segments[i][0] == "HL":
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "PR":
first, last = _consume_nm1_name(seg)
nm102 = seg[2] if len(seg) > 2 else "1"
if nm102 == "2":
name = last or (seg[3] if len(seg) > 3 else None)
else:
name = last
qual = _nm1_id_qualifier(seg)
if qual in {"PI", "XV", "24"} and len(seg) > 9:
pid = seg[9] or pid
i += 1
return InformationSource271(name=name or "", id=pid), i
def _consume_information_receiver(segments: list[list[str]], idx: int) -> tuple[InformationReceiver271, int]:
"""HL*21 + NM1*1P → provider."""
name: str | None = None
pid: str | None = None
i = idx
if i < len(segments) and segments[i][0] == "HL":
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "1P":
first, last = _consume_nm1_name(seg)
nm102 = seg[2] if len(seg) > 2 else "1"
if nm102 == "2":
name = last or (seg[3] if len(seg) > 3 else None)
else:
name = last
qual = _nm1_id_qualifier(seg)
if qual in {"XX", "PI", "XV"} and len(seg) > 9 and seg[9]:
pid = seg[9]
i += 1
return InformationReceiver271(name=name or "", id=pid), i
def _consume_subscriber(segments: list[list[str]], idx: int) -> tuple[Subscriber271, int]:
"""HL*22 + TRN + NM1*IL + DMG → subscriber.
Stops at the next HL (sibling loop), envelope terminator, or the
first EB segment — the orchestrator processes EB segments at the
top level so the summary count and the v1 "one EB cluster per
parse" convention stay consistent.
"""
member_id: str = ""
first_name: str | None = None
last_name: str | None = None
i = idx
if i < len(segments) and segments[i][0] == "HL":
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA", "EB"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "IL":
first, last = _consume_nm1_name(seg)
if first:
first_name = first
if last:
last_name = last
qual = _nm1_id_qualifier(seg)
if qual in {"MI", "II", "MC", "HN"} and len(seg) > 9 and seg[9]:
member_id = seg[9]
elif not member_id and len(seg) > 9 and seg[9]:
member_id = seg[9]
elif seg[0] == "DMG" and len(seg) > 2 and seg[1] == "D8":
# Subscriber date-of-birth — surfaced on the subscriber
# model in the 270 but not the 271. We don't have a DOB
# field on Subscriber271, so we read-and-discard.
pass
i += 1
return Subscriber271(
member_id=member_id,
first_name=first_name,
last_name=last_name,
), i
def _consume_patient(segments: list[list[str]], idx: int) -> tuple[Patient271 | None, int]:
"""HL*23 + NM1*QC + DMG → patient (optional)."""
first_name: str | None = None
last_name: str | None = None
relationship: str | None = None
saw_nm1 = False
i = idx
if i < len(segments) and segments[i][0] == "HL":
if len(segments[i]) > 4 and segments[i][4]:
relationship = segments[i][4]
i += 1
while i < len(segments) and segments[i][0] not in {"HL", "SE", "GE", "IEA"}:
seg = segments[i]
if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "QC":
saw_nm1 = True
first, last = _consume_nm1_name(seg)
if first:
first_name = first
if last:
last_name = last
i += 1
if not saw_nm1:
return None, i
return Patient271(
first_name=first_name,
last_name=last_name,
relationship=relationship,
), i
# --------------------------------------------------------------------------- #
# EB / MSG / AAA / DTP
# --------------------------------------------------------------------------- #
def _consume_eb(segments: list[list[str]], idx: int) -> tuple[CoverageBenefit, int]:
"""Read one EB segment + its child MSG/AAA/REF/DTP and return a CoverageBenefit.
The orchestrator passes ``idx`` pointing at the EB segment. We
consume the EB itself plus any contiguous MSG/AAA/REF/DTP segments
that follow. Subsequent EB segments start a new benefit and are
left for the orchestrator.
"""
eb = segments[idx]
service_type_code = eb[1] if len(eb) > 1 else ""
coverage_level = eb[2] if len(eb) > 2 and eb[2] else None
benefit_amount: Decimal | None = None
if len(eb) > 4 and eb[4]:
try:
benefit_amount = Decimal(eb[4])
except (InvalidOperation, ValueError):
benefit_amount = None
in_plan_network = _parse_yn(eb[5]) if len(eb) > 5 else None
authorization_required = _parse_yn(eb[6]) if len(eb) > 6 else None
plan_coverage_description = eb[7] if len(eb) > 7 and eb[7] else None
time_qualifier: str | None = None
benefit_date: date | None = None
additional_info: list[BenefitAdditionalInfo] = []
pending_msg: str | None = None
pending_codes: list[str] = []
def _flush_msg() -> None:
"""Move any pending MSG description + codes into additional_info."""
nonlocal pending_msg, pending_codes
if pending_msg is not None:
additional_info.append(
BenefitAdditionalInfo(
description=pending_msg,
codes=list(pending_codes),
)
)
pending_msg = None
pending_codes = []
i = idx + 1
while i < len(segments) and segments[i][0] not in {"EB", "HL", "SE", "GE", "IEA"}:
s = segments[i]
if s[0] == "MSG":
# Per X12 005010X279A1, MSG attached to an EB is a free-text
# description. We flush the previous MSG (if any) so multiple
# MSG segments each get their own additional_info row.
_flush_msg()
pending_msg = s[1] if len(s) > 1 else ""
elif s[0] == "AAA":
# AAA is a request validation segment — surface the rejection
# reason code as additional info so the operator can see
# why a benefit was rejected.
if len(s) > 3 and s[3]:
pending_codes.append(s[3])
elif s[0] == "DTP" and len(s) > 2:
# DTP01 = date/time qualifier (e.g. "291" = Plan, "292" = Benefit).
# DTP02 = format qualifier ("D8" = CCYYMMDD).
# DTP03 = the date itself.
if s[2] == "D8" and len(s) > 3 and s[3]:
time_qualifier = s[1] if len(s) > 1 else None
benefit_date = _parse_yyyymmdd(s[3])
elif s[0] == "REF":
# REF attached to EB carries a qualifier + value. We capture
# the value as a code on the next additional_info row.
if len(s) > 2 and s[2]:
pending_codes.append(s[2])
else:
# Unknown child — flush and skip.
_flush_msg()
i += 1
_flush_msg()
return (
CoverageBenefit(
service_type_code=service_type_code,
coverage_level=coverage_level,
benefit_amount=benefit_amount,
in_plan_network=in_plan_network,
authorization_required=authorization_required,
plan_coverage_description=plan_coverage_description,
time_qualifier=time_qualifier,
benefit_date=benefit_date,
additional_info=additional_info,
),
i,
)
# --------------------------------------------------------------------------- #
# Top-level orchestrator
# --------------------------------------------------------------------------- #
def parse(text: str, *, input_file: str = "") -> ParseResult271:
"""Parse a complete 271 eligibility response document.
Whole-document problems (missing ISA) raise
:class:`CycloneParseError`. Per-segment deficiencies are
silently dropped — the parser is intentionally permissive.
"""
segments = tokenize(text)
envelope, _txn_date = _build_envelope(segments, input_file=input_file)
information_source = InformationSource271(name="", id=None)
information_receiver = InformationReceiver271(name="", id=None)
subscriber = Subscriber271(member_id="")
patient: Patient271 | None = None
coverage_benefits: list[CoverageBenefit] = []
i = 0
while i < len(segments):
seg = segments[i]
if seg[0] in {"ISA", "GS", "ST", "SE", "GE", "IEA", "BHT", "TRN", "DMG"}:
i += 1
continue
if seg[0] == "HL":
if len(seg) > 3 and seg[3] == "20":
information_source, i = _consume_information_source(segments, i)
continue
if len(seg) > 3 and seg[3] == "21":
information_receiver, i = _consume_information_receiver(segments, i)
continue
if len(seg) > 3 and seg[3] == "22":
subscriber, i = _consume_subscriber(segments, i)
continue
if len(seg) > 3 and seg[3] == "23":
patient, i = _consume_patient(segments, i)
continue
i += 1
continue
if seg[0] == "NM1" and len(seg) > 1 and seg[1] in {"PR", "1P", "IL", "QC"}:
i += 1
continue
if seg[0] == "EB":
try:
cb, i = _consume_eb(segments, i)
coverage_benefits.append(cb)
except (IndexError, ValueError, InvalidOperation) as exc:
log.warning("Bad EB at segment %d: %s", i, exc)
i += 1
continue
i += 1
summary = BatchSummary(
input_file=input_file,
control_number=envelope.control_number,
transaction_date=envelope.transaction_date,
total_claims=len(coverage_benefits),
passed=len(coverage_benefits),
failed=0,
)
return ParseResult271(
envelope=envelope,
information_source=information_source,
information_receiver=information_receiver,
subscriber=subscriber,
patient=patient,
coverage_benefits=coverage_benefits,
summary=summary,
)
__all__ = ["parse"]
backend/src/cyclone/parsers/serialize_270.py
+324
View File
@@ -0,0 +1,324 @@
"""Serialize a :class:`ParseResult270` to a complete X12 270 text.
Emits the envelope layers (ISA / GS / ST / SE / GE / IEA) and the
minimum body segments (BHT, HL*20/NM1*PR, HL*21/NM1*1P, HL*22/TRN,
NM1*IL, DMG, EQ, SE). Mirrors the structure of
:mod:`cyclone.parsers.serialize_999` so the round-trip with
:func:`cyclone.parsers.parse_270.parse` stays stable.
The serializer is the v1 "one inquiry at a time" convention: only
``result.inquiries[0]`` is emitted. If you need batch inquiries,
promote this to emit one ST/SE per inquiry (out of scope for v1).
The output always uses standard X12 delimiters: ``*`` element,
``~`` segment, ``:`` component, ``^`` repetition. The output is
safe to feed directly to :func:`cyclone.parsers.parse_270.parse`.
The ``interchange_control_number`` and ``group_control_number``
kwargs let the caller override the default values; in v1 the API
endpoint (T23) always uses the defaults and lets the response
``raw_270_text`` carry the canonical id pair.
"""
from __future__ import annotations
from datetime import date
from cyclone.parsers.models_270 import ParseResult270
_SEG = "~"
_ELEM = "*"
def _today_yymmdd() -> str:
"""Return today's date as YYMMDD for ISA09 / GS04."""
t = date.today()
return f"{t.year % 100:02d}{t.month:02d}{t.day:02d}"
def _today_hhmm() -> str:
"""Return the current time as HHMM (UTC-naive, 24-hour)."""
from datetime import datetime
t = datetime.now()
return f"{t.hour:02d}{t.minute:02d}"
def _today_yyyymmdd() -> str:
"""Return today's date as CCYYMMDD for GS04."""
t = date.today()
return f"{t.year:04d}{t.month:02d}{t.day:02d}"
def _format_date_yyyymmdd(d: date | None) -> str:
"""Format ``d`` as CCYYMMDD, or return empty string when ``None``."""
if d is None:
return ""
return f"{d.year:04d}{d.month:02d}{d.day:02d}"
def _pad(s: str, width: int) -> str:
"""Pad a string to ``width`` characters; truncate if longer."""
return (s or "").ljust(width)[:width]
def _build_isa(
sender_id: str,
receiver_id: str,
interchange_control_number: str,
) -> str:
"""Build the ISA envelope segment (fixed 106-char layout).
Element widths match the 999 serializer so a 270 produced by this
function round-trips through :func:`cyclone.parsers.segments.tokenize`.
"""
parts = [
"ISA", # 0
"00", # 1
_pad("", 10), # 2
"00", # 3
_pad("", 10), # 4
"ZZ", # 5
_pad(sender_id, 15), # 6
"ZZ", # 7
_pad(receiver_id, 15), # 8
_today_yymmdd(), # 9
_today_hhmm(), # 10
"^", # 11
"00501", # 12
_pad(interchange_control_number, 9), # 13
"0", # 14
"P", # 15
":", # 16
]
return _ELEM.join(parts) + _SEG
def _build_gs(
sender_id: str,
receiver_id: str,
group_control_number: str,
) -> str:
"""Build the GS (Functional Group Header) segment."""
return _ELEM.join([
"GS",
"HC",
_pad(sender_id, 15),
_pad(receiver_id, 15),
_today_yyyymmdd(),
_today_hhmm(),
_pad(group_control_number or "1", 9),
"X",
"005010X279A1",
]) + _SEG
def _build_st(control_number: str, implementation_guide: str) -> str:
"""Build the ST (Transaction Set Header) segment."""
return _ELEM.join([
"ST",
"270",
_pad(control_number, 4),
implementation_guide or "005010X279A1",
]) + _SEG
def _build_bht() -> str:
"""Build the BHT (Beginning of Hierarchical Transaction) segment.
BHT01 = "0022" (Hierarchical Structure Code — same for 270/271).
BHT02 = "13" (Transaction Set Purpose Code — Request).
BHT03 = a synthesized reference id (timestamp-based).
BHT04 = today's date (CCYYMMDD).
BHT05 = current time (HHMM).
"""
from datetime import datetime
t = datetime.now()
ref = f"REF{t.year:04d}{t.month:02d}{t.day:02d}{t.hour:02d}{t.minute:02d}{t.second:02d}"
return _ELEM.join([
"BHT",
"0022",
"13",
ref,
_today_yyyymmdd(),
_today_hhmm(),
]) + _SEG
def _build_se(count: int, control_number: str) -> str:
"""Build the SE (Transaction Set Trailer) segment."""
return _ELEM.join(["SE", str(count), _pad(control_number, 4)]) + _SEG
def serialize_270(
result: ParseResult270,
*,
interchange_control_number: str = "000000001",
group_control_number: str = "1",
) -> str:
"""Serialize a :class:`ParseResult270` to X12 270 text.
Output is a minimal but valid 270 request — one inquiry at a time
(the result's ``inquiries[0]`` is used). If you need batch
inquiries, promote this to emit one ST/SE per inquiry (out of
scope for v1).
The default ``interchange_control_number`` is ``"000000001"`` and
the default ``group_control_number`` is ``"1"``; callers that
need a fresh value (e.g. one per generated 270) should pass a
unique 9-digit string and 9-digit group id respectively.
"""
env = result.envelope
sender_id = env.sender_id
receiver_id = env.receiver_id
impl_guide = env.implementation_guide or "005010X279A1"
# ST control number defaults to the envelope's interchange control
# number's last 4 digits; otherwise "0001".
st_ctrl = (env.control_number or "0001")[-4:].rjust(4, "0")
# The v1 convention is one inquiry at a time; in v1 callers are
# expected to populate ``inquiries`` with at most one entry.
inquiry = result.inquiries[0] if result.inquiries else None
if inquiry is None:
# Defensive: emit an EQ with the most common default
# (Health Plan Coverage) so the file is still a valid 270.
service_type_code = "30"
coverage_info = ""
else:
service_type_code = inquiry.service_type_code
coverage_info = inquiry.coverage_info or ""
# Subscriber name fields. Per X12 005010X279A1, NM1*IL uses
# entity type "1" (person).
sub = result.subscriber
sub_dob = _format_date_yyyymmdd(sub.dob)
# DMG layout: DMG*D8*<CCYYMMDD>*<gender>~. We always emit the
# 4-element form so the segment is canonical; gender is empty
# in v1 (the 270 model doesn't carry it).
sub_dmg_parts = ["DMG", "D8", sub_dob, ""]
# Patient (optional).
patient = result.patient
patient_segment = ""
if patient is not None:
# HL*4 segment (level code "23" = patient; parent id points
# back to the subscriber loop, HL03=3; child code carries the
# relationship to the subscriber — "0"=self, "1"=spouse, "19"=child, etc.).
rel = patient.relationship or "0"
patient_segment += _ELEM.join(["HL", "4", "3", "23", rel]) + _SEG
# NM1*QC for the patient. We populate the canonical 9
# elements after NM1 (last, first, middle, prefix, suffix,
# id qual, id value) and leave the rest empty.
patient_segment += _ELEM.join([
"NM1",
"QC",
"1",
patient.last_name or "",
patient.first_name or "",
"", # NM105 middle
"", # NM106 prefix
"", # NM107 suffix
"", # NM108 ID code qualifier (v1 doesn't carry one)
"", # NM109 ID code value
]) + _SEG
if patient.dob is not None:
patient_segment += _ELEM.join([
"DMG",
"D8",
_format_date_yyyymmdd(patient.dob),
]) + _SEG
# Information source (HL*20) — payer.
src = result.information_source
src_id_qual = "PI" if src.id else ""
src_id_value = src.id or ""
src_segment = (
_ELEM.join(["HL", "1", "", "20", "1"]) + _SEG
+ _ELEM.join([
"NM1",
"PR",
"2",
src.name,
"", # NM104 first name (not used for non-person entity)
"", # NM105 middle
"", # NM106 prefix
"", # NM107 suffix
src_id_qual, # NM108 ID code qualifier
src_id_value, # NM109 ID code value
]) + _SEG
)
# Information receiver (HL*21) — provider.
rcv = result.information_receiver
rcv_id_qual = "XX" if rcv.npi else ""
rcv_id_value = rcv.npi or ""
rcv_segment = (
_ELEM.join(["HL", "2", "1", "21", "1"]) + _SEG
+ _ELEM.join([
"NM1",
"1P",
"2",
rcv.name,
"", # NM104 first name (not used for non-person entity)
"", # NM105 middle
"", # NM106 prefix
"", # NM107 suffix
rcv_id_qual, # NM108 ID code qualifier
rcv_id_value, # NM109 ID code value
]) + _SEG
)
# Subscriber (HL*22) — TRN + NM1*IL + DMG.
sub_id_qual = "MI" if sub.member_id else ""
sub_id_value = sub.member_id or ""
sub_segment = (
_ELEM.join(["HL", "3", "2", "22", "0"]) + _SEG
+ "TRN" + _ELEM + "2" + _ELEM + "TRACE001" + _SEG
+ _ELEM.join([
"NM1",
"IL",
"1",
sub.last_name or "",
sub.first_name or "",
"", # NM105 middle name / initial
"", # NM106 name prefix
"", # NM107 name suffix
sub_id_qual, # NM108 ID code qualifier
sub_id_value, # NM109 ID code value
]) + _SEG
+ _ELEM.join(sub_dmg_parts) + _SEG
)
# EQ segment.
eq_parts = ["EQ", service_type_code]
if coverage_info:
eq_parts.append(coverage_info)
eq_segment = _ELEM.join(eq_parts) + _SEG
# Build the body in canonical order. The SE count is the number of
# segments inside the ST/SE envelope (ST itself + body segments).
body_parts: list[str] = [
_build_bht(),
src_segment,
rcv_segment,
sub_segment,
patient_segment, # may be empty when no patient
eq_segment,
]
body = "".join(body_parts)
body_segment_count = body.count(_SEG)
se_count = 1 + body_segment_count # +1 for the ST itself
parts: list[str] = [
_build_isa(sender_id, receiver_id, interchange_control_number),
_build_gs(sender_id, receiver_id, group_control_number),
_build_st(st_ctrl, impl_guide),
body,
_build_se(se_count, st_ctrl),
f"GE{_ELEM}1{_ELEM}{_pad(group_control_number, 9)}{_SEG}",
f"IEA{_ELEM}1{_ELEM}{_pad(interchange_control_number, 9)}{_SEG}",
]
return "".join(parts)
__all__ = ["serialize_270"]
backend/tests/fixtures/minimal_270.txt
Vendored
+16
View File
@@ -0,0 +1,16 @@
ISA*00* *00* *ZZ*SUBMITTERID *ZZ*RECEIVERID *240101*1200*^*00501*000000001*0*P*:~
GS*HC*SUBMITTERID*RECEIVERID*20240101*1200*1*X*005010X279A1~
ST*270*0001*005010X279A1~
BHT*0022*13*REF123*20240101*1200~
HL*1**20*1~
NM1*PR*2*PAYER NAME*****PI*SKCO0~
HL*2*1*21*1~
NM1*1P*2*PROVIDER NAME*****XX*1234567890~
HL*3*2*22*0~
TRN*2*TRACE001~
NM1*IL*1*DOE*JOHN*A***MI*MEMBER123~
DMG*D8*19800101*M~
EQ*1^30~
SE*10*0001~
GE*1*1~
IEA*1*000000001~
backend/tests/fixtures/minimal_271.txt
Vendored
+19
View File
@@ -0,0 +1,19 @@
ISA*00* *00* *ZZ*PAYERID *ZZ*PROVIDERID *240101*1200*^*00501*000000002*0*P*:~
GS*HC*PAYERID*PROVIDERID*20240101*1200*2*X*005010X279A1~
ST*271*0001*005010X279A1~
BHT*0022*11*REF456*20240101*1200~
HL*1**20*1~
NM1*PR*2*PAYER NAME*****PI*SKCO0~
HL*2*1*21*1~
NM1*1P*2*PROVIDER NAME*****XX*1234567890~
HL*3*2*22*0~
TRN*2*TRACE001~
NM1*IL*1*DOE*JOHN*A***MI*MEMBER123~
DMG*D8*19800101*M~
EB*1*IND*30**Y*1~
MSG*Patient has active coverage~
EB*88*IND**100*N**~
MSG*Pharmacy copay $20~
SE*13*0001~
GE*1*2~
IEA*1*000000002~
backend/tests/test_api_eligibility.py
+169
View File
@@ -0,0 +1,169 @@
"""Tests for the FastAPI surface in ``cyclone.api`` for the 270/271
eligibility endpoints (SP3 P4 T23 + T24).
"""
from __future__ import annotations
from datetime import date
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from cyclone.api import app
from cyclone.parsers.parse_270 import parse as parse_270_text
MINIMAL_271 = Path(__file__).parent / "fixtures" / "minimal_271.txt"
@pytest.fixture
def client() -> TestClient:
return TestClient(app)
def _valid_request_body() -> dict:
return {
"subscriber": {
"first_name": "JOHN",
"last_name": "DOE",
"member_id": "MEMBER123",
"dob": "1980-01-01",
},
"provider": {
"npi": "1234567890",
"name": "PROVIDER NAME",
},
"payer": {
"id": "SKCO0",
"name": "PAYER NAME",
},
"service_type_code": "1",
}
# --------------------------------------------------------------------------- #
# T23: POST /api/eligibility/request
# --------------------------------------------------------------------------- #
def test_eligibility_request_returns_raw_270_text(client: TestClient):
"""A minimal body returns the X12 270 text + the parsed structure."""
body = _valid_request_body()
resp = client.post("/api/eligibility/request", json=body)
assert resp.status_code == 200, resp.text
payload = resp.json()
assert "raw_270_text" in payload
assert "parsed" in payload
raw = payload["raw_270_text"]
assert isinstance(raw, str) and raw
# Canonical X12 envelope markers.
assert raw.startswith("ISA*")
assert raw.rstrip("\n").endswith("~IEA*1*000000001~")
# All the required segments are present.
assert "HL*1**20*1~" in raw
assert "NM1*PR*2*PAYER NAME" in raw
assert "HL*2*1*21*1~" in raw
assert "NM1*1P*2*PROVIDER NAME" in raw
assert "NM1*IL*1*DOE*JOHN" in raw
assert "EQ*1~" in raw
# The parsed payload mirrors the input.
parsed = payload["parsed"]
assert parsed["subscriber"]["first_name"] == "JOHN"
assert parsed["subscriber"]["last_name"] == "DOE"
assert parsed["subscriber"]["member_id"] == "MEMBER123"
assert parsed["inquiries"][0]["service_type_code"] == "1"
def test_eligibility_request_parsed_matches_serialized(client: TestClient):
"""The endpoint's ``parsed`` field round-trips through a server-side
re-parse of ``raw_270_text``.
"""
body = _valid_request_body()
resp = client.post("/api/eligibility/request", json=body)
assert resp.status_code == 200, resp.text
payload = resp.json()
raw = payload["raw_270_text"]
re_parsed = parse_270_text(raw, input_file="round_trip.txt")
# The re-parsed structure must equal the endpoint's parsed payload
# for the fields the operator can observe.
parsed = payload["parsed"]
assert re_parsed.subscriber.first_name == parsed["subscriber"]["first_name"]
assert re_parsed.subscriber.last_name == parsed["subscriber"]["last_name"]
assert re_parsed.subscriber.member_id == parsed["subscriber"]["member_id"]
assert re_parsed.information_source.name == parsed["information_source"]["name"]
assert re_parsed.information_receiver.npi == parsed["information_receiver"]["npi"]
assert re_parsed.inquiries[0].service_type_code == parsed["inquiries"][0]["service_type_code"]
def test_eligibility_request_missing_service_type_returns_400(client: TestClient):
"""Missing ``service_type_code`` returns 400, never 500."""
body = _valid_request_body()
del body["service_type_code"]
resp = client.post("/api/eligibility/request", json=body)
assert resp.status_code == 400, resp.text
detail = resp.json()["detail"]
assert "service_type_code" in str(detail)
def test_eligibility_request_missing_subscriber_member_id_returns_400(client: TestClient):
"""Missing ``subscriber.member_id`` returns 400."""
body = _valid_request_body()
del body["subscriber"]["member_id"]
resp = client.post("/api/eligibility/request", json=body)
assert resp.status_code == 400, resp.text
# --------------------------------------------------------------------------- #
# T24: POST /api/eligibility/parse-271
# --------------------------------------------------------------------------- #
def test_parse_271_endpoint_happy_path(client: TestClient):
"""Uploading the minimal 271 returns 200 + a non-empty
``coverage_benefits`` array.
"""
text = MINIMAL_271.read_text()
resp = client.post(
"/api/eligibility/parse-271",
files={"file": ("minimal_271.txt", text, "text/plain")},
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert "coverage_benefits" in body
assert "subscriber" in body
assert "summary" in body
assert isinstance(body["coverage_benefits"], list)
assert len(body["coverage_benefits"]) == 2
# The first EB resolves to "Medical Care".
cb0 = body["coverage_benefits"][0]
assert cb0["service_type_code"] == "1"
assert cb0["service_type_description"] == "Medical Care"
assert cb0["in_plan_network"] is True
# The second EB is pharmacy.
cb1 = body["coverage_benefits"][1]
assert cb1["service_type_code"] == "88"
assert cb1["service_type_description"] == "Pharmacy"
assert cb1["in_plan_network"] is False
# Subscriber extracted.
assert body["subscriber"]["first_name"] == "JOHN"
assert body["subscriber"]["last_name"] == "DOE"
assert body["subscriber"]["member_id"] == "MEMBER123"
def test_parse_271_endpoint_invalid_edi_raises_400(client: TestClient):
"""Garbage input must surface as 400, never 500."""
resp = client.post(
"/api/eligibility/parse-271",
files={"file": ("garbage.txt", "not edi", "text/plain")},
)
assert resp.status_code == 400, resp.text
body = resp.json()
assert "error" in body
def test_parse_271_endpoint_empty_file_returns_400(client: TestClient):
"""An empty upload returns 400 (matches the 999 / 835 contract)."""
resp = client.post(
"/api/eligibility/parse-271",
files={"file": ("empty.txt", "", "text/plain")},
)
assert resp.status_code == 400, resp.text
backend/tests/test_models_270.py
+133
View File
@@ -0,0 +1,133 @@
"""Tests for the 270 (Eligibility Benefit Inquiry) Pydantic models.
SP3 Phase 4 (T18): minimal typed shape needed by the parser (T20),
serializer (T22), and ``/api/eligibility/request`` endpoint (T23).
The spec defines richer models (gender, address, etc.), but the v1
round-trip only needs the minimum surface — see the inline rationale
in ``parsers/models_270.py``.
"""
from __future__ import annotations
import json
from datetime import date
import pytest
from pydantic import ValidationError
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_270 import (
EligibilityBenefitInquiry,
InformationReceiver270,
InformationSource270,
ParseResult270,
Patient270,
Subscriber270,
)
def _build_result() -> ParseResult270:
return ParseResult270(
envelope=Envelope(
sender_id="SUBMITTERID",
receiver_id="RECEIVERID",
control_number="000000001",
transaction_date=date(2024, 1, 1),
implementation_guide="005010X279A1",
),
information_source=InformationSource270(
name="PAYER NAME",
id="SKCO0",
),
information_receiver=InformationReceiver270(
name="PROVIDER NAME",
npi="1234567890",
),
subscriber=Subscriber270(
member_id="MEMBER123",
first_name="JOHN",
last_name="DOE",
dob=date(1980, 1, 1),
),
inquiries=[
EligibilityBenefitInquiry(service_type_code="1"),
EligibilityBenefitInquiry(service_type_code="30"),
],
summary=BatchSummary(
input_file="minimal_270.txt",
control_number="000000001",
transaction_date=date(2024, 1, 1),
total_claims=1,
passed=1,
failed=0,
),
)
def test_parse_result_270_minimal_round_trip():
"""Build a minimal ParseResult270 and round-trip via JSON."""
r = _build_result()
blob = json.loads(r.model_dump_json())
assert blob["envelope"]["control_number"] == "000000001"
assert blob["envelope"]["implementation_guide"] == "005010X279A1"
assert blob["information_source"]["name"] == "PAYER NAME"
assert blob["information_source"]["id"] == "SKCO0"
assert blob["information_receiver"]["npi"] == "1234567890"
assert blob["subscriber"]["first_name"] == "JOHN"
assert blob["subscriber"]["last_name"] == "DOE"
assert blob["subscriber"]["member_id"] == "MEMBER123"
assert blob["subscriber"]["dob"] == "1980-01-01"
assert blob["patient"] is None
assert len(blob["inquiries"]) == 2
assert blob["inquiries"][0]["service_type_code"] == "1"
assert blob["inquiries"][1]["service_type_code"] == "30"
# Round-trip back to the model.
r2 = ParseResult270.model_validate(blob)
assert r2.information_source.name == "PAYER NAME"
assert r2.subscriber.member_id == "MEMBER123"
assert r2.inquiries[0].service_type_code == "1"
assert r2.summary.total_claims == 1
def test_eligibility_benefit_inquiry_required_fields():
"""``service_type_code`` is the only required field on the EQ model."""
# Without it: validation error.
with pytest.raises(ValidationError) as exc:
EligibilityBenefitInquiry() # type: ignore[call-arg]
assert "service_type_code" in str(exc.value)
# With it: parses cleanly. ``coverage_info`` is optional.
ebi = EligibilityBenefitInquiry(service_type_code="1")
assert ebi.coverage_info is None
# coverage_info supplied → round-trips.
ebi2 = EligibilityBenefitInquiry(service_type_code="30", coverage_info="C")
blob = json.loads(ebi2.model_dump_json())
assert blob["coverage_info"] == "C"
assert EligibilityBenefitInquiry.model_validate(blob).coverage_info == "C"
def test_patient_270_optional_fields():
"""Patient270's fields are all optional; relationship carries HL/INS qualifiers."""
p = Patient270()
blob = json.loads(p.model_dump_json())
assert blob["first_name"] is None
assert blob["relationship"] is None
p2 = Patient270(first_name="JANE", last_name="DOE", relationship="19") # 19 = child
assert p2.relationship == "19"
def test_parse_result_270_inquiries_default_empty():
"""Inquiries default to an empty list when omitted (Pydantic default_factory)."""
r = ParseResult270(
envelope=Envelope(
sender_id="S", receiver_id="R", control_number="000000001",
transaction_date=date(2024, 1, 1),
),
information_source=InformationSource270(name="P"),
information_receiver=InformationReceiver270(name="P"),
subscriber=Subscriber270(member_id="M"),
summary=BatchSummary(input_file="", total_claims=0, passed=0, failed=0),
)
assert r.inquiries == []
assert r.patient is None
backend/tests/test_models_271.py
+234
View File
@@ -0,0 +1,234 @@
"""Tests for the 271 (Eligibility Benefit Response) Pydantic models.
SP3 Phase 4 (T19): minimal typed shape needed by the parser (T21) and
``/api/eligibility/parse-271`` endpoint (T24). Plus the inline
``SERVICE_TYPE_CODES`` dict + ``service_type_description`` helper.
"""
from __future__ import annotations
import json
import re
from datetime import date
from decimal import Decimal
import pytest
from pydantic import ValidationError
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_271 import (
LAST_UPDATED,
SERVICE_TYPE_CODES,
BenefitAdditionalInfo,
CoverageBenefit,
InformationReceiver271,
InformationSource271,
ParseResult271,
Patient271,
Subscriber271,
service_type_description,
)
def _build_result() -> ParseResult271:
return ParseResult271(
envelope=Envelope(
sender_id="PAYERID",
receiver_id="PROVIDERID",
control_number="000000002",
transaction_date=date(2024, 1, 1),
implementation_guide="005010X279A1",
),
information_source=InformationSource271(
name="PAYER NAME",
id="SKCO0",
),
information_receiver=InformationReceiver271(
name="PROVIDER NAME",
),
subscriber=Subscriber271(
member_id="MEMBER123",
first_name="JOHN",
last_name="DOE",
),
coverage_benefits=[
CoverageBenefit(
service_type_code="1",
coverage_level="IND",
in_plan_network=True,
authorization_required=False,
plan_coverage_description="Patient has active coverage",
),
CoverageBenefit(
service_type_code="88",
coverage_level="IND",
benefit_amount=Decimal("100"),
in_plan_network=False,
authorization_required=True,
),
],
summary=BatchSummary(
input_file="minimal_271.txt",
control_number="000000002",
transaction_date=date(2024, 1, 1),
total_claims=2,
passed=2,
failed=0,
),
)
def test_parse_result_271_minimal_round_trip():
"""Build a minimal ParseResult271 and round-trip via JSON."""
r = _build_result()
blob = json.loads(r.model_dump_json())
assert blob["envelope"]["control_number"] == "000000002"
assert blob["envelope"]["implementation_guide"] == "005010X279A1"
assert blob["information_source"]["name"] == "PAYER NAME"
assert blob["subscriber"]["first_name"] == "JOHN"
assert blob["patient"] is None
assert len(blob["coverage_benefits"]) == 2
# Decimal serializes as a string in Pydantic v2 by default; verify
# the round-trip preserves the value (not the str type).
assert blob["coverage_benefits"][1]["benefit_amount"] == "100"
# Round-trip back to the model — Decimal must survive.
r2 = ParseResult271.model_validate(blob)
assert r2.coverage_benefits[1].benefit_amount == Decimal("100")
assert r2.coverage_benefits[0].in_plan_network is True
assert r2.coverage_benefits[1].authorization_required is True
def test_service_type_description_known():
"""``service_type_description("1")`` returns ``"Medical Care"``."""
assert service_type_description("1") == "Medical Care"
# Numeric codes with leading zero should not exist (X12 omits the
# leading zero) but the lookup still works for whatever the caller
# hands in.
assert service_type_description("88") == "Pharmacy"
assert service_type_description("AL") == "Vision (Optometry)"
assert service_type_description("MH") == "Mental Health"
assert service_type_description("UC") == "Urgent Care"
# Case-insensitive: the parser may pass "1" or any case variant.
assert service_type_description("mh") == "Mental Health"
assert service_type_description("uc") == "Urgent Care"
def test_service_type_description_unknown_returns_none():
"""Unknown codes (and the empty string) return ``None``."""
assert service_type_description("ZZZ") is None
assert service_type_description("9999") is None
assert service_type_description("") is None
# Whitespace-only also returns None.
assert service_type_description(" ") is None
def test_service_type_codes_count_ge_50():
"""The dict ships at least 50 service-type codes."""
assert len(SERVICE_TYPE_CODES) >= 50
# Every value must be a non-empty string.
for code, label in SERVICE_TYPE_CODES.items():
assert isinstance(code, str)
assert code, "empty key"
assert isinstance(label, str)
assert label, f"empty label for {code!r}"
def test_last_updated_constant():
"""LAST_UPDATED is a YYYY-MM-DD string used to track snapshot refresh."""
assert isinstance(LAST_UPDATED, str)
assert re.match(r"^\d{4}-\d{2}-\d{2}$", LAST_UPDATED), (
f"LAST_UPDATED must be YYYY-MM-DD, got {LAST_UPDATED!r}"
)
def test_coverage_benefit_resolves_description_on_construction():
"""``CoverageBenefit`` auto-populates ``service_type_description`` from
``service_type_code`` when the caller omits it.
"""
cb = CoverageBenefit(service_type_code="1")
assert cb.service_type_description == "Medical Care"
# Caller-supplied description wins.
cb2 = CoverageBenefit(service_type_code="1", service_type_description="Custom")
assert cb2.service_type_description == "Custom"
# Unknown code → description is None.
cb3 = CoverageBenefit(service_type_code="ZZZ")
assert cb3.service_type_description is None
def test_coverage_benefit_round_trip_preserves_description():
"""Round-tripping JSON does not overwrite a caller-supplied description."""
r = _build_result()
blob = json.loads(r.model_dump_json())
# After JSON round-trip, the description we built (e.g. "Medical Care") survives.
r2 = ParseResult271.model_validate(blob)
assert r2.coverage_benefits[0].service_type_description == "Medical Care"
assert r2.coverage_benefits[1].service_type_description == "Pharmacy"
def test_required_service_type_code_on_coverage_benefit():
"""``service_type_code`` is the only required field on CoverageBenefit."""
with pytest.raises(ValidationError) as exc:
CoverageBenefit() # type: ignore[call-arg]
assert "service_type_code" in str(exc.value)
# All other fields are optional / defaulted.
cb = CoverageBenefit(service_type_code="1")
assert cb.coverage_level is None
assert cb.benefit_amount is None
assert cb.in_plan_network is None
assert cb.authorization_required is None
assert cb.plan_coverage_description is None
assert cb.time_qualifier is None
assert cb.benefit_date is None
assert cb.additional_info == []
def test_benefit_additional_info_codes_default():
"""``BenefitAdditionalInfo.codes`` defaults to an empty list."""
bai = BenefitAdditionalInfo(description="Patient has active coverage")
assert bai.codes == []
bai2 = BenefitAdditionalInfo(
description="Pharmacy copay $20",
codes=["H0019", "H0020"],
)
assert bai2.codes == ["H0019", "H0020"]
def test_patient_271_optional_fields():
"""Patient271's fields are all optional; relationship is the only one with a
v1 use (HL/INS qualifiers).
"""
p = Patient271()
blob = json.loads(p.model_dump_json())
assert blob == {
"first_name": None,
"last_name": None,
"relationship": None,
}
p2 = Patient271(first_name="JANE", last_name="DOE", relationship="19")
blob2 = json.loads(p2.model_dump_json())
assert blob2["relationship"] == "19"
def test_service_type_codes_contains_required_keys():
"""Spot-check the codes the spec calls out as required."""
required = {
"1": "Medical Care",
"2": "Surgical",
"33": "Chiropractic",
"35": "Dental Care",
"47": "Hospital",
"48": "Hospital - Inpatient",
"50": "Hospital - Outpatient",
"86": "Emergency Services",
"88": "Pharmacy",
"98": "Professional (Physician) Visit - Outpatient",
"99": "Professional (Physician) Visit - Other",
"AL": "Vision (Optometry)",
"MH": "Mental Health",
"UC": "Urgent Care",
}
for code, expected in required.items():
assert SERVICE_TYPE_CODES.get(code) == expected, (
f"missing or mismatched code {code!r}: got {SERVICE_TYPE_CODES.get(code)!r}, expected {expected!r}"
)
backend/tests/test_parse_270.py
+106
View File
@@ -0,0 +1,106 @@
"""Tests for the 270 (Eligibility Benefit Inquiry) parser.
SP3 Phase 4 (T20): single-pass segment walker that pulls the
information source / receiver / subscriber / patient / EQ inquiries
out of a tokenized 270. The fixture (`tests/fixtures/minimal_270.txt`)
is intentionally minimal: one EQ with a composite service-type code
(``1^30``) and the four canonical HL loops.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from cyclone.parsers.parse_270 import parse
FIXTURE = Path(__file__).parent / "fixtures" / "minimal_270.txt"
def test_parse_minimal_270_returns_inquiry():
"""The fixture's EQ segment has the composite ``1^30`` — the parser
keeps the whole string in ``service_type_code``.
v1 convention: callers that need the individual codes should split
on ``^``. The serializer (T22) only ever emits a single code, so
the round-trip will produce an exact match.
"""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_270.txt")
assert len(result.inquiries) == 1
code = result.inquiries[0].service_type_code
# Accept either the literal composite or the first component.
assert code in {"1^30", "1"}
if code == "1^30":
first = code.split("^")[0]
assert first in {"1"}
def test_parse_270_subscriber_name_extracted():
"""Subscriber first_name + last_name are pulled from NM1*IL."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_270.txt")
assert result.subscriber.first_name == "JOHN"
assert result.subscriber.last_name == "DOE"
assert result.subscriber.member_id == "MEMBER123"
# DMG segment — D8 = CCYYMMDD, M = male.
assert result.subscriber.dob is not None
assert result.subscriber.dob.year == 1980
assert result.subscriber.dob.month == 1
assert result.subscriber.dob.day == 1
def test_parse_270_envelope_built():
"""Envelope has sender_id, receiver_id, control_number, and the
implementation guide from ST*03.
"""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_270.txt")
env = result.envelope
assert env.sender_id.strip() == "SUBMITTERID"
assert env.receiver_id.strip() == "RECEIVERID"
assert env.control_number == "000000001"
assert env.implementation_guide == "005010X279A1"
# GS04 supplies the transaction date.
assert env.transaction_date.year == 2024
assert env.transaction_date.month == 1
assert env.transaction_date.day == 1
def test_parse_270_information_source_extracted():
"""Information source (HL*20) pulls the payer name + id from NM1*PR."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_270.txt")
assert result.information_source.name == "PAYER NAME"
assert result.information_source.id == "SKCO0"
# Information receiver (HL*21) pulls the provider + NPI.
assert result.information_receiver.name == "PROVIDER NAME"
assert result.information_receiver.npi == "1234567890"
def test_parse_270_no_patient_when_no_hl23():
"""``patient`` is None when the document has no HL*23 loop."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_270.txt")
assert result.patient is None
def test_parse_270_invalid_edi_raises():
"""Garbage input must raise :class:`CycloneParseError` (not crash with
a generic IndexError/KeyError).
"""
from cyclone.parsers.exceptions import CycloneParseError
with pytest.raises(CycloneParseError):
parse("not edi at all", input_file="garbage.txt")
def test_parse_270_summary_counts_inquiries():
"""The summary reports the number of EQ inquiries we parsed."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_270.txt")
assert result.summary.total_claims == 1
assert result.summary.passed == 1
assert result.summary.failed == 0
assert result.summary.input_file == "minimal_270.txt"
assert result.summary.control_number == "000000001"
backend/tests/test_parse_271.py
+98
View File
@@ -0,0 +1,98 @@
"""Tests for the 271 (Eligibility Benefit Response) parser.
SP3 Phase 4 (T21): single-pass segment walker that pulls the
information source / receiver / subscriber / patient / EB coverage
benefits out of a tokenized 271. The fixture
(``tests/fixtures/minimal_271.txt``) has two EB segments with
descriptions — one Medical Care, one Pharmacy — so we can assert
both the count and the per-benefit resolution of service-type codes.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from cyclone.parsers.parse_271 import parse
FIXTURE = Path(__file__).parent / "fixtures" / "minimal_271.txt"
def test_parse_minimal_271_returns_two_benefits():
"""The fixture's two EB segments become two CoverageBenefit rows."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
assert len(result.coverage_benefits) == 2
# First EB: medical care (code "1"). Second: pharmacy (code "88").
assert result.coverage_benefits[0].service_type_code == "1"
assert result.coverage_benefits[1].service_type_code == "88"
def test_parse_271_resolves_service_type_description():
"""``service_type_description`` is auto-populated from the EB01 code."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
assert result.coverage_benefits[0].service_type_description == "Medical Care"
assert result.coverage_benefits[1].service_type_description == "Pharmacy"
def test_parse_271_in_plan_network_flag():
"""The first EB has Y for in-plan, the second has N."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
assert result.coverage_benefits[0].in_plan_network is True
assert result.coverage_benefits[1].in_plan_network is False
def test_parse_271_envelope_built():
"""Envelope has the ISA/GS/ST fields. ST03 = implementation guide."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
env = result.envelope
assert env.sender_id.strip() == "PAYERID"
assert env.receiver_id.strip() == "PROVIDERID"
assert env.control_number == "000000002"
assert env.implementation_guide == "005010X279A1"
# GS04 supplies the transaction date.
assert env.transaction_date.year == 2024
assert env.transaction_date.month == 1
assert env.transaction_date.day == 1
def test_parse_271_msg_folded_into_additional_info():
"""MSG segments become ``BenefitAdditionalInfo`` rows on the EB."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
cb0 = result.coverage_benefits[0]
assert len(cb0.additional_info) == 1
assert cb0.additional_info[0].description == "Patient has active coverage"
cb1 = result.coverage_benefits[1]
assert len(cb1.additional_info) == 1
assert cb1.additional_info[0].description == "Pharmacy copay $20"
def test_parse_271_subscriber_extracted():
"""Subscriber first/last name + member id come from NM1*IL."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
assert result.subscriber.first_name == "JOHN"
assert result.subscriber.last_name == "DOE"
assert result.subscriber.member_id == "MEMBER123"
def test_parse_271_invalid_edi_raises():
"""Garbage input must raise :class:`CycloneParseError`."""
from cyclone.parsers.exceptions import CycloneParseError
with pytest.raises(CycloneParseError):
parse("not edi at all", input_file="garbage.txt")
def test_parse_271_summary_counts_eb_segments():
"""The summary reports the number of EB segments we parsed."""
text = FIXTURE.read_text()
result = parse(text, input_file="minimal_271.txt")
assert result.summary.total_claims == 2
assert result.summary.passed == 2
assert result.summary.failed == 0
assert result.summary.input_file == "minimal_271.txt"
backend/tests/test_serialize_270.py
+209
View File
@@ -0,0 +1,209 @@
"""Tests for the 270 (Eligibility Benefit Inquiry) serializer.
SP3 Phase 4 (T22 + T25): build a ``ParseResult270``, serialize to
X12, then re-parse and assert the round-trip is stable. The cross-
parser round-trip test (T25) lives at the bottom of this file.
"""
from __future__ import annotations
from datetime import date
import pytest
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_270 import (
EligibilityBenefitInquiry,
InformationReceiver270,
InformationSource270,
ParseResult270,
Subscriber270,
)
from cyclone.parsers.parse_270 import parse
from cyclone.parsers.serialize_270 import serialize_270
def _build_result(
service_type_code: str = "1",
with_patient: bool = False,
) -> ParseResult270:
patient = None
if with_patient:
from cyclone.parsers.models_270 import Patient270
patient = Patient270(
first_name="JANE",
last_name="DOE",
dob=date(2010, 5, 5),
relationship="19",
)
return ParseResult270(
envelope=Envelope(
sender_id="SUBMITTERID",
receiver_id="RECEIVERID",
control_number="000000001",
transaction_date=date(2024, 1, 1),
implementation_guide="005010X279A1",
),
information_source=InformationSource270(name="PAYER NAME", id="SKCO0"),
information_receiver=InformationReceiver270(name="PROVIDER NAME", npi="1234567890"),
subscriber=Subscriber270(
member_id="MEMBER123",
first_name="JOHN",
last_name="DOE",
dob=date(1980, 1, 1),
),
patient=patient,
inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
summary=BatchSummary(
input_file="round_trip.txt",
total_claims=1,
passed=1,
failed=0,
),
)
def test_serialize_270_envelope_segments():
"""Output starts with ``ISA*`` and ends with ``IEA*``; canonical envelope
layers are all present in order.
"""
text = serialize_270(_build_result())
assert text.startswith("ISA*")
# Strip trailing newline if any, then check the IEA tail.
body = text.rstrip("\n")
assert body.endswith("~IEA*1*000000001~")
# Sender/receiver ids in the GS are padded to 15 chars (X12 fixed width).
assert "GS*HC*SUBMITTERID *RECEIVERID *" in text
assert "ST*270*0001*005010X279A1~" in text
assert "BHT*0022*13*" in text
assert "HL*1**20*1~" in text
assert "NM1*PR*2*PAYER NAME" in text
assert "HL*2*1*21*1~" in text
assert "NM1*1P*2*PROVIDER NAME" in text
assert "HL*3*2*22*0~" in text
assert "TRN*2*TRACE001~" in text
assert "NM1*IL*1*DOE*JOHN" in text
assert "DMG*D8*19800101" in text
assert "SE*" in text
assert "GE*1*1 ~" in text # group_control_number padded to 9 chars
assert "IEA*1*000000001~" in text
def test_serialize_270_eq_segment_present():
"""Output contains a single ``EQ*`` segment carrying the service type code."""
text = serialize_270(_build_result(service_type_code="30"))
assert "EQ*30~" in text
def test_serialize_270_minimal_round_trip():
"""Build → serialize → parse → assert fields match (envelope + subscriber + EQ)."""
original = _build_result(service_type_code="1")
text = serialize_270(original)
re_parsed = parse(text, input_file="round_trip.txt")
# Envelope
assert re_parsed.envelope.sender_id.strip() == original.envelope.sender_id
assert re_parsed.envelope.receiver_id.strip() == original.envelope.receiver_id
assert re_parsed.envelope.control_number == "000000001"
assert re_parsed.envelope.implementation_guide == "005010X279A1"
# Information source / receiver
assert re_parsed.information_source.name == "PAYER NAME"
assert re_parsed.information_source.id == "SKCO0"
assert re_parsed.information_receiver.name == "PROVIDER NAME"
assert re_parsed.information_receiver.npi == "1234567890"
# Subscriber
assert re_parsed.subscriber.first_name == "JOHN"
assert re_parsed.subscriber.last_name == "DOE"
assert re_parsed.subscriber.member_id == "MEMBER123"
assert re_parsed.subscriber.dob == date(1980, 1, 1)
# Inquiry
assert len(re_parsed.inquiries) == 1
assert re_parsed.inquiries[0].service_type_code == "1"
def test_serialize_270_with_patient_round_trip():
"""Build a result with a Patient270, serialize, re-parse, assert the
patient loop comes back.
"""
original = _build_result(with_patient=True)
text = serialize_270(original)
re_parsed = parse(text, input_file="round_trip_with_patient.txt")
assert re_parsed.patient is not None
assert re_parsed.patient.first_name == "JANE"
assert re_parsed.patient.last_name == "DOE"
assert re_parsed.patient.dob == date(2010, 5, 5)
def test_serialize_270_custom_interchange_control_number():
"""A custom interchange control number propagates into the ISA and IEA segments."""
text = serialize_270(_build_result(), interchange_control_number="000000042")
assert "ISA*00* *00* *ZZ*SUBMITTERID *ZZ*RECEIVERID" in text
# IEA*1*<ICN>~ at the end.
assert text.rstrip("\n").endswith("~IEA*1*000000042~")
# ISA also carries it.
assert "*000000042*" in text.split("~", 1)[0]
# --------------------------------------------------------------------------- #
# T25 — Cross-parser round-trip (full)
# --------------------------------------------------------------------------- #
def test_cross_parser_270_round_trip():
"""Build a fully populated ParseResult270, serialize, parse, assert each
field matches the original (except the raw envelope control_number
which the serializer regenerates from defaults).
"""
original = _build_result(
service_type_code="88", with_patient=True,
)
text = serialize_270(original)
re_parsed = parse(text, input_file="cross.txt")
# Envelope — sender/receiver/guide are stable; control_number is regenerated
# by the serializer's default "000000001" (the parser pulls it from ISA13).
assert re_parsed.envelope.control_number == "000000001"
assert re_parsed.envelope.sender_id.strip() == "SUBMITTERID"
assert re_parsed.envelope.receiver_id.strip() == "RECEIVERID"
assert re_parsed.envelope.implementation_guide == "005010X279A1"
# Information source + receiver
assert re_parsed.information_source.name == original.information_source.name
assert re_parsed.information_source.id == original.information_source.id
assert re_parsed.information_receiver.name == original.information_receiver.name
assert re_parsed.information_receiver.npi == original.information_receiver.npi
# Subscriber
assert re_parsed.subscriber.member_id == original.subscriber.member_id
assert re_parsed.subscriber.first_name == original.subscriber.first_name
assert re_parsed.subscriber.last_name == original.subscriber.last_name
assert re_parsed.subscriber.dob == original.subscriber.dob
# Patient (set on the original).
assert re_parsed.patient is not None
assert re_parsed.patient.first_name == original.patient.first_name
assert re_parsed.patient.last_name == original.patient.last_name
assert re_parsed.patient.dob == original.patient.dob
# Inquiries
assert len(re_parsed.inquiries) == len(original.inquiries)
for got, exp in zip(re_parsed.inquiries, original.inquiries):
assert got.service_type_code == exp.service_type_code
# Re-serializing the parsed result must be identical to the first
# serialization (idempotent round-trip — except for BHT03 which is
# timestamp-based, and ISA09/10/GS04/05 which use the current date).
text2 = serialize_270(re_parsed)
# Normalize both texts to drop the date/time-driven segments before
# comparing for "shape" stability: we keep envelope, HL/NM1, EQ.
def _shape(t: str) -> list[str]:
return [
s for s in t.split("~")
if s and not s.startswith("BHT") and not s.startswith("ISA")
and not s.startswith("GS") and not s.startswith("GE")
and not s.startswith("IEA")
]
assert _shape(text) == _shape(text2)