Blame for dashboard_api.py

"""FastAPI dashboard API — REST + WebSocket.

Start with:
    uvicorn dashboard_api:app --host 0.0.0.0 --port 8000 --reload

Or run alongside the traffic controller:
    python dashboard_api.py
"""

from __future__ import annotations

import asyncio
import json
import os
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
SRC = ROOT / "src"
for p in (str(SRC), str(ROOT)):
    if p not in sys.path:
        sys.path.insert(0, p)

from contextlib import asynccontextmanager
from typing import Optional

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from dashboard_state import state
from dashboard_runner import start_simulation, stop_simulation, is_simulation_running
from video_analysis import (
    get_sample_videos,
    process_video,
    is_video_processing_active,
    get_current_job_status,
    UPLOADS_DIR,
    analyze_perception_frames,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    _seed_from_latest_jsonl()
    broadcast_task = asyncio.create_task(_broadcast_loop())
    yield
    broadcast_task.cancel()


app = FastAPI(title="AI Traffic Adaptation Dashboard API", version="1.0.0", lifespan=lifespan)


# -----------------------------------------------------------------------
# Seed state from the most recent JSONL run log (so dashboard is never empty)
# -----------------------------------------------------------------------

def _seed_from_latest_jsonl():
    """Load the most recent runtime.jsonl into DashboardState."""
    runs_dir = ROOT / "results" / "final_runs"
    if not runs_dir.exists():
        return

    # Find newest runtime.jsonl
    jsonl_files = sorted(runs_dir.glob("*/runtime.jsonl"), key=lambda p: p.stat().st_mtime)
    if not jsonl_files:
        return
    latest = jsonl_files[-1]

    decisions = []
    junctions: dict = {}
    logs = []
    max_time = 0.0
    mode = "normal"
    has_emergency = False

    with open(latest, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                record = json.loads(line)
            except json.JSONDecodeError:
                continue

            event = record.get("event")
            logs.append(record)

            if event == "system_start":
                mode = record.get("mode", "normal")

            elif event == "decision":
                decisions.append(record)
                sim_time = float(record.get("time", 0))
                if sim_time > max_time:
                    max_time = sim_time

                jid = record.get("junction")
                if jid:
                    safety = record.get("safety", {})
                    executed = record.get("executed", {})
                    if record.get("emergency_active"):
                        has_emergency = True
                    junctions[jid] = {
                        "junction_id": jid,
                        "current_phase_index": executed.get("phase"),
                        "current_phase_state": executed.get("state"),
                        "remaining_time": None,
                        "last_decision": record.get("selected"),
                        "decision_method": record.get("method"),
                        "safety_approved": safety.get("approved"),
                        "safety_checks": safety.get("checks"),
                        "emergency_active": record.get("emergency_active", False),
                        "emergency_vehicle_id": record.get("emergency_vehicle_id"),
                        "simulation_time": sim_time,
                    }

    # Populate state (most recent first)
    for record in reversed(decisions):
        state.add_decision(record)
    for jid, jdata in junctions.items():
        state.update_junction(jid, jdata)
    for log in reversed(logs):
        state.logs.appendleft(log)

    # Derive summary stats from decisions
    emergency_count = sum(1 for d in decisions if d.get("emergency_active"))
    state.update_system(
        status="stopped",
        sim_time=max_time,
        mode=mode,
        prediction_available=True,
    )
    state.emergency_vehicles = (
        [{"vehicle_id": "ambulance (historical)", "source": "log"}]
        if has_emergency else []
    )
    state.started_at = logs[0].get("timestamp") if logs else None
    print(f"[dashboard] Seeded from {latest.name}: {len(decisions)} decisions, "
          f"{len(junctions)} junctions, sim_time={max_time}")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# -----------------------------------------------------------------------
# WebSocket connection manager
# -----------------------------------------------------------------------

class ConnectionManager:
    def __init__(self):
        self._clients: list[WebSocket] = []
        self._lock = asyncio.Lock()

    async def connect(self, ws: WebSocket):
        await ws.accept()
        async with self._lock:
            self._clients.append(ws)

    async def disconnect(self, ws: WebSocket):
        async with self._lock:
            if ws in self._clients:
                self._clients.remove(ws)

    async def broadcast(self, data: dict):
        message = json.dumps(data)
        async with self._lock:
            dead = []
            for ws in self._clients:
                try:
                    await ws.send_text(message)
                except Exception:
                    dead.append(ws)
            for ws in dead:
                self._clients.remove(ws)


manager = ConnectionManager()


# -----------------------------------------------------------------------
# Background task — push live state to all WebSocket clients
# -----------------------------------------------------------------------

async def _broadcast_loop():
    while True:
        await asyncio.sleep(1.0)
        if manager._clients:
            payload = {
                "type": "live_update",
                "status": state.snapshot(),
                "junctions": state.junctions_snapshot(),
                "emergencies": state.emergencies_snapshot(),
                "safety": state.safety_snapshot(),
                "latest_decision": state.decisions_snapshot()[:1],
                "logs": state.logs_snapshot()[:20],
            }
            await manager.broadcast(payload)


# Lifespan context manager is registered at app instantiation.


# -----------------------------------------------------------------------
# REST endpoints
# -----------------------------------------------------------------------

@app.get("/api/status")
def get_status():
    return state.snapshot()


@app.get("/api/junctions")
def get_junctions():
    return state.junctions_snapshot()


@app.get("/api/junctions/{junction_id}")
def get_junction(junction_id: str):
    data = state.junction_snapshot(junction_id)
    if data is None:
        return JSONResponse(status_code=404, content={"detail": "Junction not found"})
    return data


@app.get("/api/decisions")
def get_decisions():
    return state.decisions_snapshot()


@app.get("/api/safety")
def get_safety():
    return state.safety_snapshot()


@app.get("/api/emergencies")
def get_emergencies():
    return state.emergencies_snapshot()


@app.get("/api/metrics")
def get_metrics():
    return state.metrics_snapshot()


@app.get("/api/logs")
def get_logs():
    return state.logs_snapshot()


@app.get("/api/comparison")
def get_comparison():
    return state.comparison_snapshot()


# -----------------------------------------------------------------------
# Perception & Simulation Control endpoints
# -----------------------------------------------------------------------

_perception_data = None
_perception_mtime = 0.0


def _get_perception_data():
    global _perception_data, _perception_mtime
    path = ROOT / "outputs" / "perception_output.json"
    if not path.exists():
        return None
    try:
        mtime = path.stat().st_mtime
        if _perception_data is None or mtime != _perception_mtime:
            with open(path, "r", encoding="utf-8") as f:
                _perception_data = json.load(f)
            _perception_mtime = mtime
    except Exception as exc:
        print(f"[dashboard_api] Error loading perception data: {exc}")
        return None
    return _perception_data


@app.get("/api/perception")
def get_perception(frame: Optional[int] = Query(None, description="Optional frame number")):
    data = _get_perception_data()
    if data is None:
        return JSONResponse(status_code=404, content={"detail": "Perception data not found"})

    total_frames = len(data)
    active_frames = [f for f in data if f.get("detections")]
    active_frame_numbers = [f["frame"] for f in active_frames]

    unique_ids = set()
    emergency_count = 0
    total_detections = 0
    for f in active_frames:
        for d in f.get("detections", []):
            total_detections += 1
            if "vehicle_id" in d:
                unique_ids.add(d["vehicle_id"])
            if d.get("emergency"):
                emergency_count += 1

    summary = {
        "total_frames": total_frames,
        "active_frames_count": len(active_frames),
        "unique_vehicles": len(unique_ids),
        "total_detections": total_detections,
        "emergency_detections": emergency_count,
    }

    if frame is not None:
        target = next((f for f in data if f.get("frame") == frame), None)
        if target is None:
            return JSONResponse(status_code=404, content={"detail": f"Frame {frame} not found"})
        return {
            "frame": frame,
            "total_frames": total_frames,
            "detections": target.get("detections", []),
            "summary": summary,
        }

    sample_frame = active_frames[-1] if active_frames else (data[0] if data else {"frame": 0, "detections": []})
    return {
        "summary": summary,
        "active_frame_numbers": active_frame_numbers[:100],
        "sample_frame": sample_frame,
    }


class SimulationStartRequest(BaseModel):
    steps: int = 200
    emergency: bool = False
    gui: bool = False


@app.get("/api/simulation/status")
def get_simulation_status():
    return {
        "is_running": is_simulation_running(),
        "system_status": state.system_status,
        "simulation_time": state.simulation_time,
        "simulation_mode": state.simulation_mode,
    }


@app.post("/api/simulation/start")
def post_simulation_start(req: SimulationStartRequest = SimulationStartRequest()):
    if is_simulation_running():
        return JSONResponse(status_code=400, content={"detail": "Simulation is already running"})
    started = start_simulation(emergency=req.emergency, steps=req.steps, gui=req.gui)
    return {"started": started, "steps": req.steps, "emergency": req.emergency, "gui": req.gui}


@app.post("/api/simulation/stop")
def post_simulation_stop():
    stopped = stop_simulation()
    return {"stopped": stopped}


# -----------------------------------------------------------------------
# Video Upload & Pipeline Analysis Endpoints
# -----------------------------------------------------------------------

_latest_video_analysis: dict | None = None


class AnalyzeSampleRequest(BaseModel):
    sample_path: str


@app.get("/api/video/samples")
def get_video_samples():
    """Return available footage from data/real_footage_test."""
    return get_sample_videos()


@app.get("/api/video/status")
def get_video_status():
    """Return current video pipeline processing status."""
    return {
        "is_processing": is_video_processing_active(),
        "job": get_current_job_status(),
        "has_analysis": _latest_video_analysis is not None,
    }


@app.get("/api/video/latest-analysis")
def get_latest_video_analysis():
    """Return the most recently computed video analysis."""
    global _latest_video_analysis
    if _latest_video_analysis is None:
        # Fall back to analyzing whatever is currently in outputs/perception_output.json
        data = _get_perception_data()
        if data:
            _latest_video_analysis = {
                "status": "cached",
                "filename": "outputs/perception_output.json",
                "analysis": analyze_perception_frames(data),
            }
    if _latest_video_analysis is None:
        return JSONResponse(status_code=404, content={"detail": "No video analysis available yet"})
    return _latest_video_analysis


@app.post("/api/video/analyze-sample")
def post_analyze_sample(req: AnalyzeSampleRequest):
    """Run perception pipeline on a selected sample video."""
    global _latest_video_analysis
    if is_video_processing_active():
        return JSONResponse(status_code=409, content={"detail": "Another video is currently being processed"})

    target = ROOT / req.sample_path
    if not target.exists():
        return JSONResponse(status_code=404, content={"detail": f"Sample file not found: {req.sample_path}"})

    try:
        result = process_video(target)
        _latest_video_analysis = result
        return result
    except Exception as exc:
        return JSONResponse(status_code=500, content={"detail": f"Video analysis failed: {str(exc)}"})


@app.post("/api/video/upload-and-analyze")
async def post_upload_and_analyze(file: UploadFile = File(...)):
    """Upload a video file from the browser, process it, and return full telemetry."""
    global _latest_video_analysis
    if is_video_processing_active():
        return JSONResponse(status_code=409, content={"detail": "Another video is currently being processed"})

    # Validate file extension
    ext = Path(file.filename).suffix.lower()
    if ext not in [".mp4", ".avi", ".mov", ".mkv"]:
        return JSONResponse(status_code=400, content={"detail": "Invalid video format. Use MP4, AVI, MOV, or MKV."})

    destination = UPLOADS_DIR / file.filename
    try:
        content = await file.read()
        with open(destination, "wb") as f:
            f.write(content)

        result = process_video(destination)
        _latest_video_analysis = result
        return result
    except Exception as exc:
        return JSONResponse(status_code=500, content={"detail": f"Video processing failed: {str(exc)}"})



# -----------------------------------------------------------------------
# WebSocket endpoint
# -----------------------------------------------------------------------

@app.websocket("/ws/live")
async def websocket_live(ws: WebSocket):
    await manager.connect(ws)
    # Send initial full state immediately on connect
    await ws.send_text(json.dumps({
        "type": "initial_state",
        "status": state.snapshot(),
        "junctions": state.junctions_snapshot(),
        "decisions": state.decisions_snapshot(),
        "emergencies": state.emergencies_snapshot(),
        "safety": state.safety_snapshot(),
        "metrics": state.metrics_snapshot(),
        "logs": state.logs_snapshot()[:50],
    }))
    try:
        while True:
            # Keep connection alive; client can send pings
            await ws.receive_text()
    except WebSocketDisconnect:
        await manager.disconnect(ws)
    except Exception:
        await manager.disconnect(ws)


# -----------------------------------------------------------------------
# Static Frontend Serving (Single-Port & Cloud Container Deployment)
# -----------------------------------------------------------------------

DIST_DIR = ROOT / "frontend" / "dist"
if DIST_DIR.exists():
    assets_dir = DIST_DIR / "assets"
    if assets_dir.exists():
        app.mount("/assets", StaticFiles(directory=str(assets_dir)), name="frontend_assets")

    @app.get("/{full_path:path}")
    async def serve_spa(full_path: str):
        # Never intercept API or WebSocket paths
        if full_path.startswith("api/") or full_path.startswith("ws/") or full_path == "api" or full_path == "ws":
            return JSONResponse(status_code=404, content={"detail": "Not found"})
        target = DIST_DIR / full_path
        if full_path and target.is_file():
            return FileResponse(target)
        index_file = DIST_DIR / "index.html"
        if index_file.exists():
            return FileResponse(index_file)
        return JSONResponse(status_code=404, content={"detail": "Frontend build not found"})


# -----------------------------------------------------------------------
# Entry point for standalone run
# -----------------------------------------------------------------------

if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 8000))
    uvicorn.run("dashboard_api:app", host="0.0.0.0", port=port, reload=False)