import google.generativeai as genai import json import os from datetime import datetime, date from typing import Dict, Any, Optional import logging # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class GeminiAIService: """Service class for Gemini AI integration""" def __init__(self, api_key: str): """Initialize Gemini AI service""" self.api_key = api_key genai.configure(api_key=api_key) self.model = genai.GenerativeModel('gemini-2.0-flash') def generate_daily_advisory(self, farmer_data: Dict[str, Any], farm_data: Dict[str, Any], soil_data: Dict[str, Any], weather_data: Dict[str, Any], current_date: str = None) -> Dict[str, str]: """ Generate daily farming advisory using Gemini AI Args: farmer_data: Farmer information farm_data: Farm details including crops soil_data: Current soil parameters weather_data: Current weather data current_date: Date for the advisory (default: today) Returns: Dictionary with task_to_do, task_to_avoid, and reason_explanation """ if current_date is None: current_date = date.today().strftime('%Y-%m-%d') # Prepare the prompt for Gemini prompt = self._create_advisory_prompt(farmer_data, farm_data, soil_data, weather_data, current_date) try: # Generate response from Gemini response = self.model.generate_content(prompt) # Parse the response advisory = self._parse_gemini_response(response.text) logger.info(f"Generated advisory for farmer {farmer_data.get('name')} on {current_date}") return advisory except Exception as e: logger.error(f"Error generating advisory: {str(e)}") return self._get_fallback_advisory() def _create_advisory_prompt(self, farmer_data: Dict[str, Any], farm_data: Dict[str, Any], soil_data: Dict[str, Any], weather_data: Dict[str, Any], current_date: str) -> str: """Create a detailed prompt for Gemini AI""" crop_types = ', '.join(farm_data.get('crop_types', [])) prompt = f""" You are an expert agricultural advisor. Generate a daily farming advisory for the following farmer: FARMER INFORMATION: - Name: {farmer_data.get('name', 'Unknown')} - Farm Name: {farm_data.get('farm_name', 'Unknown')} - Farm Size: {farm_data.get('farm_size', 0)} acres - Crops: {crop_types} - Irrigation Type: {farm_data.get('irrigation_type', 'Unknown')} SOIL CONDITIONS: - Soil Type: {soil_data.get('soil_type', 'Unknown')} - pH Level: {soil_data.get('ph_level', 'Unknown')} - Nitrogen: {soil_data.get('nitrogen_level', 'Unknown')} ppm - Phosphorus: {soil_data.get('phosphorus_level', 'Unknown')} ppm - Potassium: {soil_data.get('potassium_level', 'Unknown')} ppm - Moisture: {soil_data.get('moisture_percentage', 'Unknown')}% WEATHER CONDITIONS: - Date: {current_date} - Temperature: {weather_data.get('main', {}).get('temp_min', 'Unknown')}°C - {weather_data.get('main', {}).get('temp_max', 'Unknown')}°C - Humidity: {weather_data.get('main', {}).get('humidity', 'Unknown')}% - Wind Speed: {weather_data.get('wind', {}).get('speed', 0) * 3.6:.1f} km/h - Weather Condition: {weather_data.get('weather', [{}])[0].get('description', 'Unknown').title()} INSTRUCTIONS: Please provide a daily farming advisory in the following JSON format: {{ "task_to_do": "Specific task the farmer should do today", "task_to_avoid": "Specific task the farmer should avoid today", "reason_explanation": "Simple, farmer-friendly explanation for the recommendations", "crop_stage": "Current estimated crop growth stage" }} Consider the weather conditions, soil parameters, crop requirements, and seasonal farming practices. Provide practical, actionable advice that a farmer can easily understand and implement. Make the language simple and direct. """ return prompt def _parse_gemini_response(self, response_text: str) -> Dict[str, str]: """Parse Gemini's response and extract advisory information""" try: # Try to extract JSON from the response # Look for JSON-like content in the response start_idx = response_text.find('{') end_idx = response_text.rfind('}') + 1 if start_idx != -1 and end_idx != -1: json_text = response_text[start_idx:end_idx] advisory = json.loads(json_text) # Validate required fields required_fields = ['task_to_do', 'task_to_avoid', 'reason_explanation'] for field in required_fields: if field not in advisory: advisory[field] = "No recommendation available" return advisory else: # If no JSON found, try to parse manually return self._manual_parse_response(response_text) except json.JSONDecodeError: # If JSON parsing fails, try manual parsing return self._manual_parse_response(response_text) except Exception as e: logger.error(f"Error parsing Gemini response: {str(e)}") return self._get_fallback_advisory() def _manual_parse_response(self, response_text: str) -> Dict[str, str]: """Manually parse response if JSON parsing fails""" lines = response_text.split('\n') advisory = { 'task_to_do': 'No specific task recommended', 'task_to_avoid': 'No specific task to avoid', 'reason_explanation': 'Advisory not available', 'crop_stage': 'Unknown' } for line in lines: line = line.strip() if line.startswith('✅') or 'task_to_do' in line.lower(): advisory['task_to_do'] = line.replace('✅', '').strip() elif line.startswith('❌') or 'task_to_avoid' in line.lower(): advisory['task_to_avoid'] = line.replace('❌', '').strip() elif line.startswith('ℹ️') or 'reason' in line.lower(): advisory['reason_explanation'] = line.replace('ℹ️', '').strip() return advisory def _get_fallback_advisory(self) -> Dict[str, str]: """Provide a fallback advisory when Gemini fails""" return { 'task_to_do': 'Check crop condition and water levels', 'task_to_avoid': 'Avoid heavy farm work during extreme weather', 'reason_explanation': 'General farming best practices for safety and crop health', 'crop_stage': 'Unknown' } def generate_year_plan(self, farmer_data: Dict[str, Any], farm_data: Dict[str, Any], soil_data: Dict[str, Any]) -> Dict[str, Any]: """Generate a comprehensive year-long farming plan with detailed HTML formatting""" crop_types = ', '.join(farm_data.get('crop_types', [])) farm_size = farm_data.get('farm_size', 0) irrigation_type = farm_data.get('irrigation_type', 'Unknown') soil_type = soil_data.get('soil_type', 'Unknown') # Construct crop details with areas (assume equal distribution if not specified) crops = farm_data.get('crop_types', []) if crops and farm_size: area_per_crop = farm_size / len(crops) crop_details = ", ".join([f"{crop} ({area_per_crop:.1f} acres)" for crop in crops]) else: crop_details = crop_types or "Mixed crops" prompt = f""" You are a skilled agriculture expert and data analyst. I am providing you with the following farm details: FARM INFORMATION: - Farmer Name: {farmer_data.get('name', 'Unknown')} - Farm Name: {farm_data.get('farm_name', 'Unknown')} - Total Farm Area: {farm_size} acres - Location/Address: {farmer_data.get('address', 'Unknown')} - Irrigation Type: {irrigation_type} - Soil Type: {soil_type} - Soil pH: {soil_data.get('ph_level', 'Unknown')} - Nitrogen Level: {soil_data.get('nitrogen_level', 'Unknown')} ppm - Phosphorus Level: {soil_data.get('phosphorus_level', 'Unknown')} ppm - Potassium Level: {soil_data.get('potassium_level', 'Unknown')} ppm - Current Crops: {crop_details} Using the above input, please generate a fully formatted HTML page with enhanced CSS styling that includes the following sections: 1. **Main Heading:** - A centrally aligned, bold heading in green titled "Comprehensive Yearly Farming Plan 2025" with proper spacing and clear font sizes. 2. **Farm Overview Statistics:** - A left-aligned, blue bold subheading "Farm Overview & Current Status". - Below it, include a green table with columns for: - Parameter - Current Value - Recommended Range - Status (Good/Needs Improvement) - Include farm size, soil parameters, irrigation type, etc. 3. **Monthly Farming Calendar (12-Month Plan):** - A left-aligned, blue bold subheading titled "Monthly Farming Calendar". - Create a comprehensive green table displaying month-wise activities for the entire year: - Month - Primary Activities - Crop Operations - Fertilizer Schedule - Irrigation Requirements - Expected Weather Considerations - Include specific activities for each month based on crop cycles, seasons (Rabi/Kharif), and local agricultural practices. 4. **Crop-wise Annual Strategy:** - A left-aligned, blue bold subheading "Crop-wise Annual Strategy". - Add a detailed green table showing for each crop: - Crop Name - Sowing Period - Growing Duration - Harvest Period - Expected Yield (per acre) - Estimated Revenue - Key Care Instructions 5. **Financial Projections:** - Include a section with a blue left-aligned subheading "Annual Financial Forecast". - Present tables showing: - Expected Production Costs (seeds, fertilizers, labor, etc.) - Projected Revenue by crop - Estimated Profit Margins - Monthly cash flow predictions 6. **Soil Management Plan:** - A blue left-aligned subheading "Soil Health & Fertilizer Schedule". - Create tables for: - Soil testing schedule - Fertilizer application timeline - Organic matter enhancement plan - pH management strategies 7. **Risk Management & Weather Planning:** - A blue left-aligned subheading "Risk Management Strategies". - Include tables for: - Seasonal weather challenges - Pest and disease prevention calendar - Backup crop strategies - Insurance and financial protection 8. **Key Recommendations & Action Items:** - At the bottom, include a bullet-point list with specific, actionable recommendations. - Style each bullet point in bold with yellow highlights on key dates, quantities, and financial figures. - Include immediate actions, seasonal priorities, and long-term improvements. IMPORTANT REQUIREMENTS: - Generate the entire response in Hindi language. All text, headings, table headers, and content should be in Hindi. - Use realistic data based on Indian agricultural practices and the provided farm details. - Make tables interactive, beautiful, and colorful with proper spacing and margins. - Increase font size for headings and optimize spacing for readability. - Do not include any extra spacing at the beginning or end of the response. - Ensure all content is properly formatted as HTML with appropriate CSS styling. - Include specific dates, quantities, and actionable advice suitable for Indian farmers. - Consider Indian seasons (Rabi, Kharif, Zayad) and local agricultural practices. - Use metric measurements and Indian rupee currency where applicable. Generate a comprehensive, professional, and farmer-friendly yearly plan that can guide the farmer throughout the entire agricultural year. """ try: response = self.model.generate_content(prompt) # Return the comprehensive HTML plan html_plan = response.text.strip() return { 'plan': html_plan, 'type': 'comprehensive_html', 'generated_at': datetime.now().isoformat(), 'ai_generated': True, 'farmer_name': farmer_data.get('name'), 'farm_name': farm_data.get('farm_name') } except Exception as e: logger.error(f"Error generating comprehensive year plan: {str(e)}") return { 'plan': self._get_fallback_yearly_plan(farmer_data, farm_data, soil_data), 'type': 'fallback_html', 'generated_at': datetime.now().isoformat(), 'ai_generated': False } def _get_fallback_yearly_plan(self, farmer_data: Dict[str, Any], farm_data: Dict[str, Any], soil_data: Dict[str, Any]) -> str: """Generate a fallback yearly plan in HTML format when AI fails""" crop_types = ', '.join(farm_data.get('crop_types', ['Mixed crops'])) return f"""
वार्षिक खेती योजना 2025
खेत की जानकारी
विवरणमान
किसान का नाम{farmer_data.get('name', 'अज्ञात')}
खेत का नाम{farm_data.get('farm_name', 'अज्ञात')}
कुल क्षेत्रफल{farm_data.get('farm_size', 0)} एकड़
फसलें{crop_types}
सिंचाई प्रकार{farm_data.get('irrigation_type', 'अज्ञात')}
मासिक गतिविधि कैलेंडर
महीनामुख्य गतिविधियांसिंचाई
जनवरीरबी फसल की देखभाल, खाद डालनाआवश्यकता अनुसार
फरवरीफसल की निगरानी, कीट नियंत्रणनियमित
मार्चरबी फसल की कटाई तैयारीकम
अप्रैलरबी फसल कटाई, खरीफ की तैयारीगर्मी के कारण अधिक
मईखेत की तैयारी, बीज खरीदारीअधिक
जूनखरीफ फसल बुआईमानसून शुरुआत
जुलाईखरीफ फसल देखभालमानसून
अगस्तनिराई-गुड़ाई, खादमानसून
सितंबरफसल निगरानीमध्यम
अक्टूबरखरीफ कटाई, रबी तैयारीआवश्यकता अनुसार
नवंबररबी फसल बुआईनियमित
दिसंबररबी फसल देखभालठंड में कम
मुख्य सुझाव:
• मिट्टी की जांच वर्ष में दो बार कराएं
• उन्नत बीजों का प्रयोग करें
• समय पर सिंचाई और खाद डालें
• कीट-रोग की नियमित निगरानी करें
• मौसम की जानकारी रखें
""" def format_sms_message(farmer_name: str, advisory: Dict[str, str]) -> str: """Format the advisory as an SMS message""" message = f"Good Morning, {farmer_name} 🌱\n" message += f"✅ Task: {advisory.get('task_to_do', 'No task')}\n" message += f"❌ Avoid: {advisory.get('task_to_avoid', 'No restrictions')}\n" if advisory.get('reason_explanation'): message += f"ℹ️ {advisory.get('reason_explanation')}" return message