from flask import Flask, render_template, request, jsonify import requests import google.generativeai as genai import os import json from gtts import gTTS import io import base64 app = Flask(__name__) # Mapping of SoilGrids parameter codes PARAM_MAP = { "bdod": "Bulk Density", "cec": "Cation Exchange Capacity", "cfvo": "Coarse Fragment Volume", "clay": "Clay Content", "nitrogen": "Nitrogen Content", "ocd": "Organic Carbon Density", "ocs": "Organic Carbon Stock", "phh2o": "Soil pH", "sand": "Sand Content", "silt": "Silt Content", "soc": "Soil Organic Carbon", "wv0010": "Water Content (0-10cm)", "wv0033": "Water Content (0-33cm)", "wv1500": "Water Content (1500mm)" } LANG_MAP = { "English": "en", "Hindi": "hi", "Bengali": "bn", "Telugu": "te", "Marathi": "mr", "Tamil": "ta", "Gujarati": "gu", "Urdu": "ur", "Kannada": "kn", "Odia": "or", "Malayalam": "ml" } @app.route('/') def index(): return render_template('index.html') @app.route('/get_soil_report', methods=['POST']) def get_soil_report(): data = request.get_json() lat, lon = data.get("lat"), data.get("lon") if not lat or not lon: return jsonify({"error": "Latitude and Longitude are required"}), 400 headers = {"accept": "application/json"} # Retry configuration max_retries = 3 retry_delay = 2 # seconds for attempt in range(max_retries): try: print(f"Attempt {attempt + 1}/{max_retries} to fetch soil data...") # Fetch classification data class_response = requests.get( "https://rest.isric.org/soilgrids/v2.0/classification/query", params={"lon": lon, "lat": lat, "number_classes": 5}, headers=headers, timeout=30 ) class_response.raise_for_status() class_data = class_response.json() soil_classification = { "soil_type": class_data.get("wrb_class_name", "Unknown"), "probabilities": class_data.get("wrb_class_probability", []) } # Fetch properties data prop_response = requests.get( "https://rest.isric.org/soilgrids/v2.0/properties/query", params={ "lon": lon, "lat": lat, "property": list(PARAM_MAP.keys()), "depth": "5-15cm", "value": "mean" }, headers=headers, timeout=30 ) prop_response.raise_for_status() prop_data = prop_response.json() properties_list = [] for layer in prop_data.get("properties", {}).get("layers", []): param_code = layer.get("name") name = PARAM_MAP.get(param_code, param_code.upper()) value = layer.get("depths", [{}])[0].get("values", {}).get("mean") unit = layer.get("unit_measure", {}).get("mapped_units", "") if value is not None: if param_code == "phh2o": value /= 10.0 unit = "pH" elif param_code in ["wv0010", "wv0033", "wv1500"]: value /= 100.0 unit = "cm³/cm³" properties_list.append({"parameter": name, "value": value, "unit": unit}) print("Successfully fetched soil data from API") return jsonify({"classification": soil_classification, "properties": properties_list}) except requests.exceptions.HTTPError as e: if e.response.status_code == 502: print(f"502 Bad Gateway error on attempt {attempt + 1}") if attempt < max_retries - 1: import time time.sleep(retry_delay) retry_delay *= 2 # Exponential backoff continue else: # Use fallback mock data after all retries fail print("API unavailable, using fallback mock data") return jsonify({ "classification": { "soil_type": "Cambisol (Sample Data - API Unavailable)", "probabilities": [ ["Cambisol", 45.2], ["Luvisol", 23.8], ["Vertisol", 18.5], ["Regosol", 8.3], ["Fluvisol", 4.2] ] }, "properties": [ {"parameter": "Bulk Density", "value": 1.42, "unit": "kg/dm³"}, {"parameter": "Cation Exchange Capacity", "value": 18.5, "unit": "cmol/kg"}, {"parameter": "Clay Content", "value": 28.3, "unit": "%"}, {"parameter": "Sand Content", "value": 42.1, "unit": "%"}, {"parameter": "Silt Content", "value": 29.6, "unit": "%"}, {"parameter": "Soil pH", "value": 6.8, "unit": "pH"}, {"parameter": "Soil Organic Carbon", "value": 12.4, "unit": "g/kg"}, {"parameter": "Nitrogen Content", "value": 1.2, "unit": "g/kg"} ], "_note": "⚠️ The ISRIC SoilGrids API is currently unavailable. This is sample data for demonstration purposes only. Please try again later for actual soil data for your location." }) else: raise except requests.exceptions.RequestException as e: print(f"Request error on attempt {attempt + 1}: {e}") if attempt < max_retries - 1: import time time.sleep(retry_delay) retry_delay *= 2 continue else: # Use fallback mock data after all retries fail print("API unavailable due to connection error, using fallback mock data") return jsonify({ "classification": { "soil_type": "Cambisol (Sample Data - API Unavailable)", "probabilities": [ ["Cambisol", 45.2], ["Luvisol", 23.8], ["Vertisol", 18.5], ["Regosol", 8.3], ["Fluvisol", 4.2] ] }, "properties": [ {"parameter": "Bulk Density", "value": 1.42, "unit": "kg/dm³"}, {"parameter": "Cation Exchange Capacity", "value": 18.5, "unit": "cmol/kg"}, {"parameter": "Clay Content", "value": 28.3, "unit": "%"}, {"parameter": "Sand Content", "value": 42.1, "unit": "%"}, {"parameter": "Silt Content", "value": 29.6, "unit": "%"}, {"parameter": "Soil pH", "value": 6.8, "unit": "pH"}, {"parameter": "Soil Organic Carbon", "value": 12.4, "unit": "g/kg"}, {"parameter": "Nitrogen Content", "value": 1.2, "unit": "g/kg"} ], "_note": "⚠️ The ISRIC SoilGrids API is currently unavailable (connection timeout). This is sample data for demonstration purposes only. Please try again later for actual soil data for your location." }) @app.route('/analyze_soil', methods=['POST']) def analyze_soil(): """Enhanced soil analysis with NVIDIA and Gemini fallback support.""" try: data = request.get_json() soil_report = data.get("soil_report") language = data.get("language", "English") if not soil_report: return jsonify({"error": "Soil report data is missing"}), 400 prompt = f""" Analyze the following soil report and provide recommendations. The response MUST be a single, valid JSON object, without any markdown formatting, comments, or surrounding text like ```json. The user wants the analysis in this language: {language}. Soil Report Data: {json.dumps(soil_report, indent=2)} JSON Structure to follow: {{"soilType": "Primary soil type", "generalInsights": ["Insight 1", "Insight 2"], "parameters": [{{"parameter": "Parameter Name", "value": "Value with Unit", "range": "Low/Normal/High", "comment": "Brief comment."}}], "cropRecommendations": [{{"crop": "Crop Name", "reason": "Brief reason."}}], "managementRecommendations": {{"fertilization": "Recommendation.", "irrigation": "Recommendation."}}}} """ analysis_json = None last_error = None model_used = None # Try NVIDIA models first (from config.env) try: from image_summarizer import ModelConfig from openai import OpenAI config = ModelConfig() nvidia_api_key = config.get('nvidia_api_key') nvidia_models = config.get('nvidia_models', []) if nvidia_api_key and nvidia_models: nvidia_client = OpenAI( base_url="https://integrate.api.nvidia.com/v1", api_key=nvidia_api_key ) for model_name in nvidia_models: try: print(f"Attempting NVIDIA model: {model_name}") response = nvidia_client.chat.completions.create( model=model_name, messages=[{"role": "user", "content": prompt}], max_tokens=config.get('max_tokens', 1000), temperature=config.get('temperature', 0.2) ) cleaned_text = response.choices[0].message.content.strip() json_start_index = cleaned_text.find('{') json_end_index = cleaned_text.rfind('}') + 1 if json_start_index != -1 and json_end_index > json_start_index: json_str = cleaned_text[json_start_index:json_end_index] analysis_json = json.loads(json_str) model_used = f"NVIDIA: {model_name}" print(f"✅ Successfully used NVIDIA model: {model_name}") break except Exception as e: print(f"NVIDIA model {model_name} failed: {e}") last_error = e continue except ImportError: print("⚠️ image_summarizer module not available, skipping NVIDIA models") except Exception as e: print(f"⚠️ NVIDIA fallback error: {e}") last_error = e # Fallback to Gemini models if NVIDIA failed if not analysis_json: print("Falling back to Gemini models...") api_key = os.getenv("GEMINI_API", "AIzaSyDkiYr-eSkqIXpZ1fHlik_YFsFtfQoFi0w") genai.configure(api_key=api_key) # Load Gemini models from config (all 7 models) models_to_try = config.get('gemini_models', ['gemini-2.5-flash', 'gemini-2.0-flash', 'gemini-3.0-flash']) for model_name in models_to_try: try: print(f"Attempting Gemini model: {model_name}") model = genai.GenerativeModel(model_name) response = model.generate_content(prompt) cleaned_text = response.text.strip() json_start_index = cleaned_text.find('{') json_end_index = cleaned_text.rfind('}') + 1 if json_start_index != -1 and json_end_index > json_start_index: json_str = cleaned_text[json_start_index:json_end_index] analysis_json = json.loads(json_str) model_used = f"Gemini: {model_name}" print(f"✅ Successfully used Gemini model: {model_name}") break else: raise ValueError("No valid JSON object found in the response.") except Exception as e: print(f"Gemini model {model_name} failed: {e}") last_error = e continue if not analysis_json: raise Exception("All AI models (NVIDIA + Gemini) failed to generate a valid JSON response.") from last_error # Add metadata about which model was used analysis_json['_model_used'] = model_used # Generate TTS audio print("Generating audio summary...") summary_text = f"Soil analysis complete. The soil type is {analysis_json.get('soilType', 'not specified')}. " summary_text += "Recommended crops include: " + ", ".join([c['crop'] for c in analysis_json.get('cropRecommendations', [])]) + ". " summary_text += "For fertilization, " + analysis_json.get('managementRecommendations', {}).get('fertilization', "no recommendation was given.") lang_code = LANG_MAP.get(language, 'en') tts = gTTS(text=summary_text, lang=lang_code, slow=False) mp3_fp = io.BytesIO() tts.write_to_fp(mp3_fp) mp3_fp.seek(0) base64_audio = base64.b64encode(mp3_fp.read()).decode('utf-8') analysis_json['audioContent'] = f"data:audio/mp3;base64,{base64_audio}" print("Audio generation complete.") return jsonify(analysis_json) except Exception as e: print(f"!!! AN UNHANDLED ERROR OCCURRED in /analyze_soil: {e}") return jsonify({"error": f"An unexpected server error occurred: {str(e)}"}), 500 @app.route('/analyze_image', methods=['POST']) def analyze_image(): """New endpoint for image analysis with NVIDIA fallback.""" try: # Check if image file is provided if 'image' not in request.files: return jsonify({"error": "No image file provided"}), 400 image_file = request.files['image'] if image_file.filename == '': return jsonify({"error": "No image selected"}), 400 # Get optional prompt from form data prompt = request.form.get('prompt', 'Please analyze this image and provide detailed insights.') # Save uploaded image temporarily temp_image_path = os.path.join('temp_uploads', image_file.filename) os.makedirs('temp_uploads', exist_ok=True) image_file.save(temp_image_path) try: # Use ImageAnalyzer for analysis from image_summarizer import ImageAnalyzer analyzer = ImageAnalyzer() result = analyzer.analyze_image(temp_image_path, prompt) # Clean up temp file os.remove(temp_image_path) if result['success']: return jsonify({ 'success': True, 'analysis': result['response'], 'model_used': result['model_used'], 'provider': result['provider'] }) else: return jsonify({ 'success': False, 'error': result['error'], 'suggestions': result.get('suggestions', []) }), 500 except Exception as e: # Clean up temp file on error if os.path.exists(temp_image_path): os.remove(temp_image_path) raise e except Exception as e: print(f"!!! ERROR in /analyze_image: {e}") return jsonify({"error": f"Image analysis failed: {str(e)}"}), 500 if __name__ == '__main__': app.run(debug=True, port=7860)