Spaces:
Running
Running
| """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" | |
| 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) | |
| 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 | |