Update app.py

app.py

@@ -1,7 +1,9 @@
 # app.py
 """
 Elderly HealthWatch AI Backend (FastAPI)
-Ensure this file contains only Python code — not requirements.txt content.
+This variant uses:
+- facenet-pytorch or mtcnn if available
+- otherwise falls back to OpenCV Haar cascades (fast, CPU-only, lightweight)
 """
 
 import io
@@ -15,8 +17,9 @@ from PIL import Image
 import numpy as np
 import os
 import traceback
+import cv2  # opencv-python-headless expected installed
 
-# Try facenet-pytorch first, fallback to mtcnn
+# Attempt to import facenet-pytorch MTCNN first (recommended)
 try:
     from facenet_pytorch import MTCNN as FacenetMTCNN
     _MTCNN_IMPL = "facenet_pytorch"
@@ -24,6 +27,7 @@ except Exception:
     FacenetMTCNN = None
     _MTCNN_IMPL = None
 
+# Fallback to the classic mtcnn package
 if _MTCNN_IMPL is None:
     try:
         from mtcnn import MTCNN as ClassicMTCNN
@@ -31,15 +35,45 @@ if _MTCNN_IMPL is None:
     except Exception:
         ClassicMTCNN = None
 
-def create_mtcnn():
+# We'll create a fallback "opencv" detector if neither is present.
+def create_mtcnn_or_fallback():
+    """
+    Return:
+     - facenet_pytorch.MTCNN instance if available
+     - classic mtcnn instance if available
+     - dict with OpenCV cascade detector if neither available
+     - None if something unexpected happened
+    """
     if _MTCNN_IMPL == "facenet_pytorch" and FacenetMTCNN is not None:
-        return FacenetMTCNN(keep_all=False, device="cpu")
-    elif _MTCNN_IMPL == "mtcnn" and ClassicMTCNN is not None:
-        return ClassicMTCNN()
-    else:
-        return None
+        try:
+            return FacenetMTCNN(keep_all=False, device="cpu")
+        except Exception:
+            pass
+    if _MTCNN_IMPL == "mtcnn" and ClassicMTCNN is not None:
+        try:
+            return ClassicMTCNN()
+        except Exception:
+            pass
+
+    # OpenCV fallback: use Haar cascades (bundled with cv2)
+    try:
+        face_cascade_path = os.path.join(cv2.data.haarcascades, "haarcascade_frontalface_default.xml")
+        eye_cascade_path = os.path.join(cv2.data.haarcascades, "haarcascade_eye.xml")
+        if os.path.exists(face_cascade_path) and os.path.exists(eye_cascade_path):
+            face_cascade = cv2.CascadeClassifier(face_cascade_path)
+            eye_cascade = cv2.CascadeClassifier(eye_cascade_path)
+            return {"impl": "opencv", "face_cascade": face_cascade, "eye_cascade": eye_cascade}
+    except Exception:
+        pass
 
-mtcnn = create_mtcnn()
+    return None
+
+mtcnn = create_mtcnn_or_fallback()
+# mtcnn may now be:
+# - FacenetMTCNN instance (facenet_pytorch)
+# - ClassicMTCNN instance (mtcnn)
+# - dict {"impl":"opencv", "face_cascade":..., "eye_cascade":...}
+# - None
 
 app = FastAPI(title="Elderly HealthWatch AI Backend")
 
@@ -56,14 +90,13 @@ screenings_db: Dict[str, Dict[str, Any]] = {}
 def load_image_from_bytes(bytes_data: bytes) -> Image.Image:
     return Image.open(io.BytesIO(bytes_data)).convert("RGB")
 
-def estimate_eye_openness_from_detection(detection_result: Dict[str, Any]) -> float:
+def estimate_eye_openness_from_detection(confidence: float) -> float:
+    """
+    Simple mapping from detection confidence to an "eye_openness" heuristic in [0,1].
+    (Used by facenet/mtcnn flows)
+    """
     try:
-        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:
-            conf = float(detection_result[1]) if detection_result[1] is not None else 0.0
-        else:
-            conf = 0.0
+        conf = float(confidence)
         openness = min(max((conf * 1.15), 0.0), 1.0)
         return openness
     except Exception:
@@ -75,51 +108,67 @@ async def read_root():
 
 @app.get("/health")
 async def health_check():
-    return {"status": "healthy", "mtcnn_impl": _MTCNN_IMPL}
+    impl = None
+    if mtcnn is None:
+        impl = "none"
+    elif isinstance(mtcnn, dict) and mtcnn.get("impl") == "opencv":
+        impl = "opencv_haar_fallback"
+    else:
+        impl = _MTCNN_IMPL
+    return {"status": "healthy", "detector": 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 eye_openness_score & landmarks.
+    Uses facenet/mtcnn if available; otherwise OpenCV haar cascades.
+    """
     if mtcnn is None:
-        raise HTTPException(status_code=500, detail="No MTCNN implementation available in this environment.")
+        # No detector at all
+        raise HTTPException(status_code=500, detail="No face detector available in this deployment.")
+
     try:
         content = await image.read()
         if not content:
             raise HTTPException(status_code=400, detail="Empty file uploaded.")
         pil_img = load_image_from_bytes(content)
-        img_arr = np.asarray(pil_img)
+        img_arr = np.asarray(pil_img)  # RGB
 
-        if _MTCNN_IMPL == "facenet_pytorch":
-            boxes, probs, landmarks = mtcnn.detect(pil_img, landmarks=True)
-            if boxes is None or len(boxes) == 0:
+        # facenet-pytorch branch
+        if not isinstance(mtcnn, dict) and _MTCNN_IMPL == "facenet_pytorch":
+            try:
+                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": "कोई चेहरा नहीं मिला। कृपया सुनिश्चित करें कि आपका चेहरा फ्रेम में स्पष्ट रूप से दिखाई दे रहा है।"
+                    }
+                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:
+                    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
+                eye_openness_score = estimate_eye_openness_from_detection(prob)
+                is_valid = eye_openness_score >= 0.3
                 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": "कोई चेहरा नहीं मिला। कृपया सुनिश्चित करें कि आपका चेहरा फ्रेम में स्पष्ट रूप से दिखाई दे रहा है।"
-                }
-            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:
-                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
-            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
+                    "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:
+                traceback.print_exc()
+                raise HTTPException(status_code=500, detail="Face detector failed during inference.")
 
-        elif _MTCNN_IMPL == "mtcnn":
+        # classic mtcnn branch
+        if not isinstance(mtcnn, dict) and _MTCNN_IMPL == "mtcnn":
             try:
                 detections = mtcnn.detect_faces(img_arr)
             except Exception:
@@ -137,7 +186,7 @@ async def validate_eye_photo(image: UploadFile = File(...)):
             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})
+            eye_openness_score = estimate_eye_openness_from_detection(confidence)
             is_valid = eye_openness_score >= 0.3
             return {
                 "valid": bool(is_valid),
@@ -145,13 +194,53 @@ async def validate_eye_photo(image: UploadFile = File(...)):
                 "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
-                }
+                "eye_landmarks": {"left_eye": left_eye, "right_eye": right_eye}
             }
-        else:
-            raise HTTPException(status_code=500, detail="No face detector available in this deployment.")
+
+        # OpenCV Haar cascade fallback
+        if isinstance(mtcnn, dict) and mtcnn.get("impl") == "opencv":
+            try:
+                gray = cv2.cvtColor(img_arr, cv2.COLOR_RGB2GRAY)
+                face_cascade = mtcnn["face_cascade"]
+                eye_cascade = mtcnn["eye_cascade"]
+                faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=4, minSize=(60, 60))
+                if len(faces) == 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 face
+                (x, y, w, h) = faces[0]
+                roi_gray = gray[y:y+h, x:x+w]
+                eyes = eye_cascade.detectMultiScale(roi_gray, scaleFactor=1.1, minNeighbors=5, minSize=(20, 10))
+                # Heuristic openness: if eyes detected => open
+                eye_openness_score = 1.0 if len(eyes) >= 1 else 0.0
+                is_valid = eye_openness_score >= 0.3
+                # estimate coordinates relative to full image
+                left_eye = None
+                right_eye = None
+                if len(eyes) >= 1:
+                    ex, ey, ew, eh = eyes[0]
+                    # convert to image coords center
+                    cx = float(x + ex + ew/2)
+                    cy = float(y + ey + eh/2)
+                    left_eye = {"x": cx, "y": cy}
+                return {
+                    "valid": bool(is_valid),
+                    "face_detected": True,
+                    "eye_openness_score": round(eye_openness_score, 2),
+                    "message_english": "Photo looks good! Eyes are detected." if is_valid else "Eyes not detected. 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:
+                traceback.print_exc()
+                raise HTTPException(status_code=500, detail="OpenCV fallback detector failed.")
+        # Should not reach here
+        raise HTTPException(status_code=500, detail="Invalid detector configuration.")
     except HTTPException:
         raise
     except Exception as e:
@@ -246,10 +335,14 @@ async def process_screening(screening_id: str):
             raise RuntimeError("Eye image missing")
         face_img = Image.open(face_path).convert("RGB")
         eye_img = Image.open(eye_path).convert("RGB")
+
+        # Basic detection using whichever detector is available; populate quality_metrics
         face_detected = False
         face_confidence = 0.0
         left_eye_coord = right_eye_coord = None
-        if mtcnn is not None:
+
+        # facenet/mtcnn path
+        if not isinstance(mtcnn, dict) and (_MTCNN_IMPL == "facenet_pytorch" or _MTCNN_IMPL == "mtcnn"):
             try:
                 if _MTCNN_IMPL == "facenet_pytorch":
                     boxes, probs, landmarks = mtcnn.detect(face_img, landmarks=True)
@@ -272,6 +365,28 @@ async def process_screening(screening_id: str):
                         right_eye_coord = k.get("right_eye")
             except Exception:
                 traceback.print_exc()
+
+        # OpenCV fallback path
+        if isinstance(mtcnn, dict) and mtcnn.get("impl") == "opencv":
+            try:
+                arr = np.asarray(face_img)
+                gray = cv2.cvtColor(arr, cv2.COLOR_RGB2GRAY)
+                face_cascade = mtcnn["face_cascade"]
+                eye_cascade = mtcnn["eye_cascade"]
+                faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=4, minSize=(60, 60))
+                if len(faces) > 0:
+                    face_detected = True
+                    # crude confidence proxy by face size ratio
+                    (x, y, w, h) = faces[0]
+                    face_confidence = min(1.0, (w*h) / (arr.shape[0]*arr.shape[1]) * 4.0)
+                    roi_gray = gray[y:y+h, x:x+w]
+                    eyes = eye_cascade.detectMultiScale(roi_gray, scaleFactor=1.1, minNeighbors=5, minSize=(20, 10))
+                    if len(eyes) >= 1:
+                        ex, ey, ew, eh = eyes[0]
+                        left_eye_coord = {"x": float(x + ex + ew/2), "y": float(y + ey + eh/2)}
+            except Exception:
+                traceback.print_exc()
+
         face_quality_score = 0.85 if face_detected and face_confidence > 0.6 else 0.45
         quality_metrics = {
             "face_detected": face_detected,
@@ -282,9 +397,12 @@ async def process_screening(screening_id: str):
             "face_blur_estimate": int(np.var(np.asarray(face_img.convert("L"))))
         }
         screenings_db[screening_id]["quality_metrics"] = quality_metrics
+
+        # Simulate VLM/medical model steps (kept short)
         await asyncio.sleep(1)
         vlm_face_desc = "Patient appears to have normal facial tone; no severe jaundice visible."
         vlm_eye_desc = "Sclera shows mild yellowing."
+
         await asyncio.sleep(1)
         medical_insights = {
             "hemoglobin_estimate": 11.2,
@@ -293,8 +411,10 @@ async def process_screening(screening_id: str):
             "jaundice_indicators": ["mild scleral yellowing"],
             "confidence": 0.82
         }
+
         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",
@@ -308,6 +428,7 @@ async def process_screening(screening_id: str):
             "processing_time_ms": 1200
         }
         screenings_db[screening_id]["ai_results"] = ai_results
+
         disease_predictions = [
             {
                 "condition": "Iron Deficiency Anemia",
@@ -322,16 +443,19 @@ async def process_screening(screening_id: str):
                 "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()