File size: 3,692 Bytes
7e25f7a | 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 | """
Location Intelligence service — estimates where an image was captured.
Combines evidence from:
- GPS (from EXIF metadata, via cores.metadata)
- OCR text (language + keyword detection)
- License plate format
- Scene labels
- Detected objects (vehicle presence = street/road context)
Returns a LocationEstimate with candidate countries, cities, and
conflicting evidence. No heavy geolocation models — pure heuristics.
"""
from __future__ import annotations
import time
from typing import Optional
from cores.location import LocationEvidenceCollector, LocationEstimator
from cores.metadata import extract_all
from models.jobs import JobRequest
from models.reports import LocationEstimate
from pipeline import InputValidator, ImagePreprocessor, ImageHasher
from utils.logging import execution_context, new_execution_id
class LocationIntelligenceService:
"""Estimates image capture location from multiple evidence sources."""
def __init__(
self,
validator: InputValidator,
preprocessor: ImagePreprocessor,
hasher: ImageHasher,
) -> None:
self._validator = validator
self._preprocessor = preprocessor
self._hasher = hasher
async def analyze(self, request: JobRequest) -> dict:
"""Estimate location from an image."""
eid = new_execution_id()
with execution_context(execution_id=eid, provider_id="location_intelligence_service"):
t0 = time.perf_counter()
vr = self._validator.validate(
image_url=request.image_url,
image_base64=request.image_base64,
)
if not vr.valid:
return {"success": False, "error": vr.error, "error_type": "ValidationError"}
if vr.source == "url":
pre = self._preprocessor.from_url(request.image_url)
else:
pre = self._preprocessor.from_bytes(vr.image_bytes, vr.source)
original_bytes = pre.original_bytes or vr.image_bytes
collector = LocationEvidenceCollector()
gps: Optional[dict] = None
# 1. GPS from EXIF
if original_bytes:
meta = extract_all(original_bytes)
if meta.get("gps_coords"):
gps = meta["gps_coords"]
collector.add_gps(gps["lat"], gps["lon"])
# 2. OCR text — if the caller passed it in options
# (We don't run OCR here; the caller can pass OCR results via options)
ocr_text = request.options.get("ocr_text", "")
if ocr_text:
collector.add_ocr_text(ocr_text)
# 3. License plate — if passed in options
plate_text = request.options.get("plate_text", "")
if plate_text:
collector.add_plate(plate_text)
# 4. Scene label — if passed in options
scene_label = request.options.get("scene_label", "")
if scene_label:
collector.add_scene(scene_label)
# 5. Detected objects — if passed in options
objects = request.options.get("objects", [])
if objects:
collector.add_detected_objects(objects)
# Estimate
estimator = LocationEstimator()
estimate = estimator.estimate(collector.evidence, gps=gps)
elapsed = (time.perf_counter() - t0) * 1000.0
estimate.elapsed_ms = round(elapsed, 3)
return {
"success": True,
"location_estimate": estimate.model_dump(),
"elapsed_ms": round(elapsed, 3),
}
|