Spaces:
Running
Running
| """Soil Properties Server using Google Earth Engine with lazy initialization""" | |
| import ee | |
| import os | |
| import json | |
| from typing import Dict, Any | |
| class SoilPropertiesServer: | |
| """Google Earth Engine Soil Server with lazy initialization""" | |
| def __init__(self): | |
| """Initialize server (GEE will be initialized on first use)""" | |
| self._gee_initialized = False | |
| 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 lazily (only when first request is made)""" | |
| if self._gee_initialized: | |
| return | |
| 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 (lazy)") | |
| else: | |
| # Fallback to default credentials (for local development) | |
| ee.Initialize(project='MCPprototypeGEE') | |
| print("✅ GEE initialized with default credentials (lazy)") | |
| self._gee_initialized = True | |
| except Exception as e: | |
| print(f"⚠️ GEE lazy initialization failed: {str(e)}") | |
| raise | |
| async def get_data(self, lat: float, lon: float) -> Dict[str, Any]: | |
| """Get soil properties at coordinate""" | |
| # Initialize GEE on first request (lazy loading) | |
| if not self._gee_initialized: | |
| self._initialize_gee() | |
| try: | |
| point = ee.Geometry.Point([lon, lat]) | |
| results = {} | |
| 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" | |
| } | |
| 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": str(e) | |
| } |