from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles import google.generativeai as genai from PIL import Image import io import os from typing import Optional, List import logging from pydantic import BaseModel import cv2 import numpy as np import traceback from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import time import asyncio from dotenv import load_dotenv import base64 # Load environment variables load_dotenv() # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) app = FastAPI( title="Tree Species and DBH Detection API", description="API for detecting tree species and estimating DBH using Google Gemini", version="1.0.0" ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Configure Gemini API - EXACT SAME AS YOUR PLANT DISEASE APP GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") if not GEMINI_API_KEY: logger.warning("GEMINI_API_KEY not found in environment variables") raise ValueError("GEMINI_API_KEY not found in environment variables!") else: genai.configure(api_key=GEMINI_API_KEY) # Using the SAME model as your plant disease app model = genai.GenerativeModel('models/gemini-2.5-flash') logger.info("Gemini API configured successfully with gemini-2.5-flash") class AnalysisResponse(BaseModel): species: str species_hindi: Optional[str] species_marathi: Optional[str] scientific_name: Optional[str] confidence: str height: Optional[str] canopy: Optional[str] girth: Optional[str] dbh_estimate: Optional[str] condition: Optional[str] characteristics: list[str] additional_info: Optional[str] recommendations: Optional[str] remarks: Optional[str] image_filename: Optional[str] = None class MultipleAnalysisResponse(BaseModel): total_images: int successful: int failed: int results: List[AnalysisResponse] errors: List[dict] def detect_blur(image: Image.Image, threshold: float = 100.0) -> tuple[bool, float]: """Detect if an image is blurry using Laplacian variance method""" try: img_array = np.array(image) if len(img_array.shape) == 3: gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY) else: gray = img_array laplacian = cv2.Laplacian(gray, cv2.CV_64F) variance = laplacian.var() is_blurry = variance < threshold return is_blurry, variance except Exception as e: logger.error(f"Error in blur detection: {str(e)}") return False, 0.0 def resize_image_if_needed(image: Image.Image, max_dimension: int = 2048) -> Image.Image: """Resize image if it's too large""" try: original_size = image.size if max(image.size) > max_dimension: image.thumbnail((max_dimension, max_dimension), Image.Resampling.LANCZOS) logger.info(f"Resized image from {original_size} to {image.size}") return image except Exception as e: logger.error(f"Error resizing image: {str(e)}") return image def parse_analysis_response(response_text: str, filename: str = None) -> AnalysisResponse: """Parse the Gemini API response text into structured data""" species = "Unknown" species_hindi = None species_marathi = None scientific_name = None confidence = "Low" height = None canopy = None girth = None dbh_estimate = None condition = None characteristics = [] additional_info = None recommendations = None remarks = None lines = response_text.split("\n") current_section = None for line in lines: line = line.strip() if line.startswith("ENGLISH NAME:"): species = line.replace("ENGLISH NAME:", "").strip() elif line.startswith("HINDI NAME:"): species_hindi = line.replace("HINDI NAME:", "").strip() elif line.startswith("MARATHI NAME:"): species_marathi = line.replace("MARATHI NAME:", "").strip() elif line.startswith("SCIENTIFIC NAME:"): scientific_name = line.replace("SCIENTIFIC NAME:", "").strip() elif line.startswith("CONFIDENCE:"): confidence = line.replace("CONFIDENCE:", "").strip() elif line.startswith("HEIGHT:"): height = line.replace("HEIGHT:", "").strip() elif line.startswith("CANOPY:"): canopy = line.replace("CANOPY:", "").strip() elif line.startswith("GIRTH:"): girth = line.replace("GIRTH:", "").strip() elif line.startswith("DBH:"): dbh_estimate = line.replace("DBH:", "").strip() elif line.startswith("CONDITION:"): condition = line.replace("CONDITION:", "").strip() current_section = "condition" elif line.startswith("CHARACTERISTICS:"): current_section = "characteristics" elif line.startswith("ADDITIONAL INFO:"): current_section = "additional_info" elif line.startswith("RECOMMENDATIONS:"): current_section = "recommendations" elif line.startswith("REMARKS:"): current_section = "remarks" elif line.startswith("- ") or line.startswith("• "): item = line[2:].strip() if item: if current_section == "characteristics": characteristics.append(item) elif current_section == "condition": if condition: condition += f" {item}" else: condition = item elif current_section == "additional_info": if additional_info: additional_info += f" {item}" else: additional_info = item elif current_section == "recommendations": if recommendations: recommendations += f" {item}" else: recommendations = item elif current_section == "remarks": if remarks: remarks += f" {item}" else: remarks = item return AnalysisResponse( species=species, species_hindi=species_hindi, species_marathi=species_marathi, scientific_name=scientific_name, confidence=confidence, height=height, canopy=canopy, girth=girth, dbh_estimate=dbh_estimate, condition=condition, characteristics=characteristics, additional_info=additional_info, recommendations=recommendations, remarks=remarks, image_filename=filename ) async def process_single_image(contents: bytes, filename: str) -> tuple[Optional[AnalysisResponse], Optional[dict]]: """Process a single image - SAME APPROACH AS YOUR PLANT DISEASE APP""" try: logger.info(f"{filename}: Processing image...") # Open and process image image = Image.open(io.BytesIO(contents)) logger.info(f"{filename}: Image opened successfully - Size: {image.size}, Mode: {image.mode}") # Convert RGBA to RGB if needed if image.mode == 'RGBA': logger.info(f"{filename}: Converting RGBA to RGB") background = Image.new('RGB', image.size, (255, 255, 255)) background.paste(image, mask=image.split()[3]) image = background elif image.mode != 'RGB': image = image.convert('RGB') # Resize if needed image = resize_image_if_needed(image) # Check for blur is_blurry, blur_score = detect_blur(image) logger.info(f"{filename}: Blur check - Score: {blur_score:.2f}, Is blurry: {is_blurry}") if is_blurry: logger.warning(f"{filename}: Image appears blurry (score: {blur_score:.2f}). Continuing anyway...") # Convert to bytes for API - SAME AS PLANT DISEASE APP img_buffer = io.BytesIO() image.save(img_buffer, format='JPEG', quality=85) img_data = img_buffer.getvalue() # Prepare the detailed prompt prompt = """Analyze this tree image and provide comprehensive information in the EXACT format below. Be as specific and detailed as possible. ENGLISH NAME: [Common English name of the tree species] HINDI NAME: [Hindi name if known, otherwise write "Not available"] MARATHI NAME: [Marathi name if known, otherwise write "Not available"] SCIENTIFIC NAME: [Binomial nomenclature / Latin name] CONFIDENCE: [High/Medium/Low - based on image clarity and visible features] HEIGHT: [Estimated height in meters or feet, e.g., "15-20 meters" or "Not clearly visible"] CANOPY: [Canopy spread/diameter, e.g., "8-10 meters" or "Not clearly visible"] GIRTH: [Circumference at breast height, e.g., "2.5 meters" or "Not measurable from image"] DBH: [Diameter at Breast Height in cm or inches, calculated from girth, e.g., "80 cm" or "Not calculable"] CONDITION: [Overall health status - Healthy/Good/Fair/Poor/Declining, with brief explanation] CHARACTERISTICS: - [Key identifying feature 1 - bark texture, color, pattern] - [Key identifying feature 2 - leaf shape, arrangement, size] - [Key identifying feature 3 - growth habit, branch structure] - [Key identifying feature 4 - any visible flowers, fruits, or unique features] - [Additional characteristics as observed] ADDITIONAL INFO: [Any ecological, cultural, or interesting facts about this species. Native range, typical habitat, growth rate, etc.] RECOMMENDATIONS: [Care suggestions - watering, pruning needs, pest management, best growing conditions for this species] REMARKS: [Any other observations about this specific tree, potential issues, notable features, or identification notes] Important: 1. For measurements like DBH, HEIGHT, and CANOPY - if not clearly measurable from the image, state "Not clearly visible" or "Not measurable from image" 2. Base all measurements on what's actually visible in the image 3. Provide detailed botanical characteristics to justify the identification 4. Include both common names and scientific nomenclature 5. Be honest about confidence level - if identification is uncertain, explain why""" # Call Gemini API - EXACT SAME METHOD AS YOUR PLANT DISEASE APP logger.info(f"{filename}: Calling Gemini API...") try: response = model.generate_content([ prompt, {"mime_type": "image/jpeg", "data": base64.b64encode(img_data).decode('utf-8')} ]) response_text = response.text.strip() logger.info(f"{filename}: Analysis response received") except Exception as api_error: logger.error(f"{filename}: Gemini API Error: {api_error}") return None, {"filename": filename, "error": f"AI API Error: {str(api_error)}"} # Parse response result = parse_analysis_response(response_text, filename) logger.info(f"{filename}: Analysis completed successfully: {result.species} ({result.scientific_name})") return result, None except Exception as e: logger.error(f"{filename}: CRITICAL ERROR: {type(e).__name__}") logger.error(f"{filename}: Error message: {str(e)}") logger.error(f"{filename}: Full traceback:\n{traceback.format_exc()}") return None, {"filename": filename, "error": str(e)} @app.get("/") async def root(): """Root endpoint""" return { "message": "Tree Species and DBH Detection API", "version": "1.0.0", "model": "gemini-2.5-flash", "endpoints": { "health": "/health", "analyze": "/api/analyze (POST)", "analyze_multiple": "/api/analyze-multiple (POST)", } } @app.get("/health") async def health_check(): """Health check endpoint""" api_configured = GEMINI_API_KEY is not None return { "status": "healthy", "gemini_api_configured": api_configured, "model": "gemini-2.5-flash", "timestamp": time.time() } @app.post("/api/analyze", response_model=AnalysisResponse) async def analyze_tree(file: UploadFile = File(...)): """Analyze tree image to detect species and estimate DBH""" logger.info("=" * 80) logger.info(f"NEW ANALYSIS REQUEST") logger.info(f"Filename: {file.filename}") logger.info(f"Content-Type: {file.content_type}") logger.info(f"API Key configured: {GEMINI_API_KEY is not None}") logger.info("=" * 80) if not GEMINI_API_KEY: logger.error("GEMINI_API_KEY not configured") raise HTTPException( status_code=500, detail="Gemini API key not configured. Please set GEMINI_API_KEY environment variable." ) # Validate file type if not file.content_type.startswith("image/"): logger.warning(f"Invalid file type: {file.content_type}") raise HTTPException( status_code=400, detail="File must be an image" ) try: # Read uploaded file contents = await file.read() # Process the image result, error = await process_single_image(contents, file.filename) if error: raise HTTPException(status_code=400, detail=error["error"]) return result except HTTPException: raise except Exception as e: logger.error(f"CRITICAL ERROR in analyze_tree: {type(e).__name__}") logger.error(f"Error message: {str(e)}") logger.error(f"Full traceback:\n{traceback.format_exc()}") raise HTTPException( status_code=500, detail=f"Error analyzing image: {str(e)}" ) @app.post("/api/analyze-multiple", response_model=MultipleAnalysisResponse) async def analyze_multiple_trees(files: List[UploadFile] = File(...)): """Analyze multiple tree images at once""" logger.info("=" * 80) logger.info(f"NEW MULTIPLE ANALYSIS REQUEST") logger.info(f"Number of files: {len(files)}") logger.info(f"API Key configured: {GEMINI_API_KEY is not None}") logger.info("=" * 80) if not GEMINI_API_KEY: logger.error("GEMINI_API_KEY not configured") raise HTTPException( status_code=500, detail="Gemini API key not configured. Please set GEMINI_API_KEY environment variable." ) # Limit number of files if len(files) > 10: raise HTTPException( status_code=400, detail="Maximum 10 images allowed per request" ) # Validate all file types for file in files: if not file.content_type.startswith("image/"): raise HTTPException( status_code=400, detail=f"All files must be images. {file.filename} is not an image." ) try: # Read all files first files_data = [] for file in files: contents = await file.read() files_data.append((contents, file.filename)) # Process all images concurrently tasks = [process_single_image(contents, filename) for contents, filename in files_data] results = await asyncio.gather(*tasks) # Separate successful results from errors successful_results = [] errors = [] for result, error in results: if result: successful_results.append(result) if error: errors.append(error) logger.info(f"Multiple analysis completed: {len(successful_results)} successful, {len(errors)} failed") return MultipleAnalysisResponse( total_images=len(files), successful=len(successful_results), failed=len(errors), results=successful_results, errors=errors ) except HTTPException: raise except Exception as e: logger.error(f"CRITICAL ERROR in analyze_multiple_trees: {type(e).__name__}") logger.error(f"Error message: {str(e)}") logger.error(f"Full traceback:\n{traceback.format_exc()}") raise HTTPException( status_code=500, detail=f"Error analyzing images: {str(e)}" ) @app.get("/api/supported-species") async def get_supported_species(): """Information about species identification capabilities""" return { "message": "This API can identify ANY tree species from around the world", "model": "gemini-2.5-flash", "capabilities": [ "Identifies trees from all continents and climate zones", "Recognizes both common and rare species", "Provides scientific names (binomial nomenclature)", "Analyzes bark texture, leaf patterns, growth habits", "Estimates tree dimensions (DBH)", "Gives species-specific care recommendations", "Automatic blur detection for quality assurance", "Image resizing for faster processing", "Support for single or multiple image analysis" ], "examples": [ "Tropical: Mahogany, Teak, Baobab, Kapok, Rubber tree", "Temperate: Oak, Maple, Pine, Birch, Elm, Ash", "Subtropical: Palm species, Eucalyptus, Acacia", "Coniferous: Spruce, Fir, Cedar, Redwood, Sequoia", "Fruit trees: Apple, Mango, Cherry, Fig, Olive", "Exotic: Jacaranda, Bougainvillea tree, Flame tree" ] } if __name__ == "__main__": import uvicorn print("\n" + "="*60) print("🌳 Tree Species and DBH Detection API v1.0.0") print("="*60) print("🤖 AI Model: Gemini 2.5 Flash") print("🌐 Web Interface: http://localhost:8000") print("📤 Single Image: POST /api/analyze") print("📤 Multiple: POST /api/analyze-multiple (max 10)") print("📚 API Docs: http://localhost:8000/docs") print("="*60 + "\n") uvicorn.run(app, host="0.0.0.0", port=8000)