File size: 3,787 Bytes
ca9ccdd b20698b ca9ccdd b20698b ca9ccdd b20698b ca9ccdd b20698b ca9ccdd b20698b ca9ccdd b20698b ca9ccdd b20698b ca9ccdd b20698b ca9ccdd |
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 |
"""
Soil Properties Server using Google Earth Engine
"""
import ee
import os
import json
from typing import Dict, Any
class SoilPropertiesServer:
"""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 at coordinate"""
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)
} |