feat (GEE soil MCP server implementation)

src/servers/soil.py

@@ -1,101 +1,100 @@
-import aiohttp
-import asyncio
+"""
+Soil Properties Server using Google Earth Engine
+"""
+
+import ee
 import os
-import xarray as xr
-import requests
-from datetime import datetime
+import json
 from typing import Dict, Any
 
 
-# ============================================================================
-# SOIL PROPERTIES SERVER (SoilGrids)
-# ============================================================================
-
 class SoilPropertiesServer:
-    """SoilGrids API Server - Enhanced for reliability"""
-
+    """Google Earth Engine Soil Server"""
+    
+    def __init__(self):
+        self._initialize_gee()
+        
+        self.layers = {
+            "clay": "OpenLandMap/SOL/SOL_CLAY-WFRACTION_USDA-3A1A1A_M/v02",
+            "sand": "OpenLandMap/SOL/SOL_SAND-WFRACTION_USDA-3A1A1A_M/v02",
+            "phh2o": "OpenLandMap/SOL/SOL_PH-H2O_USDA-4C1A2A_M/v02",
+            "soc": "OpenLandMap/SOL/SOL_ORGANIC-CARBON_USDA-6A1C_M/v02"
+        }
+        self.depth_band = "b0"
+    
+    def _initialize_gee(self):
+        """Initialize GEE with service account"""
+        try:
+            # Check if service account key is provided
+            service_account_key = os.environ.get('GEE_SERVICE_ACCOUNT_KEY')
+            
+            if service_account_key:
+                # Parse JSON key
+                credentials_dict = json.loads(service_account_key)
+                credentials = ee.ServiceAccountCredentials(
+                    credentials_dict['client_email'],
+                    key_data=service_account_key
+                )
+                ee.Initialize(credentials)
+                print("✅ GEE initialized with service account")
+            else:
+                # Fallback to default credentials (for local development)
+                ee.Initialize(project='MCPprototypeGEE')
+                print("✅ GEE initialized with default credentials")
+                
+        except Exception as e:
+            print(f"⚠️ GEE initialization failed: {str(e)}")
+            raise
+    
     async def get_data(self, lat: float, lon: float) -> Dict[str, Any]:
-        """Get soil properties with retry logic and rate limit handling"""
+        """Get soil properties at coordinate"""
         try:
-            properties = ["clay", "sand", "silt", "phh2o", "soc"]
+            point = ee.Geometry.Point([lon, lat])
             results = {}
-
-            timeout = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10)
-            connector = aiohttp.TCPConnector(limit=1, limit_per_host=1)
-
-            async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
-                for prop in properties:
-                    # Retry logic (max 2 attempts)
-                    for attempt in range(2):
-                        try:
-                            url = "https://rest.isric.org/soilgrids/v2.0/properties/query"
-                            params = {
-                                "lon": lon,
-                                "lat": lat,
-                                "property": prop,
-                                "depth": "0-5cm",
-                                "value": "mean"
-                            }
-
-                            async with session.get(url, params=params) as response:
-                                if response.status == 200:
-                                    data = await response.json()
-                                    value = data['properties']['layers'][0]['depths'][0]['values']['mean']
-
-                                    if value is not None:
-                                        if prop == 'phh2o':
-                                            results[prop] = round(value / 10, 1)
-                                        elif prop == 'soc':
-                                            results[prop] = value
-                                        else:
-                                            results[prop] = round(value / 10, 1)
-                                    else:
-                                        results[prop] = None
-                                    break
-
-                                elif response.status == 429:
-                                    if attempt == 0:
-                                        await asyncio.sleep(1)
-                                        continue
-                                    else:
-                                        results[prop] = None
-                                        break
-                                else:
-                                    results[prop] = None
-                                    break
-
-                        except asyncio.TimeoutError:
-                            if attempt == 0:
-                                await asyncio.sleep(0.5)
-                                continue
-                            else:
-                                results[prop] = None
-                                break
-                        except Exception:
-                            results[prop] = None
-                            break
-
-                    await asyncio.sleep(0.2)  # Delay between properties
-
-            if any(v is not None for v in results.values()):
-                return {
-                    "status": "success",
-                    "data": {
-                        "clay_percent": results.get("clay"),
-                        "sand_percent": results.get("sand"),
-                        "silt_percent": results.get("silt"),
-                        "pH": results.get("phh2o"),
-                        "organic_carbon_dg_per_kg": results.get("soc"),
-                        "data_source": "SoilGrids API (ISRIC)",
-                        "location": {"latitude": lat, "longitude": lon},
-                        "depth": "0-5cm (topsoil)"
-                    }
-                }
-            else:
+            
+            for prop_name, image_id in self.layers.items():
+                try:
+                    image = ee.Image(image_id).select(self.depth_band)
+                    value = image.sample(point, 250).first().get(self.depth_band).getInfo()
+                    
+                    if value is not None:
+                        if prop_name in ['clay', 'sand']:
+                            results[prop_name] = round(value / 10, 1)
+                        elif prop_name == 'phh2o':
+                            results[prop_name] = round(value / 10, 1)
+                        elif prop_name == 'soc':
+                            results[prop_name] = round(value / 10, 1)
+                    else:
+                        results[prop_name] = None
+                except:
+                    results[prop_name] = None
+            
+            if not any(v is not None for v in results.values()):
                 return {
                     "status": "error",
-                    "error": "No soil data available for this location"
+                    "error": "No soil data available"
                 }
-
+            
+            silt = None
+            if results.get("clay") and results.get("sand"):
+                silt = round(100 - results["clay"] - results["sand"], 1)
+            
+            return {
+                "status": "success",
+                "data": {
+                    "clay_percent": results.get("clay"),
+                    "sand_percent": results.get("sand"),
+                    "silt_percent": silt,
+                    "pH": results.get("phh2o"),
+                    "organic_carbon_g_kg": results.get("soc"),
+                    "data_source": "Google Earth Engine (OpenLandMap)",
+                    "location": {"latitude": lat, "longitude": lon},
+                    "depth": "0-5cm"
+                }
+            }
+            
         except Exception as e:
-            return {"status": "error", "error": f"SoilGrids error: {str(e)}"}
+            return {
+                "status": "error",
+                "error": str(e)
+            }