from fastapi import FastAPI, File, UploadFile, HTTPException, Form from fastapi.middleware.cors import CORSMiddleware from typing import Optional import pandas as pd from PIL import Image import torch import numpy as np from datetime import datetime import os import open_clip import base64 from io import BytesIO from dotenv import load_dotenv import requests from difflib import SequenceMatcher from supabase import create_client, Client # Load environment variables load_dotenv() THRESHOLD = 0.50 # Threshold for mushrooms PLANT_THRESHOLD = 0.70 # High threshold for plants (very strict to avoid false matches) MUSHROOM_THRESHOLD = 0.30 # Minimum confidence that it's actually a mushroom # Persistent storage configuration for Hugging Face Spaces # If /data exists (persistent storage), use it. Otherwise use local data/ PERSISTENT_DATA_DIR = "/data" if os.path.exists("/data") and os.access("/data", os.W_OK) else "data" DATA_DIR = PERSISTENT_DATA_DIR LOG_FILE = os.path.join(DATA_DIR, "recognition_log.csv") SUBMISSION_LOG_FILE = os.path.join(DATA_DIR, "new_species_submissions.csv") MUSHROOM_CSV = os.path.join(DATA_DIR, "mushrooms.csv") PLANT_CSV = os.path.join(DATA_DIR, "plants.csv") print(f"[INFO] Using data directory: {DATA_DIR}") if DATA_DIR == "/data": print("[INFO] Persistent storage enabled - changes will be saved across restarts") else: print("[WARNING] Using ephemeral storage - changes will be lost on restart") print("[WARNING] Enable persistent storage in HuggingFace Space settings for data persistence") # Copy initial CSV files to persistent storage if they don't exist if DATA_DIR == "/data": for filename in ["mushrooms.csv", "plants.csv", "recognition_log.csv", "new_species_submissions.csv"]: source = os.path.join("data", filename) dest = os.path.join(DATA_DIR, filename) if os.path.exists(source) and not os.path.exists(dest): import shutil shutil.copy2(source, dest) print(f"[INFO] Copied {filename} to persistent storage") # Read from environment (works with Hugging Face Spaces secrets) OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "") PLANTNET_API_KEY = os.getenv("PLANTNET_API_KEY", "") SUPABASE_URL = os.getenv("SUPABASE_URL", "") SUPABASE_KEY = os.getenv("SUPABASE_KEY", "") # Initialize Supabase client (optional - falls back to CSV if not configured) supabase: Optional[Client] = None USE_SUPABASE = False if SUPABASE_URL and SUPABASE_KEY: try: supabase = create_client(SUPABASE_URL, SUPABASE_KEY) # Test connection supabase.table("species").select("count").limit(1).execute() USE_SUPABASE = True print(f"[INFO] Supabase connected successfully") print(f"[INFO] Using Supabase for data persistence (FREE & permanent)") except Exception as e: print(f"[WARNING] Supabase connection failed: {e}") print(f"[WARNING] Falling back to CSV storage") USE_SUPABASE = False else: print("[INFO] Supabase not configured - using CSV storage") print("[INFO] Add SUPABASE_URL and SUPABASE_KEY to .env for permanent data storage") # Print API key status at startup (first 10 chars only for security) if OPENROUTER_API_KEY: print(f"[INFO] OpenRouter API Key loaded: {OPENROUTER_API_KEY[:10]}...") else: print("[WARNING] OpenRouter API Key not found. VLM fallback will be disabled.") if PLANTNET_API_KEY: print(f"[INFO] PlantNet API Key loaded: {PLANTNET_API_KEY[:10]}...") else: print("[WARNING] PlantNet API Key not found. PlantNet identification will be disabled.") # MCC Location options MCC_LOCATIONS = ["QSC", "BOTANY TANK", "MMIP", "CANTEEN", "ANDERSON HALL"] # Default coordinates for MCC locations MCC_LOCATION_COORDS = { "QSC": (12.92000, 80.12100), "BOTANY TANK": (12.92050, 80.12080), "MMIP": (12.91900, 80.12000), "CANTEEN": (12.91800, 80.11900), "ANDERSON HALL": (12.92100, 80.12200), } # Load BioCLIP model model, _, preprocess = open_clip.create_model_and_transforms('hf-hub:imageomics/bioclip') tokenizer = open_clip.get_tokenizer('hf-hub:imageomics/bioclip') # Set model to evaluation mode (important for consistent predictions) model.eval() print("[INFO] BioCLIP model loaded and set to evaluation mode") def load_species_from_supabase(): """Load species from Supabase and convert to pandas DataFrames""" try: print("[INFO] Loading species from Supabase...") # Load mushrooms mushroom_response = supabase.table("species").select("*").eq("species_type", "mushroom").execute() mushrooms = [] for item in mushroom_response.data: mushrooms.append({ "sno": item["sno"], "species": item["species_name"], "location": item["location"], "lat": item["latitude"], "lon": item["longitude"] }) mushroom_df = pd.DataFrame(mushrooms) if mushrooms else pd.DataFrame(columns=["sno", "species", "location", "lat", "lon"]) # Load plants plant_response = supabase.table("species").select("*").eq("species_type", "plant").execute() plants = [] for item in plant_response.data: plants.append({ "sno": item["sno"], "species": item["species_name"], "location": item["location"], "lat": item["latitude"], "lon": item["longitude"] }) plant_df = pd.DataFrame(plants) if plants else pd.DataFrame(columns=["sno", "species", "location", "lat", "lon"]) print(f"[INFO] Loaded from Supabase: {len(mushroom_df)} mushrooms, {len(plant_df)} plants") return mushroom_df, plant_df except Exception as e: print(f"[ERROR] Failed to load from Supabase: {e}") print(f"[INFO] Falling back to CSV...") return None, None # Load species databases (from Supabase if available, otherwise CSV) if USE_SUPABASE: mushroom_db, plant_db = load_species_from_supabase() if mushroom_db is None or plant_db is None: # Fallback to CSV mushroom_db = pd.read_csv(MUSHROOM_CSV) plant_db = pd.read_csv(PLANT_CSV) else: # Use CSV mushroom_db = pd.read_csv(MUSHROOM_CSV) plant_db = pd.read_csv(PLANT_CSV) print(f"[INFO] Loaded from CSV: {len(mushroom_db)} mushrooms, {len(plant_db)} plants") # Create labels for species identification mushroom_labels = [species for species in mushroom_db["species"]] mushroom_labels.append("unknown species") plant_labels = [species for species in plant_db["species"]] plant_labels.append("unknown species") # Labels for general classification (to detect plants vs mushrooms) general_labels = [ "mushroom", "fungus", "toadstool", "plant", "flower", "tree", "leaf", "shrub", "herb", "foliage", "vegetation", "blossom", "crop", "grass", "car", "vehicle", "person", "animal", "building", "food", "object", "landscape" ] # Initialize log file if not os.path.exists(LOG_FILE): log_df = pd.DataFrame(columns=["timestamp", "species", "confidence", "location", "lat", "lon", "status"]) log_df.to_csv(LOG_FILE, index=False) # Initialize submissions log file if not os.path.exists(SUBMISSION_LOG_FILE): submissions_df = pd.DataFrame(columns=["timestamp", "species_name", "location", "user_id", "user_email", "notes", "status"]) submissions_df.to_csv(SUBMISSION_LOG_FILE, index=False) # Initialize FastAPI app = FastAPI(title="Mushroom Recognition API", version="1.0.0") # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def log_recognition(species, confidence, location="N/A", lat="N/A", lon="N/A", status="success"): """Log recognition to Supabase (if available) and CSV""" log_entry = { "species_name": species, "confidence": float(confidence), "location": location, "latitude": str(lat), "longitude": str(lon), "status": status } # Log to Supabase if available if USE_SUPABASE: try: supabase.table("recognition_logs").insert(log_entry).execute() except Exception as e: print(f"[WARNING] Failed to log to Supabase: {e}") # Also log to CSV as backup csv_log_entry = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "species": species, "confidence": confidence, "location": location, "lat": lat, "lon": lon, "status": status } log_df = pd.read_csv(LOG_FILE) new_row_df = pd.DataFrame([csv_log_entry]) log_df = pd.concat([log_df, new_row_df], ignore_index=True) log_df.to_csv(LOG_FILE, index=False) def log_new_species_submission(species_name, location, user_id="", user_email="", notes=""): submission_entry = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "species_name": species_name, "location": location, "user_id": user_id, "user_email": user_email, "notes": notes, "status": "pending" } submissions_df = pd.read_csv(SUBMISSION_LOG_FILE) submissions_df = pd.concat([submissions_df, pd.DataFrame([submission_entry])], ignore_index=True) submissions_df.to_csv(SUBMISSION_LOG_FILE, index=False) return submission_entry def identify_with_plantnet(image: Image.Image, species_type="plant"): """ Identify plants using PlantNet API Args: image: PIL Image object species_type: Type of species (should be "plant") Returns: tuple: (species_idx, species_name, found_in_db, species_type) """ if not PLANTNET_API_KEY: print("[PlantNet] API key not configured, skipping PlantNet identification") return None, "PlantNet API key not configured", False, species_type if species_type != "plant": print(f"[PlantNet] Skipping - PlantNet only works for plants, not {species_type}") return None, "PlantNet only identifies plants", False, species_type print(f"[PlantNet] Starting PlantNet identification...") try: # Convert image to JPEG bytes buffered = BytesIO() image.save(buffered, format="JPEG", quality=90) buffered.seek(0) # PlantNet API endpoint - using "k-indian-subcontinent" project for Indian plants # Available options: "k-world-flora" (82920 species), "k-indian-subcontinent" (7469 species) # See full list at: https://my-api.plantnet.org/v2/projects api_url = f"https://my-api.plantnet.org/v2/identify/k-indian-subcontinent" # Prepare the multipart form data files = { 'images': ('plant.jpg', buffered, 'image/jpeg') } params = { 'api-key': PLANTNET_API_KEY # Note: 'organs' parameter not supported by k-indian-subcontinent project } # Make API request response = requests.post(api_url, files=files, params=params, timeout=30) # Check for errors if response.status_code == 401: print("[PlantNet ERROR] Invalid API key - check your .env file") return None, "Invalid PlantNet API key", False, species_type elif response.status_code == 400: error_detail = response.text[:200] if response.text else "Bad request" print(f"[PlantNet ERROR] 400 Bad Request: {error_detail}") # Check if API key is still placeholder if PLANTNET_API_KEY.startswith("your_"): print("[PlantNet ERROR] API key appears to be a placeholder. Get your key from https://my.plantnet.org/") return None, "PlantNet API key not configured (placeholder detected)", False, species_type return None, f"PlantNet API error 400: {error_detail}", False, species_type elif response.status_code == 429: print("[PlantNet ERROR] API quota exceeded (500/day limit)") return None, "PlantNet API quota exceeded", False, species_type elif response.status_code != 200: error_detail = response.text[:200] if response.text else "Unknown error" print(f"[PlantNet ERROR] Status {response.status_code}: {error_detail}") return None, f"PlantNet API error: {response.status_code}", False, species_type result = response.json() # Log the response for debugging print(f"[PlantNet DEBUG] Response status: {response.status_code}") # Check if we got results if 'results' not in result or len(result['results']) == 0: print("[PlantNet] No matches found") return None, "No plant matches found", False, species_type # Get the best match best_match = result['results'][0] score = best_match.get('score', 0.0) species_info = best_match.get('species', {}) # Get scientific name scientific_name = species_info.get('scientificNameWithoutAuthor', species_info.get('scientificName', '')).strip() # Get common names if available common_names = species_info.get('commonNames', []) common_name = common_names[0] if common_names else "" # Get family and genus family = species_info.get('family', {}).get('scientificNameWithoutAuthor', '') genus = species_info.get('genus', {}).get('scientificNameWithoutAuthor', '') print(f"[PlantNet DEBUG] Best match: {scientific_name}") print(f"[PlantNet DEBUG] Common names: {common_names[:5]}...") print(f"[PlantNet DEBUG] Family: {family}, Genus: {genus}") print(f"[PlantNet DEBUG] Confidence: {score:.3f}") # Try to match with our database print(f"[PlantNet DEBUG] Checking against {len(plant_db)} plants in local DB...") for idx, row in plant_db.iterrows(): db_species = row['species'].lower().strip() # Check scientific name if scientific_name.lower() == db_species: print(f"[PlantNet] Exact scientific name match: {row['species']}") return idx, row['species'], True, species_type # Check ALL common names for c_name in common_names: if c_name.lower() == db_species: print(f"[PlantNet] Common name match ('{c_name}'): {row['species']}") return idx, row['species'], True, species_type # SUBSTRING MATCHING (New Fix) # If DB has "Ocimum gratissimum (African basil)" and PlantNet gives "Ocimum gratissimum" if scientific_name.lower() in db_species and len(scientific_name) > 4: print(f"[PlantNet] Substring match (Scientific in DB): {row['species']}") return idx, row['species'], True, species_type # If DB has "Ocimum gratissimum (African basil)" and PlantNet gives "African basil" for c_name in common_names: if c_name.lower() in db_species and len(c_name) > 4: print(f"[PlantNet] Substring match (Common '{c_name}' in DB): {row['species']}") return idx, row['species'], True, species_type # Try fuzzy matching with genus/species parts if scientific_name: for idx, row in plant_db.iterrows(): db_species = row['species'].lower().strip() scientific_lower = scientific_name.lower() # Check if genus matches if genus and genus.lower() in db_species: similarity = SequenceMatcher(None, scientific_lower, db_species).ratio() if similarity > 0.7: print(f"[PlantNet] Fuzzy match (genus + similarity): {row['species']} (sim: {similarity:.2f})") return idx, row['species'], True, species_type # Not found in our database, but we have a PlantNet identification display_name = f"{scientific_name}" + (f" ({common_name})" if common_name else "") print(f"[PlantNet] Identified as '{display_name}' but NOT in MCC database") return None, display_name, False, species_type except requests.exceptions.Timeout: return None, "PlantNet API timeout", False, species_type except requests.exceptions.RequestException as e: return None, f"PlantNet request error: {str(e)}", False, species_type except Exception as e: return None, f"PlantNet error: {str(e)}", False, species_type def identify_with_vlm(image, species_type="mushroom"): if not OPENROUTER_API_KEY: print("[VLM] API key not configured, skipping VLM identification") return None, "OpenRouter API key not configured", False, species_type print(f"[VLM] Starting VLM identification for {species_type}...") try: buffered = BytesIO() image.save(buffered, format="JPEG") img_base64 = base64.b64encode(buffered.getvalue()).decode() # Use appropriate database based on species type db = mushroom_db if species_type == "mushroom" else plant_db species_list = "\n".join([f"- {s}" for s in db["species"].tolist()]) response = requests.post( url="https://openrouter.ai/api/v1/chat/completions", headers={ "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json" }, json={ "model": "nvidia/nemotron-nano-12b-v2-vl:free", "temperature": 0.1, # Low temperature for more deterministic responses "max_tokens": 100, "top_p": 0.9, "messages": [ { "role": "user", "content": [ { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{img_base64}" } }, { "type": "text", "text": f"""You are an expert {'mycologist' if species_type == 'mushroom' else 'botanist'} AI assistant. Analyze this {'mushroom/fungus' if species_type == 'mushroom' else 'plant'} image carefully and identify the species. IMPORTANT INSTRUCTIONS: 1. Look at these key features: {'- Cap shape, color, and texture' if species_type == 'mushroom' else '- Leaf shape, arrangement, and texture'} {'- Gill structure and color' if species_type == 'mushroom' else '- Flower characteristics (if present)'} {'- Stem characteristics' if species_type == 'mushroom' else '- Stem/trunk characteristics'} - Size and proportions - Growing environment 2. Carefully match the {'mushroom' if species_type == 'mushroom' else 'plant'} to ONE species from this list: {species_list} 3. Respond with ONLY the exact species name from the list above - use the EXACT spelling as shown. 4. Do NOT add any explanation, description, or additional text. 5. If no species matches confidently, respond with exactly: "Unknown" Your response (exact species name only):""" } ] } ] } ) response_data = response.json() if "error" in response_data: error_msg = response_data["error"].get("message", "Unknown API error") return None, f"API Error: {error_msg}", False, species_type if "choices" not in response_data: return None, f"Unexpected response format", False, species_type result = response_data["choices"][0]["message"]["content"].strip() # Log VLM response for debugging print(f"[VLM DEBUG] Raw response: '{result}'") # Clean up the response - remove common extra text result = result.split('\n')[0].strip() if '.' in result: result = result.split('.')[0].strip() # Remove common prefixes prefixes_to_remove = ["species:", "answer:", "identification:", "result:"] for prefix in prefixes_to_remove: if result.lower().startswith(prefix): result = result[len(prefix):].strip() print(f"[VLM DEBUG] Cleaned response: '{result}'") # Try exact or substring match first for idx, species in enumerate(db["species"]): if species.lower() == result.lower(): print(f"[VLM DEBUG] Exact match found: {species}") return idx, species, True, species_type if species.lower() in result.lower() or result.lower() in species.lower(): print(f"[VLM DEBUG] Substring match found: {species}") return idx, species, True, species_type # Try fuzzy matching with high threshold (0.85 similarity) best_match_idx = None best_match_score = 0.0 for idx, species in enumerate(db["species"]): similarity = SequenceMatcher(None, species.lower(), result.lower()).ratio() if similarity > best_match_score: best_match_score = similarity best_match_idx = idx print(f"[VLM DEBUG] Best fuzzy match: {db.iloc[best_match_idx]['species'] if best_match_idx is not None else 'None'} (similarity: {best_match_score:.2f})") # If we found a good match (85% or higher similarity) if best_match_score >= 0.85: matched_species = db.iloc[best_match_idx]["species"] print(f"[VLM DEBUG] Fuzzy match accepted: {matched_species}") return best_match_idx, matched_species, True, species_type print(f"[VLM DEBUG] No match found, returning: {result}") return None, result, False, species_type except Exception as e: return None, f"Error: {str(e)}", False, species_type def recognize_species_internal(image: Image.Image): # Clear any potential cached gradients (though we use no_grad, being safe) if torch.cuda.is_available(): torch.cuda.empty_cache() # Step 1: General classification - check if it's a plant or mushroom image_input = preprocess(image).unsqueeze(0) general_text_input = tokenizer(general_labels) with torch.no_grad(): image_features = model.encode_image(image_input) general_text_features = model.encode_text(general_text_input) image_features_norm = image_features / image_features.norm(dim=-1, keepdim=True) general_text_features_norm = general_text_features / general_text_features.norm(dim=-1, keepdim=True) general_scores = (100.0 * image_features_norm @ general_text_features_norm.T).softmax(dim=-1) # Determine what type of species it is general_idx = general_scores.argmax().item() general_confidence = general_scores[0][general_idx].item() top_prediction = general_labels[general_idx] # Calculate combined scores for mushroom and plant categories mushroom_indices = [i for i, label in enumerate(general_labels) if label in ["mushroom", "fungus", "toadstool"]] plant_indices = [i for i, label in enumerate(general_labels) if label in ["plant", "flower", "tree", "leaf", "shrub", "herb", "foliage", "vegetation", "blossom", "crop", "grass"]] mushroom_score = sum(general_scores[0][i].item() for i in mushroom_indices) plant_score = sum(general_scores[0][i].item() for i in plant_indices) # Debug logging print(f"[Classification] Top prediction: {top_prediction} ({general_confidence:.3f})") print(f"[Classification] Mushroom category score: {mushroom_score:.3f}") print(f"[Classification] Plant category score: {plant_score:.3f}") # Determine initial classification based on combined scores initial_type = "mushroom" if mushroom_score > plant_score else "plant" # If both scores are very low, it's neither a mushroom nor a plant if max(mushroom_score, plant_score) < MUSHROOM_THRESHOLD: return { "success": False, "message": f"This image appears to be a {top_prediction}, not a plant or mushroom. Please upload a plant or mushroom image.", "detected_object": top_prediction, "confidence": round(float(general_confidence), 3), "top_matches": [] } # DUAL-CHECK: Test against BOTH databases to verify classification print(f"[Classification] Initial type: {initial_type}, performing dual-check...") # Test against mushroom database mushroom_text_input = tokenizer(mushroom_labels) with torch.no_grad(): mushroom_text_features = model.encode_text(mushroom_text_input) mushroom_text_features_norm = mushroom_text_features / mushroom_text_features.norm(dim=-1, keepdim=True) mushroom_species_scores = (100.0 * image_features_norm @ mushroom_text_features_norm.T).softmax(dim=-1) mushroom_best_confidence = mushroom_species_scores.max().item() mushroom_best_idx = mushroom_species_scores.argmax().item() # Test against plant database plant_text_input = tokenizer(plant_labels) with torch.no_grad(): plant_text_features = model.encode_text(plant_text_input) plant_text_features_norm = plant_text_features / plant_text_features.norm(dim=-1, keepdim=True) plant_species_scores = (100.0 * image_features_norm @ plant_text_features_norm.T).softmax(dim=-1) plant_best_confidence = plant_species_scores.max().item() plant_best_idx = plant_species_scores.argmax().item() print(f"[Dual-Check] Mushroom DB best match confidence: {mushroom_best_confidence:.3f}") print(f"[Dual-Check] Plant DB best match confidence: {plant_best_confidence:.3f}") # If there's a significant confidence difference (>15%), use that as the final decision confidence_diff = abs(mushroom_best_confidence - plant_best_confidence) if confidence_diff > 0.15: if plant_best_confidence > mushroom_best_confidence: species_type = "plant" db = plant_db labels = plant_labels scores = plant_species_scores idx = plant_best_idx print(f"[Dual-Check] Overriding to PLANT based on higher confidence") else: species_type = "mushroom" db = mushroom_db labels = mushroom_labels scores = mushroom_species_scores idx = mushroom_best_idx print(f"[Dual-Check] Overriding to MUSHROOM based on higher confidence") else: # Use initial classification if confidence is similar species_type = initial_type if species_type == "mushroom": db = mushroom_db labels = mushroom_labels scores = mushroom_species_scores idx = mushroom_best_idx else: db = plant_db labels = plant_labels scores = plant_species_scores idx = plant_best_idx print(f"[Dual-Check] Using initial classification: {species_type}") print(f"[Classification] Final selected type: {species_type}") # Step 2: Get species-level results (already computed in dual-check) confidence = scores[0][idx].item() # UNCERTAINTY CHECK: If top matches are very close, it means system is uncertain # Get top 3 scores to check spread top_3_scores = scores[0].topk(3)[0] if len(top_3_scores) >= 2: score_diff = top_3_scores[0].item() - top_3_scores[1].item() print(f"[Uncertainty Check] Top score: {top_3_scores[0].item():.3f}, 2nd: {top_3_scores[1].item():.3f}, diff: {score_diff:.3f}") # If difference < 0.10 (10%), the model is very uncertain if score_diff < 0.10: print(f"[WARNING] Model is uncertain - top matches too similar (diff: {score_diff:.3f})") confidence = confidence * 0.7 # Reduce confidence score print(f"[Uncertainty] Adjusted confidence: {confidence:.3f}") # Get top 5 matches top_k = min(5, len(scores[0])) top_scores, top_indices = scores[0].topk(top_k) top_matches = [] for i in range(top_k): match_idx = top_indices[i].item() match_score = top_scores[i].item() if match_idx < len(labels) - 1: top_matches.append({ "species": db.iloc[match_idx]['species'], "confidence": round(float(match_score), 3) }) else: top_matches.append({ "species": "Unknown species", "confidence": round(float(match_score), 3) }) # Check if unknown or low confidence (use different thresholds for plants vs mushrooms) threshold_to_use = PLANT_THRESHOLD if species_type == "plant" else THRESHOLD print(f"[BioCLIP] Best match: {labels[idx] if idx < len(labels) else 'unknown'} with confidence {confidence:.3f}") print(f"[BioCLIP] Required threshold for {species_type}: {threshold_to_use}") if idx == len(labels)-1 or confidence < threshold_to_use: print(f"[BioCLIP] Low confidence ({confidence:.3f}) or unknown species detected (threshold: {threshold_to_use})") log_recognition("Unknown", confidence, status="not_found") # Try PlantNet for plants (if available) if species_type == "plant" and PLANTNET_API_KEY: print(f"[BioCLIP] Attempting PlantNet identification...") pn_idx, pn_species, pn_found, pn_type = identify_with_plantnet(image, species_type) if pn_idx is not None and pn_found: print(f"[PlantNet] Successfully identified: {pn_species} (found in database)") row = plant_db.iloc[pn_idx] log_recognition(row['species'], 0.90, row['location'], row['lat'], row['lon'], status="plantnet_success") return { "success": True, "found_in_database": True, "species": row['species'], "species_type": species_type, "location": row['location'], "latitude": float(row['lat']), "longitude": float(row['lon']), "confidence": 0.90, "identified_by": "PlantNet API", "top_matches": top_matches } elif pn_idx is None and not pn_found and "error" not in pn_species.lower() and "not configured" not in pn_species.lower(): # PlantNet identified but not in our database print(f"[PlantNet] Identified: {pn_species} (NOT in MCC database)") log_recognition(pn_species, 0.90, status="not_at_mcc") return { "success": True, "found_in_database": False, "identified_species": pn_species, "species_type": species_type, "confidence": 0.90, "message": f"This plant ({pn_species}) is not available in the MCC campus database", "identified_by": "PlantNet API", "top_matches": top_matches } else: print(f"[PlantNet] Failed: {pn_species}") # Try VLM fallback if PlantNet didn't work if OPENROUTER_API_KEY: print(f"[BioCLIP] Attempting VLM fallback for {species_type}...") vlm_idx, vlm_species, found_in_db, vlm_species_type = identify_with_vlm(image, species_type) if vlm_idx is not None and found_in_db: print(f"[VLM] Successfully identified: {vlm_species} (found in database)") db = mushroom_db if vlm_species_type == "mushroom" else plant_db row = db.iloc[vlm_idx] log_recognition(row['species'], 0.85, row['location'], row['lat'], row['lon'], status="vlm_success") return { "success": True, "found_in_database": True, "species": row['species'], "species_type": vlm_species_type, "location": row['location'], "latitude": float(row['lat']), "longitude": float(row['lon']), "confidence": 0.85, "identified_by": "Nvidia Nemotron VLM", "top_matches": top_matches } else: # Species identified but not in database print(f"[VLM] Identified: {vlm_species} (NOT in MCC database)") log_recognition(vlm_species, 0.85, status="not_at_mcc") return { "success": True, "found_in_database": False, "identified_species": vlm_species, "species_type": vlm_species_type, "confidence": 0.85, "message": f"This {vlm_species_type} is not available in the MCC campus database", "identified_by": "Nvidia Nemotron VLM", "top_matches": top_matches } else: print("[BioCLIP] VLM fallback skipped - API key not configured") return { "success": False, "message": "Species not found in database", "species_type": species_type, "confidence": round(float(confidence), 3), "top_matches": top_matches } # Found in database with high confidence row = db.iloc[idx] print(f"[BioCLIP] High confidence match: {row['species']} ({confidence:.3f})") # For plants, use PlantNet FIRST if available, then VLM verification if species_type == "plant": if not OPENROUTER_API_KEY and not PLANTNET_API_KEY: print(f"[WARNING] No PlantNet or VLM API key - plant identification may be inaccurate!") print(f"[WARNING] BioCLIP alone is not reliable for plants. Configure PLANTNET_API_KEY or OPENROUTER_API_KEY for better accuracy.") # Try PlantNet first (more specialized for plants) if PLANTNET_API_KEY: print(f"[PlantNet] Verifying BioCLIP result: {row['species']} ({confidence:.3f})") pn_idx, pn_species, pn_found, pn_type = identify_with_plantnet(image, species_type) if pn_idx is not None and pn_found: bioclip_suggestion = row['species'] plantnet_suggestion = plant_db.iloc[pn_idx]['species'] print(f"[PlantNet] Result: {plantnet_suggestion}") print(f"[BioCLIP] Result: {bioclip_suggestion}") # If PlantNet suggests different species, trust PlantNet (it's more specialized) if pn_idx != idx: print(f"[CONFLICT] PlantNet and BioCLIP disagree - using PlantNet result (more specialized)") row = plant_db.iloc[pn_idx] confidence = 0.90 log_recognition(row['species'], confidence, row['location'], row['lat'], row['lon'], status="plantnet_verified") return { "success": True, "found_in_database": True, "species": row['species'], "species_type": species_type, "location": row['location'], "latitude": float(row['lat']), "longitude": float(row['lon']), "confidence": confidence, "identified_by": "PlantNet API (verified)", "bioclip_suggestion": bioclip_suggestion, "top_matches": top_matches } else: print(f"[AGREEMENT] PlantNet and BioCLIP agree on: {plantnet_suggestion}") confidence = 0.95 # Higher confidence when both agree log_recognition(row['species'], confidence, row['location'], row['lat'], row['lon'], status="double_verified") return { "success": True, "found_in_database": True, "species": row['species'], "species_type": species_type, "location": row['location'], "latitude": float(row['lat']), "longitude": float(row['lon']), "confidence": confidence, "identified_by": "PlantNet + BioCLIP (both agree)", "top_matches": top_matches } elif pn_idx is None and not pn_found and "error" not in pn_species.lower() and "not configured" not in pn_species.lower(): # PlantNet identified something NOT in our database print(f"[PlantNet] Identified '{pn_species}' but NOT in MCC database") print(f"[OVERRIDE] PlantNet says this plant is not in our database") log_recognition(pn_species, 0.90, status="not_at_mcc") return { "success": True, "found_in_database": False, "identified_species": pn_species, "species_type": species_type, "confidence": 0.90, "message": f"This plant ({pn_species}) is not available in the MCC campus database", "identified_by": "PlantNet API", "bioclip_suggestion": row['species'], "top_matches": top_matches } # For plants without PlantNet, try VLM verification as before if species_type == "plant" and OPENROUTER_API_KEY: print(f"[BioCLIP] Plant identified: {row['species']} with {confidence:.3f} confidence") print(f"[VLM] Mandatory verification for plant species...") vlm_idx, vlm_species, found_in_db, vlm_species_type = identify_with_vlm(image, species_type) if vlm_idx is not None and found_in_db: bioclip_suggestion = row['species'] vlm_suggestion = db.iloc[vlm_idx]['species'] print(f"[VLM] Result: {vlm_suggestion}") print(f"[BioCLIP] Result: {bioclip_suggestion}") # If VLM suggests different species if vlm_idx != idx: print(f"[CONFLICT] VLM and BioCLIP disagree - using VLM result (more accurate for plants)") row = db.iloc[vlm_idx] confidence = 0.85 log_recognition(row['species'], confidence, row['location'], row['lat'], row['lon'], status="vlm_verified") # Get top matches for reference top_k = min(5, len(scores[0])) top_scores, top_indices = scores[0].topk(top_k) top_matches = [] for i in range(top_k): match_idx = top_indices[i].item() match_score = top_scores[i].item() if match_idx < len(labels) - 1: top_matches.append({ "species": db.iloc[match_idx]['species'], "confidence": round(float(match_score), 3) }) return { "success": True, "found_in_database": True, "species": row['species'], "species_type": species_type, "location": row['location'], "latitude": float(row['lat']), "longitude": float(row['lon']), "confidence": confidence, "identified_by": "Nvidia Nemotron VLM (verified)", "bioclip_suggestion": bioclip_suggestion, "top_matches": top_matches } else: print(f"[AGREEMENT] VLM and BioCLIP agree on: {vlm_suggestion}") elif vlm_idx is None and not found_in_db: # VLM identified a species but it's NOT in database print(f"[VLM] Identified '{vlm_species}' but NOT FOUND in MCC database") print(f"[OVERRIDE] Treating as unknown species (VLM says not in database)") # Return as not found in database log_recognition(vlm_species, 0.85, status="not_at_mcc") # Get top matches top_k = min(5, len(scores[0])) top_scores, top_indices = scores[0].topk(top_k) top_matches = [] for i in range(top_k): match_idx = top_indices[i].item() match_score = top_scores[i].item() if match_idx < len(labels) - 1: top_matches.append({ "species": db.iloc[match_idx]['species'], "confidence": round(float(match_score), 3) }) return { "success": True, "found_in_database": False, "identified_species": vlm_species, "species_type": species_type, "confidence": 0.85, "message": f"This plant ({vlm_species}) is not available in the MCC campus database", "identified_by": "Nvidia Nemotron VLM", "bioclip_suggestion": row['species'], "top_matches": top_matches } print(f"[BioCLIP] Successfully identified: {row['species']} (confidence: {confidence:.3f})") log_recognition(row['species'], confidence, row['location'], row['lat'], row['lon'], status="success") # Get top 5 matches to show alternatives top_k = min(5, len(scores[0])) top_scores, top_indices = scores[0].topk(top_k) top_matches = [] for i in range(top_k): match_idx = top_indices[i].item() match_score = top_scores[i].item() if match_idx < len(labels) - 1: top_matches.append({ "species": db.iloc[match_idx]['species'], "confidence": round(float(match_score), 3) }) return { "success": True, "found_in_database": True, "species": row['species'], "species_type": species_type, "location": row['location'], "latitude": float(row['lat']), "longitude": float(row['lon']), "confidence": round(float(confidence), 3), "identified_by": "BioCLIP", "top_matches": top_matches } @app.post("/submit-new-species") async def submit_new_species( species_name: str = Form(...), location: str = Form(...), species_type: str = Form("mushroom"), user_id: str = Form(""), user_email: str = Form(""), notes: str = Form(""), file: Optional[UploadFile] = File(None) ): """ Submit a new species that was not found in the MCC database and add it to the appropriate CSV Args: species_name: Name of the species identified location: MCC location where found (must be one of the predefined locations) species_type: Type of species - "plant" or "mushroom" user_id: Optional user ID user_email: Optional user email for follow-up notes: Optional additional notes file: Optional image file of the species Returns: JSON with submission confirmation """ try: global mushroom_db, plant_db, mushroom_labels, plant_labels # Validate location if location not in MCC_LOCATIONS: raise HTTPException( status_code=400, detail=f"Invalid location. Must be one of: {', '.join(MCC_LOCATIONS)}" ) # Validate species type if species_type not in ["plant", "mushroom"]: raise HTTPException( status_code=400, detail="Invalid species_type. Must be 'plant' or 'mushroom'" ) # Log the submission submission = log_new_species_submission( species_name=species_name, location=location, user_id=user_id, user_email=user_email, notes=notes ) # If image is provided, save it image_path = None if file: # Create submissions directory if it doesn't exist submissions_dir = os.path.join(DATA_DIR, "submissions") os.makedirs(submissions_dir, exist_ok=True) # Generate unique filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") safe_species_name = "".join(c for c in species_name if c.isalnum() or c in (' ', '_')).rstrip() safe_species_name = safe_species_name.replace(' ', '_') filename = f"{timestamp}_{safe_species_name}.jpg" image_path = os.path.join(submissions_dir, filename) # Save image contents = await file.read() image = Image.open(BytesIO(contents)) if image.mode != 'RGB': image = image.convert('RGB') image.save(image_path, 'JPEG') # Add to database (Supabase if available, otherwise CSV) db = mushroom_db if species_type == "mushroom" else plant_db # Get coordinates for the location lat, lon = MCC_LOCATION_COORDS.get(location, (12.92000, 80.12100)) # Create new entry new_sno = len(db) + 1 # Save to Supabase if available if USE_SUPABASE: try: species_data = { "sno": new_sno, "species_name": species_name, "species_type": species_type, "location": location, "latitude": float(lat), "longitude": float(lon), "added_by": user_id if user_id else "anonymous", "notes": notes } print(f"[DEBUG] Attempting to insert species data: {species_data}") response = supabase.table("species").insert(species_data).execute() # Check if the insert was successful if not response.data or len(response.data) == 0: error_msg = f"Insert failed - no data returned. Response: {response}" print(f"[ERROR] {error_msg}") raise Exception(error_msg) print(f"[INFO] Successfully saved {species_name} to Supabase") print(f"[DEBUG] Insert response: {response.data}") # Update in-memory cache immediately new_entry_dict = { "sno": new_sno, "species": species_name, "location": location, "lat": lat, "lon": lon } new_row = pd.DataFrame([new_entry_dict]) if species_type == "mushroom": mushroom_db = pd.concat([mushroom_db, new_row], ignore_index=True) else: plant_db = pd.concat([plant_db, new_row], ignore_index=True) except Exception as e: error_str = str(e) print(f"[ERROR] Failed to save to Supabase: {error_str}") # Check for specific error types if "duplicate key" in error_str.lower() or "unique constraint" in error_str.lower(): raise HTTPException( status_code=409, detail=f"This species '{species_name}' already exists at location '{location}'. Please check the database or choose a different location." ) elif "permission denied" in error_str.lower() or "policy" in error_str.lower(): raise HTTPException( status_code=403, detail=f"Permission denied. Please contact administrator. Error: {error_str}" ) else: raise HTTPException( status_code=500, detail=f"Failed to save to database: {error_str}" ) else: # Fallback to CSV csv_file = MUSHROOM_CSV if species_type == "mushroom" else PLANT_CSV new_entry = pd.DataFrame([{ "sno": new_sno, "species": species_name, "location": location, "lat": lat, "lon": lon }]) new_entry.to_csv(csv_file, mode='a', header=False, index=False) print(f"[INFO] Saved {species_name} to CSV") # Regenerate labels lists mushroom_labels = [species for species in mushroom_db["species"]] mushroom_labels.append("unknown species") plant_labels = [species for species in plant_db["species"]] plant_labels.append("unknown species") print(f"[INFO] Updated cache: {len(mushroom_db)} mushrooms, {len(plant_db)} plants") return { "success": True, "message": f"Thank you! The {species_type} '{species_name}' has been added to the MCC database at {location}.", "submission": { "species_name": species_name, "species_type": species_type, "location": location, "latitude": lat, "longitude": lon, "timestamp": submission["timestamp"], "image_saved": image_path is not None, "added_to_database": True } } except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing submission: {str(e)}") @app.get("/") async def root(): storage_info = "Supabase (permanent)" if USE_SUPABASE else "CSV/ephemeral" return { "message": "Mushroom & Plant Recognition API", "version": "2.0.0", "data_storage": storage_info, "api_status": { "bioclip": "active", "openrouter_vlm": "active" if OPENROUTER_API_KEY else "not configured", "plantnet": "active" if PLANTNET_API_KEY else "not configured", "supabase": "active" if USE_SUPABASE else "not configured" }, "database_stats": { "mushrooms": len(mushroom_db), "plants": len(plant_db), "total_species": len(mushroom_db) + len(plant_db) }, "endpoints": { "/recognize": "POST - Upload image for species recognition", "/submit-new-species": "POST - Submit a new species not in database", "/health": "GET - Health check", "/test/plantnet": "GET - Test PlantNet API connection", "/test/supabase": "GET - Test Supabase connection" } } @app.get("/health") async def health(): return { "status": "healthy", "model_loaded": True, "storage": "supabase" if USE_SUPABASE else "csv", "apis": { "bioclip": True, "vlm": OPENROUTER_API_KEY != "", "plantnet": PLANTNET_API_KEY != "", "supabase": USE_SUPABASE }, "species_count": { "mushrooms": len(mushroom_db), "plants": len(plant_db) } } @app.get("/test/supabase") async def test_supabase(): """Test Supabase connection and show database stats""" if not USE_SUPABASE: return { "success": False, "message": "Supabase not configured", "instructions": "Add SUPABASE_URL and SUPABASE_KEY to your .env file" } try: # Get species counts mushrooms = supabase.table("species").select("*").eq("species_type", "mushroom").execute() plants = supabase.table("species").select("*").eq("species_type", "plant").execute() return { "success": True, "message": "Supabase connection successful", "database_stats": { "mushrooms": len(mushrooms.data), "plants": len(plants.data), "total_species": len(mushrooms.data) + len(plants.data) }, "storage_type": "permanent", "cached_in_memory": True } except Exception as e: return { "success": False, "message": f"Supabase connection error: {str(e)}" } @app.get("/test/plantnet") async def test_plantnet(): """Test PlantNet API connection""" if not PLANTNET_API_KEY: return { "success": False, "message": "PlantNet API key not configured", "instructions": "Add PLANTNET_API_KEY to your .env file" } try: # Test with a simple request (no image, just check auth) api_url = f"https://my-api.plantnet.org/v2/projects" params = {'api-key': PLANTNET_API_KEY} response = requests.get(api_url, params=params, timeout=10) if response.status_code == 200: projects = response.json() return { "success": True, "message": "PlantNet API key is valid", "available_projects": projects if isinstance(projects, list) else ["all", "india", "weurope"], "daily_limit": "500 requests (free tier)" } elif response.status_code == 401: return { "success": False, "message": "Invalid PlantNet API key", "instructions": "Check your PLANTNET_API_KEY in .env file" } elif response.status_code == 429: return { "success": False, "message": "PlantNet API quota exceeded", "instructions": "Wait 24 hours or upgrade your plan" } else: return { "success": False, "message": f"PlantNet API error: {response.status_code}", "details": response.text } except Exception as e: return { "success": False, "message": f"Error testing PlantNet API: {str(e)}" } @app.post("/recognize") async def recognize(file: UploadFile = File(...)): """ Recognize mushroom/plant species from uploaded image Returns JSON with species information No caching to ensure fresh predictions """ try: # Read and validate image contents = await file.read() image = Image.open(BytesIO(contents)) # Convert to RGB if needed if image.mode != 'RGB': image = image.convert('RGB') # Perform recognition (fresh computation each time) result = recognize_species_internal(image) # Return with no-cache headers from fastapi import Response from fastapi.responses import JSONResponse response = JSONResponse(content=result) response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing image: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)