app.py

# =============================================================================
# DEPLOYED FILE — canonical source of truth for the Tuas Crossing Space BACKEND.
# Ships to munyew/tuas-crossing as app.py via:
#     python C:\TuasData\deploy_to_hf.py
# Edit THIS file only. Do NOT edit the archived decoys:
#   C:\TuasData\tuas_app_content.py.UNUSED  and  *.UNUSED-decoy-with-alerts
# =============================================================================
"""Tuas Crossing — FastAPI app for a Hugging Face Docker Space.

Serves a dashboard UI plus two JSON APIs:
  /api/live     — real-time corridor state from LTA DataMall (server-side key)
  /api/history  — day x hour heatmap aggregated from the public HF Dataset CSVs

The DataMall key is read from the DATAMALL_KEY env var (set as a Space secret).
It is never sent to the client. The Dataset is public, so history reads need no token.
"""
import os
import time
import json
import io
import threading
from collections import defaultdict
from datetime import datetime, timezone, timedelta

SGT = timezone(timedelta(hours=8))  # Singapore time; HF containers run in UTC


def sgt_now():
    """Current wall-clock time in Singapore, regardless of server timezone."""
    return datetime.now(timezone.utc).astimezone(SGT).strftime("%Y-%m-%d %H:%M:%S")

import httpx
import pandas as pd
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

# --- config -----------------------------------------------------------------
BASE = "https://datamall2.mytransport.sg/ltaodataservice"
KEY = os.environ.get("DATAMALL_KEY", "")
H = {"AccountKey": KEY, "accept": "application/json"}
ROADS = {"AYER RAJAH EXPRESSWAY", "TUAS SECOND CROSSING", "TUAS VIADUCT"}
BOX = dict(latmin=1.325, latmax=1.360, lonmin=103.620, lonmax=103.675)

HF_USER = os.environ.get("HF_USER", "munyew")
DATASET_REPO = os.environ.get("DATASET_REPO", f"{HF_USER}/tuas-data")
ALERTS_REPO = os.environ.get("ALERTS_REPO", f"{HF_USER}/tuas-alerts")  # PRIVATE
HF_WRITE_TOKEN = os.environ.get("HF_WRITE_TOKEN", "")  # Space secret; never sent to client

app = FastAPI(title="Tuas Crossing")

# Two-tier live cache. The "fast" part (cameras/ETA/incidents) is cheap to fetch
# and must stay fresh because camera URLs expire ~5 min. The "bands" part is the
# expensive 288-page speed-band sweep, which LTA only updates ~every 5 min, so we
# refresh it far less often. Each is kept warm by its own background thread.
_fast_cache = {"t": 0.0, "data": None}
_fast_lock = threading.Lock()
_corridor_cache = {"t": 0.0, "data": None}
_queue_cache = {"t": 0.0, "data": None}
_queue_lock = threading.Lock()
_corridor_lock = threading.Lock()
_hist_cache = {"t": 0.0, "data": None}
_hist_lock = threading.Lock()
_fresh_cache = {"t": 0.0, "data": None}
_fresh_lock = threading.Lock()
FAST_TTL = 60       # cameras / travel time / incidents (live LTA, cheap)
CORRIDOR_TTL = 180  # corridor speed bands, read from the public dataset (no LTA)
QUEUE_TTL = 120     # camera-queue estimate, read from the public dataset (no LTA)
HIST_TTL = 600      # history changes slowly; recompute every 10 min
FRESH_TTL = 300     # collection-freshness check (reads the dataset)
# Warn if the newest *collected* sample is older than this. The collector runs
# every 15 min and the push every 15 min, so normal lag is <=~30 min; 60
# tolerates a missed cycle while still catching a real stall (the earlier
# silent failure left data DAYS old).
STALE_MIN = 60


# --- LTA helpers ------------------------------------------------------------
def in_box(lat, lon):
    return (BOX["latmin"] <= lat <= BOX["latmax"]
            and BOX["lonmin"] <= lon <= BOX["lonmax"])


PAGE = 500  # DataMall fixed page size


