fu4ll/Vehicle_Detection

1from fastapi import FastAPI, UploadFile, File2from fastapi.staticfiles import StaticFiles3from fastapi.responses import HTMLResponse, FileResponse4import os5import shutil6from detector import detect as run_detect, detect_video7from database import SessionLocal, Detection, init_db8 9app = FastAPI()10 11os.makedirs("uploads", exist_ok=True)12os.makedirs("outputs", exist_ok=True)13app.mount("/static", StaticFiles(directory="static"), name="static")14 15@app.on_event("startup")16def startup():17    init_db()18 19@app.post("/detect")20async def detect_endpoint(file: UploadFile = File(...)):21    file_path = f"uploads/{file.filename}"22    with open(file_path, "wb") as buffer:23        shutil.copyfileobj(file.file, buffer)24 25    detections, img_base64 = run_detect(file_path)26 27    db = SessionLocal()28    record = Detection(29        filename=file.filename,30        type="image",31        total=len(detections),32        detections=detections33    )34    db.add(record)35    db.commit()36    db.refresh(record)37    db.close()38 39    return {40        "id": record.id,41        "filename": file.filename,42        "total": len(detections),43        "detections": detections,44        "image": img_base64,45    }46 47@app.post("/detect/video")48async def detect_video_endpoint(file: UploadFile = File(...)):49    input_path  = f"uploads/{file.filename}"50    output_path = f"outputs/annotated_{file.filename}"51 52    with open(input_path, "wb") as buffer:53        shutil.copyfileobj(file.file, buffer)54 55    stats, converted_path = detect_video(input_path, output_path)56    video_filename = os.path.basename(converted_path)57 58    db = SessionLocal()59    record = Detection(60        filename=file.filename,61        type="video",62        total=stats["unique_vehicles"],63        detections=stats64    )65    db.add(record)66    db.commit()67    db.refresh(record)68    db.close()69 70    return {71        "id": record.id,72        "filename": file.filename,73        "output_video": f"/video/{video_filename}",74        "stats": stats75    }76 77@app.get("/history")78async def get_history():79    db = SessionLocal()80    records = db.query(Detection).order_by(Detection.created_at.desc()).limit(50).all()81    db.close()82    return [83        {84            "id": r.id,85            "filename": r.filename,86            "type": r.type,87            "total": r.total,88            "created_at": r.created_at89        }90        for r in records91    ]92 93@app.get("/history/{detection_id}")94async def get_detection(detection_id: int):95    db = SessionLocal()96    record = db.query(Detection).filter(Detection.id == detection_id).first()97    db.close()98    if not record:99        return {"error": "Not found"}100    return record101 102@app.get("/video/{filename}")103async def get_video(filename: str):104    path = f"outputs/{filename}"105    return FileResponse(path, media_type="video/mp4")106 107@app.get("/", response_class=HTMLResponse)108async def root():109    with open("static/index.html") as f:110        return f.read()