Merge feat/clm05-co-validation: CLM05-1 POS + CLM05-2 qualifier validation
This commit is contained in:
@@ -59,6 +59,7 @@ class ClaimHeader(_Base):
|
||||
claim_id: str
|
||||
total_charge: Decimal
|
||||
place_of_service: str | None = None
|
||||
facility_code_qualifier: str | None = None
|
||||
frequency_code: str | None = None
|
||||
provider_signature: str | None = None
|
||||
assignment: str | None = None
|
||||
|
||||
@@ -181,17 +181,21 @@ def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, in
|
||||
claim_id = clm[1] if len(clm) > 1 else ""
|
||||
total_charge = Decimal(clm[2]) if len(clm) > 2 and clm[2] else Decimal("0")
|
||||
pos = ""
|
||||
qualifier: str | None = None
|
||||
freq = ""
|
||||
if len(clm) > 5:
|
||||
clm05 = clm[5].split(":")
|
||||
if len(clm05) > 0:
|
||||
pos = clm05[0]
|
||||
if len(clm05) > 1:
|
||||
qualifier = clm05[1] or None
|
||||
if len(clm05) > 2:
|
||||
freq = clm05[2]
|
||||
claim_header = ClaimHeader(
|
||||
claim_id=claim_id,
|
||||
total_charge=total_charge,
|
||||
place_of_service=pos or None,
|
||||
facility_code_qualifier=qualifier,
|
||||
frequency_code=freq or None,
|
||||
provider_signature=clm[6] if len(clm) > 6 else None,
|
||||
assignment=clm[7] if len(clm) > 7 else None,
|
||||
|
||||
@@ -11,6 +11,21 @@ from __future__ import annotations
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
# Full CMS Place of Service code list (X12 837P CLM05-1).
|
||||
# Two-digit zero-padded codes covering the entire CMS POS set as of 2026.
|
||||
CMS_PLACE_OF_SERVICE_CODES: set[str] = {
|
||||
"01", "02", "03", "04", "05", "06", "07", "08", "09",
|
||||
"10", "11", "12", "13", "14", "15", "16", "17", "18", "19",
|
||||
"20", "21", "22", "23", "24", "25", "26", "27", "28", "29",
|
||||
"30", "31", "32", "33", "34", "41", "42", "49", "50", "51",
|
||||
"52", "53", "54", "55", "56", "57", "58", "59", "60", "61",
|
||||
"62", "63", "64", "65", "66", "67", "68", "69", "70", "71",
|
||||
"72", "73", "74", "75", "76", "77", "78", "79", "80", "81",
|
||||
"82", "83", "84", "85", "86", "87", "88", "89", "90", "91",
|
||||
"92", "93", "94", "95", "96", "97", "98", "99",
|
||||
}
|
||||
|
||||
|
||||
class PayerConfig(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
@@ -23,6 +38,12 @@ class PayerConfig(BaseModel):
|
||||
payer_name: str = ""
|
||||
no_patient_loop: bool = False
|
||||
encounter_claim_in_same_batch: bool = False
|
||||
# CLM05-2 Facility Code Qualifier. "B" = CMS Place of Service code.
|
||||
allowed_facility_qualifiers: set[str] = Field(default_factory=lambda: {"B"})
|
||||
# CLM05-1 Facility Type Code / Place of Service. Defaults to the full CMS POS list.
|
||||
allowed_place_of_service_codes: set[str] = Field(
|
||||
default_factory=lambda: set(CMS_PLACE_OF_SERVICE_CODES)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def co_medicaid(cls) -> "PayerConfig":
|
||||
@@ -41,6 +62,8 @@ class PayerConfig(BaseModel):
|
||||
payer_name="COHCPF",
|
||||
no_patient_loop=True,
|
||||
encounter_claim_in_same_batch=False,
|
||||
allowed_facility_qualifiers={"B"},
|
||||
allowed_place_of_service_codes=set(CMS_PLACE_OF_SERVICE_CODES),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
@@ -56,4 +79,6 @@ class PayerConfig(BaseModel):
|
||||
payer_name="",
|
||||
no_patient_loop=False,
|
||||
encounter_claim_in_same_batch=True,
|
||||
allowed_facility_qualifiers={"B"},
|
||||
allowed_place_of_service_codes=set(CMS_PLACE_OF_SERVICE_CODES),
|
||||
)
|
||||
|
||||
@@ -57,6 +57,37 @@ def _r031_ref_g1_optional(claim: ClaimOutput, _: PayerConfig) -> Iterable[Valida
|
||||
return ()
|
||||
|
||||
|
||||
def _r032_clm05_2_facility_qualifier(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]:
|
||||
"""CLM05-2 Facility Code Qualifier must be recognized for the payer.
|
||||
|
||||
When the field is absent (e.g. production data with only a 2-component
|
||||
CLM05), the rule is skipped silently — the absence of a qualifier is
|
||||
tolerated, but a present-and-unknown qualifier is a hard error.
|
||||
"""
|
||||
q = claim.claim.facility_code_qualifier
|
||||
if not q:
|
||||
return
|
||||
if q not in cfg.allowed_facility_qualifiers:
|
||||
yield ValidationIssue(
|
||||
rule="R032_clm05_2_facility_qualifier",
|
||||
severity="error",
|
||||
message=f"CLM05-2 facility code qualifier {q!r} not in {sorted(cfg.allowed_facility_qualifiers)} for payer {cfg.name}",
|
||||
)
|
||||
|
||||
|
||||
def _r033_clm05_1_place_of_service_code(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]:
|
||||
"""CLM05-1 Facility Type Code must be a valid CMS Place of Service code."""
|
||||
pos = claim.claim.place_of_service
|
||||
if not pos:
|
||||
return
|
||||
if pos not in cfg.allowed_place_of_service_codes:
|
||||
yield ValidationIssue(
|
||||
rule="R033_clm05_1_place_of_service_code",
|
||||
severity="error",
|
||||
message=f"CLM05-1 place of service {pos!r} not in CMS POS code list for payer {cfg.name}",
|
||||
)
|
||||
|
||||
|
||||
def _r050_diagnosis_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]:
|
||||
if not claim.diagnoses:
|
||||
yield ValidationIssue(rule="R050_diagnosis_present", severity="error", message="HI segment missing — no diagnoses on claim")
|
||||
@@ -95,6 +126,8 @@ _RULES: list[Rule] = [
|
||||
_r020_npi_format,
|
||||
_r030_frequency_allowed,
|
||||
_r031_ref_g1_optional,
|
||||
_r032_clm05_2_facility_qualifier,
|
||||
_r033_clm05_1_place_of_service_code,
|
||||
_r050_diagnosis_present,
|
||||
_r060_service_dates_present,
|
||||
_r070_charges_sum,
|
||||
|
||||
@@ -150,3 +150,46 @@ def test_r100_payer_id_warning_only():
|
||||
report = validate(claim, cfg)
|
||||
assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings)
|
||||
assert report.passed is True
|
||||
|
||||
|
||||
def test_r032_passes_when_qualifier_b():
|
||||
cfg = PayerConfig.co_medicaid()
|
||||
claim = _build_claim()
|
||||
claim.claim.facility_code_qualifier = "B"
|
||||
report = validate(claim, cfg)
|
||||
assert not any(i.rule == "R032_clm05_2_facility_qualifier" for i in report.errors + report.warnings)
|
||||
|
||||
|
||||
def test_r032_errors_when_qualifier_unknown():
|
||||
cfg = PayerConfig.co_medicaid()
|
||||
claim = _build_claim()
|
||||
claim.claim.facility_code_qualifier = "X"
|
||||
report = validate(claim, cfg)
|
||||
assert any(i.rule == "R032_clm05_2_facility_qualifier" and i.severity == "error" for i in report.errors)
|
||||
assert report.passed is False
|
||||
|
||||
|
||||
def test_r032_skipped_when_qualifier_none():
|
||||
cfg = PayerConfig.co_medicaid()
|
||||
claim = _build_claim()
|
||||
# Default _build_claim leaves facility_code_qualifier unset (None).
|
||||
assert claim.claim.facility_code_qualifier is None
|
||||
report = validate(claim, cfg)
|
||||
assert not any(i.rule == "R032_clm05_2_facility_qualifier" for i in report.errors + report.warnings)
|
||||
|
||||
|
||||
def test_r033_passes_with_valid_pos_code():
|
||||
cfg = PayerConfig.co_medicaid()
|
||||
claim = _build_claim()
|
||||
claim.claim.place_of_service = "12" # Office — valid CMS POS
|
||||
report = validate(claim, cfg)
|
||||
assert not any(i.rule == "R033_clm05_1_place_of_service_code" for i in report.errors + report.warnings)
|
||||
|
||||
|
||||
def test_r033_errors_with_invalid_pos_code():
|
||||
cfg = PayerConfig.co_medicaid()
|
||||
claim = _build_claim()
|
||||
claim.claim.place_of_service = "999" # not a valid CMS POS code
|
||||
report = validate(claim, cfg)
|
||||
assert any(i.rule == "R033_clm05_1_place_of_service_code" and i.severity == "error" for i in report.errors)
|
||||
assert report.passed is False
|
||||
|
||||
Reference in New Issue
Block a user