app.py

import os
from dotenv import load_dotenv
load_dotenv()

import logging

# Fix: Set Hugging Face cache to writable location
# In containerized environments, /.cache may not be writable
if "HF_HOME" not in os.environ:
    os.environ["HF_HOME"] = "/tmp/huggingface"
    print(f"Set HF_HOME to {os.environ['HF_HOME']}")

# Debug/Fix: Unset CUDA_VISIBLE_DEVICES to ensure all GPUs are visible
# Some environments (like HF Spaces) might set this to "0" by default.
if "CUDA_VISIBLE_DEVICES" in os.environ:
    # Use print because logging config might not be set yet
    print(f"Found CUDA_VISIBLE_DEVICES={os.environ['CUDA_VISIBLE_DEVICES']}. Unsetting it to enable all GPUs.")
    del os.environ["CUDA_VISIBLE_DEVICES"]
else:
    print("CUDA_VISIBLE_DEVICES not set. All GPUs should be visible.")

import torch
try:
    print(f"Startup Diagnostics: Torch version {torch.__version__}, CUDA available: {torch.cuda.is_available()}, Device count: {torch.cuda.device_count()}")
except Exception as e:
    print(f"Startup Diagnostics Error: {e}")

import asyncio
import shutil
import tempfile
import uuid
from contextlib import asynccontextmanager
from datetime import timedelta
from pathlib import Path

import cv2
import numpy as np
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
import uvicorn

from inference import process_first_frame, run_inference, run_segmentation
from models.depth_estimators.model_loader import list_depth_estimators
from jobs.background import process_video_async
from jobs.models import JobInfo, JobStatus
from jobs.streaming import get_stream
from jobs.storage import (
    get_depth_output_path,
    get_first_frame_depth_path,
    get_first_frame_path,
    get_input_video_path,
    get_job_directory,
    get_job_storage,
    get_output_video_path,
)
from utils.gpt_distance import estimate_distance_gpt

logging.basicConfig(level=logging.INFO)

# Suppress noisy external libraries
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("huggingface_hub").setLevel(logging.WARNING)
logging.getLogger("transformers").setLevel(logging.WARNING)



async def _periodic_cleanup() -> None:
    while True:
        await asyncio.sleep(600)
        get_job_storage().cleanup_expired(timedelta(hours=1))


@asynccontextmanager
async def lifespan(_: FastAPI):
    cleanup_task = asyncio.create_task(_periodic_cleanup())
    try:
        yield
    finally:
        cleanup_task.cancel()


