app.py

"""
Clock It — Flask Application
Run with: python app.py
Ensure pipeline.py and ml/train.py have been run first.
"""

from flask import Flask, render_template, jsonify, request, send_file
import pandas as pd
import numpy as np
import json
import pickle
import os
import sys
import subprocess
import threading
import uuid
from pathlib import Path
from datetime import datetime

app = Flask(__name__)
app.config["JSON_SORT_KEYS"] = False

BASE_DIR  = Path(__file__).resolve().parent
PROCESSED = BASE_DIR / "data" / "processed"
ML_DIR    = BASE_DIR / "ml"
RAW_DIR   = BASE_DIR / "data" / "raw"

# In-memory job tracker for long-running pipeline/training runs.
# Keyed by job id so the frontend can poll status without holding
# the HTTP connection open for the full duration.
JOBS = {}
JOBS_LOCK = threading.Lock()

# In-memory storage for mobile app state.
# WARNING: This state is lost on every server restart. All claimed slots,
# manpower entries, and handover notes will be cleared. For production use,
# replace with a database (SQLite is sufficient) or Redis.
mobile_state = {
    "claims":      {},
    "manpower":    {},
    "handovers":   {},
    "task_status": {},
    "alerts":      [],
}


# ─────────────────────────────────────────────
# HELPERS
# ─────────────────────────────────────────────

