File size: 5,550 Bytes
a31fd7f | 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 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | 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()
# Helper to convert EXIF coordinate format to decimal degrees
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
# 1. Check if device coordinates are provided by the client
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}")
# 2. Check EXIF data if client GPS is not available
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}")
# 3. Fallback to Gemini AI Vision geolocator
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}")
# In a real environment, we would upload the file to Supabase Storage and get a URL.
# For this setup, we will mock the photo URL as a static link or base64 placeholder.
photo_url = f"https://placeholder-url.com/uploads/{file.filename}"
# Save both public damage report and private details
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))
|