Initial commit: FastAPI Incident Analyzer for HF Spaces

.gitignore

@@ -0,0 +1,6 @@
+.venv/
+__pycache__/
+*.pyc
+.env
+.ipynb_checkpoints/
+*.bat

Dockerfile

@@ -0,0 +1,26 @@
+FROM python:3.10-slim
+
+WORKDIR /code
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    libgl1-mesa-glx \
+    libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements and install dependencies
+COPY ./requirements.txt /code/requirements.txt
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+RUN pip install --no-cache-dir accelerate
+
+# Copy application files
+COPY ./api /code/api
+COPY ./app.py /code/app.py
+
+# Expose port 7860 (default port for Hugging Face Spaces)
+ENV PORT=7860
+EXPOSE 7860
+
+# Run the FastAPI server
+CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,55 @@
+---
+title: Emergency Incident Detection API
+emoji: 🚨
+colorFrom: red
+colorTo: blue
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# Emergency Incident Detection & Severity Analyzer API
+
+A REST API wrapper around Hugging Face's `grounding-dino-base` zero-shot object detection model for analyzing emergency incidents, blockages, and hazards.
+
+## API Endpoints
+
+### 1. Root Endpoint
+- **GET** `/`
+- Returns a welcome message.
+
+### 2. Health Check
+- **GET** `/health`
+- Returns the service health status and confirms whether the model has loaded successfully.
+
+### 3. Analyze Image
+- **POST** `/api/v1/incidents/analyze`
+- Upload an image file under key `file`.
+- **Response Schema**:
+```json
+{
+  "success": true,
+  "incident_type": "road_accident",
+  "severity": "high",
+  "severity_score": 78,
+  "keywords": [
+    "car",
+    "truck",
+    "ambulance",
+    "person"
+  ],
+  "counts": {
+    "car": 2,
+    "truck": 1,
+    "ambulance": 1,
+    "person": 4
+  },
+  "detections": [
+    {
+      "label": "car",
+      "confidence": 0.91,
+      "box": [10.2, 50.4, 250.7, 400.1]
+    }
+  ]
+}
+```

api/config.py

@@ -0,0 +1,30 @@
+import os
+
+class Settings:
+    MODEL_ID: str = os.getenv("MODEL_ID", "IDEA-Research/grounding-dino-base")
+    BOX_THRESHOLD: float = float(os.getenv("BOX_THRESHOLD", 0.35))
+    TEXT_THRESHOLD: float = float(os.getenv("TEXT_THRESHOLD", 0.25))
+    MIN_CONFIDENCE: float = float(os.getenv("MIN_CONFIDENCE", 0.45))
+    
+    TEXT_LABELS: list[str] = [
+        "car",
+        "truck",
+        "bus",
+        "motorcycle",
+        "person",
+        "ambulance",
+        "police vehicle",
+        "fire",
+        "smoke",
+        "tree",
+        "water",
+        "building",
+        "road",
+        "debris",
+        "damaged vehicle",
+        "traffic congestion",
+        "collapsed structure",
+        "construction barrier"
+    ]
+
+settings = Settings()

api/main.py

@@ -0,0 +1,53 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+import torch
+from api.services.dino_service import GroundingDinoService
+from api.routers.analysis import router as analysis_router
+
+# Define the lifespan context manager for startup and shutdown events
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Initialize the Grounding DINO Service (downloads model if not cached and loads onto device)
+    dino_service = GroundingDinoService()
+    app.state.dino_service = dino_service
+    yield
+    # Clean up and free device memory
+    if hasattr(app.state, "dino_service"):
+        del app.state.dino_service
+    if torch.cuda.is_available():
+        torch.cuda.empty_cache()
+
+# Create FastAPI application instance
+app = FastAPI(
+    title="Emergency Incident Detector & Analyzer API",
+    description="A REST API wrapper around Grounding DINO for real-time disaster and accident scene analysis.",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# Set up CORS middleware to support local testing and front-end integration
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include incident routers
+app.include_router(analysis_router)
+
+# Basic health-check endpoint
+@app.get("/health", tags=["Health"])
+def health_check():
+    return {
+        "status": "healthy",
+        "model_loaded": hasattr(app.state, "dino_service") and app.state.dino_service is not None
+    }
+
+@app.get("/", tags=["Root"])
+def read_root():
+    return {
+        "message": "Welcome to the Emergency Incident Detection API. Go to /docs for the interactive Swagger API documentation."
+    }

api/routers/analysis.py

@@ -0,0 +1,69 @@
+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
+    }