app = FastAPI(title="Video Object Detection", lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


from fastapi import Request

@app.middleware("http")
async def add_no_cache_header(request: Request, call_next):
    """Ensure frontend assets are not cached by the browser (important for HF Spaces updates)."""
    response = await call_next(request)
    # Apply to all static files and the root page
    if request.url.path.startswith("/laser") or request.url.path == "/":
        response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
        response.headers["Pragma"] = "no-cache"
        response.headers["Expires"] = "0"
    return response

# Optional: serve the LaserPerception frontend from this backend.
# The frontend files are now located in the 'frontend' directory.
_FRONTEND_DIR = Path(__file__).with_name("frontend")
if _FRONTEND_DIR.exists():
    # Mount the entire frontend directory at /laser (legacy path) or /frontend
    app.mount("/laser", StaticFiles(directory=_FRONTEND_DIR, html=True), name="laser")

# Valid detection modes
VALID_MODES = {"object_detection", "segmentation", "drone_detection"}


def _save_upload_to_tmp(upload: UploadFile) -> str:
    """Save uploaded file to temporary location."""
    suffix = Path(upload.filename or "upload.mp4").suffix or ".mp4"
    fd, path = tempfile.mkstemp(prefix="input_", suffix=suffix, dir="/tmp")
    os.close(fd)
    with open(path, "wb") as buffer:
        data = upload.file.read()
        buffer.write(data)
    return path


def _save_upload_to_path(upload: UploadFile, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "wb") as buffer:
        data = upload.file.read()
        buffer.write(data)


def _safe_delete(path: str) -> None:
    """Safely delete a file, ignoring errors."""
    try:
        os.remove(path)
    except FileNotFoundError:
        return
    except Exception:
        logging.exception("Failed to remove temporary file: %s", path)


def _schedule_cleanup(background_tasks: BackgroundTasks, path: str) -> None:
    """Schedule file cleanup after response is sent."""
    def _cleanup(target: str = path) -> None:
        _safe_delete(target)

    background_tasks.add_task(_cleanup)


def _default_queries_for_mode(mode: str) -> list[str]:
    if mode == "segmentation":
        return ["object"]
    if mode == "drone_detection":
        return ["drone"]
    return ["person", "car", "truck", "motorcycle", "bicycle", "bus", "train", "airplane"]


@app.get("/", response_class=HTMLResponse)
async def demo_page():
    """Redirect to LaserPerception app."""
    # The main entry point is now index.html in the mounted directory
    return RedirectResponse(url="/laser/index.html")


@app.post("/detect")
async def detect_endpoint(
    background_tasks: BackgroundTasks,
    video: UploadFile = File(...),
    mode: str = Form(...),
    queries: str = Form(""),
    detector: str = Form("hf_yolov8"),
    segmenter: str = Form("sam3"),
    enable_depth: bool = Form(False),
    enable_gpt: bool = Form(True),
):
    """
    Main detection endpoint.

    Args:
        video: Video file to process
        mode: Detection mode (object_detection, segmentation, drone_detection)
        queries: Comma-separated object classes for object_detection mode
        detector: Model to use (hf_yolov8, detr_resnet50, grounding_dino)
        segmenter: Segmentation model to use (sam3)
        enable_depth: Whether to run legacy depth estimation (default: False)
        drone_detection uses the dedicated drone_yolo model.

    Returns:
        - For object_detection: Processed video with bounding boxes
        - For segmentation: Processed video with masks rendered
        - For drone_detection: Processed video with bounding boxes
    """
    # Validate mode
    if mode not in VALID_MODES:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid mode '{mode}'. Must be one of: {', '.join(VALID_MODES)}"
        )

    if mode == "segmentation":
        if video is None:
            raise HTTPException(status_code=400, detail="Video file is required.")

        try:
            input_path = _save_upload_to_tmp(video)
        except Exception:
            logging.exception("Failed to save uploaded file.")
            raise HTTPException(status_code=500, detail="Failed to save uploaded video.")
        finally:
            await video.close()

        fd, output_path = tempfile.mkstemp(prefix="output_", suffix=".mp4", dir="/tmp")
        os.close(fd)

        # Parse queries
        query_list = [q.strip() for q in queries.split(",") if q.strip()]
        if not query_list:
            query_list = ["object"]

        try:
            output_path = run_segmentation(
                input_path,
                output_path,
                query_list,
                segmenter_name=segmenter,
            )
        except ValueError as exc:
            logging.exception("Segmentation processing failed.")
            _safe_delete(input_path)
            _safe_delete(output_path)
            raise HTTPException(status_code=500, detail=str(exc))
        except Exception as exc:
            logging.exception("Segmentation inference failed.")
            _safe_delete(input_path)
            _safe_delete(output_path)
            return JSONResponse(status_code=500, content={"error": str(exc)})

        _schedule_cleanup(background_tasks, input_path)
        _schedule_cleanup(background_tasks, output_path)

        return FileResponse(
            path=output_path,
            media_type="video/mp4",
            filename="segmented.mp4",
        )

    # Handle object detection or drone detection mode
    if video is None:
        raise HTTPException(status_code=400, detail="Video file is required.")

    # Save uploaded video
    try:
        input_path = _save_upload_to_tmp(video)
    except Exception:
        logging.exception("Failed to save uploaded file.")
        raise HTTPException(status_code=500, detail="Failed to save uploaded video.")
    finally:
        await video.close()

    # Create output path
    fd, output_path = tempfile.mkstemp(prefix="output_", suffix=".mp4", dir="/tmp")
    os.close(fd)

    # Parse queries
    query_list = [q.strip() for q in queries.split(",") if q.strip()]
    if mode == "drone_detection" and not query_list:
        query_list = ["drone"]

    # Run inference
    try:
        detector_name = "drone_yolo" if mode == "drone_detection" else detector
        
        # Determine depth estimator
        active_depth = "depth" if enable_depth else None

        output_path, _ = run_inference(
            input_path,
            output_path,
            query_list,
            detector_name=detector_name,
            depth_estimator_name=active_depth,
            depth_scale=25.0,
            enable_gpt=enable_gpt,
        )
    except ValueError as exc:
        logging.exception("Video processing failed.")
        _safe_delete(input_path)
        _safe_delete(output_path)
        raise HTTPException(status_code=500, detail=str(exc))
    except Exception as exc:
        logging.exception("Inference failed.")
        _safe_delete(input_path)
        _safe_delete(output_path)
        return JSONResponse(status_code=500, content={"error": str(exc)})

    # Schedule cleanup
    _schedule_cleanup(background_tasks, input_path)
    _schedule_cleanup(background_tasks, output_path)

    # Return processed video
    response = FileResponse(
        path=output_path,
        media_type="video/mp4",
        filename="processed.mp4",
    )
    return response