def _page(client, path, skip, tries=3):
    """Fetch one page, retrying briefly on 429/5xx (LTA throttles bursts)."""
    last = None
    for i in range(tries):
        try:
            r = client.get(f"{BASE}/{path}", headers=H, params={"$skip": skip})
            if r.status_code in (429, 500, 502, 503):
                last = httpx.HTTPStatusError(f"{r.status_code}", request=r.request, response=r)
                time.sleep(0.4 * (i + 1))
                continue
            r.raise_for_status()
            return r.json().get("value", [])
        except httpx.TransportError as e:
            last = e
            time.sleep(0.4 * (i + 1))
    raise last


def lta(client, path, keep=None):
    """Sequential paginated DataMall fetch for small/short feeds. Some feeds
    (Traffic-Imagesv2) ignore $skip and return the same page, so we stop on a
    short page (<500) and guard against a repeated first record."""
    out, skip, prev_first = [], 0, None
    while True:
        v = _page(client, path, skip)
        if not v:
            break
        cur_first = json.dumps(v[0], sort_keys=True)
        if cur_first == prev_first:   # $skip not honoured -> avoid runaway
            break
        prev_first = cur_first
        out += (v if keep is None else [x for x in v if keep(x)])
        skip += len(v)
        if len(v) < PAGE:             # last page reached
            break
    return out


def band_state(avg):
    """Map an average SpeedBand (1 worst .. 8 free-flow) to a label + colour."""
    if avg is None:
        return ("No data", "#475569")
    if avg >= 6.5:
        return ("Free flow", "#22c55e")
    if avg >= 5:
        return ("Moderate", "#84cc16")
    if avg >= 3:
        return ("Slow", "#f59e0b")
    return ("Heavy", "#ef4444")


def build_fast():
    """Cheap feeds that must stay fresh: travel time, incidents, cameras."""
    limits = httpx.Limits(max_connections=10, max_keepalive_connections=10)
    with httpx.Client(timeout=30, limits=limits) as c:
        ett = lta(c, "EstTravelTimes",
                  keep=lambda x: x["Name"] == "AYE"
                  and "TUAS" in (x.get("FarEndPoint") or ""))
        inc = lta(c, "TrafficIncidents")
        img = lta(c, "Traffic-Imagesv2",
                  keep=lambda x: in_box(float(x["Latitude"]), float(x["Longitude"])))

    eta = sum(int(x["EstTime"]) for x in ett) if ett else None

    pad = 0.02
    incidents = []
    for x in inc:
        try:
            lat, lon = float(x["Latitude"]), float(x["Longitude"])
        except (KeyError, ValueError):
            lat = lon = None
        near = (lat is not None and
                BOX["latmin"] - pad <= lat <= BOX["latmax"] + pad and
                BOX["lonmin"] - pad <= lon <= BOX["lonmax"] + pad)
        incidents.append({"type": x.get("Type"), "msg": x.get("Message"),
                          "lat": lat, "lon": lon, "near": near})
    incidents.sort(key=lambda i: not i["near"])

    cameras = [{"id": x["CameraID"], "url": x["ImageLink"],
                "lat": float(x["Latitude"]), "lon": float(x["Longitude"])}
               for x in img]

    return {"etaMinutes": eta, "ettSegments": len(ett),
            "incidents": incidents,
            "incidentsNear": sum(1 for i in incidents if i["near"]),
            "cameras": cameras}


def _newest_dataset_csv(prefix):
    """Download the newest data/<prefix>_<date>.csv from the dataset.
    Returns (DataFrame, date_str) or (None, None)."""
    import re
    from huggingface_hub import HfApi, hf_hub_download
    api = HfApi()
    try:
        files = api.list_repo_files(DATASET_REPO, repo_type="dataset")
    except Exception:
        return None, None
    dated = []
    for f in files:
        if not f.startswith("data/"):
            continue
        m = re.match(rf"{prefix}_(\d{{4}}-\d{{2}}-\d{{2}})\.csv$", os.path.basename(f))
        if m:
            dated.append((m.group(1), f))
    if not dated:
        return None, None
    date_str, fname = max(dated)
    try:
        path = hf_hub_download(DATASET_REPO, fname, repo_type="dataset")
        return pd.read_csv(path), date_str
    except Exception:
        return None, None