def load_json(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def load_csv(name):
    return pd.read_csv(PROCESSED / name)


def safe_unlink(path, retries=3, delay=0.15):
    """
    Delete a file without raising. On Windows, a file can remain briefly
    locked after a read/write handle goes out of scope even though the
    `with` block has exited, because the OS releases the lock on a
    slightly different schedule than CPython's reference counting closes
    the handle. A bare path.unlink() in that window raises PermissionError,
    which then replaces the original, more useful error in the response.
    Retrying a few times with a short delay avoids that without ever
    surfacing a cleanup failure to the caller.
    """
    import time
    for attempt in range(retries):
        try:
            path.unlink(missing_ok=True)
            return
        except PermissionError:
            if attempt == retries - 1:
                return  # give up silently — stale temp file, not worth failing the request over
            time.sleep(delay)
        except Exception:
            return


def is_job_running():
    with JOBS_LOCK:
        return any(j["status"] == "running" for j in JOBS.values())


def generate_mock_alerts(station=None):
    """Generate alerts from Critical-tier junctions for the given station."""
    try:
        pdi = load_csv("junction_pdi.csv")
        if station:
            pdi = pdi[pdi["police_station"] == station]
        criticals = pdi[pdi["pdi_tier"] == "Critical"].head(2)
        alerts = []
        now_time = datetime.now().strftime("%H:%M")
        for _, row in criticals.iterrows():
            alerts.append({
                "type":      "critical",
                "junction":  row["junction_short"],
                "message":   f"Spike in commercial vehicles blocking main road. PDI at {int(row['pdi_score'])}.",
                "time":      now_time,
                "pdi_score": int(row["pdi_score"]),
            })
        return alerts
    except Exception:
        return []


def _run_script_job(job_id, script_name, timeout):
    """Execute a script as a subprocess in a background thread and
    record its outcome in the JOBS dict. Never raises into the thread
    pool — all exceptions are captured into the job record."""
    with JOBS_LOCK:
        JOBS[job_id] = {
            "status": "running", "script": script_name,
            "stdout": "", "stderr": "", "result": None,
        }
    try:
        # encoding="utf-8" is required on Windows: subprocess.run(text=True)
        # without it decodes the child process's stdout/stderr using the
        # parent's default codepage (cp1252), and pipeline.py's own log
        # output contains an em-dash, which throws UnicodeDecodeError and
        # would otherwise silently fail every pipeline/training run on
        # Windows after the script itself completed successfully.
        # PYTHONIOENCODING forces the child interpreter's own stdout/stderr
        # streams to UTF-8 regardless of the Windows console codepage —
        # without this, pipeline.py's print() calls containing an em-dash
        # can raise UnicodeEncodeError inside the child process itself,
        # before subprocess.run even gets a chance to decode the output.
        child_env = os.environ.copy()
        child_env["PYTHONIOENCODING"] = "utf-8"

        proc = subprocess.run(
            [sys.executable, script_name],
            capture_output=True, text=True, timeout=timeout,
            encoding="utf-8", errors="replace",
            cwd=str(BASE_DIR), env=child_env,
        )
        ok = proc.returncode == 0
        result = {}
        if ok and script_name == "pipeline.py" and (PROCESSED / "summary.json").exists():
            result["summary"] = load_json(PROCESSED / "summary.json")
        if ok and script_name == "ml/train.py":
            if (ML_DIR / "metrics.json").exists():
                result["metrics"] = load_json(ML_DIR / "metrics.json")
            if (ML_DIR / "feature_importance.json").exists():
                result["feature_importance"] = load_json(ML_DIR / "feature_importance.json")
        with JOBS_LOCK:
            JOBS[job_id] = {
                "status": "done" if ok else "failed",
                "script": script_name,
                "stdout": (proc.stdout or "")[-3000:],
                "stderr": (proc.stderr or "")[-1500:],
                "result": result,
            }
    except subprocess.TimeoutExpired:
        with JOBS_LOCK:
            JOBS[job_id] = {
                "status": "failed", "script": script_name,
                "stdout": "", "stderr": f"{script_name} timed out after {timeout}s",
                "result": None,
            }
    except Exception as e:
        with JOBS_LOCK:
            JOBS[job_id] = {
                "status": "failed", "script": script_name,
                "stdout": "", "stderr": str(e), "result": None,
            }


# ─────────────────────────────────────────────
# PAGES
# ─────────────────────────────────────────────

@app.route("/")
def landing():
    return render_template("landing.html")


@app.route("/dashboard")
def index():
    summary = load_json(PROCESSED / "summary.json")
    return render_template("index.html", summary=summary)


@app.route("/map")
def map_view():
    summary = load_json(PROCESSED / "summary.json")
    return render_template("map.html", summary=summary)


@app.route("/schedule")
def schedule_view():
    summary = load_json(PROCESSED / "summary.json")
    return render_template("schedule.html", summary=summary)


@app.route("/analytics")
def analytics_view():
    summary = load_json(PROCESSED / "summary.json")
    return render_template("analytics.html", summary=summary)


@app.route("/model")
def model_view():
    summary  = load_json(PROCESSED / "summary.json")
    metrics  = load_json(ML_DIR / "metrics.json")
    features = load_json(ML_DIR / "feature_importance.json")
    return render_template("model.html", summary=summary,
                           metrics=metrics, features=features)


@app.route("/brief/<btp_code>")
def junction_brief(btp_code):
    pdi = load_csv("junction_pdi.csv")
    row = pdi[pdi["btp_code"] == btp_code]
    if row.empty:
        return "Junction not found", 404
    jxn     = row.iloc[0].to_dict()
    summary = load_json(PROCESSED / "summary.json")
    return render_template("brief.html", jxn=jxn, summary=summary)


@app.route("/mobile")
def mobile_view():
    return render_template("mobile.html")


# ─────────────────────────────────────────────
# API — CORE DATA
# ─────────────────────────────────────────────

@app.route("/api/summary")
def api_summary():
    return jsonify(load_json(PROCESSED / "summary.json"))


@app.route("/api/junctions")
def api_junctions():
    """GeoJSON of top junctions for Leaflet map."""
    geojson = load_json(PROCESSED / "junctions.geojson")
    tier    = request.args.get("tier", "all")
    if tier != "all":
        geojson["features"] = [
            f for f in geojson["features"]
            if f["properties"]["pdi_tier"] == tier
        ]
    return jsonify(geojson)


@app.route("/api/pdi-leaderboard")
def api_pdi_leaderboard():
    pdi = load_csv("junction_pdi.csv")
    n   = int(request.args.get("n", 20))
    cols = [
        "rank", "junction_short", "btp_code", "pdi_score", "pdi_tier",
        "total_violations", "police_station", "main_road_rate",
        "peak_hour_rate", "repeat_consistency", "multi_viol_rate",
        "hotspot_prob", "lat", "lon",
    ]
    cols_exist = [c for c in cols if c in pdi.columns]
    top = pdi.head(n)[cols_exist].fillna(0)
    return jsonify(top.to_dict(orient="records"))


@app.route("/api/schedule")
def api_schedule():
    """
    Returns enforcement slots for a given day/shift.
    If a station is provided, returns that station's own PDI junctions
    (sorted Critical→Low) so every station sees their real locations —
    not the same global 5-junction list.
    Falls back to the schedule CSV for the global view (desktop schedule page).
    """
    day     = request.args.get("day",     None)
    shift   = request.args.get("shift",   None)
    station = request.args.get("station", None)

    if station:
        # Per-station view: pull from full PDI leaderboard filtered to this station.
        pdi  = load_csv("junction_pdi.csv")
        mine = pdi[pdi["police_station"] == station].copy()
        mine = mine.sort_values("pdi_score", ascending=False)

        shift_label = (
            "Morning (06-12h)"
            if (shift or "").lower().startswith("m")
            else "Evening (16-22h)"
        )
        records = []
        for _, row in mine.iterrows():
            # Estimate slot violations: split total evenly across 14 slots (7 days × 2 shifts)
            slot_v    = round(float(row.get("total_violations", 0)) / 14)
            reduction = min(0.45, max(0.15, float(row.get("pdi_score", 50)) / 100 * 0.45))
            records.append({
                "day":                   day or "Today",
                "shift":                 shift_label,
                "junction_short":        row.get("junction_short", ""),
                "btp_code":              row.get("btp_code", ""),
                "police_station":        row.get("police_station", ""),
                "pdi_score":             round(float(row.get("pdi_score", 0)), 1),
                "pdi_tier":              row.get("pdi_tier", "Low"),
                "slot_violations":       slot_v,
                "expected_reduction_pct": round(reduction, 3),
                "lat":                   float(row.get("lat", 0)),
                "lon":                   float(row.get("lon", 0)),
            })
        return jsonify(records)

    # Global view (desktop schedule page): use the schedule CSV.
    sched = load_csv("enforcement_schedule.csv")
    if day:
        sched = sched[sched["day"] == day]
    if shift:
        sched = sched[sched["shift"].str.contains(shift, case=False)]
    cols = [
        "day", "shift", "junction_short", "btp_code", "police_station",
        "pdi_score", "pdi_tier", "slot_violations", "expected_reduction_pct",
        "lat", "lon",
    ]
    cols_exist = [c for c in cols if c in sched.columns]
    return jsonify(sched[cols_exist].fillna(0).to_dict(orient="records"))


@app.route("/api/hourly")
def api_hourly():
    city_hour = load_csv("city_hourly.csv")
    return jsonify(city_hour.to_dict(orient="records"))


@app.route("/api/junction-hourly/<btp_code>")
def api_junction_hourly(btp_code):
    pdi = load_csv("junction_pdi.csv")
    row = pdi[pdi["btp_code"] == btp_code]
    if row.empty:
        return jsonify([])
    jxn_name  = row.iloc[0]["junction_name"]
    jh        = load_csv("junction_hourly.csv")
    jh        = jh[jh["junction_name"] == jxn_name]
    all_hours = pd.DataFrame({"hour": range(24)})
    jh        = all_hours.merge(jh[["hour", "count"]], on="hour", how="left").fillna(0)
    return jsonify(jh.to_dict(orient="records"))


@app.route("/api/vehicle-types")
def api_vehicle_types():
    vt = load_csv("vehicle_types.csv")
    return jsonify(vt.head(10).to_dict(orient="records"))


@app.route("/api/stations")
def api_stations():
    st = load_csv("police_stations.csv")
    return jsonify(st.to_dict(orient="records"))


@app.route("/api/monthly")
def api_monthly():
    m           = load_csv("monthly_trend.csv")
    month_names = {
        1: "Jan", 2: "Feb", 3: "Mar", 4: "Apr",
        5: "May", 6: "Jun", 7: "Jul", 8: "Aug",
        9: "Sep", 10: "Oct", 11: "Nov", 12: "Dec",
    }
    m["month_name"] = m["month"].map(month_names)
    return jsonify(m.to_dict(orient="records"))


@app.route("/api/dow")
def api_dow():
    d = load_csv("city_dow.csv")
    return jsonify(d.to_dict(orient="records"))


# ─────────────────────────────────────────────
# API — ML / MODEL
# ─────────────────────────────────────────────

@app.route("/api/model-metrics")
def api_model_metrics():
    return jsonify(load_json(ML_DIR / "metrics.json"))


@app.route("/api/feature-importance")
def api_feature_importance():
    return jsonify(load_json(ML_DIR / "feature_importance.json"))


@app.route("/api/predict", methods=["POST"])
def api_predict():
    """Score a junction given its stats. Used by the officer brief."""
    data = request.json or {}
    try:
        with open(ML_DIR / "model.pkl", "rb") as f:
            model = pickle.load(f)
        features = load_json(ML_DIR / "feature_names.json")

        # Warn about any keys in the request that don't match the model's
        # expected feature list (previously silently defaulted to 0, making
        # debugging the brief page very hard).
        unknown_keys = [k for k in data if k not in features]
        missing_keys = [k for k in features if k not in data]

        X    = [[float(data.get(k, 0)) for k in features]]
        prob = model.predict_proba(X)[0][1]
        return jsonify({
            "probability":  round(prob * 100, 1),
            "features":     features,
            "unknown_keys": unknown_keys,   # keys sent but not used
            "missing_keys": missing_keys,   # features defaulted to 0
        })
    except FileNotFoundError:
        return jsonify({"error": "Model not found. Run ml/train.py first."}), 500
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/brief-pdf/<btp_code>")
def api_brief_pdf(btp_code):
    """Generate and return a PDF enforcement brief."""
    try:
        from utils.pdf_generator import generate_brief_pdf
        pdi = load_csv("junction_pdi.csv")
        row = pdi[pdi["btp_code"] == btp_code]
        if row.empty:
            return "Junction not found", 404
        jxn  = row.iloc[0].to_dict()
        path = generate_brief_pdf(jxn)
        return send_file(path, as_attachment=True,
                         download_name=f"brief_{btp_code}.pdf")
    except Exception as e:
        return jsonify({"error": str(e)}), 500


# ─────────────────────────────────────────────
# API — DATASET MANAGEMENT
# ─────────────────────────────────────────────

@app.route("/api/upload-dataset", methods=["POST"])
def api_upload_dataset():
    """
    Accept a new violations.csv upload, validate it has the expected
    columns, and replace data/raw/violations.csv. Does NOT run the
    pipeline automatically — that is a separate explicit step so the
    UI can show progress for each stage independently.
    """
    if is_job_running():
        return jsonify({
            "ok": False,
            "error": "A pipeline or training run is already in progress. Wait for it to finish before uploading a new dataset.",
        }), 409

    if "file" not in request.files:
        return jsonify({"ok": False, "error": "No file part in request"}), 400

    f = request.files["file"]
    if f.filename == "":
        return jsonify({"ok": False, "error": "No file selected"}), 400

    if not f.filename.lower().endswith(".csv"):
        return jsonify({"ok": False, "error": "File must be a .csv"}), 400

    RAW_DIR.mkdir(parents=True, exist_ok=True)
    tmp_path = RAW_DIR / f"_uploaded_tmp_{uuid.uuid4().hex[:8]}.csv"

    try:
        f.save(tmp_path)
    except Exception as e:
        safe_unlink(tmp_path)
        return jsonify({"ok": False, "error": f"Could not save upload: {e}"}), 500

    # Validate required columns before committing.
    # Row counting goes through pandas rather than a raw open()/line-count
    # loop: pandas sniffs encoding (utf-8, utf-8-sig, latin-1 fallback)
    # far more reliably than Python's open() default, which on Windows
    # is the system codepage (cp1252) and will throw UnicodeDecodeError
    # on any byte outside that codepage's mapped range.
    required_cols = {
        "id", "latitude", "longitude", "location", "violation_type",
        "vehicle_type", "created_datetime", "junction_name", "police_station",
    }
    try:
        sample  = pd.read_csv(tmp_path, nrows=5)
        missing = required_cols - set(sample.columns)
        if missing:
            safe_unlink(tmp_path)
            return jsonify({
                "ok":    False,
                "error": f"Missing required columns: {', '.join(sorted(missing))}",
            }), 400
        if len(sample) == 0:
            safe_unlink(tmp_path)
            return jsonify({"ok": False, "error": "CSV has no data rows"}), 400

        # Count rows via pandas in chunks so large files don't load fully
        # into memory just to get a count, and so encoding is handled
        # the same way the column-validation read above handled it.
        row_count = 0
        for chunk in pd.read_csv(tmp_path, usecols=[sample.columns[0]], chunksize=50_000):
            row_count += len(chunk)
    except pd.errors.EmptyDataError:
        safe_unlink(tmp_path)
        return jsonify({"ok": False, "error": "CSV file is empty"}), 400
    except UnicodeDecodeError as e:
        safe_unlink(tmp_path)
        return jsonify({
            "ok":    False,
            "error": f"CSV encoding not recognised — please save it as UTF-8 and re-upload ({e})",
        }), 400
    except Exception as e:
        safe_unlink(tmp_path)
        return jsonify({"ok": False, "error": f"Could not parse CSV: {e}"}), 400

    final_path = RAW_DIR / "violations.csv"
    try:
        tmp_path.replace(final_path)
    except Exception as e:
        safe_unlink(tmp_path)
        return jsonify({"ok": False, "error": f"Could not save dataset: {e}"}), 500

    return jsonify({
        "ok":       True,
        "filename": f.filename,
        "rows":     row_count,
        "columns":  list(sample.columns),
        "message":  f"Dataset uploaded — {row_count:,} rows. Run the pipeline to process it.",
    })


@app.route("/api/dataset-status")
def api_dataset_status():
    """Check what dataset is currently active and whether pipeline/ML outputs exist."""
    raw_path = RAW_DIR / "violations.csv"
    status = {
        "dataset_exists":  raw_path.exists(),
        "dataset_rows":    None,
        "pipeline_done":   (PROCESSED / "junction_pdi.csv").exists(),
        "model_trained":   (ML_DIR / "metrics.json").exists(),
        "job_running":     is_job_running(),
    }
    if raw_path.exists():
        try:
            status["dataset_modified"] = datetime.fromtimestamp(
                raw_path.stat().st_mtime
            ).strftime("%Y-%m-%d %H:%M:%S")
        except Exception:
            pass
        try:
            first_col = pd.read_csv(raw_path, nrows=0).columns[0]
            row_count = 0
            for chunk in pd.read_csv(raw_path, usecols=[first_col], chunksize=50_000):
                row_count += len(chunk)
            status["dataset_rows"] = row_count
        except Exception as e:
            status["dataset_rows_error"] = str(e)
    return jsonify(status)


# ─────────────────────────────────────────────
# API — PIPELINE & TRAINING (ASYNC)
# ─────────────────────────────────────────────

@app.route("/api/run-pipeline", methods=["POST"])
def api_run_pipeline():
    """Kick off pipeline.py in a background thread and return a job id immediately."""
    if not (RAW_DIR / "violations.csv").exists():
        return jsonify({"ok": False, "error": "No dataset found. Upload one first."}), 400
    if is_job_running():
        return jsonify({"ok": False, "error": "Another job is already running. Wait for it to finish."}), 409

    job_id = uuid.uuid4().hex
    thread = threading.Thread(
        target=_run_script_job, args=(job_id, "pipeline.py", 180), daemon=True
    )
    thread.start()
    return jsonify({"ok": True, "job_id": job_id})


@app.route("/api/run-training", methods=["POST"])
def api_run_training():
    """Kick off ml/train.py in a background thread and return a job id immediately."""
    if not (PROCESSED / "junction_pdi.csv").exists():
        return jsonify({"ok": False, "error": "Run the pipeline first — no processed data found."}), 400
    if is_job_running():
        return jsonify({"ok": False, "error": "Another job is already running. Wait for it to finish."}), 409

    job_id = uuid.uuid4().hex
    thread = threading.Thread(
        target=_run_script_job, args=(job_id, "ml/train.py", 300), daemon=True
    )
    thread.start()
    return jsonify({"ok": True, "job_id": job_id})


@app.route("/api/job-status/<job_id>")
def api_job_status(job_id):
    """Poll the status of a background pipeline/training job."""
    with JOBS_LOCK:
        job = JOBS.get(job_id)
    if job is None:
        return jsonify({"status": "not_found"}), 404
    return jsonify(job)


# ─────────────────────────────────────────────
# API — MOBILE (IN-MEMORY STATE)
# ─────────────────────────────────────────────

@app.route("/api/claims", methods=["GET"])
def api_get_claims():
    return jsonify(mobile_state["claims"])


@app.route("/api/claim", methods=["POST"])
def api_claim():
    data     = request.json or {}
    slot_key = data.get("slot_key")
    station  = data.get("station")
    officer  = data.get("officer")
    if not slot_key or not station:
        return jsonify({"ok": False, "error": "Missing slot_key or station"}), 400

    if slot_key in mobile_state["claims"]:
        existing = mobile_state["claims"][slot_key]
        if existing["station"] != station:
            return jsonify({"ok": False, "error": "Already claimed", "by": existing["station"]})

    mobile_state["claims"][slot_key] = {
        "station":    station,
        "officer":    officer,
        "claimed_at": datetime.now().isoformat(),
    }
    return jsonify({"ok": True})


@app.route("/api/unclaim", methods=["POST"])
def api_unclaim():
    data     = request.json or {}
    slot_key = data.get("slot_key")
    station  = data.get("station")
    if slot_key in mobile_state["claims"]:
        if mobile_state["claims"][slot_key]["station"] == station:
            del mobile_state["claims"][slot_key]
            return jsonify({"ok": True})
        return jsonify({"ok": False, "error": "Not claimed by you"}), 403
    return jsonify({"ok": True})


@app.route("/api/manpower", methods=["POST"])
def api_manpower():
    data    = request.json or {}
    station = data.get("station")
    if station:
        mobile_state["manpower"][station] = data
    return jsonify({"ok": True})


@app.route("/api/handover", methods=["GET", "POST"])
def api_handover():
    if request.method == "POST":
        data        = request.json or {}
        key         = f"{data.get('station')}::{data.get('shift')}"
        data["time"] = datetime.now().isoformat()
        mobile_state["handovers"][key] = data
        return jsonify({"ok": True})
    else:
        station = request.args.get("station")
        shift   = request.args.get("shift")
        key     = f"{station}::{shift}"
        return jsonify(mobile_state["handovers"].get(key, {}))


@app.route("/api/task-status", methods=["POST"])
def api_task_status():
    data     = request.json or {}
    slot_key = data.get("slot_key")
    if slot_key:
        mobile_state["task_status"][slot_key] = data
    return jsonify({"ok": True})


@app.route("/api/alerts", methods=["GET"])
def api_alerts():
    station = request.args.get("station")
    alerts  = generate_mock_alerts(station)
    return jsonify(alerts)


# ─────────────────────────────────────────────
# ENTRY POINT
# ─────────────────────────────────────────────

if __name__ == "__main__":
    if not (PROCESSED / "junction_pdi.csv").exists():
        print("ERROR: Run pipeline.py first!")
        exit(1)
    if not (ML_DIR / "metrics.json").exists():
        print("WARNING: ML metrics not found. Run ml/train.py for full features.")
    print("\n  ClearPath starting on http://localhost:5000\n")
    # use_reloader=False is required: the auto-reloader watches the working
    # directory, and pipeline.py / ml/train.py write into data/processed/
    # and ml/ while running as subprocesses. A file-change-triggered
    # restart mid-run kills the in-flight background job. threaded=True
    # lets /api/job-status polling requests return immediately instead of
    # queueing behind the long-running pipeline/training thread.
    app.run(debug=True, use_reloader=False, threaded=True, host="0.0.0.0", port=5000)