@app.post("/detect/async")
async def detect_async_endpoint(
    video: UploadFile = File(...),
    mode: str = Form(...),
    queries: str = Form(""),
    detector: str = Form("hf_yolov8"),
    segmenter: str = Form("sam3"),
    depth_estimator: str = Form("depth"),
    depth_scale: float = Form(25.0),
    enable_depth: bool = Form(False),
    enable_gpt: bool = Form(True),
):
    if mode not in VALID_MODES:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid mode '{mode}'. Must be one of: {', '.join(VALID_MODES)}",
        )

    if video is None:
        raise HTTPException(status_code=400, detail="Video file is required.")

    job_id = uuid.uuid4().hex
    job_dir = get_job_directory(job_id)
    input_path = get_input_video_path(job_id)
    output_path = get_output_video_path(job_id)
    first_frame_path = get_first_frame_path(job_id)
    depth_output_path = get_depth_output_path(job_id)
    first_frame_depth_path = get_first_frame_depth_path(job_id)

    try:
        _save_upload_to_path(video, input_path)
    except Exception:
        logging.exception("Failed to save uploaded file.")
        raise HTTPException(status_code=500, detail="Failed to save uploaded video.")
    finally:
        await video.close()

    query_list = [q.strip() for q in queries.split(",") if q.strip()]
    if not query_list:
        query_list = _default_queries_for_mode(mode)

    available_depth_estimators = set(list_depth_estimators())
    if depth_estimator not in available_depth_estimators:
        raise HTTPException(
            status_code=400,
            detail=(
                f"Invalid depth estimator '{depth_estimator}'. "
                f"Must be one of: {', '.join(sorted(available_depth_estimators))}"
            ),
        )

    detector_name = detector
    if mode == "drone_detection":
        detector_name = "drone_yolo"
        
    # Determine actve depth estimator (Legacy)
    active_depth = depth_estimator if enable_depth else None

    try:
        processed_frame, detections, depth_map = process_first_frame(
            str(input_path),
            query_list,
            mode=mode,
            detector_name=detector_name,
            segmenter_name=segmenter,
            depth_estimator_name=active_depth,
            depth_scale=depth_scale,
            enable_depth_estimator=enable_depth,
            enable_gpt=enable_gpt,
        )
        cv2.imwrite(str(first_frame_path), processed_frame)
        
        if depth_map is not None:
             # Simple visualization: Normalize and apply colormap
             try:
                 d_min, d_max = np.min(depth_map), np.max(depth_map)
                 if d_max - d_min > 1e-6:
                     d_norm = (depth_map - d_min) / (d_max - d_min)
                 else:
                     d_norm = np.zeros_like(depth_map)
                 d_uint8 = (d_norm * 255).astype(np.uint8)
                 d_color = cv2.applyColorMap(d_uint8, cv2.COLORMAP_INFERNO)
                 cv2.imwrite(str(first_frame_depth_path), d_color)
             except Exception as e:
                 logging.warning(f"Failed to save depth map: {e}")
    except Exception:
        logging.exception("First-frame processing failed.")
        shutil.rmtree(job_dir, ignore_errors=True)
        raise HTTPException(status_code=500, detail="Failed to process first frame.")

    job = JobInfo(
        job_id=job_id,
        status=JobStatus.PROCESSING,
        mode=mode,
        queries=query_list,
        detector_name=detector_name,
        segmenter_name=segmenter,
        input_video_path=str(input_path),
        output_video_path=str(output_path),
        first_frame_path=str(first_frame_path),
        first_frame_detections=detections,
        depth_estimator_name=active_depth,
        depth_scale=float(depth_scale),
        depth_output_path=str(depth_output_path),
        first_frame_depth_path=str(first_frame_depth_path),
        enable_gpt=enable_gpt,
    )
    get_job_storage().create(job)
    asyncio.create_task(process_video_async(job_id))

    return {
        "job_id": job_id,
        "first_frame_url": f"/detect/first-frame/{job_id}",
        "first_frame_depth_url": f"/detect/first-frame-depth/{job_id}",
        "status_url": f"/detect/status/{job_id}",
        "video_url": f"/detect/video/{job_id}",
        "depth_video_url": f"/detect/depth-video/{job_id}",
        "job_id": job_id,
        "first_frame_url": f"/detect/first-frame/{job_id}",
        "first_frame_depth_url": f"/detect/first-frame-depth/{job_id}",
        "status_url": f"/detect/status/{job_id}",
        "video_url": f"/detect/video/{job_id}",
        "depth_video_url": f"/detect/depth-video/{job_id}",
        "stream_url": f"/detect/stream/{job_id}",
        "status": job.status.value,
        "first_frame_detections": detections,
    }


