Spaces:
Runtime error
Runtime error
| """ | |
| Bio-Agronomist Agent: Climate & Agronomic Intelligence Tool | |
| Fetches Advanced Agricultural Metrics from Open-Meteo Historic & Forecast APIs. | |
| Provides massive toolset upgrades over standard weather: | |
| 1. Growing Degree Days (GDD): Crucial for crop maturity cycle calculations. | |
| 2. Evapotranspiration (ET₀): Tells the exact crop water requirement. | |
| 3. Soil Moisture (0-100cm): Tracks drought stress across differing root zones. | |
| 4. Solar Radiation (GHI/DNI): Key for photosynthetic rate/yield potential. | |
| """ | |
| import requests | |
| import json | |
| import logging | |
| from datetime import datetime, timedelta | |
| # Configure Logger | |
| logger = logging.getLogger("BioAgronomist.ClimateTool") | |
| logger.setLevel(logging.DEBUG) | |
| handler = logging.StreamHandler() | |
| handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) | |
| logger.addHandler(handler) | |
| class AgroClimateIntelligenceTool: | |
| def __init__(self): | |
| self.ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive" | |
| self.FORECAST_URL = "https://api.open-meteo.com/v1/forecast" | |
| def fetch_annual_agronomic_baseline(self, lat: float, lon: float) -> dict: | |
| """ | |
| Fetches a 1-year historical baseline of advanced agronomic variables. | |
| Used to determine base viability for crop selection. | |
| """ | |
| end_date = datetime.now() - timedelta(days=7) # Archive API has ~5 day lag | |
| start_date = end_date - timedelta(days=365) | |
| logger.info(f"Fetching 1-Year Agronomic Baseline for {lat}, {lon} ({start_date.strftime('%Y-%m-%d')} to {end_date.strftime('%Y-%m-%d')})") | |
| params = { | |
| "latitude": lat, | |
| "longitude": lon, | |
| "start_date": start_date.strftime("%Y-%m-%d"), | |
| "end_date": end_date.strftime("%Y-%m-%d"), | |
| "daily": [ | |
| "temperature_2m_max", | |
| "temperature_2m_min", | |
| "precipitation_sum", | |
| "et0_fao_evapotranspiration", # Crop Water Demand | |
| "shortwave_radiation_sum", # Photosynthetic potential | |
| ], | |
| "timezone": "auto" | |
| } | |
| try: | |
| resp = requests.get(self.ARCHIVE_URL, params=params, timeout=30) | |
| if resp.status_code != 200: | |
| logger.error(f"Archive API failed: HTTP {resp.status_code}") | |
| return {"error": "API Failure"} | |
| data = resp.json().get("daily", {}) | |
| times = data.get("time", []) | |
| p_sum = data.get("precipitation_sum", []) | |
| t_max = data.get("temperature_2m_max", []) | |
| et0 = data.get("et0_fao_evapotranspiration", []) | |
| rad = data.get("shortwave_radiation_sum", []) | |
| if not times: | |
| return {"error": "Empty data array"} | |
| # Aggregate into Monthly Summaries | |
| monthly_data = {} | |
| for i in range(len(times)): | |
| month_key = times[i][:7] # YYYY-MM | |
| if month_key not in monthly_data: | |
| monthly_data[month_key] = {"rain": [], "tmax": [], "et0": [], "rad": []} | |
| if p_sum[i] is not None: monthly_data[month_key]["rain"].append(p_sum[i]) | |
| if t_max[i] is not None: monthly_data[month_key]["tmax"].append(t_max[i]) | |
| if et0[i] is not None: monthly_data[month_key]["et0"].append(et0[i]) | |
| if rad[i] is not None: monthly_data[month_key]["rad"].append(rad[i]) | |
| summary = [] | |
| for m, vals in sorted(monthly_data.items()): | |
| summary.append({ | |
| "month": m, | |
| "avg_temp_max_c": round(sum(vals["tmax"])/len(vals["tmax"]), 1) if vals["tmax"] else 0, | |
| "total_rain_mm": round(sum(vals["rain"]), 1) if vals["rain"] else 0, | |
| "total_et0_mm": round(sum(vals["et0"]), 1) if vals["et0"] else 0, | |
| "avg_daily_radiation_mj_m2": round(sum(vals["rad"])/len(vals["rad"]), 2) if vals["rad"] else 0, | |
| # Hydrological Stress: if Evapotranspiration > Rain, crop requires irrigation | |
| "irrigation_deficit_mm": round((sum(vals["et0"]) or 0) - (sum(vals["rain"]) or 0), 1) | |
| }) | |
| logger.info("Successfully compiled 12-month agronomic baseline.") | |
| return {"monthly_agronomics": summary, "annual_analysis_ready": True} | |
| except Exception as e: | |
| logger.error(f"Exception during agronomic baseline fetch: {e}") | |
| return {"error": str(e)} | |
| def fetch_current_soil_moisture_stress(self, lat: float, lon: float) -> dict: | |
| """ | |
| Fetches LIVE soil moisture levels to evaluate immediate planting conditions. | |
| Crucial for deciding if sowing can happen now without irrigation. | |
| """ | |
| logger.info(f"Fetching Live Soil Moisture Profile for {lat}, {lon}") | |
| params = { | |
| "latitude": lat, | |
| "longitude": lon, | |
| "current": ["soil_moisture_0_to_1cm", "soil_moisture_1_to_3cm", "soil_moisture_3_to_9cm", "soil_moisture_9_to_27cm", "soil_moisture_27_to_81cm"], | |
| "timezone": "auto" | |
| } | |
| try: | |
| resp = requests.get(self.FORECAST_URL, params=params, timeout=20) | |
| if resp.status_code == 200: | |
| current = resp.json().get("current", {}) | |
| profile = { | |
| "surface_0_1cm": current.get("soil_moisture_0_to_1cm"), | |
| "shallow_1_9cm": round(((current.get("soil_moisture_1_to_3cm", 0) + current.get("soil_moisture_3_to_9cm", 0)) / 2), 3), | |
| "root_zone_9_27cm": current.get("soil_moisture_9_to_27cm"), | |
| "deep_zone_27_81cm": current.get("soil_moisture_27_to_81cm"), | |
| } | |
| logger.debug(f"Root Zone Moisture: {profile['root_zone_9_27cm']} m3/m3") | |
| return {"live_moisture_profile": profile} | |
| else: | |
| logger.warning(f"Failed to fetch live moisture: HTTP {resp.status_code}") | |
| return {} | |
| except Exception as e: | |
| logger.error(f"Moisture fetch error: {e}") | |
| return {} | |
| if __name__ == "__main__": | |
| tool = AgroClimateIntelligenceTool() | |
| print("\n--- Testing Agronomic Baseline ---") | |
| baseline = tool.fetch_annual_agronomic_baseline(18.5204, 73.8567) | |
| print(json.dumps(baseline, indent=2)) | |
| print("\n--- Testing Live Soil Moisture ---") | |
| moisture = tool.fetch_current_soil_moisture_stress(18.5204, 73.8567) | |
| print(json.dumps(moisture, indent=2)) | |