import requests import json from flask import Flask, render_template, request, jsonify, Response, send_file from google import genai from gtts import gTTS import os app = Flask(__name__) app.config['AUDIO_FOLDER'] = 'static/audio' os.makedirs(app.config['AUDIO_FOLDER'], exist_ok=True) # IMPORTANT: Replace with your actual Gemini API key api_key = os.getenv('GEMINI_API_KEY') if not api_key: api_key = os.getenv('GEMINI_API_KEY_1') if not api_key: api_key = os.getenv('GEMINI_API_KEY_2') client = genai.Client(api_key="AIzaSyDkiYr-eSkqIXpZ1fHlik_YFsFtfQoFi0w") def validate_coordinates(lat, lon): """Validate and convert latitude and longitude to float.""" try: return float(lat), float(lon) except (TypeError, ValueError): return None, None @app.route('/') def index(): return render_template('index.html') @app.route('/get_weather_data', methods=['GET']) def get_weather_data(): """ Fetch weather data using Open-Meteo's forecast endpoint. """ lat = request.args.get('lat') lon = request.args.get('lon') lat, lon = validate_coordinates(lat, lon) if lat is None or lon is None: return jsonify({"error": "Invalid coordinates"}), 400 try: forecast_url = "https://api.open-meteo.com/v1/forecast" forecast_params = { "latitude": lat, "longitude": lon, "current_weather": "true", "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum", "hourly": "relative_humidity_2m,soil_moisture_3_to_9cm,cloudcover,windspeed_10m", "timezone": "auto" } resp = requests.get(forecast_url, params=forecast_params) resp.raise_for_status() data = resp.json() daily = data.get("daily", {}) hourly = data.get("hourly", {}) current = data.get("current_weather", {}) # Daily data max_temp = daily.get("temperature_2m_max", [None])[0] min_temp = daily.get("temperature_2m_min", [None])[0] rain = daily.get("precipitation_sum", [None])[0] # Hourly data (averages) humidity_list = hourly.get("relative_humidity_2m", []) soil_list = hourly.get("soil_moisture_3_to_9cm", []) cloud_list = hourly.get("cloudcover", []) avg_humidity = sum(humidity_list)/len(humidity_list) if humidity_list else None avg_soil_moisture = sum(soil_list)/len(soil_list) if soil_list else None avg_cloud_cover = sum(cloud_list)/len(cloud_list) if cloud_list else None # Current weather current_temp = current.get("temperature") wind_speed = current.get("windspeed") weather = { "max_temp": max_temp, "min_temp": min_temp, "rainfall": rain, "humidity": avg_humidity, "soil_moisture": avg_soil_moisture, "current_temp": current_temp, "wind_speed": wind_speed, "cloud_cover": avg_cloud_cover } return jsonify(weather) except Exception as e: return jsonify({"error": str(e)}), 500 def call_gemini_api(input_data, language): """ Calls the Gemini API to get a pest outbreak report in a structured JSON format. """ prompt = f""" Analyze the provided agricultural and weather data to generate a pest outbreak report. Your response MUST be a single, valid JSON object and nothing else. Do not wrap it in markdown backticks. The entire text content within the JSON must be in the '{language}' language. This is the required JSON structure: {{ "report_title": "Pest Outbreak Dashboard Report", "location_info": {{ "latitude": "{input_data.get('latitude')}", "longitude": "{input_data.get('longitude')}", "derived_location": "A human-readable location derived from the coordinates (e.g., 'Nagpur, India')." }}, "agricultural_inputs_analysis": "A detailed paragraph analyzing the provided agricultural inputs (crop type, growth stage, irrigation, etc.) and their potential impact on pest outbreaks.", "pest_prediction_table": [ {{ "pest_name": "Name of the predicted pest", "outbreak_months": "Predicted month(s) for the outbreak", "severity": "Predicted severity level (e.g., Low, Medium, High)", "precautionary_measures": "A short description of key precautionary measures." }} ], "pest_avoidance_practices": [ "A detailed, specific pest avoidance practice based on the inputs.", "Another specific recommendation.", "Provide 10-12 detailed bullet points." ], "agricultural_best_practices": [ "A specific agricultural best practice based on the inputs.", "Another specific recommendation related to crop management." ], "predicted_pest_damage_info": "A paragraph describing the potential damage the predicted pests could cause to the specified crop." }} Use the following data for your analysis: - Crop Type: {input_data.get('crop_type')} - Sowing Date: {input_data.get('sowing_date')} - Harvest Date: {input_data.get('harvest_date')} - Current Growth Stage: {input_data.get('growth_stage')} - Irrigation Frequency: {input_data.get('irrigation_freq')} - Irrigation Method: {input_data.get('irrigation_method')} - Soil Type: {input_data.get('soil_type')} - Max Temp: {input_data.get('max_temp')}°C - Min Temp: {input_data.get('min_temp')}°C - Humidity: {input_data.get('humidity')}% - Rainfall: {input_data.get('rain')}mm - Soil Moisture: {input_data.get('soil_moisture')}% - Wind Speed: {input_data.get('wind_speed')} km/h """ try: response = client.models.generate_content( model="gemini-2.5-flash", contents=prompt ) json_text = response.text.strip().replace("```json", "").replace("```", "") return json.loads(json_text) except Exception as e: print(f"Error calling Gemini or parsing JSON: {e}") return {"error": "Failed to generate a valid report from the AI model. The model may have returned an unexpected format. Please try again."} @app.route('/predict', methods=['POST']) def predict(): form_data = request.form.to_dict() language = form_data.get("language", "English") report_data = call_gemini_api(form_data, language) report_html = "" # Check for an error from the API call if "error" in report_data: report_html = f"

