Update app.py

app.py

@@ -4,84 +4,315 @@ import pandas as pd
 import tempfile
 import cv2
 import os
+from typing import List, Dict, Any, Optional
+
+# For live video capture via WebRTC
+from streamlit_webrtc import webrtc_streamer, WebRtcMode, RTCConfiguration
+import av
+import time
+from collections import deque
 
 st.set_page_config(page_title="AI Health Lab", layout="wide")
 
-st.title("🧬 Face-based Health Lab")
-st.markdown("Upload a **facial image** (for Hb) and a **20-30 s face video** (for HR).\n\n"
-            "This is a baseline app – models will be added for each test step-by-step.")
-
-# -------------------------------------------------
-# Sidebar instructions
-# -------------------------------------------------
-st.sidebar.header("Instructions")
-st.sidebar.info(
-    "1. Upload a good quality **face / eye image** for Hemoglobin.\n"
-    "2. Upload a **20-30 s face video** for Heart-Rate.\n"
-    "3. Click **Run Analysis** to see test results.\n\n"
-    "Later we will add more tests and replace the dummy predictions with real models."
+# =========================================================
+# Config
+# =========================================================
+# ICE config for WebRTC (use public Google STUN; replace with your TURN for prod)
+RTC_CONFIGURATION = RTCConfiguration(
+    {"iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]}
 )
 
-# -------------------------------------------------
-# Input widgets
-# -------------------------------------------------
-uploaded_img = st.file_uploader("Upload Face / Eye Image for Hemoglobin", type=["jpg","jpeg","png"])
-uploaded_vid = st.file_uploader("Upload Face Video for Heart-Rate (20-30 s)", type=["mp4","avi","mov"])
+# List your 27 tests here (we’ll fill values as models arrive)
+TESTS = [
+    "Hemoglobin (Hb)",           # 1
+    "Heart Rate (HR)",           # 2
+    "SpO₂",                      # 3
+    "Respiration Rate (RR)",     # 4
+    "Systolic BP",               # 5
+    "Diastolic BP",              # 6
+    "Body Temperature",          # 7
+    "BMI",                       # 8
+    "Blood Glucose (Fasting)",   # 9
+    "Blood Glucose (PP)",        #10
+    "Total Cholesterol",         #11
+    "LDL",                       #12
+    "HDL",                       #13
+    "Triglycerides",             #14
+    "HbA1c",                     #15
+    "Hematocrit (HCT)",          #16
+    "RBC Count",                 #17
+    "WBC Count",                 #18
+    "Platelet Count",            #19
+    "Serum Creatinine",          #20
+    "eGFR",                      #21
+    "Uric Acid",                 #22
+    "Vitamin D",                 #23
+    "Calcium",                   #24
+    "Sodium",                    #25
+    "Potassium",                 #26
+    "CRP"                        #27
+]
 
-run_btn = st.button("Run Analysis")
+UNITS = {
+    "Hemoglobin (Hb)": "g/dL",
+    "Heart Rate (HR)": "bpm",
+    "SpO₂": "%",
+    "Respiration Rate (RR)": "breaths/min",
+    "Systolic BP": "mmHg",
+    "Diastolic BP": "mmHg",
+    "Body Temperature": "°C",
+    "BMI": "kg/m²",
+    "Blood Glucose (Fasting)": "mg/dL",
+    "Blood Glucose (PP)": "mg/dL",
+    "Total Cholesterol": "mg/dL",
+    "LDL": "mg/dL",
+    "HDL": "mg/dL",
+    "Triglycerides": "mg/dL",
+    "HbA1c": "%",
+    "Hematocrit (HCT)": "%",
+    "RBC Count": "10^6/µL",
+    "WBC Count": "10^3/µL",
+    "Platelet Count": "10^3/µL",
+    "Serum Creatinine": "mg/dL",
+    "eGFR": "mL/min/1.73m²",
+    "Uric Acid": "mg/dL",
+    "Vitamin D": "ng/mL",
+    "Calcium": "mg/dL",
+    "Sodium": "mEq/L",
+    "Potassium": "mEq/L",
+    "CRP": "mg/L"
+}
 
-# -------------------------------------------------
+# =========================================================
 # Placeholder prediction functions
-# -------------------------------------------------
-def predict_hemoglobin(image: np.ndarray) -> float:
-    # TODO: replace with actual Hb model
-    return float(np.random.uniform(11.0,15.0))
-
-def predict_heart_rate(video_path: str) -> float:
-    # TODO: replace with real rPPG inference
-    return float(np.random.uniform(65,85))
-
-# -------------------------------------------------
-# Run inference
-# -------------------------------------------------
-if run_btn:
-    results = []
-
-    # ---- Hemoglobin ----
-    if uploaded_img is not None:
-        file_bytes = np.asarray(bytearray(uploaded_img.read()), dtype=np.uint8)
-        img = cv2.imdecode(file_bytes, cv2.IMREAD_COLOR)
-        hb = predict_hemoglobin(img)
-        results.append({"Test": "Hemoglobin", "Result": f"{hb:.2f} g/dL"})
-    else:
-        results.append({"Test": "Hemoglobin", "Result": "No image uploaded"})
+# (Swap these with real model calls as you implement tests)
+# =========================================================
+def predict_hemoglobin_from_image(image_bgr: np.ndarray) -> Optional[float]:
+    """
+    TODO: Replace with real Hb inference (e.g., conjunctiva crop + CNN/ViT model).
+    Returns float(g/dL) or None if unavailable.
+    """
+    if image_bgr is None:
+        return None
+    # Simple sanity check: require "face-like" size
+    h, w = image_bgr.shape[:2]
+    if h < 64 or w < 64:
+        return None
+    # Placeholder random (stable per run by seeding if you want)
+    return float(np.random.uniform(11.0, 15.0))
+
+
+def estimate_hr_from_video_file(video_path: str) -> Optional[float]:
+    """
+    TODO: Replace with actual rPPG pipeline (e.g., MTTS-CAN/PhysNet).
+    Returns float(bpm) or None.
+    """
+    if not video_path or not os.path.exists(video_path):
+        return None
+
+    cap = cv2.VideoCapture(video_path)
+    frames = 0
+    while True:
+        ret, _ = cap.read()
+        if not ret:
+            break
+        frames += 1
+    cap.release()
+    if frames < 30:  # ~1s at 30fps; require more in real use
+        return None
+
+    return float(np.random.uniform(65, 85))
+
+
+# =========================================================
+# Live Video Processor (WebRTC): collect frames for N seconds
+# =========================================================
+class HRCollectorVideoProcessor:
+    """
+    Minimal frame collector. We buffer incoming frames for the duration the user records,
+    then write them to a temporary .mp4 to feed into estimate_hr_from_video_file().
+    Replace with a live rPPG inference pipeline when ready.
+    """
+    def __init__(self):
+        self.frames = deque(maxlen=30 * 60)  # up to ~60s at 30fps
+        self.recording = False
+
+    def recv(self, frame: av.VideoFrame) -> av.VideoFrame:
+        img = frame.to_ndarray(format="bgr24")
+        if self.recording:
+            self.frames.append(img)
+        # pass-through (optional: draw overlay)
+        return av.VideoFrame.from_ndarray(img, format="bgr24")
+
+    def start(self):
+        self.recording = True
+        self.frames.clear()
+
+    def stop_and_dump_to_file(self) -> Optional[str]:
+        self.recording = False
+        if len(self.frames) < 30:
+            return None
+
+        # Write temp mp4
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+        tmp_path = tmp.name
+        tmp.close()
+
+        # Guess size from first frame
+        h, w = self.frames[0].shape[:2]
+        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+        out = cv2.VideoWriter(tmp_path, fourcc, 30.0, (w, h))
+        for frm in self.frames:
+            out.write(frm)
+        out.release()
+        return tmp_path
+
+
+# =========================================================
+# UI
+# =========================================================
+st.title("🧬 Face-based Health Lab")
+st.caption("Capture via camera or upload files. Results table lists **27 tests** with numbering. "
+           "Swap out placeholders with real models as you go.")
 
-    # ---- Heart Rate ----
-    if uploaded_vid is not None:
-        # save video temporarily
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmpf:
-            tmpf.write(uploaded_vid.read())
-            tmp_path = tmpf.name
+with st.sidebar:
+    st.header("Capture Options")
+    cap_hb_mode = st.radio(
+        "Hemoglobin image source:",
+        ["Camera", "Upload"],
+        index=0,
+        help="Capture a face/eye image for Hb from the camera or upload a file"
+    )
+    cap_hr_mode = st.radio(
+        "Heart-rate video source:",
+        ["Camera (Live)", "Upload"],
+        index=0,
+        help="Record a short (20–30s) face video via camera or upload a video file"
+    )
+    st.info("Tip: Good lighting and steady face improve HR/Hb quality. "
+            "This app is a scaffold; accuracy depends on the models you integrate.")
 
-        hr = predict_heart_rate(tmp_path)
-        results.append({"Test": "Heart Rate", "Result": f"{hr:.1f} bpm"})
+# ==== Columns for inputs ====
+col_img, col_vid = st.columns(2, gap="large")
 
-        # clean up
-        os.remove(tmp_path)
+with col_img:
+    st.subheader("Hemoglobin (Image)")
+    img_np_bgr = None
+
+    if cap_hb_mode == "Camera":
+        cam_img = st.camera_input("Capture Face / Eye Image")
+        if cam_img is not None:
+            file_bytes = np.asarray(bytearray(cam_img.read()), dtype=np.uint8)
+            img_np_bgr = cv2.imdecode(file_bytes, cv2.IMREAD_COLOR)
+            st.image(cv2.cvtColor(img_np_bgr, cv2.COLOR_BGR2RGB), caption="Captured Image", use_column_width=True)
+    else:
+        up_img = st.file_uploader("Upload Image", type=["jpg", "jpeg", "png"])
+        if up_img is not None:
+            file_bytes = np.asarray(bytearray(up_img.read()), dtype=np.uint8)
+            img_np_bgr = cv2.imdecode(file_bytes, cv2.IMREAD_COLOR)
+            st.image(cv2.cvtColor(img_np_bgr, cv2.COLOR_BGR2RGB), caption="Uploaded Image", use_column_width=True)
+
+with col_vid:
+    st.subheader("Heart Rate (Video)")
+    temp_video_path = None
+
+    if cap_hr_mode == "Upload":
+        up_vid = st.file_uploader("Upload Video (mp4/avi/mov)", type=["mp4", "avi", "mov"])
+        if up_vid is not None:
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmpf:
+                tmpf.write(up_vid.read())
+                temp_video_path = tmpf.name
+            st.video(temp_video_path)
     else:
-        results.append({"Test": "Heart Rate", "Result": "No video uploaded"})
-
-    # ---- Future tests placeholders ----
-    results.append({"Test": "SpO₂", "Result": "—"})
-    results.append({"Test": "Respiration Rate", "Result": "—"})
-    # ... add more as needed later
-
-    # -------------------------------------------------
-    # Show results table
-    # -------------------------------------------------
-    df = pd.DataFrame(results)
-    st.subheader("🩺 Test Results")
-    st.table(df)
-
-else:
-    st.info("⬆ Upload image and/or video, then click **Run Analysis**.")
+        st.markdown("**Live Camera (WebRTC)** — Click **Start Recording**, wait 20–30s, then **Stop & Use**.")
+        if "webrtc_ctx" not in st.session_state:
+            st.session_state.webrtc_ctx = None
+            st.session_state.hr_processor = HRCollectorVideoProcessor()
+            st.session_state.is_recording = False
+            st.session_state.record_start = 0.0
+
+        # Start WebRTC streamer
+        ctx = webrtc_streamer(
+            key="hr-webrtc",
+            mode=WebRtcMode.SENDRECV,
+            rtc_configuration=RTC_CONFIGURATION,
+            media_stream_constraints={"video": True, "audio": False},
+            video_processor_factory=lambda: st.session_state.hr_processor,
+        )
+
+        col_btn1, col_btn2, col_btn3 = st.columns([1,1,2])
+        with col_btn1:
+            if st.button("Start Recording", disabled=not ctx.state.playing or st.session_state.is_recording):
+                st.session_state.hr_processor.start()
+                st.session_state.is_recording = True
+                st.session_state.record_start = time.time()
+        with col_btn2:
+            if st.button("Stop & Use", disabled=not st.session_state.is_recording):
+                dump_path = st.session_state.hr_processor.stop_and_dump_to_file()
+                st.session_state.is_recording = False
+                if dump_path and os.path.exists(dump_path):
+                    temp_video_path = dump_path
+                    st.success("Video captured.")
+                    st.video(temp_video_path)
+                else:
+                    st.warning("Captured video too short. Please record ~20–30 seconds.")
+
+        if st.session_state.is_recording:
+            elapsed = time.time() - st.session_state.record_start
+            st.info(f"Recording… {int(elapsed)} s")
+
+# ==== Run Analysis ====
+run = st.button("Run Analysis", type="primary", use_container_width=True)
+
+def init_results_table() -> List[Dict[str, Any]]:
+    rows = []
+    for i, test in enumerate(TESTS, start=1):
+        rows.append({
+            "No.": i,
+            "Test": test,
+            "Result": "—",
+            "Unit": UNITS.get(test, ""),
+            "Status": "Pending"
+        })
+    return rows
+
+if "results_rows" not in st.session_state:
+    st.session_state.results_rows = init_results_table()
+
+if run:
+    # Reset results table each run
+    st.session_state.results_rows = init_results_table()
+
+    # ---- Hemoglobin from image ----
+    hb_value = predict_hemoglobin_from_image(img_np_bgr)
+    for row in st.session_state.results_rows:
+        if row["Test"] == "Hemoglobin (Hb)":
+            if hb_value is None:
+                row["Result"] = "No image"
+                row["Status"] = "No input"
+            else:
+                row["Result"] = f"{hb_value:.2f}"
+                row["Status"] = "OK"
+            break
+
+    # ---- Heart Rate from video ----
+    hr_value = estimate_hr_from_video_file(temp_video_path) if temp_video_path else None
+    for row in st.session_state.results_rows:
+        if row["Test"] == "Heart Rate (HR)":
+            if hr_value is None:
+                row["Result"] = "No video"
+                row["Status"] = "No input"
+            else:
+                row["Result"] = f"{hr_value:.1f}"
+                row["Status"] = "OK"
+            break
+
+# ==== Display Results Table ====
+st.subheader("🩺 Test Results (27)")
+df = pd.DataFrame(st.session_state.results_rows, columns=["No.", "Test", "Result", "Unit", "Status"])
+st.dataframe(df, use_container_width=True, hide_index=True)
+
+st.caption(
+    "Note: Only Hb and HR have placeholder logic today. "
+    "As we build each test, swap in real model inference calls while keeping this UI and table."
+)