Spaces:
Runtime error
Runtime error
| """ | |
| Deep Soil Intelligence & Analysis Engine | |
| ========================================= | |
| Provides comprehensive soil analysis with: | |
| - Current soil data display (moisture, pH, nutrients, etc.) | |
| - Soil health summary | |
| - Key issues detection | |
| - Actionable recommendations (irrigation, fertilizer, organic improvements) | |
| - Yield optimization tips | |
| - Priority levels for each action | |
| """ | |
| import json | |
| import logging | |
| from typing import Dict, List, Any, Tuple | |
| logger = logging.getLogger(__name__) | |
| # Soil parameter reference ranges for different crops | |
| SOIL_RANGES = { | |
| "TOMATO": { | |
| "moisture": {"min": 40, "max": 70, "ideal": 55}, | |
| "ph": {"min": 6.0, "max": 7.5, "ideal": 6.8}, | |
| "nitrogen": {"min": 1.5, "max": 3.0, "ideal": 2.2}, # units based on soil type | |
| "organic_matter": {"min": 2, "max": 5, "ideal": 3.5}, | |
| "potassium": {"min": 1.5, "max": 3.0, "ideal": 2.0}, | |
| "phosphorus": {"min": 1.0, "max": 2.5, "ideal": 1.8}, | |
| }, | |
| "WHEAT": { | |
| "moisture": {"min": 35, "max": 60, "ideal": 50}, | |
| "ph": {"min": 6.5, "max": 8.0, "ideal": 7.2}, | |
| "nitrogen": {"min": 1.0, "max": 2.5, "ideal": 1.8}, | |
| "organic_matter": {"min": 1.5, "max": 3.5, "ideal": 2.5}, | |
| "potassium": {"min": 1.2, "max": 2.5, "ideal": 1.8}, | |
| "phosphorus": {"min": 0.8, "max": 2.0, "ideal": 1.5}, | |
| }, | |
| "RICE": { | |
| "moisture": {"min": 60, "max": 90, "ideal": 75}, | |
| "ph": {"min": 6.0, "max": 7.5, "ideal": 6.8}, | |
| "nitrogen": {"min": 1.0, "max": 3.0, "ideal": 2.0}, | |
| "organic_matter": {"min": 2.0, "max": 4.5, "ideal": 3.2}, | |
| "potassium": {"min": 1.2, "max": 2.8, "ideal": 2.0}, | |
| "phosphorus": {"min": 1.0, "max": 2.5, "ideal": 1.8}, | |
| }, | |
| "MAIZE": { | |
| "moisture": {"min": 45, "max": 70, "ideal": 60}, | |
| "ph": {"min": 6.0, "max": 7.5, "ideal": 6.8}, | |
| "nitrogen": {"min": 1.2, "max": 2.8, "ideal": 2.0}, | |
| "organic_matter": {"min": 1.5, "max": 4.0, "ideal": 2.8}, | |
| "potassium": {"min": 1.0, "max": 2.5, "ideal": 1.8}, | |
| "phosphorus": {"min": 0.8, "max": 2.2, "ideal": 1.5}, | |
| }, | |
| "ONION": { | |
| "moisture": {"min": 50, "max": 75, "ideal": 65}, | |
| "ph": {"min": 6.0, "max": 7.5, "ideal": 6.5}, | |
| "nitrogen": {"min": 1.5, "max": 3.0, "ideal": 2.2}, | |
| "organic_matter": {"min": 2.5, "max": 5.0, "ideal": 3.5}, | |
| "potassium": {"min": 1.5, "max": 3.0, "ideal": 2.2}, | |
| "phosphorus": {"min": 1.2, "max": 2.5, "ideal": 2.0}, | |
| }, | |
| "DEFAULT": { | |
| "moisture": {"min": 40, "max": 70, "ideal": 55}, | |
| "ph": {"min": 6.0, "max": 7.5, "ideal": 6.8}, | |
| "nitrogen": {"min": 1.0, "max": 2.5, "ideal": 1.8}, | |
| "organic_matter": {"min": 2.0, "max": 4.0, "ideal": 3.0}, | |
| "potassium": {"min": 1.0, "max": 2.5, "ideal": 1.8}, | |
| "phosphorus": {"min": 0.8, "max": 2.0, "ideal": 1.5}, | |
| } | |
| } | |
| def extract_soil_parameters(farm_data: Dict[str, Any], soil_properties: List[Dict] = None) -> Dict[str, Any]: | |
| """ | |
| Extract relevant soil parameters from farm data and processed properties. | |
| Returns dict with available soil parameters: | |
| - moisture (from farmer input or estimated) | |
| - pH (from farmer input or ISRIC) | |
| - nitrogen, phosphorus, potassium (from ISRIC) | |
| - organic_matter | |
| - temperature (from weather data) | |
| """ | |
| extracted = {} | |
| # Get directly provided values | |
| extracted["moisture"] = farm_data.get("soil_moisture", None) | |
| extracted["ph"] = farm_data.get("soil_ph", None) | |
| extracted["temperature"] = farm_data.get("temp", None) | |
| # Try to extract from soil properties (ISRIC) | |
| if soil_properties: | |
| for prop in soil_properties: | |
| param = prop.get("parameter", "").lower() | |
| value = prop.get("value") | |
| if value == "N/A" or value is None: | |
| continue | |
| if "nitrogen" in param: | |
| extracted["nitrogen"] = value | |
| elif "phosphorus" in param or "phosphate" in param: | |
| extracted["phosphorus"] = value | |
| elif "potassium" in param or "potash" in param: | |
| extracted["potassium"] = value | |
| elif "organic" in param or "carbon" in param: | |
| extracted["organic_matter"] = value | |
| elif "ph" in param and not extracted["ph"]: | |
| extracted["ph"] = value | |
| elif "water" in param or "moisture" in param: | |
| if not extracted["moisture"]: | |
| extracted["moisture"] = value | |
| return {k: v for k, v in extracted.items() if v is not None} | |
| def detect_key_issues(crop_type: str, extracted_params: Dict[str, Any]) -> List[Dict[str, Any]]: | |
| """ | |
| Detect key soil issues based on crop-specific ranges. | |
| Returns list of issue dicts: | |
| { | |
| "issue": "Issue name", | |
| "severity": "HIGH|MEDIUM|LOW", | |
| "current_value": value, | |
| "ideal_range": "min-max", | |
| "description": "What this means" | |
| } | |
| """ | |
| crop = crop_type.upper() | |
| ranges = SOIL_RANGES.get(crop, SOIL_RANGES["DEFAULT"]) | |
| issues = [] | |
| for param_name, param_value in extracted_params.items(): | |
| if param_name not in ranges: | |
| continue | |
| range_info = ranges[param_name] | |
| min_val = range_info["min"] | |
| max_val = range_info["max"] | |
| ideal_val = range_info.get("ideal", (min_val + max_val) / 2) | |
| # Determine severity | |
| severity = "LOW" | |
| if param_value < min_val or param_value > max_val: | |
| severity = "HIGH" | |
| elif abs(param_value - ideal_val) > (max_val - min_val) * 0.25: | |
| severity = "MEDIUM" | |
| else: | |
| severity = "LOW" | |
| if severity == "LOW": | |
| continue # Only report issues | |
| # Get issue description | |
| if param_name == "moisture": | |
| if param_value < min_val: | |
| desc = f"Soil is too dry ({param_value}%). Risk of plant stress and reduced water uptake." | |
| else: | |
| desc = f"Soil is too wet ({param_value}%). Risk of waterlogging and root rot." | |
| elif param_name == "ph": | |
| if param_value < min_val: | |
| desc = f"Soil is too acidic (pH {param_value}). Nutrient availability is reduced." | |
| else: | |
| desc = f"Soil is too alkaline (pH {param_value}). Iron and other micronutrients may be locked." | |
| elif param_name == "nitrogen": | |
| desc = f"Nitrogen levels are low ({param_value}). Plants may show stunted growth and pale color." | |
| elif param_name == "phosphorus": | |
| desc = f"Phosphorus levels are low ({param_value}). Root development and flowering may be affected." | |
| elif param_name == "potassium": | |
| desc = f"Potassium levels are low ({param_value}). Fruit quality and disease resistance reduced." | |
| elif param_name == "organic_matter": | |
| desc = f"Organic matter is low ({param_value}%). Soil structure and water retention are poor." | |
| else: | |
| desc = f"{param_name.capitalize()} levels are outside ideal range ({param_value})." | |
| issues.append({ | |
| "issue": param_name.capitalize(), | |
| "severity": severity, | |
| "current_value": param_value, | |
| "ideal_range": f"{min_val}-{max_val}", | |
| "description": desc | |
| }) | |
| # Sort by severity | |
| severity_order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2} | |
| issues.sort(key=lambda x: severity_order.get(x["severity"], 3)) | |
| return issues | |
| def generate_soil_health_summary(crop_type: str, extracted_params: Dict[str, Any], issues: List[Dict]) -> str: | |
| """ | |
| Generate a brief, farmer-friendly soil health summary. | |
| """ | |
| if not issues: | |
| return "✅ **Excellent Soil Condition** - Your soil is in good health with balanced nutrients and proper moisture. Continue current management practices." | |
| high_issues = [i for i in issues if i["severity"] == "HIGH"] | |
| medium_issues = [i for i in issues if i["severity"] == "MEDIUM"] | |
| if high_issues: | |
| main_issue = high_issues[0]["issue"].lower() | |
| if len(high_issues) == 1: | |
| summary = f"⚠️ **Soil Health: CRITICAL** - Your soil has **{main_issue}** problem that needs urgent attention. Additionally, there are {len(medium_issues)} moderate issues to address." | |
| else: | |
| summary = f"🚨 **Soil Health: POOR** - Multiple critical issues detected including {main_issue}. Immediate intervention required." | |
| elif medium_issues: | |
| main_issue = medium_issues[0]["issue"].lower() | |
| summary = f"⚡ **Soil Health: FAIR** - Your soil has moderate {main_issue} and needs optimization. Follow the recommendations below." | |
| else: | |
| summary = "✅ **Soil Health: GOOD** - Minor issues detected but overall soil is suitable for {crop_type}." | |
| return summary | |
| def generate_actionable_recommendations(crop_type: str, extracted_params: Dict[str, Any], issues: List[Dict]) -> List[Dict[str, Any]]: | |
| """ | |
| Generate practical, farmer-friendly recommendations. | |
| Returns list of dicts: | |
| { | |
| "recommendation": "Action to take", | |
| "priority": "HIGH|MEDIUM|LOW", | |
| "category": "Irrigation|Fertilizer|Organic|pH|Other", | |
| "steps": ["Step 1", "Step 2", ...], | |
| "expected_result": "What farmer should expect", | |
| "timeline": "When to see results" | |
| } | |
| """ | |
| recommendations = [] | |
| for issue in issues: | |
| param = issue["issue"].lower() | |
| if param == "moisture": | |
| if extracted_params.get("moisture", 0) < SOIL_RANGES.get(crop_type, SOIL_RANGES["DEFAULT"])["moisture"]["min"]: | |
| # Low moisture | |
| recommendations.append({ | |
| "recommendation": "Increase Watering Frequency", | |
| "priority": issue["severity"], | |
| "category": "Irrigation", | |
| "steps": [ | |
| "Water early morning (6-7 AM) or late evening (5-6 PM)", | |
| "Apply water slowly at base of plants, not on leaves", | |
| "Water 3-4 times per week instead of 2 times", | |
| "Check soil moisture by pressing hand 2 inches deep - should feel damp", | |
| "Mulch around plants to retain moisture" | |
| ], | |
| "expected_result": "Soil moisture will improve within 2-3 days, plant stress will reduce", | |
| "timeline": "1-3 days" | |
| }) | |
| else: | |
| # High moisture | |
| recommendations.append({ | |
| "recommendation": "Improve Drainage & Reduce Watering", | |
| "priority": issue["severity"], | |
| "category": "Irrigation", | |
| "steps": [ | |
| "Reduce watering to 1-2 times per week", | |
| "Ensure field has proper drainage channels", | |
| "Add raised beds if drainage is poor", | |
| "Aerify soil by loosening with fork (5-6 inches deep)", | |
| "Remove standing water if any" | |
| ], | |
| "expected_result": "Waterlogging will reduce, root rot risk decreases", | |
| "timeline": "5-7 days" | |
| }) | |
| elif param == "ph": | |
| current_ph = extracted_params.get("ph", 7) | |
| if current_ph < SOIL_RANGES.get(crop_type, SOIL_RANGES["DEFAULT"])["ph"]["min"]: | |
| # Acidic | |
| recommendations.append({ | |
| "recommendation": "Apply Agricultural Lime to Raise pH", | |
| "priority": issue["severity"], | |
| "category": "pH", | |
| "steps": [ | |
| f"Measure soil pH: Current is {current_ph} (acidic)", | |
| "Apply agricultural lime (calcium carbonate) at 1-2 tons per acre", | |
| "Mix lime into top 6 inches of soil", | |
| "Water thoroughly after application", | |
| "Wait 3-4 weeks before next planting" | |
| ], | |
| "expected_result": "Soil pH will increase by 0.5-1 point, nutrient availability improves", | |
| "timeline": "3-4 weeks" | |
| }) | |
| else: | |
| # Alkaline | |
| recommendations.append({ | |
| "recommendation": "Apply Sulfur to Lower pH", | |
| "priority": issue["severity"], | |
| "category": "pH", | |
| "steps": [ | |
| f"Current soil pH is {current_ph} (alkaline)", | |
| "Apply elemental sulfur at 500 kg per acre", | |
| "Incorporate sulfur into top 4-6 inches of soil", | |
| "Water well after application", | |
| "Monitor pH changes every 2 weeks" | |
| ], | |
| "expected_result": "Soil pH will gradually decrease by 0.5-1 point over time", | |
| "timeline": "4-8 weeks" | |
| }) | |
| elif param == "nitrogen": | |
| recommendations.append({ | |
| "recommendation": "Apply Nitrogen-Rich Fertilizer", | |
| "priority": issue["severity"], | |
| "category": "Fertilizer", | |
| "steps": [ | |
| "Use urea (46-0-0) or ammonium nitrate (33-0-0)", | |
| "Apply 50-100 kg per acre for moderate deficiency", | |
| "Split application: Half now, half after 3-4 weeks", | |
| "Mix thoroughly with top 2-3 inches of soil", | |
| "Water immediately after application" | |
| ], | |
| "expected_result": "Plants will show greener leaves and faster growth within 10-15 days", | |
| "timeline": "10-15 days" | |
| }) | |
| recommendations.append({ | |
| "recommendation": "Add Organic Nitrogen Sources", | |
| "priority": "MEDIUM", | |
| "category": "Organic", | |
| "steps": [ | |
| "Apply farm yard manure at 10-15 tons per acre", | |
| "Or use compost at 5-10 tons per acre", | |
| "Or incorporate legume-based green manure", | |
| "Mix well into soil 2-3 weeks before sowing", | |
| "This builds long-term soil fertility" | |
| ], | |
| "expected_result": "Gradual nitrogen release, improved soil structure and microbes", | |
| "timeline": "2-4 weeks" | |
| }) | |
| elif param == "phosphorus": | |
| recommendations.append({ | |
| "recommendation": "Apply Phosphorus Fertilizer", | |
| "priority": issue["severity"], | |
| "category": "Fertilizer", | |
| "steps": [ | |
| "Use single super phosphate (SSP) or DAP (18-46-0)", | |
| "Apply 50-75 kg per acre", | |
| "Mix into root zone (top 4-6 inches)", | |
| "Phosphorus moves slowly, apply early", | |
| "Water after application" | |
| ], | |
| "expected_result": "Root development improves, flowering and seed formation enhance", | |
| "timeline": "20-30 days" | |
| }) | |
| elif param == "potassium": | |
| recommendations.append({ | |
| "recommendation": "Apply Potassium Fertilizer", | |
| "priority": issue["severity"], | |
| "category": "Fertilizer", | |
| "steps": [ | |
| "Use muriate of potash (0-0-60) or potassium nitrate", | |
| "Apply 40-60 kg per acre", | |
| "Best applied during fruiting stage", | |
| "Mix with irrigation water for faster absorption", | |
| "Apply in 2 splits if possible" | |
| ], | |
| "expected_result": "Larger fruits, better color, improved disease resistance", | |
| "timeline": "15-25 days" | |
| }) | |
| recommendations.append({ | |
| "recommendation": "Use Wood Ash as Organic Potassium Source", | |
| "priority": "MEDIUM", | |
| "category": "Organic", | |
| "steps": [ | |
| "Collect clean wood ash (no treated wood)", | |
| "Apply 2-3 tons per acre", | |
| "Spread evenly and mix into soil", | |
| "Use only if soil pH is not already high", | |
| "Avoid overuse (causes salt accumulation)" | |
| ], | |
| "expected_result": "Gradual potassium release plus calcium benefits", | |
| "timeline": "2-3 weeks" | |
| }) | |
| elif param == "organic matter": | |
| recommendations.append({ | |
| "recommendation": "Build Organic Matter with Compost & Manure", | |
| "priority": issue["severity"], | |
| "category": "Organic", | |
| "steps": [ | |
| "Apply 8-12 tons of farm yard manure per acre", | |
| "Or 5-8 tons of well-made compost per acre", | |
| "Mix into top 6 inches of soil", | |
| "Continue annual additions to build reserves", | |
| "Use crop residue mulching" | |
| ], | |
| "expected_result": "Better water retention, improved soil structure, more microbes", | |
| "timeline": "4-8 weeks to see major benefits" | |
| }) | |
| recommendations.append({ | |
| "recommendation": "Plant Green Manure Crops in Off-Season", | |
| "priority": "MEDIUM", | |
| "category": "Organic", | |
| "steps": [ | |
| "Sow legumes (dhaincha, sesbania) in off-season", | |
| "Grow for 6-8 weeks", | |
| "Plow into soil 2 weeks before main crop", | |
| "This adds nitrogen AND organic matter", | |
| "Repeat every year for maximum benefit" | |
| ], | |
| "expected_result": "Organic matter builds over time, nitrogen increases naturally", | |
| "timeline": "Ongoing, benefits compound yearly" | |
| }) | |
| return recommendations | |
| def generate_optimization_tips(crop_type: str, extracted_params: Dict[str, Any]) -> List[str]: | |
| """ | |
| Generate smart optimization tips for better yield. | |
| """ | |
| tips = [] | |
| crop = crop_type.upper() | |
| ranges = SOIL_RANGES.get(crop, SOIL_RANGES["DEFAULT"]) | |
| # Moisture optimization | |
| moisture = extracted_params.get("moisture") | |
| if moisture and moisture < ranges["moisture"]["ideal"]: | |
| tips.append("💧 **Irrigation Timing**: Water early morning to minimize evaporation and maximize absorption.") | |
| if moisture and moisture > ranges["moisture"]["ideal"] * 0.7: | |
| tips.append("💧 **Soil Moisture**: Your soil has good moisture. Drip irrigation saves 30-40% water vs flood irrigation.") | |
| # pH optimization | |
| ph = extracted_params.get("ph") | |
| if ph and ranges["ph"]["min"] < ph < ranges["ph"]["max"]: | |
| tips.append("🧪 **pH Balance**: Your soil pH is optimal. Maintain it by avoiding excessive lime or sulfur applications.") | |
| # Organic matter | |
| organic = extracted_params.get("organic_matter") | |
| if not organic or organic < ranges["organic_matter"]["ideal"]: | |
| tips.append("🌿 **Soil Biology**: Add mulch around plants to feed microorganisms and improve soil structure over time.") | |
| # General tips | |
| if extracted_params.get("temperature", 0) > 35: | |
| tips.append("🌡️ **Heat Protection**: Use shade cloth or intercroppings to reduce soil temperature stress during extreme heat.") | |
| tips.append("📊 **Soil Testing**: Test soil every 2-3 years to track changes and adjust fertilizer application.") | |
| tips.append("🌱 **Crop Rotation**: Rotate crops to break pest cycles and naturally restore soil fertility.") | |
| tips.append("🚜 **Minimum Tillage**: Reduce deep plowing to preserve soil structure and protect microorganisms.") | |
| return tips | |
| def deep_soil_intelligence(farm_data: Dict[str, Any], soil_report: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Main function to generate comprehensive soil analysis. | |
| Returns: | |
| { | |
| "extracted_parameters": {...}, | |
| "current_data_display": [...], # For UI display | |
| "health_summary": "...", | |
| "key_issues": [...], | |
| "actionable_recommendations": [...], | |
| "optimization_tips": [...], | |
| "overall_priority": "HIGH|MEDIUM|LOW" | |
| } | |
| """ | |
| crop_type = farm_data.get("crop_type", "DEFAULT") | |
| # Extract parameters | |
| soil_properties = soil_report.get("properties", []) | |
| extracted = extract_soil_parameters(farm_data, soil_properties) | |
| # Create display-friendly version of all parameters | |
| current_data_display = [] | |
| # Add farmer-input parameters | |
| if farm_data.get("soil_moisture") is not None: | |
| current_data_display.append({ | |
| "name": "Soil Moisture", | |
| "value": farm_data.get("soil_moisture"), | |
| "unit": "%", | |
| "source": "Farmer Input" | |
| }) | |
| if farm_data.get("soil_ph") is not None: | |
| current_data_display.append({ | |
| "name": "Soil pH", | |
| "value": farm_data.get("soil_ph"), | |
| "unit": "pH", | |
| "source": "Farmer Input" | |
| }) | |
| if farm_data.get("temp") is not None: | |
| current_data_display.append({ | |
| "name": "Soil Temperature", | |
| "value": farm_data.get("temp"), | |
| "unit": "°C", | |
| "source": "Weather Data" | |
| }) | |
| if farm_data.get("humidity") is not None: | |
| current_data_display.append({ | |
| "name": "Soil Humidity", | |
| "value": farm_data.get("humidity"), | |
| "unit": "%", | |
| "source": "Weather Data" | |
| }) | |
| # Add ISRIC properties (if available) | |
| for prop in soil_properties: | |
| if prop.get("value") != "N/A" and prop.get("value") is not None: | |
| current_data_display.append({ | |
| "name": prop.get("parameter", "Unknown"), | |
| "value": prop.get("value"), | |
| "unit": prop.get("unit", ""), | |
| "source": "ISRIC SoilGrids" | |
| }) | |
| # Detect issues | |
| issues = detect_key_issues(crop_type, extracted) | |
| # Generate health summary | |
| health_summary = generate_soil_health_summary(crop_type, extracted, issues) | |
| # Generate recommendations | |
| recommendations = generate_actionable_recommendations(crop_type, extracted, issues) | |
| # Generate optimization tips | |
| tips = generate_optimization_tips(crop_type, extracted) | |
| # Determine overall priority | |
| if any(i["severity"] == "HIGH" for i in issues): | |
| overall_priority = "HIGH" | |
| elif any(i["severity"] == "MEDIUM" for i in issues): | |
| overall_priority = "MEDIUM" | |
| else: | |
| overall_priority = "LOW" | |
| return { | |
| "extracted_parameters": extracted, | |
| "current_data_display": current_data_display, | |
| "health_summary": health_summary, | |
| "key_issues": issues, | |
| "actionable_recommendations": recommendations, | |
| "optimization_tips": tips, | |
| "overall_priority": overall_priority | |
| } | |