@app.get("/detect/status/{job_id}")
async def detect_status(job_id: str):
    job = get_job_storage().get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found or expired.")
    return {
        "job_id": job.job_id,
        "status": job.status.value,
        "created_at": job.created_at.isoformat(),
        "completed_at": job.completed_at.isoformat() if job.completed_at else None,
        "error": job.error,
    }


@app.get("/detect/tracks/{job_id}/{frame_idx}")
async def get_frame_tracks(job_id: str, frame_idx: int):
    """Retrieve detections (with tracking info) for a specific frame."""
    # This requires us to store detections PER FRAME in JobStorage or similar.
    # Currently, inference.py returns 'sorted_detections' at the end.
    # But during streaming, where is it?
    # We can peek into the 'stream_queue' logic or we need a shared store.
    # Ideally, inference should write to a map/db that we can read.
    
    # Quick fix: If job is done, we might have it. If running, it's harder absent a DB.
    # BUT, 'stream_queue' sends frames.
    # Let's use a global cache in memory for active jobs?
    # See inference.py: 'all_detections_map' is local to that function.
    
    # BETTER APPROACH for this demo:
    # Use a simple shared dictionary in jobs/storage.py or app.py used by inference.
    # We will pass a callback or shared dict to run_inference.
    
    # For now, let's just return 404 if not implemented, but I need to implement it.
    # I'll add a cache in app.py for active job tracks?
    from jobs.storage import get_track_data
    data = get_track_data(job_id, frame_idx)
    return data or []


@app.delete("/detect/job/{job_id}")
async def cancel_job(job_id: str):
    """Cancel a running job."""
    job = get_job_storage().get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found or expired.")
    if job.status != JobStatus.PROCESSING:
        return {
            "message": f"Job already {job.status.value}",
            "status": job.status.value,
        }

    get_job_storage().update(job_id, status=JobStatus.CANCELLED)
    return {
        "message": "Job cancellation requested",
        "status": "cancelled",
    }


@app.get("/detect/first-frame/{job_id}")
async def detect_first_frame(job_id: str):
    job = get_job_storage().get(job_id)
    if not job or not Path(job.first_frame_path).exists():
        raise HTTPException(status_code=404, detail="First frame not found.")
    return FileResponse(
        path=job.first_frame_path,
        media_type="image/jpeg",
        filename="first_frame.jpg",
    )


@app.get("/detect/video/{job_id}")
async def detect_video(job_id: str):
    job = get_job_storage().get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found or expired.")
    if job.status == JobStatus.FAILED:
        raise HTTPException(status_code=500, detail=f"Job failed: {job.error}")
    if job.status == JobStatus.CANCELLED:
        raise HTTPException(status_code=410, detail="Job was cancelled")
    if job.status == JobStatus.PROCESSING:
        return JSONResponse(
            status_code=202,
            content={"detail": "Video still processing", "status": "processing"},
        )
    if not job.output_video_path or not Path(job.output_video_path).exists():
        raise HTTPException(status_code=404, detail="Video file not found.")
    return FileResponse(
        path=job.output_video_path,
        media_type="video/mp4",
        filename="processed.mp4",
    )


@app.get("/detect/depth-video/{job_id}")
async def detect_depth_video(job_id: str):
    """Return depth estimation video."""
    job = get_job_storage().get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found or expired.")
    if not job.depth_output_path:
        # Check if depth failed (partial success)
        if job.partial_success and job.depth_error:
            raise HTTPException(status_code=404, detail=f"Depth unavailable: {job.depth_error}")
        raise HTTPException(status_code=404, detail="No depth video for this job.")
    if job.status == JobStatus.FAILED:
        raise HTTPException(status_code=500, detail=f"Job failed: {job.error}")
    if job.status == JobStatus.CANCELLED:
        raise HTTPException(status_code=410, detail="Job was cancelled")
    if job.status == JobStatus.PROCESSING:
        return JSONResponse(
            status_code=202,
            content={"detail": "Video still processing", "status": "processing"},
        )
    if not Path(job.depth_output_path).exists():
        raise HTTPException(status_code=404, detail="Depth video file not found.")
    return FileResponse(
        path=job.depth_output_path,
        media_type="video/mp4",
        filename="depth.mp4",
    )