api/services/analyzer_service.py

@@ -0,0 +1,181 @@
+class IncidentAnalyzer:
+    @staticmethod
+    def classify_incident(labels: set[str]) -> str:
+        """
+        Determines the specific incident classification type based on the detected labels.
+        """
+        vehicles = {
+            "car",
+            "truck",
+            "bus",
+            "motorcycle",
+            "damaged vehicle"
+        }
+        vehicle_count = len(labels.intersection(vehicles))
+
+        # ==========================
+        # FIRE RELATED
+        # ==========================
+        if "fire" in labels and "building" in labels:
+            return "building_fire"
+        if "fire" in labels and vehicle_count > 0:
+            return "vehicle_fire"
+        if "fire" in labels:
+            return "fire_incident"
+        if "smoke" in labels and "building" in labels:
+            return "possible_building_fire"
+        if "smoke" in labels:
+            return "smoke_hazard"
+
+        # ==========================
+        # FLOOD RELATED
+        # ==========================
+        if "water" in labels and "road" in labels:
+            return "road_flooding"
+        if "water" in labels and "building" in labels:
+            return "urban_flooding"
+        if "water" in labels and "tree" in labels:
+            return "storm_damage"
+        if "water" in labels:
+            return "water_hazard"
+
+        # ==========================
+        # BUILDING DAMAGE
+        # ==========================
+        if "collapsed structure" in labels and "person" in labels:
+            return "major_building_collapse"
+        if "collapsed structure" in labels:
+            return "building_collapse"
+
+        # ==========================
+        # ROAD ACCIDENTS
+        # ==========================
+        if "damaged vehicle" in labels and "ambulance" in labels:
+            return "critical_road_accident"
+        if vehicle_count >= 2 and "person" in labels:
+            return "road_accident"
+        if vehicle_count >= 2:
+            return "possible_vehicle_collision"
+        if "motorcycle" in labels and "ambulance" in labels:
+            return "motorcycle_accident"
+        if "truck" in labels and "ambulance" in labels:
+            return "truck_accident"
+        if "bus" in labels and "ambulance" in labels:
+            return "bus_accident"
+
+        # ==========================
+        # TRAFFIC RELATED
+        # ==========================
+        if "traffic congestion" in labels and vehicle_count >= 2:
+            return "heavy_traffic"
+        if "construction barrier" in labels and "road" in labels:
+            return "road_construction"
+
+        # ==========================
+        # ROAD BLOCKAGE
+        # ==========================
+        if "tree" in labels and "road" in labels:
+            return "fallen_tree_blockage"
+        if "debris" in labels and "road" in labels:
+            return "road_debris"
+        if "tree" in labels and "debris" in labels:
+            return "storm_road_blockage"
+
+        # ==========================
+        # MEDICAL
+        # ==========================
+        if "ambulance" in labels and "person" in labels:
+            return "medical_emergency"
+
+        # ==========================
+        # LAW ENFORCEMENT
+        # ==========================
+        if "police vehicle" in labels and vehicle_count > 0:
+            return "traffic_enforcement"
+        if "police vehicle" in labels:
+            return "police_activity"
+
+        # ==========================
+        # GENERAL EVENTS
+        # ==========================
+        if vehicle_count > 0:
+            return "vehicle_activity"
+        if "building" in labels:
+            return "building_related_event"
+        if "person" in labels:
+            return "crowd_activity"
+
+        return "unknown_incident"
+
+    @classmethod
+    def analyze(cls, labels_found: list[str]) -> dict:
+        """
+        Calculates severity scores, severity categorizations, and determines
+        the incident type classification.
+        """
+        labels = set(x.lower().strip() for x in labels_found)
+        score = 0
+
+        # Object weight mapping for severity scoring
+        weights = {
+            "ambulance": 35,
+            "fire": 40,
+            "smoke": 20,
+            "water": 20,
+            "person": 5,
+            "debris": 15,
+            "tree": 10,
+            "police vehicle": 15,
+            "damaged vehicle": 25
+        }
+
+        # Accumulate weights of detected labels
+        for label in labels:
+            score += weights.get(label, 0)
+
+        vehicles = {
+            "car",
+            "truck",
+            "bus",
+            "motorcycle",
+            "damaged vehicle"
+        }
+        vehicle_count = len(labels.intersection(vehicles))
+
+        # Scoring modifiers based on incident context
+        # Fire
+        if "fire" in labels:
+            score += 20
+        # Flood
+        elif "water" in labels and "road" in labels:
+            score += 20
+        # Accident
+        elif vehicle_count >= 2 and "person" in labels:
+            score += 25
+        elif "ambulance" in labels and vehicle_count >= 1:
+            score += 30
+        # Obstruction
+        elif "tree" in labels and "road" in labels:
+            score += 15
+
+        # Cap score at 100
+        score = min(score, 100)
+
+        # Categorize severity based on score thresholds
+        if score >= 80:
+            severity = "critical"
+        elif score >= 60:
+            severity = "high"
+        elif score >= 30:
+            severity = "medium"
+        else:
+            severity = "low"
+
+        # Get classified incident type
+        incident_type = cls.classify_incident(labels)
+
+        return {
+            "incident_type": incident_type,
+            "severity": severity,
+            "severity_score": score
+        }

