"""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