def build_corridor():
    """Corridor speed-band cards, read from the PUBLIC DATASET — the collector on
    the laptop is the single v4/TrafficSpeedBands client, so the Space never
    sweeps that feed. Uses the most recent snapshot in the newest speedbands CSV,
    filtered to the 3 corridor roads inside the box, and reports its age."""
    df, _ = _newest_dataset_csv("speedbands")
    empty = {"corridor": [], "corridorTs": None, "corridorAgeMin": None}
    if df is None or df.empty or "ts" not in df.columns:
        return empty
    df = df.copy()
    df["ts"] = pd.to_datetime(df["ts"], errors="coerce")
    df = df.dropna(subset=["ts"])
    if df.empty:
        return empty
    latest = df["ts"].max()
    snap = df[df["ts"] == latest].copy()
    snap = snap[snap["RoadName"].isin(ROADS)]
    for col in ("StartLat", "StartLon", "EndLat", "EndLon", "SpeedBand"):
        snap[col] = pd.to_numeric(snap[col], errors="coerce")
    snap = snap[(snap["StartLat"] >= BOX["latmin"]) & (snap["StartLat"] <= BOX["latmax"]) &
                (snap["StartLon"] >= BOX["lonmin"]) & (snap["StartLon"] <= BOX["lonmax"])]
    snap = snap.dropna(subset=["EndLat", "EndLon"])

    # OUTBOUND filter: segment END closer to JB than START → heading SG→JB
    _JB_LAT, _JB_LON = 1.422, 103.629
    def _dist_jb(lat, lon):
        return (lat - _JB_LAT) ** 2 + (lon - _JB_LON) ** 2
    snap = snap[_dist_jb(snap["EndLat"], snap["EndLon"]) <
                _dist_jb(snap["StartLat"], snap["StartLon"])]

    corridor = []
    for road in ["AYER RAJAH EXPRESSWAY", "TUAS VIADUCT", "TUAS SECOND CROSSING"]:
        bands = snap[snap["RoadName"] == road]["SpeedBand"].dropna()
        avg = round(float(bands.mean()), 2) if len(bands) else None
        label, color = band_state(avg)
        corridor.append({"road": road, "avgBand": avg, "n": int(len(bands)),
                         "state": label, "color": color,
                         "subtitle": "outbound (SG→JB) only"})
    now_sgt = pd.Timestamp(datetime.now(timezone.utc).astimezone(SGT).replace(tzinfo=None))
    age_min = max(0, int((now_sgt - latest).total_seconds() // 60))
    return {"corridor": corridor,
            "corridorTs": latest.strftime("%Y-%m-%d %H:%M:%S"),
            "corridorAgeMin": age_min}


def refresh_fast():
    with _fast_lock:
        data = build_fast()
        _fast_cache.update(t=time.time(), data=data)
        return data


def refresh_corridor():
    with _corridor_lock:
        data = build_corridor()
        _corridor_cache.update(t=time.time(), data=data)
        return data


def build_queue():
    """Booth-queue estimate, read from the public dataset (computed on the laptop
    by Queue-Estimate.py from saved cam 4713 frames). Age is recomputed here from
    the frame timestamp so a stalled job shows as stale, not fresh."""
    from huggingface_hub import hf_hub_download
    try:
        p = hf_hub_download(DATASET_REPO, "queue_estimate.json",
                            repo_type="dataset", force_download=True)
        with open(p, "r", encoding="utf-8") as f:
            q = json.load(f)
    except Exception:
        return {"available": False}
    if not q.get("available"):
        return {"available": False}
    age_sec = None
    try:
        ft = datetime.strptime(q["frameTs"], "%Y-%m-%d %H:%M:%S")
        now = datetime.now(timezone.utc).astimezone(SGT).replace(tzinfo=None)
        age_sec = max(0, int((now - ft).total_seconds()))
    except Exception:
        pass
    return {"available": True, "bucket": q.get("bucket"),
            "minutesBand": q.get("minutesBand"), "score": q.get("score"),
            "ageSec": age_sec, "frameTs": q.get("frameTs"),
            "cross4703Score": q.get("cross4703Score"),
            "source": q.get("source", "cam4713-density"), "provisional": True}


def refresh_queue():
    with _queue_lock:
        data = build_queue()
        _queue_cache.update(t=time.time(), data=data)
        return data


@app.get("/api/live")
def live():
    if not KEY:
        return JSONResponse({"error": "DATAMALL_KEY not configured"}, status_code=503)
    # The fast tier (ETA / cameras / incidents) is the headline and is reliable;
    # build it inline on cold start. A bands failure must NOT take down the live
    # view, so bands are best-effort and the warmer keeps retrying in the dark.
    if not _fast_cache["data"]:
        try:
            refresh_fast()
        except Exception as e:
            if not _fast_cache["data"]:
                return JSONResponse({"warming": True, "error": str(e)}, status_code=503)
    if not _corridor_cache["data"]:
        try:
            refresh_corridor()            # reads the dataset, no LTA
        except Exception:
            pass  # serve ETA + cameras now; corridor fills in from the dataset
    if not _queue_cache["data"]:
        try:
            refresh_queue()               # reads the dataset, no LTA
        except Exception:
            pass

    now = time.time()
    fast = _fast_cache["data"] or {}
    corridor = _corridor_cache["data"] or {}
    rows = corridor.get("corridor", [])
    data = {
        "updated": sgt_now(),
        "corridor": rows,
        "corridorTs": corridor.get("corridorTs"),
        "corridorAgeMin": corridor.get("corridorAgeMin"),
        "corridorSource": "dataset",
        "etaMinutes": fast.get("etaMinutes"),
        "ettSegments": fast.get("ettSegments"),
        "incidents": fast.get("incidents", []),
        "incidentsNear": fast.get("incidentsNear", 0),
        "cameras": fast.get("cameras", []),
        "ageSec": round(now - _fast_cache["t"]),
        "corridorAvailable": any(c.get("avgBand") is not None for c in rows),
        "collected": _fresh_cache["data"],   # populated by the freshness warmer
        "queueEstimate": _queue_cache["data"],  # cam-4713 booth-queue estimate
    }
    return data


def _warm_loop(refresh_fn, ttl, err_backoff=10):
    while True:
        try:
            refresh_fn()
        except Exception:
            # Upstream error: back off before retrying. Bands uses a long backoff
            # so we don't hammer LTA's TrafficSpeedBands while it is flapping/500ing.
            time.sleep(err_backoff)
            continue
        time.sleep(max(15, ttl - 10))


@app.on_event("startup")
def _start_warmers():
    if KEY:
        threading.Thread(target=_warm_loop, args=(refresh_fast, FAST_TTL, 10), daemon=True).start()
    # corridor + freshness read the PUBLIC DATASET (no LTA), so they run even
    # without the key. The Space no longer touches v4/TrafficSpeedBands at all.
    threading.Thread(target=_warm_loop, args=(refresh_corridor, CORRIDOR_TTL, 60), daemon=True).start()
    threading.Thread(target=_warm_loop, args=(refresh_queue, QUEUE_TTL, 60), daemon=True).start()
    threading.Thread(target=_warm_loop, args=(refresh_freshness, FRESH_TTL, 60), daemon=True).start()


# --- collection freshness ---------------------------------------------------
def _latest_collected_ts():
    """Newest sample timestamp across the most recent date's collected CSVs in
    the dataset. Uses the feeds that stamp every 15-min run (travel time /
    images / incidents) plus speed bands, taking the MAX so an intermittent
    speed-band feed never makes collection look stalled."""
    import re
    from huggingface_hub import HfApi, hf_hub_download
    api = HfApi()
    try:
        files = api.list_repo_files(DATASET_REPO, repo_type="dataset")
    except Exception:
        return None
    prefixes = ("esttraveltimes", "images", "incidents", "speedbands")
    cand = []
    for f in files:
        b = os.path.basename(f)
        for p in prefixes:
            m = re.match(rf"{p}_(\d{{4}}-\d{{2}}-\d{{2}})\.csv$", b)
            if m:
                cand.append((m.group(1), f))
    if not cand:
        return None
    maxdate = max(d for d, _ in cand)
    maxts = None
    for d, f in cand:
        if d != maxdate:          # only read the newest date's files (cheap)
            continue
        try:
            path = hf_hub_download(DATASET_REPO, f, repo_type="dataset")
            df = pd.read_csv(path, usecols=["ts"])
            t = pd.to_datetime(df["ts"], errors="coerce").max()
            if pd.notna(t) and (maxts is None or t > maxts):
                maxts = t
        except Exception:
            continue
    return maxts


def build_freshness():
    maxts = _latest_collected_ts()
    if maxts is None:
        return {"lastCollected": None, "ageMin": None, "stale": True,
                "staleAfterMin": STALE_MIN}
    now_sgt = datetime.now(timezone.utc).astimezone(SGT).replace(tzinfo=None)
    age_min = max(0, int((pd.Timestamp(now_sgt) - maxts).total_seconds() // 60))
    return {"lastCollected": maxts.strftime("%Y-%m-%d %H:%M:%S"),
            "ageMin": age_min, "stale": age_min > STALE_MIN,
            "staleAfterMin": STALE_MIN}


def refresh_freshness():
    with _fresh_lock:
        data = build_freshness()
        _fresh_cache.update(t=time.time(), data=data)
        return data


@app.get("/api/freshness")
def freshness():
    if _fresh_cache["data"]:
        return _fresh_cache["data"]
    try:
        return refresh_freshness()
    except Exception as e:
        return JSONResponse({"error": str(e), "stale": None}, status_code=502)


# --- history ----------------------------------------------------------------
def _read_dataset_csvs(prefix):
    """Download and concat all data/<prefix>_*.csv files from the public dataset."""
    from huggingface_hub import HfApi, hf_hub_download
    api = HfApi()
    try:
        files = api.list_repo_files(DATASET_REPO, repo_type="dataset")
    except Exception:
        return pd.DataFrame()
    targets = [f for f in files
               if f.startswith("data/") and os.path.basename(f).startswith(prefix)
               and f.endswith(".csv")]
    frames = []
    for f in targets:
        try:
            path = hf_hub_download(DATASET_REPO, f, repo_type="dataset")
            frames.append(pd.read_csv(path))
        except Exception:
            continue
    return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()


def build_history():
    sb = _read_dataset_csvs("speedbands")
    DOW = ["Monday", "Tuesday", "Wednesday", "Thursday",
           "Friday", "Saturday", "Sunday"]
    heatmap = {d: {} for d in DOW}
    today = {"dow": None, "hours": {}}
    days = 0
    rows = 0
    if not sb.empty and "ts" in sb.columns:
        sb = sb[sb["RoadName"].isin(ROADS)].copy()
        sb["ts"] = pd.to_datetime(sb["ts"], errors="coerce")
        sb = sb.dropna(subset=["ts"])
        sb["hour"] = sb["ts"].dt.hour
        sb["date"] = sb["ts"].dt.date
        sb["dow"] = sb["ts"].dt.day_name()
        sb["SpeedBand"] = pd.to_numeric(sb["SpeedBand"], errors="coerce")
        rows = int(len(sb))
        days = int(sb["date"].nunique())
        grp = sb.groupby(["dow", "hour"])["SpeedBand"].mean()
        for (d, h), val in grp.items():
            if d in heatmap:
                heatmap[d][int(h)] = round(float(val), 2)
        # today's sparkline = most recent date present
        if rows:
            last_date = sb["date"].max()
            tdf = sb[sb["date"] == last_date]
            today["dow"] = tdf["dow"].iloc[0]
            for h, val in tdf.groupby("hour")["SpeedBand"].mean().items():
                today["hours"][int(h)] = round(float(val), 2)

    # estimated travel-time-by-hour (predicted crossing minutes)
    eta_by_hour = {d: {} for d in DOW}
    ett = _read_dataset_csvs("esttraveltimes")
    if not ett.empty and "ts" in ett.columns:
        col = "EstTimeMin" if "EstTimeMin" in ett.columns else (
            "EstTime" if "EstTime" in ett.columns else None)
        if col:
            ett = ett.copy()
            ett["ts"] = pd.to_datetime(ett["ts"], errors="coerce")
            ett = ett.dropna(subset=["ts"])
            ett[col] = pd.to_numeric(ett[col], errors="coerce")
            ett["dow"] = ett["ts"].dt.day_name()
            ett["hour"] = ett["ts"].dt.hour
            ett["minute"] = ett["ts"].dt.floor("min")
            # sum segments per snapshot, then average by dow,hour
            per_snap = ett.groupby(["minute", "dow", "hour"])[col].sum().reset_index()
            grp = per_snap.groupby(["dow", "hour"])[col].mean()
            for (d, h), val in grp.items():
                if d in eta_by_hour:
                    eta_by_hour[d][int(h)] = round(float(val), 1)

    return {
        "updated": sgt_now(),
        "heatmap": heatmap,
        "etaByHour": eta_by_hour,
        "today": today,
        "days": days,
        "rows": rows,
        "dowOrder": DOW,
    }


@app.get("/api/history")
def history():
    now = time.time()
    if _hist_cache["data"] and now - _hist_cache["t"] < HIST_TTL:
        return _hist_cache["data"]
    with _hist_lock:
        now = time.time()
        if _hist_cache["data"] and now - _hist_cache["t"] < HIST_TTL:
            return _hist_cache["data"]
        try:
            data = build_history()
        except Exception as e:
            return JSONResponse({"error": str(e)}, status_code=502)
        _hist_cache.update(t=time.time(), data=data)
        return data


@app.get("/api/health")
def health():
    return {"ok": True, "keyConfigured": bool(KEY), "dataset": DATASET_REPO,
            "alertsConfigured": bool(HF_WRITE_TOKEN)}


# --- Web Push alerts: receive subscription + test request -------------------
# The Space only STORES these in the PRIVATE tuas-alerts dataset. The always-on
# laptop (which holds the VAPID private key) actually sends pushes via
# Send-Alerts.py — free Spaces sleep, so the Space is never the sender.
def _alerts_write(path_in_repo, data_bytes, msg):
    from huggingface_hub import HfApi
    HfApi(token=HF_WRITE_TOKEN).upload_file(
        path_or_fileobj=data_bytes, path_in_repo=path_in_repo,
        repo_id=ALERTS_REPO, repo_type="dataset", commit_message=msg)


@app.post("/api/subscribe")
async def subscribe(req: Request):
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    try:
        body = await req.json()
    except Exception:
        return JSONResponse({"error": "bad json"}, status_code=400)
    sub = body.get("subscription") or {}
    if not sub.get("endpoint"):
        return JSONResponse({"error": "missing subscription"}, status_code=400)
    payload = {"subscription": sub, "ua": body.get("ua"), "savedAt": sgt_now()}
    try:
        _alerts_write("subscription.json",
                      json.dumps(payload, indent=2).encode(), "save push subscription")
    except Exception as e:
        return JSONResponse({"error": f"save failed: {e}"}, status_code=502)
    return {"ok": True}


@app.post("/api/request-test")
async def request_test(req: Request):
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    try:
        _alerts_write("test_request.json",
                      json.dumps({"ts": sgt_now()}).encode(), "test push requested")
    except Exception as e:
        return JSONResponse({"error": f"request failed: {e}"}, status_code=502)
    return {"ok": True}


# --- Alert RULE (day/time window + GO/WAIT thresholds) ----------------------
DAYS = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]


def _alerts_read(name):
    """Read a JSON file from the PRIVATE alerts dataset (needs the write token to
    read a private repo). Returns the parsed object or None."""
    from huggingface_hub import hf_hub_download
    try:
        path = hf_hub_download(ALERTS_REPO, name, repo_type="dataset",
                               token=HF_WRITE_TOKEN, force_download=True)
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def _alerts_delete(name):
    from huggingface_hub import HfApi
    HfApi(token=HF_WRITE_TOKEN).delete_file(
        name, ALERTS_REPO, repo_type="dataset", commit_message=f"delete {name}")


def _valid_rule(r):
    """Validate + normalise a posted rule. Returns (rule, error)."""
    if not isinstance(r, dict):
        return None, "rule must be an object"
    days = [d for d in (r.get("days") or []) if d in DAYS]
    if not days:
        return None, "pick at least one day"
    import re
    hhmm = re.compile(r"^([01]?\d|2[0-3]):[0-5]\d$")
    frm, to = str(r.get("from", "")), str(r.get("to", ""))
    if not (hhmm.match(frm) and hhmm.match(to)):
        return None, "from/to must be HH:MM"

    def num(v):
        try:
            return int(v)
        except (TypeError, ValueError):
            return None
    go, wait = num(r.get("go")), num(r.get("wait"))
    if go is None and wait is None:
        return None, "set a GO and/or a WAIT threshold"
    return ({"days": [d for d in DAYS if d in days], "from": frm, "to": to,
             "go": go, "wait": wait, "savedAt": sgt_now()}, None)


@app.post("/api/rule")
async def save_rule(req: Request):
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    try:
        body = await req.json()
    except Exception:
        return JSONResponse({"error": "bad json"}, status_code=400)
    rule, err = _valid_rule(body)
    if err:
        return JSONResponse({"error": err}, status_code=400)
    try:
        _alerts_write("rule.json", json.dumps(rule, indent=2).encode(), "save alert rule")
        # a new/edited rule resets the debounce so it can fire fresh
        try:
            _alerts_delete("alert_state.json")
        except Exception:
            pass
    except Exception as e:
        return JSONResponse({"error": f"save failed: {e}"}, status_code=502)
    return {"ok": True, "rule": rule}


@app.get("/api/rule")
def get_rule():
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    return {"rule": _alerts_read("rule.json")}


@app.delete("/api/rule")
def delete_rule():
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    try:
        _alerts_delete("rule.json")
    except Exception:
        pass
    try:
        _alerts_delete("alert_state.json")
    except Exception:
        pass
    return {"ok": True}


# --- Calibration: log my ACTUAL crossing time -------------------------------
# Stored in the PRIVATE dataset (personal data). Each entry pairs the camera
# bucket/score at crossing time with the real minutes I took, so Stage-2
# calibration can replace the placeholder bucket->minutes bands with my data.
@app.post("/api/crossing-log")
async def crossing_log(req: Request):
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    try:
        body = await req.json()
    except Exception:
        return JSONResponse({"error": "bad json"}, status_code=400)
    try:
        minutes = int(round(float(body.get("minutes"))))
    except (TypeError, ValueError):
        return JSONResponse({"error": "minutes required"}, status_code=400)
    if not (1 <= minutes <= 300):
        return JSONResponse({"error": "minutes out of range"}, status_code=400)
    entry = {
        "loggedAt": sgt_now(),
        "minutes": minutes,
        "bucket": body.get("bucket"),          # camera bucket at crossing time
        "score": body.get("score"),
        "startTs": body.get("startTs"),
        "endTs": body.get("endTs"),
        "mode": body.get("mode", "manual"),
    }
    logs = _alerts_read("crossing_logs.json") or []
    if not isinstance(logs, list):
        logs = []
    logs.append(entry)
    try:
        _alerts_write("crossing_logs.json",
                      json.dumps(logs, indent=2).encode(), "log crossing")
    except Exception as e:
        return JSONResponse({"error": f"save failed: {e}"}, status_code=502)
    return {"ok": True, "count": len(logs)}


@app.get("/api/crossing-log")
def crossing_log_list():
    if not HF_WRITE_TOKEN:
        return JSONResponse({"error": "alerts not configured"}, status_code=503)
    logs = _alerts_read("crossing_logs.json") or []
    if not isinstance(logs, list):
        logs = []
    return {"count": len(logs), "recent": logs[-5:]}


# --- ETT chart: today vs last week same weekday --------------------------------
DATA_DIR = os.environ.get("ETT_DATA_DIR", r"C:\TuasData")


def _ett_hourly_from_csv(csv_path):
    """Read one esttraveltimes CSV and return {hour: avg_summed_minutes}.
    Filters AYE -> TUAS CHECKPOINT, sums segments per snapshot, averages by hour."""
    try:
        df = pd.read_csv(csv_path)
    except Exception:
        return {}
    col = "EstTimeMin" if "EstTimeMin" in df.columns else (
          "EstTime" if "EstTime" in df.columns else None)
    if col is None:
        return {}
    df = df[(df["Expressway"] == "AYE") &
            (df["FarEndPoint"].str.contains("TUAS", na=False))].copy()
    if df.empty:
        return {}
    df["ts"] = pd.to_datetime(df["ts"], errors="coerce")
    df = df.dropna(subset=["ts"])
    df[col] = pd.to_numeric(df[col], errors="coerce")
    df["minute"] = df["ts"].dt.floor("min")
    df["hour"] = df["ts"].dt.hour
    per_snap = df.groupby(["minute", "hour"])[col].sum().reset_index()
    by_hour = per_snap.groupby("hour")[col].mean().round(1)
    return {int(h): float(v) for h, v in by_hour.items()}


def build_ett():
    """Return today's hourly ETT and (if available) last-week same-weekday ETT."""
    now_sgt = datetime.now(timezone.utc).astimezone(SGT)
    today_str = now_sgt.strftime("%Y-%m-%d")
    lastweek_str = (now_sgt - timedelta(days=7)).strftime("%Y-%m-%d")

    def _csv_path(date_str):
        return os.path.join(DATA_DIR, f"esttraveltimes_{date_str}.csv")

    today_hours = {}
    lastweek_hours = {}

    # Try local filesystem first
    today_local = _csv_path(today_str)
    if os.path.isfile(today_local):
        today_hours = _ett_hourly_from_csv(today_local)

    lastweek_local = _csv_path(lastweek_str)
    if os.path.isfile(lastweek_local):
        lastweek_hours = _ett_hourly_from_csv(lastweek_local)

    # Fall back to HF dataset for any missing date
    if not today_hours or not lastweek_hours:
        try:
            from huggingface_hub import HfApi, hf_hub_download
            api = HfApi()
            files = api.list_repo_files(DATASET_REPO, repo_type="dataset")
            for f in files:
                b = os.path.basename(f)
                if b == f"esttraveltimes_{today_str}.csv" and not today_hours:
                    path = hf_hub_download(DATASET_REPO, f, repo_type="dataset")
                    today_hours = _ett_hourly_from_csv(path)
                elif b == f"esttraveltimes_{lastweek_str}.csv" and not lastweek_hours:
                    path = hf_hub_download(DATASET_REPO, f, repo_type="dataset")
                    lastweek_hours = _ett_hourly_from_csv(path)
        except Exception:
            pass

    return {
        "today": today_str,
        "lastWeek": lastweek_str if lastweek_hours else None,
        "todayHours": today_hours,
        "lastWeekHours": lastweek_hours,
    }


_ett_cache = {"t": 0.0, "data": None}
_ett_lock = threading.Lock()
ETT_TTL = 300  # recompute every 5 min


@app.get("/api/ett")
def ett_chart():
    now = time.time()
    if _ett_cache["data"] and now - _ett_cache["t"] < ETT_TTL:
        return _ett_cache["data"]
    with _ett_lock:
        now = time.time()
        if _ett_cache["data"] and now - _ett_cache["t"] < ETT_TTL:
            return _ett_cache["data"]
        try:
            data = build_ett()
        except Exception as e:
            return JSONResponse({"error": str(e)}, status_code=502)
        _ett_cache.update(t=time.time(), data=data)
        return data


# static UI (must be mounted last so /api/* wins)
app.mount("/", StaticFiles(directory="static", html=True), name="static")