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