app.py

import logging
import os
import tempfile
from pathlib import Path
from typing import Optional

from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
import uvicorn

from inference import run_inference

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="Video Processing Backend")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["x-mission-summary"],
)


def _save_upload_to_tmp(upload: UploadFile) -> str:
    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 _safe_delete(path: str) -> None:
    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:
    def _cleanup(target: str = path) -> None:
        _safe_delete(target)

    background_tasks.add_task(_cleanup)


def _validate_inputs(video: UploadFile | None, prompt: str | None) -> None:
    if video is None:
        raise HTTPException(status_code=400, detail="Video file is required.")
    if not prompt:
        raise HTTPException(status_code=400, detail="Prompt is required.")


@app.post("/process_video")
async def process_video(
    background_tasks: BackgroundTasks,
    video: UploadFile = File(...),
    prompt: str = Form(...),
    detector: Optional[str] = Form(None),
):
    _validate_inputs(video, prompt)

    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)

    try:
        output_path, _, _ = run_inference(
            input_path,
            output_path,
            prompt,
            max_frames=10,
            detector_name=detector,
            generate_summary=False,
        )
    except ValueError as exc:
        logging.exception("Video decoding 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(background_tasks, input_path)
    _schedule_cleanup(background_tasks, output_path)

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


@app.post("/mission_summary")
async def mission_summary(
    video: UploadFile = File(...),
    prompt: str = Form(...),
    detector: Optional[str] = Form(None),
):
    _validate_inputs(video, prompt)
    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()

    try:
        _, mission_plan, mission_summary = run_inference(
            input_path,
            output_video_path=None,
            mission_prompt=prompt,
            max_frames=10,
            detector_name=detector,
            write_output_video=False,
            generate_summary=True,
        )
    except ValueError as exc:
        logging.exception("Video decoding failed.")
        _safe_delete(input_path)
        raise HTTPException(status_code=500, detail=str(exc))
    except Exception as exc:
        logging.exception("Summary generation failed.")
        _safe_delete(input_path)
        return JSONResponse(status_code=500, content={"error": str(exc)})

    _safe_delete(input_path)
    return {
        "mission_plan": mission_plan.to_dict(),
        "mission_summary": mission_summary or "",
    }

if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)
@app.get("/", response_class=HTMLResponse)
async def demo_page() -> str:
    demo_path = Path(__file__).with_name("demo.html")
    try:
        return demo_path.read_text(encoding="utf-8")
    except FileNotFoundError:
        return "<h1>Demo page missing</h1>"