Error

{report_data['error']}

" audio_url = None else: # Build the HTML report dynamically from the JSON data location = report_data.get('location_info', {}) report_html += f"

{report_data.get('report_title', 'Pest Outbreak Report')}

" report_html += f"

Location: {location.get('derived_location', 'N/A')} (Lat: {location.get('latitude', 'N/A')}, Lon: {location.get('longitude', 'N/A')})

" report_html += f"

Agricultural Input Parameter Analysis

{report_data.get('agricultural_inputs_analysis', 'No analysis available.')}

" report_html += "

Pest Outbreak Prediction

" pest_table = report_data.get('pest_prediction_table', []) if pest_table: for pest in pest_table: report_html += f""" """ else: report_html += '' report_html += "
Pest NamePredicted Outbreak Month(s)SeverityPrecautionary Measures
{pest.get('pest_name', 'N/A')} {pest.get('outbreak_months', 'N/A')} {pest.get('severity', 'N/A')} {pest.get('precautionary_measures', 'N/A')}
No specific pest predictions available.
" report_html += "

Pest Avoidance Practices

" report_html += "

Specific Agricultural Best Practices

" report_html += f"

Potential Damage from Predicted Pests

{report_data.get('predicted_pest_damage_info', 'No information available.')}

" # Generate summary for voice (short summary) summary = f"Pest Outbreak Report for {location.get('derived_location', 'your location')}. " summary += report_data.get('agricultural_inputs_analysis', '')[:200] + "... " if pest_table: summary += f"Predicted pests: " + ', '.join([p.get('pest_name', '') for p in pest_table]) + ". " summary += f"Severity: " + ', '.join([p.get('severity', '') for p in pest_table]) + ". " summary += report_data.get('predicted_pest_damage_info', '')[:200] # Generate audio file using gTTS tts = gTTS(summary, lang='en') audio_filename = f"pest_report_{location.get('latitude', 'lat')}_{location.get('longitude', 'lon')}.mp3" audio_path = os.path.join(app.config['AUDIO_FOLDER'], audio_filename) tts.save(audio_path) audio_url = f"/static/audio/{audio_filename}" # Build audio button HTML separately to avoid nested f-string and backslash issues audio_player_html = "" if audio_url: audio_player_html = ( '
' '' '
' '
' ).format(audio_url) html_output = f""" Pest Outbreak Dashboard Report
{audio_player_html} {report_html}
""" return Response(html_output, mimetype="text/html") if __name__ == '__main__': app.run(debug=True)