Create app.py

app.py

@@ -0,0 +1,717 @@
+# app.py - FastAPI version
+
+import json
+import logging
+import os
+from typing import Dict, Any, Optional, List, Union
+from dataclasses import dataclass, asdict
+from contextlib import asynccontextmanager
+import uuid
+from datetime import datetime, timedelta
+
+from fastapi import FastAPI, HTTPException, Depends, status, BackgroundTasks
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field, validator
+from openai import OpenAI, AsyncOpenAI
+from dotenv import load_dotenv
+import asyncio
+from cachetools import TTLCache
+
+# Import your modules
+from easy_agents import EASYFARMS_FUNCTION_SCHEMAS, EasyFarmsAgent, execute_easyfarms_function
+from alert import chat_with_weather_assistant, WEATHER_TOOLS
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Load environment variables
+load_dotenv()
+
+# ===================== Configuration =====================
+
+@dataclass
+class Config:
+    """Configuration settings"""
+    api_key: str
+    api_url: str
+    model_name: str
+    max_retries: int = 3
+    temperature: float = 0.7
+    
+    @classmethod
+    def from_env(cls):
+        """Load configuration from environment variables"""
+        return cls(
+            api_key=os.getenv("API_KEY"),
+            api_url=os.getenv("API_URL"),
+            model_name=os.getenv("MODEL_NAME", "gpt-4-turbo-preview")
+        )
+
+# ===================== Pydantic Models =====================
+
+class QueryRequest(BaseModel):
+    """Request model for query endpoint"""
+    message: str = Field(..., min_length=1, max_length=5000, description="User's query message")
+    session_id: Optional[str] = Field(None, description="Session ID for conversation continuity")
+    save_history: bool = Field(True, description="Whether to save this interaction in history")
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "message": "What crop should I grow with N=80, P=45, K=120, temperature 25°C, humidity 70%?",
+                "session_id": "123e4567-e89b-12d3-a456-426614174000",
+                "save_history": True
+            }
+        }
+
+class QueryResponse(BaseModel):
+    """Response model for query endpoint"""
+    response: str
+    session_id: str
+    timestamp: datetime
+    function_calls: Optional[List[Dict[str, Any]]] = None
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "response": "Based on the soil conditions...",
+                "session_id": "123e4567-e89b-12d3-a456-426614174000",
+                "timestamp": "2024-01-15T10:30:00",
+                "function_calls": []
+            }
+        }
+
+class CropRecommendationRequest(BaseModel):
+    """Request model for crop recommendation"""
+    nitrogen: int = Field(..., ge=0, le=200, description="Nitrogen content (N)")
+    phosphorus: int = Field(..., ge=0, le=200, description="Phosphorus content (P)")
+    potassium: int = Field(..., ge=0, le=200, description="Potassium content (K)")
+    temperature: float = Field(..., ge=-10, le=50, description="Temperature in Celsius")
+    humidity: float = Field(..., ge=0, le=100, description="Humidity percentage")
+    ph: float = Field(6.5, ge=0, le=14, description="Soil pH value")
+    
+    class Config:
+        schema_extra = {
+            "example": {
+                "nitrogen": 80,
+                "phosphorus": 45,
+                "potassium": 120,
+                "temperature": 25.0,
+                "humidity": 70.0,
+                "ph": 6.5
+            }
+        }
+
+class FertilizerRecommendationRequest(BaseModel):
+    """Request model for fertilizer recommendation"""
+    crop: str = Field(..., description="Crop type")
+    soil_type: str = Field(..., description="Type of soil")
+    nitrogen: int = Field(..., ge=0, le=200, description="Current nitrogen content")
+    phosphorus: int = Field(..., ge=0, le=200, description="Current phosphorus content")
+    potassium: int = Field(..., ge=0, le=200, description="Current potassium content")
+    temperature: Optional[float] = Field(None, description="Temperature in Celsius")
+    humidity: Optional[float] = Field(None, description="Humidity percentage")
+    moisture: Optional[float] = Field(None, description="Soil moisture percentage")
+
+class WeatherAlertRequest(BaseModel):
+    """Request model for weather alerts"""
+    location: Optional[str] = Field(None, description="Location for weather alert")
+    include_forecast: bool = Field(True, description="Include weather forecast")
+
+class PlantDiseaseRequest(BaseModel):
+    """Request model for plant disease detection"""
+    symptoms: str = Field(..., description="Description of plant symptoms")
+    crop_type: Optional[str] = Field(None, description="Type of crop affected")
+    image_url: Optional[str] = Field(None, description="URL to plant image")
+
+class SessionInfo(BaseModel):
+    """Session information"""
+    session_id: str
+    created_at: datetime
+    last_activity: datetime
+    message_count: int
+    
+class HealthResponse(BaseModel):
+    """Health check response"""
+    status: str
+    version: str
+    timestamp: datetime
+
+# ===================== Session Manager =====================
+
+class SessionManager:
+    """Manages user sessions and conversation history"""
+    
+    def __init__(self, ttl_hours: int = 24):
+        self.sessions: Dict[str, Dict[str, Any]] = {}
+        self.ttl = timedelta(hours=ttl_hours)
+    
+    def create_session(self) -> str:
+        """Create a new session"""
+        session_id = str(uuid.uuid4())
+        self.sessions[session_id] = {
+            "id": session_id,
+            "created_at": datetime.now(),
+            "last_activity": datetime.now(),
+            "history": [],
+            "message_count": 0
+        }
+        return session_id
+    
+    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
+        """Get session by ID"""
+        if session_id in self.sessions:
+            session = self.sessions[session_id]
+            # Check if session has expired
+            if datetime.now() - session["last_activity"] > self.ttl:
+                del self.sessions[session_id]
+                return None
+            session["last_activity"] = datetime.now()
+            return session
+        return None
+    
+    def update_session(self, session_id: str, history_entry: Dict[str, Any]):
+        """Update session history"""
+        if session_id in self.sessions:
+            self.sessions[session_id]["history"].append(history_entry)
+            self.sessions[session_id]["message_count"] += 1
+            self.sessions[session_id]["last_activity"] = datetime.now()
+    
+    def clear_session(self, session_id: str):
+        """Clear session history"""
+        if session_id in self.sessions:
+            self.sessions[session_id]["history"] = []
+            self.sessions[session_id]["message_count"] = 0
+    
+    def cleanup_expired_sessions(self):
+        """Remove expired sessions"""
+        current_time = datetime.now()
+        expired = [
+            sid for sid, session in self.sessions.items()
+            if current_time - session["last_activity"] > self.ttl
+        ]
+        for sid in expired:
+            del self.sessions[sid]
+        return len(expired)
+
+# ===================== EasyFarms Assistant =====================
+
+class EasyFarmsAssistant:
+    """Enhanced EasyFarms AI Assistant with async support"""
+    
+    def __init__(self, config: Optional[Config] = None):
+        """Initialize the assistant with configuration"""
+        self.config = config or Config.from_env()
+        
+        # Async client for FastAPI
+        self.async_client = AsyncOpenAI(
+            api_key=self.config.api_key,
+            base_url=self.config.api_url
+        )
+        
+        # Sync client for backward compatibility
+        self.sync_client = OpenAI(
+            api_key=self.config.api_key,
+            base_url=self.config.api_url
+        )
+        
+        # Initialize tools
+        self.tools = self._initialize_tools()
+        
+        # System prompts
+        self.system_prompt = """
+You are the HAl AI assistant for EasyForms Agritech Solutions. Your task is to provide users with clear, concise, and actionable responses regarding agriculture, crop management, production, treatment, weather alerts, and related queries.
+
+Core Capabilities:
+- Crop recommendations based on soil and weather conditions
+- Fertilizer recommendations for specific crops
+- Plant disease detection and treatment advice
+- Weather alerts and forecasts for farming decisions
+- General agricultural guidance
+
+Rules:
+1. Check if any relevant function_tools or datasets are available for this query.
+2. If available, use the functions to fetch information and generate the final user-facing response.
+3. If the functions or data are unavailable, do not stop; instead, generate a general, well-reasoned response based on your own knowledge.
+4. Keep the response simple, smooth, well-pointed, and concise.
+5. Structure the response with bullet points or numbered steps where helpful.
+6. Provide practical, actionable advice a user can implement immediately.
+7. Use English or Hindi based on user preference.
+8. If any information is uncertain, mention it clearly and suggest alternatives.
+9. For weather-related queries, prioritize safety and timely alerts.
+"""
+        
+        self.final_system = """
+You are the final response assistant for EasyForms Agritech Solutions and your name is HAL AI. Use the outputs from previous function calls to generate a clear, concise, actionable response for the user.
+
+Rules:
+1. Combine the function outputs and your own reasoning to answer the query.
+2. Keep responses simple, smooth, well-pointed, and concise.
+3. Structure response with headings or bullet points if helpful.
+4. Provide practical advice that a farmer or user can implement immediately.
+5. If some data is missing, clearly state it and offer alternatives.
+6. Use English or Hindi based on the user preference.
+7. For weather alerts, emphasize urgency and protective measures.
+"""
+        
+        # Cache for function results
+        self.cache = TTLCache(maxsize=100, ttl=300)  # 5 minute cache
+    
+    def _initialize_tools(self) -> List[Dict]:
+        """Initialize and convert function schemas to new tools format"""
+        tools = []
+        
+        # Convert EasyFarms schemas
+        for schema in EASYFARMS_FUNCTION_SCHEMAS:
+            tool = {
+                "type": "function",
+                "function": {
+                    "name": schema["name"],
+                    "description": schema["description"],
+                    "parameters": schema["parameters"]
+                }
+            }
+            tools.append(tool)
+        
+        # Add weather tools
+        tools.extend(WEATHER_TOOLS)
+        
+        return tools
+    
+    async def call_function_async(self, function_name: str, arguments: Dict) -> Any:
+        """Async version of function calling"""
+        try:
+            # Check cache first
+            cache_key = f"{function_name}:{json.dumps(arguments, sort_keys=True)}"
+            if cache_key in self.cache:
+                logger.info(f"Cache hit for {function_name}")
+                return self.cache[cache_key]
+            
+            # Execute function in thread pool to avoid blocking
+            loop = asyncio.get_event_loop()
+            
+            if function_name.startswith("get_weather"):
+                result = await loop.run_in_executor(
+                    None, 
+                    self._execute_weather_function, 
+                    function_name, 
+                    arguments
+                )
+            else:
+                result = await loop.run_in_executor(
+                    None,
+                    execute_easyfarms_function,
+                    function_name,
+                    arguments
+                )
+            
+            # Cache result
+            self.cache[cache_key] = result
+            return result
+            
+        except Exception as e:
+            logger.error(f"Error executing function {function_name}: {e}")
+            return {"error": str(e)}
+    
+    def _execute_weather_function(self, function_name: str, kwargs: Dict):
+        """Execute weather functions"""
+        from alert import execute_function
+        return execute_function(function_name, kwargs)
+    
+    async def process_query_async(
+        self, 
+        user_message: str, 
+        conversation_history: List[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """Process user query asynchronously"""
+        try:
+            messages = [
+                {"role": "system", "content": self.system_prompt},
+                {"role": "user", "content": user_message}
+            ]
+            
+            # Add conversation history if exists
+            if conversation_history:
+                # Include last 5 interactions for context
+                messages = [messages[0]] + conversation_history[-10:] + [messages[1]]
+            
+            # First API call
+            response = await self.async_client.chat.completions.create(
+                model=self.config.model_name,
+                messages=messages,
+                tools=self.tools,
+                tool_choice="auto",
+                temperature=self.config.temperature
+            )
+            
+            message = response.choices[0].message
+            function_calls_made = []
+            
+            # Check if tool needs to be called
+            if hasattr(message, 'tool_calls') and message.tool_calls:
+                messages.append(message)
+                
+                # Execute all tool calls
+                for tool_call in message.tool_calls:
+                    function_name = tool_call.function.name
+                    function_args = json.loads(tool_call.function.arguments)
+                    
+                    logger.info(f"Calling function: {function_name} with args: {function_args}")
+                    
+                    # Call function asynchronously
+                    function_result = await self.call_function_async(function_name, function_args)
+                    
+                    function_calls_made.append({
+                        "name": function_name,
+                        "arguments": function_args,
+                        "result": function_result
+                    })
+                    
+                    # Add function result to messages
+                    messages.append({
+                        "role": "tool",
+                        "tool_call_id": tool_call.id,
+                        "content": json.dumps(function_result)
+                    })
+                
+                # Add final system prompt
+                messages.append({
+                    "role": "system",
+                    "content": self.final_system
+                })
+                
+                # Get final response
+                final_response = await self.async_client.chat.completions.create(
+                    model=self.config.model_name,
+                    messages=messages,
+                    temperature=self.config.temperature
+                )
+                
+                response_content = final_response.choices[0].message.content
+            else:
+                response_content = message.content
+            
+            return {
+                "response": response_content,
+                "function_calls": function_calls_made
+            }
+            
+        except Exception as e:
+            logger.error(f"Error processing query: {e}")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=f"Error processing query: {str(e)}"
+            )
+
+# ===================== FastAPI App =====================
+
+# App lifespan manager
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Manage app lifecycle"""
+    # Startup
+    logger.info("Starting EasyFarms API...")
+    app.state.assistant = EasyFarmsAssistant()
+    app.state.session_manager = SessionManager()
+    
+    # Background task to cleanup sessions
+    async def cleanup_sessions():
+        while True:
+            await asyncio.sleep(3600)  # Run every hour
+            expired = app.state.session_manager.cleanup_expired_sessions()
+            if expired > 0:
+                logger.info(f"Cleaned up {expired} expired sessions")
+    
+    # Start background task
+    cleanup_task = asyncio.create_task(cleanup_sessions())
+    
+    yield
+    
+    # Shutdown
+    cleanup_task.cancel()
+    logger.info("Shutting down EasyFarms API...")
+
+# Create FastAPI app
+app = FastAPI(
+    title="EasyFarms Agricultural Assistant API",
+    description="AI-powered agricultural assistant for crop management, fertilizer recommendations, and weather alerts",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Configure appropriately for production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ===================== Dependencies =====================
+
+def get_assistant() -> EasyFarmsAssistant:
+    """Get assistant instance"""
+    return app.state.assistant
+
+def get_session_manager() -> SessionManager:
+    """Get session manager instance"""
+    return app.state.session_manager
+
+# ===================== API Endpoints =====================
+
+@app.get("/", response_model=HealthResponse)
+async def root():
+    """Root endpoint - health check"""
+    return HealthResponse(
+        status="healthy",
+        version="1.0.0",
+        timestamp=datetime.now()
+    )
+
+@app.get("/health", response_model=HealthResponse)
+async def health_check():
+    """Health check endpoint"""
+    return HealthResponse(
+        status="healthy",
+        version="1.0.0",
+        timestamp=datetime.now()
+    )
+
+@app.post("/api/query", response_model=QueryResponse)
+async def process_query(
+    request: QueryRequest,
+    assistant: EasyFarmsAssistant = Depends(get_assistant),
+    session_manager: SessionManager = Depends(get_session_manager)
+):
+    """
+    Process a general query to the agricultural assistant
+    """
+    # Handle session
+    if request.session_id:
+        session = session_manager.get_session(request.session_id)
+        if not session:
+            request.session_id = session_manager.create_session()
+            session = session_manager.get_session(request.session_id)
+    else:
+        request.session_id = session_manager.create_session()
+        session = session_manager.get_session(request.session_id)
+    
+    # Process query
+    result = await assistant.process_query_async(
+        request.message,
+        conversation_history=session["history"] if session else None
+    )
+    
+    # Update session history if requested
+    if request.save_history and session:
+        session_manager.update_session(
+            request.session_id,
+            {"role": "user", "content": request.message}
+        )
+        session_manager.update_session(
+            request.session_id,
+            {"role": "assistant", "content": result["response"]}
+        )
+    
+    return QueryResponse(
+        response=result["response"],
+        session_id=request.session_id,
+        timestamp=datetime.now(),
+        function_calls=result.get("function_calls")
+    )
+
+@app.post("/api/crop-recommendation")
+async def get_crop_recommendation(
+    request: CropRecommendationRequest,
+    assistant: EasyFarmsAssistant = Depends(get_assistant)
+):
+    """
+    Get crop recommendation based on soil and weather conditions
+    """
+    query = f"What crop should I grow with N={request.nitrogen}, P={request.phosphorus}, K={request.potassium}, temperature {request.temperature}°C, humidity {request.humidity}%, pH {request.ph}?"
+    
+    result = await assistant.process_query_async(query)
+    
+    return {
+        "recommendation": result["response"],
+        "input_parameters": request.dict(),
+        "timestamp": datetime.now()
+    }
+
+@app.post("/api/fertilizer-recommendation")
+async def get_fertilizer_recommendation(
+    request: FertilizerRecommendationRequest,
+    assistant: EasyFarmsAssistant = Depends(get_assistant)
+):
+    """
+    Get fertilizer recommendation for specific crop and soil conditions
+    """
+    query_parts = [
+        f"I need fertilizer recommendation for {request.crop} in {request.soil_type} soil",
+        f"with N={request.nitrogen}, P={request.phosphorus}, K={request.potassium}"
+    ]
+    
+    if request.temperature:
+        query_parts.append(f"temperature {request.temperature}°C")
+    if request.humidity:
+        query_parts.append(f"humidity {request.humidity}%")
+    if request.moisture:
+        query_parts.append(f"moisture {request.moisture}%")
+    
+    query = ", ".join(query_parts)
+    result = await assistant.process_query_async(query)
+    
+    return {
+        "recommendation": result["response"],
+        "crop": request.crop,
+        "soil_type": request.soil_type,
+        "timestamp": datetime.now()
+    }
+
+@app.post("/api/weather-alert")
+async def get_weather_alert(
+    request: WeatherAlertRequest,
+    assistant: EasyFarmsAssistant = Depends(get_assistant)
+):
+    """
+    Get weather alerts for farming
+    """
+    location_str = f" for {request.location}" if request.location else ""
+    query = f"What are the current weather alerts and conditions{location_str}? How will this affect farming?"
+    
+    if request.include_forecast:
+        query += " Include the weather forecast."
+    
+    result = await assistant.process_query_async(query)
+    
+    return {
+        "alerts": result["response"],
+        "location": request.location or "Default location",
+        "timestamp": datetime.now()
+    }
+
+@app.post("/api/plant-disease")
+async def detect_plant_disease(
+    request: PlantDiseaseRequest,
+    assistant: EasyFarmsAssistant = Depends(get_assistant)
+):
+    """
+    Detect plant disease based on symptoms
+    """
+    query_parts = [f"My plants have these symptoms: {request.symptoms}"]
+    
+    if request.crop_type:
+        query_parts.append(f"The crop is {request.crop_type}.")
+    
+    query_parts.append("What could be the problem and how should I treat it?")
+    
+    query = " ".join(query_parts)
+    result = await assistant.process_query_async(query)
+    
+    return {
+        "diagnosis": result["response"],
+        "symptoms": request.symptoms,
+        "crop_type": request.crop_type,
+        "timestamp": datetime.now()
+    }
+
+@app.get("/api/session/{session_id}")
+async def get_session_info(
+    session_id: str,
+    session_manager: SessionManager = Depends(get_session_manager)
+):
+    """
+    Get information about a specific session
+    """
+    session = session_manager.get_session(session_id)
+    
+    if not session:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Session not found or expired"
+        )
+    
+    return SessionInfo(
+        session_id=session["id"],
+        created_at=session["created_at"],
+        last_activity=session["last_activity"],
+        message_count=session["message_count"]
+    )
+
+@app.delete("/api/session/{session_id}")
+async def clear_session(
+    session_id: str,
+    session_manager: SessionManager = Depends(get_session_manager)
+):
+    """
+    Clear session history
+    """
+    session = session_manager.get_session(session_id)
+    
+    if not session:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Session not found or expired"
+        )
+    
+    session_manager.clear_session(session_id)
+    
+    return {"message": "Session history cleared", "session_id": session_id}
+
+@app.get("/api/supported-options")
+async def get_supported_options(
+    assistant: EasyFarmsAssistant = Depends(get_assistant)
+):
+    """
+    Get all supported options for crops, fertilizers, etc.
+    """
+    result = await assistant.process_query_async("Show me all supported crop types and options")
+    
+    return {
+        "options": result["response"],
+        "timestamp": datetime.now()
+    }
+
+# ===================== Error Handlers =====================
+
+@app.exception_handler(HTTPException)
+async def http_exception_handler(request, exc):
+    """Handle HTTP exceptions"""
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={
+            "error": exc.detail,
+            "status_code": exc.status_code,
+            "timestamp": datetime.now().isoformat()
+        }
+    )
+
+@app.exception_handler(Exception)
+async def general_exception_handler(request, exc):
+    """Handle general exceptions"""
+    logger.error(f"Unhandled exception: {exc}")
+    return JSONResponse(
+        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        content={
+            "error": "Internal server error",
+            "status_code": 500,
+            "timestamp": datetime.now().isoformat()
+        }
+    )
+
+# ===================== Main =====================
+
+if __name__ == "__main__":
+    import uvicorn
+    
+    # Run the FastAPI app
+    uvicorn.run(
+        "app:app",
+        host="0.0.0.0",
+        port=7860,
+        reload=True,
+        log_level="info"
+    )