Update app.py

app.py

@@ -1,459 +1,9 @@
-"""
-Final `app.py` — ready to run (designed for HF Spaces / Docker).
-- Uses facenet-pytorch's MTCNN (preferred) but will attempt to fall back to mtcnn if needed.
-- Binds to port 7860 in the __main__ block for local testing (Spaces expects port 7860).
-- Writes only to /tmp for any temporary files.
-- Keeps BackgroundTasks short and resilient.
-
-Install dependencies (recommended):
-    fastapi uvicorn[standard] pillow numpy opencv-python-headless facenet-pytorch torch
-
-If you must use the older `mtcnn` (tensorflow-backed) package, install it explicitly and
-the fallback will try to use it.
-"""
-
-import io
-import uuid
-import asyncio
-from typing import Dict, Any, Optional
-from datetime import datetime
-from fastapi import FastAPI, UploadFile, File, BackgroundTasks, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from PIL import Image
-import numpy as np
-import os
-import traceback
-
-# Try facenet-pytorch MTCNN first (recommended)
-try:
-    from facenet_pytorch import MTCNN as FacenetMTCNN
-    _MTCNN_IMPL = "facenet_pytorch"
-except Exception:
-    FacenetMTCNN = None
-    _MTCNN_IMPL = None
-
-# Fallback to the classic mtcnn package if facenet-pytorch is not available
-if _MTCNN_IMPL is None:
-    try:
-        from mtcnn import MTCNN as ClassicMTCNN
-        _MTCNN_IMPL = "mtcnn"
-    except Exception:
-        ClassicMTCNN = None
-
-# Initialize MTCNN detector depending on availability
-def create_mtcnn():
-    if _MTCNN_IMPL == "facenet_pytorch" and FacenetMTCNN is not None:
-        # keep device CPU by default; Spaces typically doesn't provide GPUs
-        return FacenetMTCNN(keep_all=False, device="cpu")
-    elif _MTCNN_IMPL == "mtcnn" and ClassicMTCNN is not None:
-        return ClassicMTCNN()
-    else:
-        return None
-
-mtcnn = create_mtcnn()
-
-app = FastAPI(title="Elderly HealthWatch AI Backend")
-
-# CORS for demo
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# In-memory store for demo (replace with persistent DB in prod)
-screenings_db: Dict[str, Dict[str, Any]] = {}
-
-# Utility: safe image load from UploadFile bytes -> PIL.Image
-def load_image_from_bytes(bytes_data: bytes) -> Image.Image:
-    return Image.open(io.BytesIO(bytes_data)).convert("RGB")
-
-# Heuristic eye openness: uses detection probability/confidence and face size
-def estimate_eye_openness_from_detection(detection_result: Dict[str, Any]) -> float:
-    """
-    Returns a float in [0.0, 1.0] estimating eye openness.
-    For facenet-pytorch, detection_result may be (box, prob, landmarks) depending on API.
-    For classic mtcnn, the detect_faces() dict is used.
-    We keep a conservative, simple heuristic: rely on detection probability and landmark presence.
-    """
-    try:
-        # facenet-pytorch path: we might get prob and landmarks separately upstream,
-        # but this helper expects a uniform dict-like structure if possible.
-        if isinstance(detection_result, dict) and "confidence" in detection_result:
-            conf = float(detection_result.get("confidence", 0.0))
-        elif isinstance(detection_result, (list, tuple)) and len(detection_result) >= 2:
-            # some APIs return (boxes, probs) or similar structures — guard against that upstream
-            conf = float(detection_result[1]) if detection_result[1] is not None else 0.0
-        else:
-            conf = 0.0
-
-        # Basic scaling: make confidence map into [0,1], nudge slightly
-        openness = min(max((conf * 1.15), 0.0), 1.0)
-        return openness
-    except Exception:
-        return 0.0
-
-@app.get("/")
-async def read_root():
-    return {"message": "Elderly HealthWatch AI Backend"}
-
-@app.get("/health")
-async def health_check():
-    return {"status": "healthy", "mtcnn_impl": _MTCNN_IMPL}
-
-@app.post("/api/v1/validate-eye-photo")
-async def validate_eye_photo(image: UploadFile = File(...)):
-    """
-    Validate an eye photo: detects a face and returns an eye_openness_score and landmarks.
-    Uses MTCNN implementation available in the container.
-    """
-    if mtcnn is None:
-        raise HTTPException(status_code=500, detail="No MTCNN implementation available in this environment.")
-
-    try:
-        content = await image.read()
-        if not content:
-            raise HTTPException(status_code=400, detail="Empty file uploaded.")
-        pil_img = load_image_from_bytes(content)
-
-        # Convert to numpy RGB for some detectors that expect np arrays
-        img_arr = np.asarray(pil_img)
-
-        # Two possible MTCNN APIs:
-        # - facenet_pytorch.MTCNN: has detect and forward methods (detect returns boxes, probs, landmarks)
-        # - mtcnn (older): has detect_faces(image) returning list of dicts with 'confidence' and 'keypoints'
-        if _MTCNN_IMPL == "facenet_pytorch":
-            try:
-                # facenet_pytorch.MTCNN.detect returns (boxes, probs, landmarks)
-                boxes, probs, landmarks = mtcnn.detect(pil_img, landmarks=True)
-                if boxes is None or len(boxes) == 0:
-                    return {
-                        "valid": False,
-                        "face_detected": False,
-                        "eye_openness_score": 0.0,
-                        "message_english": "No face detected. Please ensure your face is clearly visible in the frame.",
-                        "message_hindi": "कोई चेहरा नहीं मिला। कृपया सुनिश्चित करें कि आपका चेहरा फ्रेम में स्पष्ट रूप से दिखाई दे रहा है।"
-                    }
-
-                # Use first detection
-                prob = float(probs[0]) if probs is not None else 0.0
-                lm = landmarks[0] if landmarks is not None else None
-                if lm is not None and len(lm) >= 2:
-                    # facenet landmarks are [[left_eye_x, left_eye_y], [right_eye_x, right_eye_y], ...]
-                    left_eye = {"x": float(lm[0][0]), "y": float(lm[0][1])}
-                    right_eye = {"x": float(lm[1][0]), "y": float(lm[1][1])}
-                else:
-                    left_eye = right_eye = None
-
-                # Estimate openness
-                eye_openness_score = estimate_eye_openness_from_detection((None, prob))
-                is_valid = eye_openness_score >= 0.3
-
-                return {
-                    "valid": bool(is_valid),
-                    "face_detected": True,
-                    "eye_openness_score": round(eye_openness_score, 2),
-                    "message_english": "Photo looks good! Eyes are properly open." if is_valid else "Eyes appear to be closed or partially closed. Please open your eyes wide and try again.",
-                    "message_hindi": "फोटो अच्छी है! आंखें ठीक से खुली हैं।" if is_valid else "आंखें बंद या आंशिक रूप से बंद दिखाई दे रही हैं। कृपया अपनी आंखें चौड़ी खोलें और पुनः प्रयास करें।",
-                    "eye_landmarks": {
-                        "left_eye": left_eye,
-                        "right_eye": right_eye
-                    }
-                }
-            except Exception as e:
-                # Facenet path failed unexpectedly; fall through to generic error handling
-                raise
-
-        elif _MTCNN_IMPL == "mtcnn":
-            # classic mtcnn: detect_faces returns list of dicts
-            try:
-                detections = mtcnn.detect_faces(img_arr)
-            except Exception:
-                # some mtcnn implementations accept PIL images
-                detections = mtcnn.detect_faces(pil_img)
-
-            if not detections:
-                return {
-                    "valid": False,
-                    "face_detected": False,
-                    "eye_openness_score": 0.0,
-                    "message_english": "No face detected. Please ensure your face is clearly visible in the frame.",
-                    "message_hindi": "कोई चेहरा नहीं मिला। कृपया सुनिश्चित करें कि आपका चेहरा फ्रेम में स्पष्ट रूप से दिखाई दे रहा है।"
-                }
-
-            face = detections[0]
-            keypoints = face.get("keypoints", {})
-            left_eye = keypoints.get("left_eye")
-            right_eye = keypoints.get("right_eye")
-            confidence = float(face.get("confidence", 0.0))
-
-            eye_openness_score = estimate_eye_openness_from_detection({"confidence": confidence})
-            is_valid = eye_openness_score >= 0.3
-
-            return {
-                "valid": bool(is_valid),
-                "face_detected": True,
-                "eye_openness_score": round(eye_openness_score, 2),
-                "message_english": "Photo looks good! Eyes are properly open." if is_valid else "Eyes appear to be closed or partially closed. Please open your eyes wide and try again.",
-                "message_hindi": "फोटो अच्छी है! आंखें ठीक से खुली हैं।" if is_valid else "आंखें बंद या आंशिक रूप से बंद दिखाई दे रही हैं। कृपया अपनी आंखें चौड़ी खोलें और पुनः प्रयास करें।",
-                "eye_landmarks": {
-                    "left_eye": left_eye,
-                    "right_eye": right_eye
-                }
-            }
-        else:
-            raise HTTPException(status_code=500, detail="No face detector available in this deployment.")
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        # Log traceback to container logs for debugging
-        traceback.print_exc()
-        return {
-            "valid": False,
-            "face_detected": False,
-            "eye_openness_score": 0.0,
-            "message_english": "Error processing image. Please try again.",
-            "message_hindi": "छवि प्रोसेस करने में त्रुटि। कृपया पुनः प्रयास करें।",
-            "error": str(e)
-        }
-
-
-@app.post("/api/v1/upload")
-async def upload_images(
-    background_tasks: BackgroundTasks,
-    face_image: UploadFile = File(...),
-    eye_image: UploadFile = File(...)
-):
-    """
-    Accept face and eye images and enqueue background processing.
-    Stores minimal metadata in an in-memory dict. Use external storage in production.
-    """
-    try:
-        screening_id = str(uuid.uuid4())
-        now = datetime.utcnow().isoformat() + "Z"
-
-        # In production you would persist the bytes to S3 or to a DB.
-        # For demo we store temp bytes in /tmp/<screening_id> (ephemeral)
-        tmp_dir = "/tmp/elderly_healthwatch"
-        os.makedirs(tmp_dir, exist_ok=True)
-
-        face_path = os.path.join(tmp_dir, f"{screening_id}_face.jpg")
-        eye_path = os.path.join(tmp_dir, f"{screening_id}_eye.jpg")
-
-        # Save raw bytes quickly
-        face_bytes = await face_image.read()
-        eye_bytes = await eye_image.read()
-        with open(face_path, "wb") as f:
-            f.write(face_bytes)
-        with open(eye_path, "wb") as f:
-            f.write(eye_bytes)
-
-        screenings_db[screening_id] = {
-            "id": screening_id,
-            "timestamp": now,
-            "face_image_path": face_path,
-            "eye_image_path": eye_path,
-            "status": "queued",
-            "quality_metrics": {},
-            "ai_results": {},
-            "disease_predictions": [],
-            "recommendations": {}
-        }
-
-        # Start background processing (short tasks preferred)
-        background_tasks.add_task(process_screening, screening_id)
-
-        return {"screening_id": screening_id}
-    except Exception as e:
-        traceback.print_exc()
-        raise HTTPException(status_code=500, detail=f"Failed to upload images: {e}")
-
-
-@app.post("/api/v1/analyze/{screening_id}")
-async def analyze_screening(screening_id: str, background_tasks: BackgroundTasks):
-    """Trigger analysis for an existing screening (re-run)."""
-    if screening_id not in screenings_db:
-        raise HTTPException(status_code=404, detail="Screening not found")
-    if screenings_db[screening_id].get("status") == "processing":
-        return {"message": "Already processing"}
-    screenings_db[screening_id]["status"] = "queued"
-    background_tasks.add_task(process_screening, screening_id)
-    return {"message": "Analysis enqueued"}
-
-
-@app.get("/api/v1/status/{screening_id}")
-async def get_status(screening_id: str):
-    if screening_id not in screenings_db:
-        raise HTTPException(status_code=404, detail="Screening not found")
-    status = screenings_db[screening_id].get("status", "unknown")
-    progress = 50 if status == "processing" else (100 if status == "completed" else 0)
-    return {
-        "screening_id": screening_id,
-        "status": status,
-        "progress": progress
-    }
-
-
-@app.get("/api/v1/results/{screening_id}")
-async def get_results(screening_id: str):
-    if screening_id not in screenings_db:
-        raise HTTPException(status_code=404, detail="Screening not found")
-    return screenings_db[screening_id]
-
-
-@app.get("/api/v1/history/{user_id}")
-async def get_history(user_id: str):
-    # This demo does not link screenings to users by default
-    history = [s for s in screenings_db.values() if s.get("user_id") == user_id]
-    return {"screenings": history}
-
-
-async def process_screening(screening_id: str):
-    """
-    Background processing pipeline:
-    - quick image checks using MTCNN
-    - placeholder VLM / medical LM stages simulated with sleeps
-    - populates quality_metrics, ai_results, disease_predictions, recommendations
-    Keep tasks reasonably short to avoid container restarts killing long jobs.
-    """
-    try:
-        if screening_id not in screenings_db:
-            print(f"[process_screening] screening {screening_id} not found")
-            return
-
-        screenings_db[screening_id]["status"] = "processing"
-        print(f"[process_screening] Starting {screening_id}")
-
-        entry = screenings_db[screening_id]
-        face_path = entry.get("face_image_path")
-        eye_path = entry.get("eye_image_path")
-
-        # Basic file checks
-        if not (face_path and os.path.exists(face_path)):
-            raise RuntimeError("Face image missing")
-        if not (eye_path and os.path.exists(eye_path)):
-            raise RuntimeError("Eye image missing")
-
-        # Load images
-        face_img = Image.open(face_path).convert("RGB")
-        eye_img = Image.open(eye_path).convert("RGB")
-
-        # Stage 1: face detection + quality metrics (fast)
-        face_detected = False
-        face_confidence = 0.0
-        left_eye_coord = right_eye_coord = None
-
-        if mtcnn is not None:
-            try:
-                if _MTCNN_IMPL == "facenet_pytorch":
-                    boxes, probs, landmarks = mtcnn.detect(face_img, landmarks=True)
-                    if boxes is not None and len(boxes) > 0:
-                        face_detected = True
-                        face_confidence = float(probs[0]) if probs is not None else 0.0
-                        if landmarks is not None:
-                            lm = landmarks[0]
-                            if len(lm) >= 2:
-                                left_eye_coord = {"x": float(lm[0][0]), "y": float(lm[0][1])}
-                                right_eye_coord = {"x": float(lm[1][0]), "y": float(lm[1][1])}
-                else:
-                    # classic mtcnn
-                    arr = np.asarray(face_img)
-                    detections = mtcnn.detect_faces(arr)
-                    if detections:
-                        face_detected = True
-                        face_confidence = float(detections[0].get("confidence", 0.0))
-                        k = detections[0].get("keypoints", {})
-                        left_eye_coord = k.get("left_eye")
-                        right_eye_coord = k.get("right_eye")
-            except Exception:
-                traceback.print_exc()
-
-        # Simple quality metrics (placeholders but useful)
-        face_quality_score = 0.85 if face_detected and face_confidence > 0.6 else 0.45
-        quality_metrics = {
-            "face_detected": face_detected,
-            "face_confidence": round(face_confidence, 3),
-            "face_quality_score": round(face_quality_score, 2),
-            "eye_coords": {"left_eye": left_eye_coord, "right_eye": right_eye_coord},
-            "face_brightness": int(np.mean(np.asarray(face_img.convert("L")))),
-            "face_blur_estimate": int(np.var(np.asarray(face_img.convert("L"))))  # crude proxy
-        }
-        screenings_db[screening_id]["quality_metrics"] = quality_metrics
-
-        # Stage 2/3: placeholder Visual and Medical model steps (simulate with sleeps)
-        await asyncio.sleep(1)  # simulate feature extraction
-        vlm_face_desc = "Patient appears to have normal facial tone; no severe jaundice visible."
-        vlm_eye_desc = "Sclera shows mild yellowing."  # placeholder
-
-        await asyncio.sleep(1)  # simulate medical LM analysis
-        medical_insights = {
-            "hemoglobin_estimate": 11.2,
-            "bilirubin_estimate": 1.8,
-            "anemia_indicators": ["pale skin"],
-            "jaundice_indicators": ["mild scleral yellowing"],
-            "confidence": 0.82
-        }
-
-        # Stage 4: disease inference and recommendations
-        hem = medical_insights["hemoglobin_estimate"]
-        bil = medical_insights["bilirubin_estimate"]
-
-        ai_results = {
-            "hemoglobin_g_dl": hem,
-            "anemia_status": "Mild Anemia" if hem < 12 else "Normal",
-            "anemia_confidence": medical_insights["confidence"],
-            "bilirubin_mg_dl": bil,
-            "jaundice_status": "Normal" if bil < 2.5 else "Elevated",
-            "jaundice_confidence": medical_insights["confidence"],
-            "vlm_face_description": vlm_face_desc,
-            "vlm_eye_description": vlm_eye_desc,
-            "medical_insights": medical_insights,
-            "processing_time_ms": 1200
-        }
-        screenings_db[screening_id]["ai_results"] = ai_results
-
-        disease_predictions = [
-            {
-                "condition": "Iron Deficiency Anemia",
-                "risk_level": "Medium" if hem < 12 else "Low",
-                "probability": 0.76 if hem < 12 else 0.23,
-                "confidence": medical_insights["confidence"]
-            },
-            {
-                "condition": "Jaundice",
-                "risk_level": "Low" if bil < 2.5 else "Medium",
-                "probability": 0.23 if bil < 2.5 else 0.45,
-                "confidence": medical_insights["confidence"]
-            }
-        ]
-
-        recommendations = {
-            "action_needed": "consult" if hem < 12 else "monitor",
-            "message_english": f"Your hemoglobin is {hem} g/dL. Please consult a doctor within 2 weeks for blood tests.",
-            "message_hindi": f"आपका हीमोग्लोबिन {hem} g/dL है। कृपया 2 सप्ताह में डॉक्टर से परामर्श करें।"
-        }
-
-        screenings_db[screening_id].update({
-            "status": "completed",
-            "disease_predictions": disease_predictions,
-            "recommendations": recommendations
-        })
-
-        print(f"[process_screening] Completed {screening_id}")
-
-    except Exception as e:
-        traceback.print_exc()
-        if screening_id in screenings_db:
-            screenings_db[screening_id]["status"] = "failed"
-            screenings_db[screening_id]["error"] = str(e)
-        else:
-            print(f"[process_screening] Failed for unknown screening {screening_id}: {e}")
-
-
-if __name__ == "__main__":
-    # Local debug run (Spaces will run uvicorn via Dockerfile/CMD)
-    import uvicorn
-    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)
+fastapi==0.95.2
+uvicorn[standard]==0.22.0
+numpy==1.26.0
+opencv-python-headless==4.8.0.74
+Pillow==10.0.1
+facenet-pytorch==3.4.0   # optional, keep if you use facenet-pytorch
+torch==2.2.0            # optional, keep if you use facenet-pytorch
+python-multipart==0.0.6
+aiofiles==23.1.0