core.py

"""core.py — Drive & Save: pure computation module.

No network calls. No Gradio imports. Takes data as function arguments.
All tunable parameters live in config.yaml.
"""
from __future__ import annotations

import math
from dataclasses import dataclass
from pathlib import Path

import h3
import yaml

CONFIG_PATH = Path(__file__).parent / "config.yaml"


def load_config(path: "Path | str" = CONFIG_PATH) -> dict:
    """Read config.yaml into a dict. All tunable params live there."""
    with open(path, "r") as f:
        return yaml.safe_load(f)


# ---------------------------------------------------------------------------
# H3 distance helpers
# ---------------------------------------------------------------------------


def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    R = 6371.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlmb = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2
    return 2 * R * math.asin(math.sqrt(a))


def grid_distance_km(home_cell: str, target_cell: str, cfg: dict) -> float:
    """Approx straight-line km between two res-8 H3 cells.

    Primary: H3 grid distance × per-step spacing.
    Falls back to haversine on cell centroids if grid_distance raises.
    """
    if home_cell == target_cell:
        return 0.0
    km_per_step = cfg["h3"]["km_per_step"]
    try:
        steps = h3.grid_distance(home_cell, target_cell)
        return steps * km_per_step
    except Exception:
        lat1, lon1 = h3.cell_to_latlng(home_cell)
        lat2, lon2 = h3.cell_to_latlng(target_cell)
        return _haversine_km(lat1, lon1, lat2, lon2)


# ---------------------------------------------------------------------------
# Drive-time estimate
# ---------------------------------------------------------------------------


def _round_to(value: float, nearest: int) -> int:
    return int(round(value / nearest) * nearest)


def drive_minutes(dist_km: float, cfg: dict) -> int:
    """Approx driving minutes from a straight-line distance.

    Explicitly an estimate (UI labels it 'approx').
    """
    d = cfg["distance"]
    minutes = dist_km * d["road_factor"] / d["avg_speed_kmh"] * 60.0
    return _round_to(minutes, d["drive_time_round_min"])


# ---------------------------------------------------------------------------
# Ranking, suppression, savings
# ---------------------------------------------------------------------------


@dataclass
class MetroRow:
    metro: str
    median_price: float
    n_hospitals: int
    drive_min: int
    is_home: bool
    suppressed: bool


@dataclass
class Ranking:
    procedure_name: str
    home_metro: str
    home_median: "float | None"
    cheapest_metro: "str | None"
    cheapest_median: "float | None"
    savings: float
    savings_pct: float
    drive_min_to_cheapest: int
    home_is_cheapest: bool
    rows: list  # list[MetroRow], ascending by price


def rank_metros(medians_df, metros_df, procedure_code: str, home_metro: str, cfg: dict) -> Ranking:
    """Rank same-state metros for a procedure; compute savings vs. the cheapest
    non-suppressed metro and the drive time to it from the home metro.
    """
    min_hosp = cfg["ranking"]["min_hospitals"]

    cells = dict(zip(metros_df["metro"], metros_df["h3_cell"]))
    states = dict(zip(metros_df["metro"], metros_df["state"]))
    home_state = states.get(home_metro)
    home_cell = cells.get(home_metro)

    df = medians_df[
        (medians_df["procedure_code"] == procedure_code)
        & (medians_df["state"] == home_state)
    ].copy()
    df = df.sort_values("median_price", ascending=True)

    procedure_name = str(df["procedure_name"].iloc[0]) if len(df) else procedure_code

    rows: list = []
    for _, m in df.iterrows():
        suppressed = int(m["n_hospitals"]) < min_hosp
        target_cell = cells.get(m["metro"])
        drive = (
            drive_minutes(grid_distance_km(home_cell, target_cell, cfg), cfg)
            if home_cell and target_cell else 0
        )
        rows.append(MetroRow(
            metro=str(m["metro"]),
            median_price=float(m["median_price"]),
            n_hospitals=int(m["n_hospitals"]),
            drive_min=drive,
            is_home=(m["metro"] == home_metro),
            suppressed=suppressed,
        ))

    visible = [r for r in rows if not r.suppressed]
    home_row = next((r for r in rows if r.is_home), None)
    home_median = home_row.median_price if home_row else None
    home_suppressed = home_row.suppressed if home_row else True

    candidates = [r for r in visible if not r.is_home]
    cheapest = candidates[0] if candidates else None  # rows already price-asc
    home_is_cheapest = bool(
        home_row and not home_row.suppressed and visible and visible[0].is_home
    )

    if cheapest and home_median is not None and not home_suppressed and not home_is_cheapest:
        savings = home_median - cheapest.median_price
        savings_pct = (savings / home_median * 100.0) if home_median else 0.0
        drive_to = cheapest.drive_min
        cheapest_metro, cheapest_median = cheapest.metro, cheapest.median_price
    elif home_is_cheapest and cheapest:
        savings, savings_pct, drive_to = 0.0, 0.0, cheapest.drive_min
        cheapest_metro, cheapest_median = cheapest.metro, cheapest.median_price
    else:
        savings, savings_pct, drive_to = 0.0, 0.0, 0
        cheapest_metro = cheapest.metro if cheapest else None
        cheapest_median = cheapest.median_price if cheapest else None

    return Ranking(
        procedure_name=procedure_name,
        home_metro=home_metro,
        home_median=home_median,
        cheapest_metro=cheapest_metro,
        cheapest_median=cheapest_median,
        savings=savings,
        savings_pct=savings_pct,
        drive_min_to_cheapest=drive_to,
        home_is_cheapest=home_is_cheapest,
        rows=rows,
    )


# ---------------------------------------------------------------------------
# CTA URL builder
# ---------------------------------------------------------------------------


def _slugify(text: str) -> str:
    out = "".join(c if c.isalnum() else "-" for c in text.lower())
    while "--" in out:
        out = out.replace("--", "-")
    return out.strip("-")


def build_cta_url(procedure_code: str, metro: str, cfg: dict) -> str:
    """Build the CPH CTA link.

    Deep-links if a pattern is configured, else the homepage.
    Always UTM-tagged.
    """
    f = cfg["funnel"]
    base = f["cta_base_url"].rstrip("/")
    pattern = f.get("deep_link_pattern") or ""
    utm = f["utm"]
    if pattern:
        path = pattern.format(code=procedure_code, metro_slug=_slugify(metro))
        sep = "&" if "?" in path else "?"
        return f"{base}{path}{sep}{utm}"
    return f"{base}/?{utm}"