| """AOI Advisor — Claude-powered region insight for analysis recommendations.""" |
| from __future__ import annotations |
|
|
| import json |
| import logging |
| from datetime import date |
|
|
| from app.config import ANTHROPIC_API_KEY |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _SYSTEM_PROMPT = """\ |
| You are a remote sensing advisor for humanitarian programme teams. Given a geographic location, provide a brief analysis recommendation. |
| |
| Available EO products: |
| - ndvi: Vegetation health from Sentinel-2. Detects drought, crop stress, deforestation. |
| - water: Water extent (MNDWI) from Sentinel-2. Detects flooding, drought, reservoir changes. |
| - sar: Radar backscatter from Sentinel-1. Detects ground surface changes, flooding, construction. |
| - buildup: Settlement extent (NDBI) from Sentinel-2. Detects urban growth, displacement camps. |
| |
| Coverage: East Africa region. Resolution: 100m. Max analysis window: 3 years. |
| |
| Respond with JSON only, no markdown. Structure: |
| { |
| "context": "1-3 sentences about this region and recent relevant events", |
| "recommended_start": "YYYY-MM-DD", |
| "recommended_end": "YYYY-MM-DD", |
| "product_priorities": ["product_id", ...], |
| "reasoning": "1 sentence per EO product explaining why it is relevant here" |
| }""" |
|
|
| _EMPTY_RESPONSE = { |
| "context": None, |
| "recommended_start": None, |
| "recommended_end": None, |
| "product_priorities": None, |
| "reasoning": None, |
| } |
|
|
| |
| _client = None |
|
|
|
|
| def _get_client(): |
| global _client |
| if _client is None: |
| import anthropic |
| _client = anthropic.AsyncAnthropic(api_key=ANTHROPIC_API_KEY) |
| return _client |
|
|
|
|
| async def get_aoi_advice(bbox: list[float]) -> dict: |
| """Call Claude to get region-aware analysis recommendations. |
| |
| Returns structured advice dict. On any failure, returns all-null dict. |
| """ |
| if not ANTHROPIC_API_KEY: |
| logger.warning("ANTHROPIC_API_KEY not set — skipping AOI advisor") |
| return _EMPTY_RESPONSE |
|
|
| center_lng = (bbox[0] + bbox[2]) / 2 |
| center_lat = (bbox[1] + bbox[3]) / 2 |
| today = date.today().isoformat() |
|
|
| lat_label = f"{abs(center_lat):.4f}°{'N' if center_lat >= 0 else 'S'}" |
| lng_label = f"{abs(center_lng):.4f}°{'E' if center_lng >= 0 else 'W'}" |
|
|
| user_prompt = ( |
| f"Location: {lat_label}, {lng_label}\n" |
| f"Current date: {today}\n" |
| f"Recommend an analysis timeframe and indicator priorities for this area." |
| ) |
|
|
| try: |
| client = _get_client() |
| message = await client.messages.create( |
| model="claude-haiku-4-5-20251001", |
| max_tokens=500, |
| temperature=0, |
| system=_SYSTEM_PROMPT, |
| messages=[{"role": "user", "content": user_prompt}], |
| ) |
|
|
| raw = message.content[0].text |
| advice = json.loads(raw) |
|
|
| |
| for key in ("context", "recommended_start", "recommended_end", "product_priorities", "reasoning"): |
| if key not in advice: |
| logger.warning("AOI advisor response missing key: %s", key) |
| return _EMPTY_RESPONSE |
|
|
| |
| date.fromisoformat(advice["recommended_start"]) |
| date.fromisoformat(advice["recommended_end"]) |
|
|
| |
| valid_ids = {"ndvi", "water", "sar", "buildup"} |
| advice["product_priorities"] = [ |
| i for i in advice["product_priorities"] if i in valid_ids |
| ] |
|
|
| return advice |
|
|
| except Exception as exc: |
| logger.warning("AOI advisor failed: %s", exc) |
| return _EMPTY_RESPONSE |
|
|