api/services/dino_service.py

@@ -0,0 +1,71 @@
+import torch
+from PIL import Image
+from transformers import AutoProcessor, AutoModelForZeroShotObjectDetection
+from api.config import settings
+
+class GroundingDinoService:
+    def __init__(self):
+        self.model_id = settings.MODEL_ID
+        print(f"Initializing Grounding DINO model '{self.model_id}'...")
+        
+        # Load processor and model
+        self.processor = AutoProcessor.from_pretrained(self.model_id)
+        self.model = AutoModelForZeroShotObjectDetection.from_pretrained(
+            self.model_id,
+            device_map="auto"
+        )
+        print("Grounding DINO model loaded successfully!")
+
+    def detect(self, image: Image.Image) -> list[dict]:
+        """
+        Executes object detection on the PIL Image using the configured labels and thresholds.
+        Returns a list of detections containing label, confidence score, and bounding boxes.
+        """
+        # Format labels as expected by processor: nested list containing label strings
+        formatted_labels = [settings.TEXT_LABELS]
+        
+        # Prepare inputs
+        inputs = self.processor(
+            images=image,
+            text=formatted_labels,
+            return_tensors="pt"
+        ).to(self.model.device)
+
+        # Run inference
+        with torch.no_grad():
+            outputs = self.model(**inputs)
+
+        # Post-process detections
+        results = self.processor.post_process_grounded_object_detection(
+            outputs,
+            inputs.input_ids,
+            threshold=settings.BOX_THRESHOLD,
+            text_threshold=settings.TEXT_THRESHOLD,
+            target_sizes=[image.size[::-1]]
+        )
+
+        detections = []
+        result = results[0]
+
+        # Iterate over output boxes, scores, and labels
+        for box, score, label in zip(
+            result["boxes"],
+            result["scores"],
+            result["labels"]
+        ):
+            confidence = round(score.item(), 3)
+            
+            # Filter detections by minimum confidence threshold
+            if confidence < settings.MIN_CONFIDENCE:
+                continue
+
+            # Convert box coordinates to float list and round
+            box_coords = [round(x, 2) for x in box.tolist()]
+
+            detections.append({
+                "label": label,
+                "confidence": confidence,
+                "box": box_coords
+            })
+
+        return detections

app.py

@@ -0,0 +1,14 @@
+import uvicorn
+import os
+
+if __name__ == "__main__":
+    port = int(os.getenv("PORT", 8000))
+    host = os.getenv("HOST", "0.0.0.0")
+    
+    print(f"Starting FastAPI server on http://{host}:{port}...")
+    uvicorn.run(
+        "api.main:app",
+        host=host,
+        port=port,
+        reload=False  # Disabled by default for ML models to prevent duplicate model loads in memory
+    )

requirements.txt

@@ -0,0 +1,8 @@
+fastapi>=0.110.0
+uvicorn>=0.28.0
+python-multipart>=0.0.9
+transformers>=4.38.0
+torch>=2.0.0
+pillow>=10.0.0
+requests>=2.31.0
+icrawler>=0.6.0