Spaces:
Running
Running
File size: 3,572 Bytes
5cceba0 | 1 2 3 4 5 6 7 8 9 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 35 36 37 38 39 40 41 42 43 44 45 46 47 48 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 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """Parser for the OHIP Physician Fee Schedule Master (fixed-width, 75 chars).
Official record layout (OHIP Fee Schedule Master Record Layout, 1-based):
Fee Schedule code start 1 len 4 (ANNN)
Effective Date start 5 len 8 (YYYYMMDD)
Termination Date start 13 len 8 (99999999 = indefinite)
Provider Fee start 21 len 11 (N, 4 decimal places)
Assistant's fee start 32 len 11 (N)
Specialist fee start 43 len 11 (N, 4 decimal places)
Anaesthetist's fee start 54 len 11 (N)
Non-Anaesthetist's fee start 65 len 11 (N)
"""
from __future__ import annotations
import datetime as dt
import logging
from dataclasses import dataclass, field
from pathlib import Path
from .config import settings
logger = logging.getLogger(__name__)
_INDEFINITE = "99999999"
@dataclass
class FeeRecord:
billing_code: str
effective_date: str # YYYY-MM-DD
termination_date: str | None # YYYY-MM-DD or None (indefinite)
provider_fee: float
assistant_fee: float
specialist_fee: float
anaesthetist_fee: float
non_anaesthetist_fee: float
all_fees: dict = field(default_factory=dict)
@property
def primary_fee(self) -> float:
"""Best single fee for display: provider, else specialist, else max."""
for f in (self.provider_fee, self.specialist_fee):
if f > 0:
return f
return max(self.all_fees.values(), default=0.0)
def _fee(segment: str) -> float:
seg = segment.strip() or "0"
try:
return int(seg) / 10000.0 # 4 implied decimal places
except ValueError:
return 0.0
def _date(segment: str) -> str | None:
seg = segment.strip()
if not seg or seg == _INDEFINITE:
return None
try:
return dt.datetime.strptime(seg, "%Y%m%d").strftime("%Y-%m-%d")
except ValueError:
return None
def parse_line(line: str) -> FeeRecord | None:
if len(line) < 75:
line = line.ljust(75)
code = line[0:4].strip()
if not code:
return None
eff = _date(line[4:12]) or "1990-01-01"
term = _date(line[12:20])
provider = _fee(line[20:31])
assistant = _fee(line[31:42])
specialist = _fee(line[42:53])
anaesthetist = _fee(line[53:64])
non_anaesthetist = _fee(line[64:75])
return FeeRecord(
billing_code=code,
effective_date=eff,
termination_date=term,
provider_fee=provider,
assistant_fee=assistant,
specialist_fee=specialist,
anaesthetist_fee=anaesthetist,
non_anaesthetist_fee=non_anaesthetist,
all_fees={
"provider": provider,
"assistant": assistant,
"specialist": specialist,
"anaesthetist": anaesthetist,
"non_anaesthetist": non_anaesthetist,
},
)
def _is_active(rec: FeeRecord, today: dt.date) -> bool:
if rec.termination_date is None:
return True
return dt.date.fromisoformat(rec.termination_date) >= today
def parse_fsm(path: Path) -> list[FeeRecord]:
today = dt.date.today()
records: list[FeeRecord] = []
text = Path(path).read_text(encoding="latin-1")
for raw in text.splitlines():
if not raw.strip():
continue
rec = parse_line(raw)
if rec is None:
continue
if settings.active_codes_only and not _is_active(rec, today):
continue
records.append(rec)
logger.info("Parsed %d active fee records from FSM", len(records))
return records
|