GeoSight MCP Server

""" Satellite imagery search and retrieval tools. """ import asyncio from datetime import datetime, timedelta from typing import Optional, Literal import structlog from geosight.config import settings from geosight.tools.geocoding import resolve_location from geosight.services.sentinel_hub import SentinelHubService from geosight.services.planetary_computer import PlanetaryComputerService logger = structlog.get_logger(__name__) async def search_imagery( start_date: str, end_date: str, latitude: Optional[float] = None, longitude: Optional[float] = None, location_name: Optional[str] = None, max_cloud_cover: float = 20.0, data_source: Literal["sentinel-2", "landsat-8", "landsat-9", "sentinel-1"] = "sentinel-2", ) -> dict: """ Search for available satellite imagery for a given location and date range. Args: start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format latitude: Center latitude longitude: Center longitude location_name: Location name to geocode max_cloud_cover: Maximum cloud cover percentage data_source: Satellite data source Returns: Dictionary with search results and imagery metadata """ try: # Resolve location lat, lon, display_name = await resolve_location( location_name=location_name, latitude=latitude, longitude=longitude, ) logger.info( "searching_imagery", location=display_name, lat=lat, lon=lon, start_date=start_date, end_date=end_date, data_source=data_source, ) # Parse dates start = datetime.strptime(start_date, "%Y-%m-%d") end = datetime.strptime(end_date, "%Y-%m-%d") # Use Planetary Computer for free data access service = PlanetaryComputerService() results = await service.search_imagery( latitude=lat, longitude=lon, start_date=start, end_date=end, collection=data_source, max_cloud_cover=max_cloud_cover, ) # Format results if not results["items"]: return { "summary": f"🔍 **No imagery found**\n" f" Location: {display_name}\n" f" Date range: {start_date} to {end_date}\n" f" Data source: {data_source}\n" f" Max cloud cover: {max_cloud_cover}%\n\n" f"💡 Try expanding the date range or increasing max cloud cover.", "items": [], "statistics": { "total_results": 0, "date_range_days": (end - start).days, }, } # Build summary items = results["items"] summary_lines = [ f"🛰️ **Found {len(items)} image{'s' if len(items) != 1 else ''}**", f" 📍 Location: {display_name}", f" 📅 Date range: {start_date} to {end_date}", f" 🛰️ Data source: {data_source.upper()}", f" ☁️ Max cloud cover: {max_cloud_cover}%", "", "**Available imagery:**", ] for i, item in enumerate(items[:10], 1): # Show top 10 date = item.get("datetime", "Unknown date") cloud = item.get("cloud_cover", "N/A") cloud_str = f"{cloud:.1f}%" if isinstance(cloud, (int, float)) else cloud summary_lines.append(f" {i}. {date} | Cloud: {cloud_str}") if len(items) > 10: summary_lines.append(f" ... and {len(items) - 10} more") # Calculate statistics cloud_covers = [ item["cloud_cover"] for item in items if isinstance(item.get("cloud_cover"), (int, float)) ] stats = { "total_results": len(items), "date_range_days": (end - start).days, "avg_cloud_cover_pct": sum(cloud_covers) / len(cloud_covers) if cloud_covers else None, "min_cloud_cover_pct": min(cloud_covers) if cloud_covers else None, "best_image_date": items[0].get("datetime") if items else None, } return { "summary": "\n".join(summary_lines), "items": items, "location": { "name": display_name, "latitude": lat, "longitude": lon, }, "search_params": { "start_date": start_date, "end_date": end_date, "data_source": data_source, "max_cloud_cover": max_cloud_cover, }, "statistics": stats, } except ValueError as e: return { "summary": f"❌ **Error:** {str(e)}", "error": str(e), } except Exception as e: logger.error("imagery_search_error", error=str(e), exc_info=True) return { "summary": f"❌ **Search failed:** {str(e)}", "error": str(e), }