Spaces:
Sleeping
Sleeping
File size: 2,372 Bytes
8e6f164 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | import io
from fastapi import APIRouter, File, UploadFile, HTTPException, Request
from PIL import Image
from api.services.analyzer_service import IncidentAnalyzer
router = APIRouter(prefix="/api/v1/incidents", tags=["Incidents"])
@router.post("/analyze", response_model=dict)
async def analyze_incident_image(request: Request, file: UploadFile = File(...)):
"""
Accepts an emergency incident image, runs zero-shot object detection using
Grounding DINO, and computes an incident type and severity score.
"""
# Validate uploaded file type
if not file.content_type or not file.content_type.startswith("image/"):
raise HTTPException(
status_code=400,
detail="Invalid file format. Please upload an image file."
)
try:
# Read file contents and open as PIL Image
file_bytes = await file.read()
image = Image.open(io.BytesIO(file_bytes)).convert("RGB")
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Failed to process image file: {str(e)}"
)
# Get Grounding DINO service instance from app state
dino_service = getattr(request.app.state, "dino_service", None)
if dino_service is None:
raise HTTPException(
status_code=503,
detail="Model service is currently initializing. Please try again shortly."
)
try:
# Run inference
detections = dino_service.detect(image)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Error executing object detection: {str(e)}"
)
# Count frequencies of each detected label
counts = {}
for detection in detections:
label = detection["label"]
counts[label] = counts.get(label, 0) + 1
# Extract unique labels list
keywords = list(counts.keys())
# Analyze incident characteristics (severity and type classification)
analysis_result = IncidentAnalyzer.analyze(keywords)
# Return structured API response
return {
"success": True,
"incident_type": analysis_result["incident_type"],
"severity": analysis_result["severity"],
"severity_score": analysis_result["severity_score"],
"keywords": keywords,
"counts": counts,
"detections": detections
}
|