import asyncio import io from typing import TYPE_CHECKING import google.generativeai as genai from PIL import Image from .config import get_settings if TYPE_CHECKING: from app.models.yield_model import YieldInput # Lazily built after .env is present (import order can run this before main.py runs) _gemini_vision_model = None _gemini_text_model = None # Re-init when API key or model id changes _configured_signature: tuple[str | None, str | None] = (None, None) def _ensure_gemini_models(): """Configure google-generativeai on first use; pick up new keys after .env edits.""" global _gemini_vision_model, _gemini_text_model, _configured_signature settings = get_settings() key = settings.GEMINI_API_KEY model_name = settings.GEMINI_MODEL if not key or key == "YOUR_GEMINI_API_KEY_NOT_SET": _gemini_vision_model = None _gemini_text_model = None _configured_signature = (None, None) return None, None if _configured_signature == (key, model_name) and _gemini_vision_model is not None: return _gemini_vision_model, _gemini_text_model try: genai.configure(api_key=key) _gemini_vision_model = genai.GenerativeModel(model_name) _gemini_text_model = genai.GenerativeModel(model_name) _configured_signature = (key, model_name) except Exception as e: print(f"AI_SERVICES: Error configuring Gemini: {e}") _gemini_vision_model = None _gemini_text_model = None _configured_signature = (None, None) return None, None return _gemini_vision_model, _gemini_text_model async def run_gemini_text_prompt(prompt: str) -> str: """Single-turn Gemini text call (lazy init, off event loop). Used by agents.""" _, model = _ensure_gemini_models() if not model: return "Error: Gemini text model is not configured." def _run(): r = model.generate_content(prompt) return r.text try: return await asyncio.to_thread(_run) except ValueError: return "Error: No text returned (content may have been blocked)." except Exception as e: return f"Error: {str(e)}" # --- Disease Prediction Function --- async def get_disease_prediction(image_bytes: bytes) -> str: """ Analyzes an image using Gemini Vision model to detect crop diseases. """ gemini_vision_model, _ = _ensure_gemini_models() if not gemini_vision_model: return "Error: Gemini Vision model is not configured." try: img = Image.open(io.BytesIO(image_bytes)) img = img.convert("RGB") w, h = img.size max_side = 1280 if max(w, h) > max_side: scale = max_side / float(max(w, h)) img = img.resize( (max(1, int(w * scale)), max(1, int(h * scale))), Image.Resampling.LANCZOS, ) prompt = f"""Analyze the attached image of a plant leaf from the Baramati, Maharashtra region. Identify potential diseases or pests. Describe the visible symptoms clearly. Suggest 1-2 brief, practical, and sustainable management/treatment options suitable for a local farmer. If the image is unclear or not a plant, state that clearly. Format the response clearly, perhaps using bullet points for symptoms and suggestions. Respond in English. """ # Run sync SDK off the asyncio event loop so other API requests stay responsive def _generate(): return gemini_vision_model.generate_content([prompt, img]) response = await asyncio.to_thread(_generate) try: return response.text except ValueError: return ( "Error: Could not read model response (content may have been blocked). " "Try a clearer plant photo." ) except Exception as e: print(f"Error in Gemini disease prediction: {e}") # Consider more specific error handling based on potential Gemini exceptions return f"Error analyzing image with Gemini: {str(e)}" # --- Yield Prediction Function --- async def get_yield_estimate(yield_input: 'YieldInput') -> str: """ Generates a yield estimate using Gemini based on farmer's input. """ _, gemini_text_model = _ensure_gemini_models() if not gemini_text_model: return "Error: Gemini text model is not configured." try: # Include regional context in the prompt location_context = "Baramati, Maharashtra, India" prompt = f"""Act as an agricultural assistant for a farmer in {location_context}. Based on the following inputs: - Crop: {yield_input.crop_type} - Area: {yield_input.area} - Region Details: {yield_input.region} (within {location_context}) - Soil Type: {yield_input.soil or 'Not specified'} - Recent/Expected Weather: {yield_input.weather or 'Not specified'} Provide a realistic estimated yield range (e.g., in quintals per acre or tonnes per hectare, specify the unit clearly). Briefly explain the key factors (like weather, soil, crop type in this region) influencing this estimate in 2-3 short bullet points. Keep the explanation simple and practical for a farmer. Respond in English. """ # Generate content # Note: Check Gemini API documentation for latest recommended methods response = gemini_text_model.generate_content(prompt) # Check for safety ratings or blocks if necessary (depends on API version/config) # if response.prompt_feedback and response.prompt_feedback.block_reason: # return f"Error: Content blocked due to {response.prompt_feedback.block_reason}" return response.text except Exception as e: print(f"Error in Gemini yield prediction: {e}") # Consider more specific error handling based on potential Gemini exceptions return f"Error generating yield estimate with Gemini: {str(e)}" # --- Voice Command Processing Function --- async def process_voice_command_ai(transcript: str, language: str = "en") -> str: """ Processes a voice transcript using Gemini to understand intent and generate a response. """ _, gemini_text_model = _ensure_gemini_models() if not gemini_text_model: return "Error: Gemini text model is not configured." # Basic language code mapping (expand as needed) lang_map = {"en": "English", "hi": "Hindi", "mr": "Marathi"} language_name = lang_map.get(language, "English") # Default to English if code unknown # Context about the app's capabilities app_capabilities = """ The CropPilot app can: 1. Analyze an uploaded image of a plant to detect diseases (triggered by asking about the 'last image' or 'this picture'). 2. Predict crop yield based on inputs like crop type, area, region, soil, weather. 3. Provide mock information about local market prices for crops like Wheat and Onion in Baramati. """ try: prompt = f"""You are the voice interface for the CropPilot agricultural app, assisting a farmer in Baramati, Maharashtra. The user, speaking {language_name}, said: "{transcript}" App Capabilities: {app_capabilities} Your tasks: 1. Understand the user's intent based on their statement. Does the user want to: - Get analysis of the last uploaded image? - Ask for a yield prediction (they might mention crop, area etc.)? - Ask about market prices (they might mention crop names)? - Something else (greet, ask for help)? 2. Generate a concise and helpful response **in {language_name}**. 3. If the intent is clear and relates to an app capability: - For image analysis: Respond like "Okay, analyzing the last uploaded image..." or ask for the image if needed. (The actual analysis happens separately). - For yield prediction: If they provided details, acknowledge them. If not, ask for necessary details like crop type, area, etc. - For market prices: Provide the mock data if they ask for Wheat/Onion prices in Baramati, otherwise state which prices are available. 4. If the intent is unclear or unrelated to app capabilities, politely state what the app can do or ask for clarification. 5. Keep the response relatively short and easy to understand for a voice interaction. """ # Generate content # Note: Check Gemini API documentation for latest recommended methods response = gemini_text_model.generate_content(prompt) # Check for safety ratings or blocks if necessary (depends on API version/config) # if response.prompt_feedback and response.prompt_feedback.block_reason: # return f"Error: Content blocked due to {response.prompt_feedback.block_reason}" return response.text except Exception as e: print(f"Error in Gemini voice processing: {e}") # Consider more specific error handling based on potential Gemini exceptions return f"Error processing voice command with Gemini: {str(e)}" # --- (Optional) Market Data AI Summary --- async def get_market_summary_ai(market_data: list) -> str: """ Generates a brief summary of market data using Gemini. """ _, gemini_text_model = _ensure_gemini_models() if not gemini_text_model: return "Error: Gemini text model is not configured." if not market_data: return "No market data available to summarize." try: data_string = "\n".join([f"- {item['crop']}: {item['price_per_quintal']} INR/quintal at {item['location']}" for item in market_data]) prompt = f"""Here is some recent market data from Baramati Mandi: {data_string} Provide a very brief (1-2 sentence) summary highlighting any notable price points or trends based ONLY on this data. Respond in English. """ # Generate content # Note: Check Gemini API documentation for latest recommended methods response = gemini_text_model.generate_content(prompt) # Check for safety ratings or blocks if necessary (depends on API version/config) # if response.prompt_feedback and response.prompt_feedback.block_reason: # return f"Error: Content blocked due to {response.prompt_feedback.block_reason}" return response.text except Exception as e: print(f"Error generating market summary: {e}") # Consider more specific error handling based on potential Gemini exceptions return f"Error generating market summary with Gemini: {str(e)}"