| import logging |
| from typing import Optional |
| from fastapi import APIRouter, File, UploadFile, Form, Query, HTTPException |
| from pydantic import BaseModel |
| from exif import Image as ExifImage |
| from services.gemini_service import estimate_location_from_photo |
| from services.supabase_service import save_damage_report, get_damage_reports_in_bbox |
|
|
| logger = logging.getLogger("api_damage") |
| router = APIRouter() |
|
|
| |
| def convert_to_decimal(coords, ref) -> float: |
| try: |
| degrees = float(coords[0]) |
| minutes = float(coords[1]) |
| seconds = float(coords[2]) |
| |
| decimal = degrees + (minutes / 60.0) + (seconds / 3600.0) |
| if ref in ['S', 'W']: |
| decimal = -decimal |
| return decimal |
| except Exception as e: |
| logger.error(f"Error converting EXIF coordinates: {e}") |
| return 0.0 |
|
|
| def get_exif_gps(image_bytes: bytes) -> tuple[Optional[float], Optional[float]]: |
| """ |
| Attempts to read GPS coordinates from image EXIF metadata. |
| """ |
| try: |
| img = ExifImage(image_bytes) |
| if not img.has_exif: |
| return None, None |
| |
| lat = getattr(img, "gps_latitude", None) |
| lat_ref = getattr(img, "gps_latitude_ref", None) |
| lng = getattr(img, "gps_longitude", None) |
| lng_ref = getattr(img, "gps_longitude_ref", None) |
| |
| if lat and lat_ref and lng and lng_ref: |
| lat_dec = convert_to_decimal(lat, lat_ref) |
| lng_dec = convert_to_decimal(lng, lng_ref) |
| return lat_dec, lng_dec |
| except Exception as e: |
| logger.error(f"Failed to extract EXIF data: {e}") |
| return None, None |
|
|
| @router.post("/report") |
| async def report_damage( |
| file: UploadFile = File(...), |
| severity: str = Form(...), |
| owner_name: str = Form(...), |
| owner_contact: str = Form(...), |
| exact_address: str = Form(...), |
| description: Optional[str] = Form(None), |
| device_lat: Optional[float] = Form(None), |
| device_lng: Optional[float] = Form(None) |
| ): |
| """ |
| Submits a new damage report. Cascades coordinates lookup: |
| 1. Direct device coordinates from PWA (HTML5 Geolocation). |
| 2. EXIF metadata coordinates from the uploaded image. |
| 3. Gemini Vision AI analysis of the photo context. |
| """ |
| try: |
| contents = await file.read() |
| |
| final_lat = None |
| final_lng = None |
| detection_method = "unknown" |
| ai_reasoning = None |
|
|
| |
| if device_lat is not None and device_lng is not None: |
| final_lat = device_lat |
| final_lng = device_lng |
| detection_method = "device_gps" |
| logger.info(f"Using device GPS coordinates: {final_lat}, {final_lng}") |
| |
| |
| if final_lat is None or final_lng is None: |
| exif_lat, exif_lng = get_exif_gps(contents) |
| if exif_lat is not None and exif_lng is not None: |
| final_lat = exif_lat |
| final_lng = exif_lng |
| detection_method = "exif_metadata" |
| logger.info(f"Using EXIF GPS coordinates: {final_lat}, {final_lng}") |
| |
| |
| if final_lat is None or final_lng is None: |
| logger.info("No GPS data found. Calling Gemini Vision AI...") |
| mime_type = file.content_type or "image/jpeg" |
| ai_estimate = estimate_location_from_photo(contents, mime_type) |
| final_lat = ai_estimate.latitude |
| final_lng = ai_estimate.longitude |
| ai_reasoning = ai_estimate.reasoning |
| detection_method = "gemini_vision_ai" |
| logger.info(f"Gemini estimated coordinates: {final_lat}, {final_lng}") |
|
|
| |
| |
| photo_url = f"https://placeholder-url.com/uploads/{file.filename}" |
| |
| |
| saved_report = save_damage_report( |
| severity=severity, |
| latitude=final_lat, |
| longitude=final_lng, |
| photo_url=photo_url, |
| status="pending", |
| owner_name=owner_name, |
| owner_contact=owner_contact, |
| exact_address=exact_address, |
| description=description |
| ) |
|
|
| return { |
| "success": True, |
| "report_id": saved_report["id"], |
| "coordinates": { |
| "latitude": final_lat, |
| "longitude": final_lng |
| }, |
| "detection_method": detection_method, |
| "ai_reasoning": ai_reasoning, |
| "severity": severity, |
| "status": "pending" |
| } |
| |
| except Exception as e: |
| logger.error(f"Error reporting damage: {e}") |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
| @router.get("/bbox") |
| def get_damage_in_view( |
| min_lat: float = Query(...), |
| min_lng: float = Query(...), |
| max_lat: float = Query(...), |
| max_lng: float = Query(...) |
| ): |
| """ |
| Fetches public damage reports within the visible map bounding box. |
| """ |
| try: |
| reports = get_damage_reports_in_bbox(min_lat, min_lng, max_lat, max_lng) |
| return reports |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=str(e)) |
|
|