@app.get("/detect/first-frame-depth/{job_id}")
async def detect_first_frame_depth(job_id: str):
    """Return first frame depth visualization."""
    job = get_job_storage().get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found or expired.")
    if not job.first_frame_depth_path:
        # Return placeholder or error if depth not available
        if job.partial_success and job.depth_error:
            raise HTTPException(status_code=404, detail=f"Depth unavailable: {job.depth_error}")
        raise HTTPException(status_code=404, detail="First frame depth not found.")
    if not Path(job.first_frame_depth_path).exists():
        raise HTTPException(status_code=404, detail="First frame depth file not found.")
    return FileResponse(
        path=job.first_frame_depth_path,
        media_type="image/jpeg",
        filename="first_frame_depth.jpg",
    )


@app.get("/detect/stream/{job_id}")
async def stream_video(job_id: str):
    """MJPEG stream of the processing video (optimized)."""
    import queue
    
    async def stream_generator():
        loop = asyncio.get_running_loop()
        buffered = False
        
        while True:
            q = get_stream(job_id)
            if not q:
                break
                
            try:
                # Initial Buffer: Wait until we have enough frames or job is done
                if not buffered:
                    if q.qsize() < 30:
                        # If queue is empty, wait a bit
                        await asyncio.sleep(0.1)
                        # Check if job is still running? For now just wait for buffer or stream close
                        continue
                    buffered = True

                # Get ONE frame (no skipping)
                # Use wait to allow generator to yield cleanly
                try:
                    # Blocking get in executor to avoid hanging async loop? 
                    # Actually standard queue.get() is blocking. get_nowait is not.
                    # We can sleep-poll for async compatibility
                    while q.empty():
                        await asyncio.sleep(0.01)
                        if not get_stream(job_id): # Stream closed
                            return

                    frame = q.get_nowait()
                except queue.Empty:
                    continue
                
                # Resize if too big (e.g. > 640 width)
                # Optimization: Only resize if needed
                h, w = frame.shape[:2]
                if w > 640:
                    scale = 640 / w
                    new_h = int(h * scale)
                    frame = cv2.resize(frame, (640, new_h), interpolation=cv2.INTER_NEAREST)

                # Encode in thread
                # JPEG Quality = 60 (Better quality for smooth video)
                encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 60]
                success, buffer = await loop.run_in_executor(None, cv2.imencode, '.jpg', frame, encode_param)
                
                if success:
                    yield (b'--frame\r\n'
                           b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')
                
                # Control playback speed? 
                # If we blast frames as fast as possible, it might play accelerated.
                # Ideally we want to sync to ~30fps.
                await asyncio.sleep(0.033) # Simple pacer (~30fps)

            except Exception:
                await asyncio.sleep(0.1)

    return StreamingResponse(
        stream_generator(), 
        media_type="multipart/x-mixed-replace; boundary=frame"
    )


@app.post("/reason/track")
async def reason_track(
    frame: UploadFile = File(...),
    tracks: str = Form(...)  # JSON string of tracks: [{"id": "T01", "bbox": [x,y,w,h], "label": "car"}, ...]
):
    """
    Reason about specific tracks in a frame using GPT.
    Returns distance and description for each object ID.
    """
    import json
    try:
        input_path = _save_upload_to_tmp(frame)
    except Exception:
        raise HTTPException(status_code=500, detail="Failed to save uploaded frame")

    try:
        track_list = json.loads(tracks)
    except json.JSONDecodeError:
        _safe_delete(input_path)
        raise HTTPException(status_code=400, detail="Invalid tracks JSON")

    # Run GPT estimation
    # This is blocking, but that's expected for this endpoint structure. 
    # For high concurrency, might want to offload to threadpool or async wrapper.
    try:
        # estimate_distance_gpt reads the file from disk
        results = await asyncio.to_thread(estimate_distance_gpt, input_path, track_list)
        logging.info(f"GPT Output for Video Track Update:\n{results}")
    except Exception as e:
        logging.exception("GPT reasoning failed")
        _safe_delete(input_path)
        raise HTTPException(status_code=500, detail=str(e))
    
    _safe_delete(input_path)
    return results



if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)