Pvpres/SkinSightModel

1from fastapi import FastAPI, HTTPException2from fastapi.responses import HTMLResponse3from pydantic import BaseModel4from typing import List, Optional, Dict, Any5import numpy as np6import time7import base648import io9from PIL import Image10from face_scanner import FaceScanner11import logging12import os13from fastapi.middleware.cors import CORSMiddleware14import uvicorn15 16# Configure logging17logging.basicConfig(level=logging.INFO)18logger = logging.getLogger(__name__)19 20# Initialize FastAPI app21app = FastAPI(22    title="SkinSight API",23    description="Real-time skin condition analysis using computer vision and deep learning",24    version="1.0.0"25)26 27# Enable CORS for deployed static site(s)28app.add_middleware(29    CORSMiddleware,30    allow_origins=[31        "https://pvpres.github.io",32        "https://skinsight-kgf9.onrender.com",33        "http://localhost:8080",34        "http://localhost:8000",35    ],36    allow_origin_regex=None,  # set to None to rely on explicit origins37    allow_credentials=True,38    allow_methods=["GET", "POST", "OPTIONS"],39    allow_headers=["*"],40    expose_headers=["*"],41    max_age=600,42)43 44# Global scanner instance45scanner: FaceScanner | None = None46device = None47 48# Pydantic models for API requests/responses49class ScanRequest(BaseModel):50    image_data: str  # Base64 encoded image51    scan_duration: Optional[float] = 3.0  # Duration in seconds52 53class BatchScanRequest(BaseModel):54    image_data_list: List[str]  # List of base64 encoded images55    scan_duration: Optional[float] = 3.0  # Duration in seconds56 57class ScanResponse(BaseModel):58    success: bool59    message: str60    results: Optional[Dict[str, Any]] = None61    processing_time: Optional[float] = None62 63class HealthResponse(BaseModel):64    status: str65    model_loaded: bool66    device: str67 68# Initialize scanner on startup69@app.on_event("startup")70async def startup_event():71    """Initialize the face scanner and device on application startup"""72    global scanner, device73    74    try:75        # Instantiate a single FaceScanner (handles device, model, transforms)76        model_path = os.environ.get(77            "MODEL_PATH",78            "prod_model/best_model_twohead_0922_031215.pth"79        )80        scanner = FaceScanner(model_path=model_path)81        device = scanner.device82        logger.info("FaceScanner initialized successfully!")83        84    except Exception as e:85        logger.error(f"Failed to initialize model: {e}")86        raise e87 88# Utility function89def decode_base64_image(image_data: str) -> np.ndarray:90    """Decode base64 image data to numpy array"""91    try:92        # Remove data URL prefix if present93        if ',' in image_data:94            image_data = image_data.split(',')[1]95        96        # Decode base6497        image_bytes = base64.b64decode(image_data)98        image = Image.open(io.BytesIO(image_bytes))99        100        # Convert to RGB if needed101        if image.mode != 'RGB':102            image = image.convert('RGB')103        104        # Convert to numpy array105        return np.array(image)106    except Exception as e:107        raise HTTPException(status_code=400, detail=f"Invalid image data: {e}")108 109def analyze_skin_condition(face_crops: List[np.ndarray]) -> Dict[str, Any]:110    """Delegate analysis to FaceScanner for consistency"""111    if scanner is None:112        raise HTTPException(status_code=500, detail="Scanner not initialized")113    try:114        return scanner.analyze_skin_condition(face_crops)115    except ValueError as e:116        raise HTTPException(status_code=400, detail=str(e))117    except Exception as e:118        logger.error(f"Skin analysis error: {e}")119        raise HTTPException(status_code=500, detail=f"Analysis failed: {e}")120 121# API Endpoints122@app.get("/", response_class=HTMLResponse)123async def root():124    """Serve the main web interface"""125    try:126        with open("web_interface.html", "r") as f:127            html_content = f.read()128        return HTMLResponse(content=html_content)129    except FileNotFoundError:130        return HTMLResponse(content="""131        <html><body>132        <h1>SkinSight API</h1>133        <p>API is running! Web interface not found.</p>134        <p>Available endpoints:</p>135        <ul>136            <li><a href="/docs">API Documentation</a></li>137            <li><a href="/health">Health Check</a></li>138            <li>POST /analyze - Single image analysis</li>139            <li>POST /analyze-batch - Batch image analysis</li>140        </ul>141        </body></html>142        """)143 144@app.get("/health", response_model=HealthResponse)145async def health_check():146    """Health check endpoint"""147    return HealthResponse(148        status="healthy" if scanner is not None else "unhealthy",149        model_loaded=scanner is not None,150        device=str(device) if device else "unknown"151    )152 153@app.post("/analyze", response_model=ScanResponse)154async def analyze_skin(request: ScanRequest):155    """Analyze skin condition from uploaded image"""156    start_time = time.time()157    158    try:159        # Decode image160        image = decode_base64_image(request.image_data)161        162        # Detect and crop face using FaceScanner helper for consistency163        face_crop = scanner.detect_and_crop_face(image) if scanner else None164        165        if face_crop is None:166            return ScanResponse(167                success=False,168                message="No face detected or multiple faces detected. Please ensure exactly one face is visible.",169                processing_time=time.time() - start_time170            )171        172        # Analyze skin condition173        results = analyze_skin_condition([face_crop])174        175        processing_time = time.time() - start_time176        177        return ScanResponse(178            success=True,179            message="Analysis completed successfully",180            results=results,181            processing_time=round(processing_time, 2)182        )183        184    except HTTPException:185        raise186    except Exception as e:187        logger.error(f"Unexpected error in analyze_skin: {e}")188        return ScanResponse(189            success=False,190            message=f"Unexpected error: {e}",191            processing_time=time.time() - start_time192        )193 194@app.post("/analyze-batch", response_model=ScanResponse)195async def analyze_skin_batch(request: BatchScanRequest):196    """Analyze skin condition from multiple images (proper batching)"""197    start_time = time.time()198    199    try:200        if not request.image_data_list:201            return ScanResponse(202                success=False,203                message="No images provided for batch analysis",204                processing_time=time.time() - start_time205            )206        207        num_images = len(request.image_data_list)208        logger.info(f"📥 Received batch request with {num_images} images")209        210        # Limit number of frames to prevent excessive processing211        max_frames = 15212        images_to_process = request.image_data_list[:max_frames]213        if num_images > max_frames:214            logger.warning(f"⚠️ Limiting batch processing from {num_images} to {max_frames} frames")215        216        # Decode all images and detect faces217        decode_start = time.time()218        face_crops = []219        for i, image_data in enumerate(images_to_process):220            try:221                frame_start = time.time()222                image = decode_base64_image(image_data)223                decode_time = time.time() - frame_start224                225                detect_start = time.time()226                face_crop = scanner.detect_and_crop_face(image) if scanner else None227                detect_time = time.time() - detect_start228                229                if face_crop is not None:230                    face_crops.append(face_crop)231                    logger.info(f"✅ Processed frame {i+1}/{len(images_to_process)}: decode={decode_time:.3f}s, detect={detect_time:.3f}s")232                else:233                    logger.warning(f"⚠️ No face detected in image {i+1}")234                    235            except Exception as e:236                logger.warning(f"❌ Error processing image {i+1}: {e}")237                continue238        239        decode_time = time.time() - decode_start240        logger.info(f"📊 Decoded {len(face_crops)}/{len(images_to_process)} faces in {decode_time:.2f}s")241        242        if not face_crops:243            return ScanResponse(244                success=False,245                message="No valid faces detected in any of the provided images",246                processing_time=time.time() - start_time247            )248        249        # Analyze skin condition with proper batching via FaceScanner250        analysis_start = time.time()251        logger.info(f"🧠 Starting model inference on {len(face_crops)} face crops...")252        results = analyze_skin_condition(face_crops)253        analysis_time = time.time() - analysis_start254        logger.info(f"✅ Model inference completed in {analysis_time:.2f}s")255        256        processing_time = time.time() - start_time257        logger.info(f"🎯 Total batch processing time: {processing_time:.2f}s (decode: {decode_time:.2f}s, analysis: {analysis_time:.2f}s)")258        259        return ScanResponse(260            success=True,261            message=f"Batch analysis completed successfully on {len(face_crops)} face crops",262            results=results,263            processing_time=round(processing_time, 2)264        )265        266    except HTTPException:267        raise268    except Exception as e:269        logger.error(f"❌ Unexpected error in analyze_skin_batch: {e}", exc_info=True)270        return ScanResponse(271            success=False,272            message=f"Batch analysis error: {e}",273            processing_time=time.time() - start_time274        )275 276if __name__ == "__main__":277    uvicorn.run(app, host="0.0.0.0", port=8000)