""" Professional Alert Decision Engine =================================== Generates real-time alerts based on weather, pest, and water data Follows climate-resilient agriculture best practices """ import logging from datetime import datetime from typing import List, Dict, Optional logger = logging.getLogger("DecisionEngine") # Alert severity constants SEVERITY_CRITICAL = "CRITICAL" SEVERITY_HIGH = "HIGH" SEVERITY_MEDIUM = "MEDIUM" SEVERITY_LOW = "LOW" SEVERITY_INFO = "INFO" def calculate_climate_resilience_score(w_intel: Dict, p_intel: Dict, water_intel: Dict, farmer: Dict) -> Dict: """Calculate comprehensive climate resilience index (0-100)""" score_components = {} # Weather Risk (30 points max) curr = w_intel.get("current", {}) temp = curr.get("temp", 25) wind = curr.get("wind", 0) forecast = w_intel.get("forecast", []) weather_score = 30 if temp > farmer.get("threshold_settings", {}).get("temp_high", 40): weather_score -= 15 elif temp < farmer.get("threshold_settings", {}).get("temp_low", 5): weather_score -= 10 if wind > farmer.get("threshold_settings", {}).get("wind_speed", 30): weather_score -= 15 heavy_rain_days = len([d for d in forecast if "heavy" in d.get("description", "").lower()]) if heavy_rain_days > 2: weather_score -= 5 * heavy_rain_days score_components["weather_risk"] = max(0, weather_score) # Pest Risk (30 points max) pest_score = 30 for pest in p_intel.get("pest_prediction_table", []): severity = pest.get("severity", "").upper() if severity == "CRITICAL": pest_score -= 15 elif severity == "HIGH": pest_score -= 10 elif severity == "MEDIUM": pest_score -= 5 score_components["pest_risk"] = max(0, pest_score) # Water Management Risk (20 points max) water_score = 20 irrigation_minutes = water_intel.get("irrigation_minutes", 0) water_m3 = water_intel.get("water_m3_sqm", 0) # Optimal range: 20-40 minutes if irrigation_minutes < 20 or irrigation_minutes > 50: water_score -= 10 score_components["water_risk"] = max(0, water_score) # Resource Efficiency (20 points max) efficiency_score = 20 rain_next_7days = len([d for d in forecast if "rain" in d.get("description", "").lower()]) # More rain = less irrigation needed = better efficiency if rain_next_7days > 2: efficiency_score = 20 elif irrigation_minutes > 40: efficiency_score = 10 score_components["resource_efficiency"] = efficiency_score total_score = sum([score_components["weather_risk"], score_components["pest_risk"], score_components["water_risk"], score_components["resource_efficiency"]]) resilience_level = "๐ŸŸข GREEN" if total_score >= 75 else "๐ŸŸก YELLOW" if total_score >= 50 else "๐Ÿ”ด RED" return { "total_score": total_score, "components": score_components, "resilience_level": resilience_level, "critical_factors": [k for k, v in score_components.items() if v < 15] } def get_alerts(w_intel: Dict, p_intel: Dict, water_intel: Dict, farmer: Dict) -> List[Dict]: """ Generate actionable alerts with severity levels Returns list of professional farm alerts """ alerts = [] # Core Data Extraction curr = w_intel.get("current", {}) temp = curr.get("temp", 25) humidity = curr.get("humidity", 60) wind = curr.get("wind", 0) forecast = w_intel.get("forecast", []) thresholds = farmer.get("threshold_settings", {}) rain_next_24h = any("rain" in d.get("description", "").lower() for d in forecast[:2]) heavy_rain_forecast = [d for d in forecast if "heavy" in d.get("description", "").lower()] # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # 1๏ธโƒฃ CRITICAL WEATHER ALERTS # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• if temp > thresholds.get("temp_high", 40): alerts.append({ "type": "WEATHER", "id": "EXTREME_HEAT", "severity": SEVERITY_CRITICAL, "title": "๐ŸŒก๏ธ EXTREME HEAT WARNING", "message": f"Temperature reached {temp}ยฐC. Immediate irrigation required.", "action": "IRRIGATE_IMMEDIATELY", "details": { "current_temp": temp, "threshold": thresholds.get("temp_high", 40), "expected_crop_loss": "25-35%" if temp > 45 else "15-20%", "hydration_urgency": "CRITICAL" }, "action_items": [ f"๐Ÿšจ Start irrigation within next 30 minutes", "๐Ÿ’ง Increase water supply by 20-30%", "๐ŸŒณ Consider temporary shade if available", "๐Ÿ“Š Monitor soil moisture every 2 hours" ] }) if wind > thresholds.get("wind_speed", 30): alerts.append({ "type": "WEATHER", "id": "SEVERE_WIND", "severity": SEVERITY_CRITICAL, "title": "๐ŸŒช๏ธ SEVERE WIND WARNING", "message": f"Wind speed {wind} km/h. High crop damage risk.", "action": "AVOID_ALL_OPERATIONS", "details": { "wind_speed": wind, "risk_level": "CRITICAL", "expected_damage": "Physical damage to plants, fruit drop" }, "action_items": [ "โ›” Stop all spraying and pesticide application", "๐Ÿ”ง Secure loose equipment and stakes", "๐Ÿ“ Postpone harvesting activities", "๐ŸŒพ Support weak branches with stakes" ] }) if temp < thresholds.get("temp_low", 5): alerts.append({ "type": "WEATHER", "id": "FROST_RISK", "severity": SEVERITY_HIGH, "title": "โ„๏ธ FROST RISK - CROP PROTECTION NEEDED", "message": f"Temperature dropping to {temp}ยฐC. Frost damage imminent.", "action": "PROTECT_CROP", "details": { "current_temp": temp, "frost_risk": "HIGH", "vulnerable_stages": "Flowering and young fruits" }, "action_items": [ "๐Ÿ”ฅ Prepare frost protection measures (smudging/overhead irrigation)", "๐Ÿ’จ Use wind machines if available", "๐ŸŒซ๏ธ Apply antitranspirants to sensitive crops", "๐Ÿ“ž Contact agricultural extension for emergency support" ] }) # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # 2๏ธโƒฃ HEAVY RAIN & FLOODING ALERTS # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• if heavy_rain_forecast: alerts.append({ "type": "WEATHER", "id": "HEAVY_RAIN_WARNING", "severity": SEVERITY_HIGH, "title": f"๐ŸŒง๏ธ HEAVY RAIN WARNING - {len(heavy_rain_forecast)} days", "message": f"Heavy rainfall expected. Prepare drainage systems.", "action": "PREPARE_DRAINAGE", "details": { "days_affected": len(heavy_rain_forecast), "flooding_risk": "HIGH", "drainage_priority": "URGENT" }, "action_items": [ "๐Ÿ’ง Clear all drainage channels and ditches", "๐Ÿ”จ Repair broken drainage pipes immediately", "๐Ÿชด Avoid planting in low-lying areas", "โš™๏ธ Test pump systems before heavy rain", "๐Ÿ“Š Monitor water levels in fields" ] }) if not rain_next_24h or wind > thresholds.get("wind_alert", 15): alerts.append({ "type": "SPRAY_OPERATIONS", "id": "UNSAFE_SPRAY_WINDOW", "severity": SEVERITY_HIGH, "title": "๐Ÿšซ UNSAFE TO SPRAY", "message": f"Rain or high wind (Wind: {wind} km/h). Postpone spraying.", "action": "POSTPONE_SPRAYING", "details": { "wind_speed": wind, "rain_probability": "HIGH", "spray_loss_rate": "60-70%" }, "action_items": [ "โฐ Postpone pesticide/fertilizer application", "๐Ÿ“… Reschedule for 3-5 days later", "๐Ÿ’ฐ Higher wind causes 60-70% spray drift loss", "๐Ÿ“ Mark postponed tasks for follow-up" ] }) elif not rain_next_24h and wind < thresholds.get("wind_alert", 15): alerts.append({ "type": "SPRAY_OPERATIONS", "id": "OPTIMAL_SPRAY_WINDOW", "severity": SEVERITY_INFO, "title": "โœ… OPTIMAL SPRAY WINDOW", "message": "Ideal conditions for spraying. Good wind and no rain.", "action": "APPLY_INPUTS", "details": { "wind_speed": wind, "spray_efficiency": "95-98%", "window_duration": "5-7 PM optimal" }, "action_items": [ "โฑ๏ธ Plan spraying activities for 5-7 PM (optimal wind)", "๐ŸŽฏ Use this window for pesticide/herbicide application", "๐Ÿ’ก Higher efficiency = reduced chemical usage = cost savings", "๐Ÿ“Š Monitor weather updates continuously" ] }) # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # 3๏ธโƒฃ PEST & DISEASE ALERTS # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• for pest in p_intel.get("pest_prediction_table", []): severity = pest.get("severity", "").upper() pest_name = pest.get("pest_name", "Unknown Pest") confidence = pest.get("confidence", 50) if severity in ["CRITICAL", "HIGH"]: alert_severity = SEVERITY_CRITICAL if severity == "CRITICAL" else SEVERITY_HIGH alerts.append({ "type": "PEST_ALERT", "id": f"PEST_{pest_name.upper().replace(' ', '_')}", "severity": alert_severity, "title": f"๐Ÿ› {pest_name.upper()} ALERT", "message": f"High {pest_name} infestation risk detected ({confidence}% confidence).", "action": "PEST_MANAGEMENT", "details": { "pest": pest_name, "risk_level": severity, "confidence": confidence, "favorable_temp": f"{temp}ยฐC is {'FAVORABLE' if 20 <= temp <= 30 else 'UNFAVORABLE'} for {pest_name}", "humidity_level": f"Current: {humidity}% - {'High risk' if humidity > 70 else 'Moderate risk'}" }, "action_items": [ f"๐Ÿ” Scout fields immediately for {pest_name} symptoms", f"๐Ÿงช Start bioagent/chemical control within 48 hours", f"๐Ÿ“ธ Document infestation level (% affected plants)", f"๐Ÿ“ž Contact agricultural extension for spray recommendations", f"โฐ Plan early morning or late evening spraying (5-7 PM optimal)" ] }) # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # 4๏ธโƒฃ IRRIGATION MANAGEMENT ALERTS # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• water_m3 = water_intel.get("water_m3_sqm", 0) irrigation_minutes = water_intel.get("irrigation_minutes", 0) if water_m3 > 0.15 or (temp > 35 and water_m3 > 0.1): alerts.append({ "type": "IRRIGATION", "id": "IMMEDIATE_IRRIGATION_NEEDED", "severity": SEVERITY_CRITICAL if temp > 35 else SEVERITY_HIGH, "title": "๐Ÿ’ง IRRIGATION REQUIRED", "message": f"Soil water deficit: {water_m3:.3f} mยณ. Start irrigation NOW.", "action": "IRRIGATE_IMMEDIATELY", "details": { "water_needed": f"{water_m3:.3f} mยณ/mยฒ", "irrigation_duration_minutes": round(irrigation_minutes), "soil_moisture_status": "CRITICAL" if water_m3 > 0.15 else "LOW", "temperature_stress": "YES" if temp > 35 else "NO" }, "action_items": [ f"โฑ๏ธ Start irrigation for {round(irrigation_minutes)} minutes", f"๐Ÿ’ง Deliver {water_m3:.2f} mยณ per mยฒ of field", "๐ŸŒก๏ธ Best timing: Early morning (5-7 AM) or evening (6-8 PM)", "๐Ÿ“Š Check soil moisture after irrigation completion", "๐Ÿ“ˆ Monitor evapotranspiration for next irrigation schedule" ] }) elif 0.05 < water_m3 <= 0.15: alerts.append({ "type": "IRRIGATION", "id": "IRRIGATION_SCHEDULED", "severity": SEVERITY_MEDIUM, "title": "๐Ÿ’ง IRRIGATION SCHEDULED", "message": f"Plan irrigation soon. Water needed: {water_m3:.3f} mยณ", "action": "SCHEDULE_IRRIGATION", "details": { "water_needed": f"{water_m3:.3f} mยณ/mยฒ", "irrigation_duration_minutes": round(irrigation_minutes), "best_window": "Next 48-72 hours" }, "action_items": [ f"๐Ÿ“… Schedule irrigation in next 2-3 days", f"โฑ๏ธ Duration: {round(irrigation_minutes)} minutes", f"๐Ÿ’ง Water quantity: {water_m3:.2f} mยณ/mยฒ", "๐ŸŒž Avoid midday irrigation (high evaporation)", "๐Ÿงน Remove weeds before irrigation (better water use)" ] }) else: alerts.append({ "type": "IRRIGATION", "id": "IRRIGATION_OPTIMAL", "severity": SEVERITY_INFO, "title": "โœ… SOIL MOISTURE OPTIMAL", "message": "Current soil moisture is adequate. Monitor regularly.", "action": "MONITOR", "details": { "water_status": "OPTIMAL", "next_check": "48 hours" } }) # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• # 5๏ธโƒฃ CLIMATE RESILIENCE & SUSTAINABILITY ALERTS # โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ•โ• resilience = calculate_climate_resilience_score(w_intel, p_intel, water_intel, farmer) if resilience["total_score"] < 50: alerts.append({ "type": "CLIMATE_RESILIENCE", "id": "LOW_RESILIENCE_ALERT", "severity": SEVERITY_HIGH, "title": "โš ๏ธ LOW CLIMATE RESILIENCE INDEX", "message": f"Farm resilience score: {resilience['total_score']:.0f}/100. Multiple risks detected.", "action": "IMPLEMENT_RESILIENCE_MEASURES", "details": resilience, "action_items": [ "๐ŸŒพ Implement mulching to reduce water loss 30-40%", "๐ŸŒฑ Plant nitrogen-fixing legumes for soil health", "๐Ÿ’จ Establish windbreaks for wind protection", "๐Ÿ”„ Diversify crops to reduce single-pest risk", "๐Ÿ’ฐ Consider insurance schemes for climate risks" ] }) # Sustainability Tips based on conditions if irrigation_minutes > 40: alerts.append({ "type": "SUSTAINABILITY", "id": "WATER_CONSERVATION_TIP", "severity": SEVERITY_INFO, "title": "๐Ÿ’ง WATER CONSERVATION OPPORTUNITY", "message": "High irrigation need. Implement water-saving techniques.", "action": "CONSERVATION_MEASURE", "details": { "current_irrigation_time": round(irrigation_minutes), "potential_savings": "30-40%" }, "action_items": [ "๐ŸŒพ Apply mulch (straw/leaves) to reduce evaporation by 40%", "๐ŸŒฑ Use drip irrigation if available (saves 50% water vs. flood)", "๐Ÿ’จ Install wind breaks to reduce evaporative stress", "๐Ÿ• Irrigate early morning (5-7 AM) to minimize evaporation", "๐Ÿ“Š Monitor soil moisture with meter for precision" ] }) else: alerts.append({ "type": "SUSTAINABILITY", "id": "SUSTAINABLE_PRACTICE_TIP", "severity": SEVERITY_INFO, "title": "๐ŸŒ SUSTAINABLE FARMING TIP", "message": "Implement intercropping for natural pest control.", "action": "IMPLEMENT_PRACTICE", "details": { "practice": "Intercropping with legumes", "benefit": "30-40% reduction in pest incidence, soil nitrogen boost" }, "action_items": [ "๐ŸŒฑ Intercrop with legumes (beans, peas) to fix nitrogen", "๐Ÿ Attract pollinators with flowering plants at field edges", "๐Ÿ› Encourage beneficial insects (ladybugs, parasitoid wasps)", "โ™ป๏ธ Use crop residue for compost instead of burning", "๐Ÿ“ˆ Expected yield increase: 15-20% with better soil health" ] }) logger.info(f"โœ… Generated {len(alerts)} alerts for {farmer.get('name', 'Farmer')}") return alerts