Create app.py

app.py

@@ -0,0 +1,459 @@
+"""
+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)