from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi import Form from PIL import Image import base64 from io import BytesIO import os import time from src.image_desc import ImageAnalyzer from src.agent import ReportGenerator from src.envllm_service import env_llm_service, get_current_outdoor_values import asyncio import logging import requests import re from fastapi.middleware.cors import CORSMiddleware channel_id = "2916848" read_api_key = "4TSSCTPIYPXFZW9H" app = FastAPI() # Configure CORS for frontend access app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def get_tds_value(): try: url = f'https://api.thingspeak.com/channels/{channel_id}/feeds.json?api_key={read_api_key}&results=2' response = requests.get(url) data = response.json() feed = data['feeds'][0]['field1'] print("FFF", feed) return f"The TDS value is {feed}" except Exception as e: print(f"Error fetching TDS value: {e}") return "The TDS value is 300" # MULTI SUPPORT MODELS FOR FALLBACK IF ONE FAILS VLM_MODEL_NAMES = [ "meta-llama/llama-4-maverick-17b-128e-instruct", "meta-llama/llama-4-scout-17b-16e-instruct", ] LLM_MODEL_NAMES = [ "llama-3.3-70b-versatile", "llama3-70b-8192", "mistral-saba-24b", "qwen-qwq-32b", "gemma2-9b-it", "llama-3.1-8b-instant" ] # Initialize analyzers with multiple models analyzer = ImageAnalyzer(VLM_MODEL_NAMES) report_gen = ReportGenerator(LLM_MODEL_NAMES) # Format model names for report naming (using first model in each list) VLM_MODEL_NAME = "_".join(VLM_MODEL_NAMES[0].split("-")[:2]) LLM_MODEL_NAME = "_".join(LLM_MODEL_NAMES[0].split("-")[:2]) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # def get_tds_value(): async def process_image(image: UploadFile, additional_info=None): """Processes image and generates water quality report.""" try: image_data = await image.read() img = Image.open(BytesIO(image_data)) if img.mode != 'RGB': img = img.convert('RGB') buffered = BytesIO() img.save(buffered, format="JPEG") image_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8') start_time = time.time() report_message = await asyncio.to_thread(analyzer.generate_water_quality_report, image_base64, additional_info) end_time = time.time() logger.info(f"Time taken for water quality report generation: {end_time - start_time:.5f} seconds") return report_message except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing the image: {e}") @app.post("/analyze") async def analyze_water_quality(image: UploadFile = File(...), additional_info: str = Form(None)): """Endpoint to analyze water quality based on an uploaded image.""" try: report_message = await process_image(image, additional_info) if additional_info == "": additional_info = get_tds_value() print("additional_info", additional_info) start_time = time.time() combined_report = await asyncio.to_thread(report_gen.forward, report_message, additional_info) end_time = time.time() logger.info(f"Time taken for combined VLM and LLM report generation: {end_time - start_time:.5f} seconds") if isinstance(combined_report, str): combined_report = combined_report.replace("Corrected Report:", "").replace("Report:", "").replace("Water Quality Analysis Report", "") else: combined_report = combined_report.corrected_report.replace("Corrected Report:", "").replace("Report:", "").replace("Water Quality Analysis Report", "") print(type(combined_report)) return {"status": "success", "report": combined_report} except HTTPException as http_exc: raise http_exc except Exception as e: logger.error(f"Unhandled exception in /analyze: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {e}") @app.get("/environment-values") async def get_environment_values(location: str = None): """Endpoint to fetch the current environmental values for a location without generating a report. Parameters: - location: Location name to get values for (city, country, etc.) """ try: if not location: raise HTTPException(status_code=400, detail="Location parameter is required") # Get values from API api_values = get_current_outdoor_values(location) # Check if we got valid values if api_values.get('error') or None in [api_values.get('temp'), api_values.get('humid'), api_values.get('air')]: # If API failed, return the error logger.warning(f"API returned incomplete data for location '{location}': {api_values}") raise HTTPException(status_code=404, detail=f"Could not retrieve environmental data for location: {location}") # Return the parameters without generating a report return { "status": "success", "parameters": { "temperature": api_values['temp'], "humidity": api_values['humid'], "air_quality_index": api_values['air'], "location": api_values['location'], "lat": api_values['lat'], "lon": api_values['lon'] } } except HTTPException as http_exc: raise http_exc except Exception as e: logger.error(f"Error fetching environment values: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}") @app.get("/environment-report") async def get_environment_report( use_api: bool = True, temperature: int = None, humidity: int = None, air_quality: int = None, location: str = None, latitude: float = None, longitude: float = None ): """Endpoint to generate environmental reports based on weather parameters. Parameters: - use_api: Whether to use the OpenWeatherMap API for real-time data (default: True) - temperature: Optional temperature in Celsius if not using API - humidity: Optional humidity percentage if not using API - air_quality: Optional air quality index if not using API (1-5) - location: Optional location name (city, country, etc.) - latitude: Optional latitude coordinate - longitude: Optional longitude coordinate """ try: start_time = time.time() # We use asyncio.to_thread to move the model inference to a separate thread # to avoid blocking the FastAPI event loop result = await asyncio.to_thread( env_llm_service.generate_environment_report, location=location, temp=temperature, humid=humidity, air=air_quality, use_api=use_api, lat=latitude, lon=longitude ) end_time = time.time() logger.info(f"Time taken for environment report generation: {end_time - start_time:.5f} seconds") return result except Exception as e: logger.error(f"Error generating environment report: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}") @app.get("/environment-stats") async def get_environment_stats(location: str = None, limit: int = 10, format: str = "table"): """Endpoint to retrieve historical environmental data. Parameters: - location: Optional location name to filter data - limit: Number of records to return (default: 10) - format: Return format - "table" for HTML table or "json" for raw data (default: table) """ try: if format.lower() == "table": result = env_llm_service.get_realtime_values_table(location, limit) return {"status": "success", "data": result} else: result = env_llm_service.get_historical_data(location, format="json") return {"status": "success", "data": result} except Exception as e: logger.error(f"Error retrieving environmental stats: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")