Spaces:
Running on Zero
Running on Zero
| import json | |
| import logging | |
| import os | |
| import re | |
| import unicodedata | |
| import uuid | |
| from typing import Any, Dict | |
| import gradio as gr | |
| import spaces | |
| import database | |
| from report_workflow import ( | |
| list_reports_api, | |
| get_report_api, | |
| review_report_api, | |
| update_report_content_api, | |
| update_report_bilingual_api, | |
| home_queue_api, | |
| department_stats_api, | |
| team_stats_api, | |
| _identity, | |
| ) | |
| from inference import classes_info, health_info, predict_bytes, root_info | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", | |
| ) | |
| logger = logging.getLogger("featurex-zerogpu-server") | |
| SERVER_VERSION = "3.5.1-authenticated-workflow-status" | |
| MAX_UPLOAD = 15 * 1024 * 1024 | |
| def _runtime_info(): | |
| info = {} | |
| try: | |
| import torch | |
| info["torch"] = torch.__version__ | |
| info["cuda_available"] = bool(torch.cuda.is_available()) | |
| info["cuda_device"] = ( | |
| torch.cuda.get_device_name(0) if torch.cuda.is_available() else None | |
| ) | |
| except Exception as exc: | |
| info["torch_error"] = str(exc) | |
| try: | |
| import transformers | |
| info["transformers"] = transformers.__version__ | |
| except Exception as exc: | |
| info["transformers_error"] = str(exc) | |
| info["janus_enabled"] = os.environ.get("JANUS_ENABLED", "true") | |
| info["janus_model_id"] = os.environ.get("JANUS_MODEL_ID", "ZrH42/Janus-Pro-CXR-Final") | |
| info["model_path"] = os.environ.get( | |
| "MODEL_PATH", | |
| os.path.join(os.path.dirname(os.path.abspath(__file__)), "best_efficientnet.pth"), | |
| ) | |
| info["model_file_exists"] = os.path.isfile(info["model_path"]) | |
| return info | |
| def _json(payload: Any) -> str: | |
| return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) | |
| def _read_uploaded_image(image_path: str) -> tuple[bytes, str]: | |
| if not image_path: | |
| raise ValueError("Please upload an X-ray image.") | |
| path = os.path.abspath(os.fspath(image_path)) | |
| if not os.path.isfile(path): | |
| raise ValueError("Uploaded image file was not found on the server.") | |
| size = os.path.getsize(path) | |
| if size <= 0: | |
| raise ValueError("Uploaded image is empty.") | |
| if size > MAX_UPLOAD: | |
| raise ValueError("Image exceeds 15MB.") | |
| with open(path, "rb") as fh: | |
| raw = fh.read() | |
| filename = os.path.basename(path) or "xray_upload" | |
| return raw, filename | |
| def _metadata( | |
| patient_name: str = "", | |
| birth_date: str = "", | |
| gender: str = "", | |
| age: str = "", | |
| user_id: str = "", | |
| username: str = "", | |
| department: str = "", | |
| role: str = "doctor", | |
| supervisor_id: str = "", | |
| ) -> tuple[Dict[str, Any], Dict[str, Any]]: | |
| def s(value: Any, default: str = "") -> str: | |
| if value is None: | |
| value = default | |
| return unicodedata.normalize("NFC", str(value)) | |
| age_value = None | |
| age_text = s(age) | |
| if age_text: | |
| try: | |
| age_value = int(float(age_text)) | |
| except Exception: | |
| age_value = None | |
| patient = { | |
| "patient_name": s(patient_name), | |
| "birth_date": s(birth_date), | |
| "gender": s(gender), | |
| "age": age_value, | |
| } | |
| user = { | |
| "user_id": s(user_id), | |
| "username": s(username), | |
| "department": s(department), | |
| "role": s(role, "doctor") or "doctor", | |
| "supervisor_id": s(supervisor_id), | |
| } | |
| return patient, user | |
| def _build_unified(result: Dict[str, Any]) -> Dict[str, Any]: | |
| reports_ar = result.get("medical_report_ar") or {} | |
| reports_en = result.get("medical_report_en") or {} | |
| selected = reports_en if result.get("language") == "en" else reports_ar | |
| pred = selected.get("prediction") or {} | |
| quality = result.get("quality_metrics") or {} | |
| reliability = selected.get("reliability") or {} | |
| confidence_flags = selected.get("confidence_flags") or [] | |
| confidence = float(pred.get("confidence") or 0.0) | |
| safety_flags = [] | |
| if confidence < 0.60: | |
| safety_flags.append("low_confidence") | |
| if not quality.get("is_acceptable", True): | |
| safety_flags.append("image_quality") | |
| if not reliability.get("is_reliable", True): | |
| safety_flags.append("prediction_not_reliable") | |
| report_integrity = result.get("report_integrity") or {} | |
| bilingual_quality = report_integrity.get("bilingual_quality") or {} | |
| safety_flags.extend(bilingual_quality.get("flags") or []) | |
| if report_integrity.get("manual_review_required"): | |
| safety_flags.append("manual_review_required_by_report_integrity") | |
| return { | |
| "success": True, | |
| "language": result.get("language"), | |
| "schema_version": "featurex-xray-v5-hybrid-bilingual-supabase", | |
| "status": "success", | |
| "analysis_id": result.get("analysis_id"), | |
| "model": { | |
| "classifier": "EfficientNet-B4", | |
| "report_model": (result.get("model_info") or {}).get( | |
| "report_model", "ZrH42/Janus-Pro-CXR-Final" | |
| ), | |
| "architecture": (result.get("model_info") or {}).get("architecture"), | |
| "version": (result.get("model_info") or {}).get("version"), | |
| }, | |
| "classification": { | |
| "predicted_class": pred.get("primary_diagnosis"), | |
| "confidence": confidence, | |
| "probabilities": { | |
| x.get("disease"): float(x.get("probability", 0.0)) | |
| for x in result.get("predictions", []) | |
| if x.get("disease") | |
| }, | |
| "top_predictions": result.get("predictions", []), | |
| }, | |
| "predictions": result.get("predictions", []), | |
| "medical_report": selected, | |
| "medical_report_ar": reports_ar, | |
| "medical_report_en": reports_en, | |
| "quality_metrics": quality, | |
| "heatmap_stats": result.get("heatmap_stats", {}), | |
| "model_info": result.get("model_info", {}), | |
| "visualizations": { | |
| **((result.get("database") or {}).get("visualization_urls") or {}), | |
| }, | |
| "report": { | |
| "language": result.get("language"), | |
| "source": selected.get("report_source", "EFFICIENTNET_TEMPLATE_FALLBACK"), | |
| "model": selected.get("report_model", "ZrH42/Janus-Pro-CXR-Final"), | |
| "janus_raw_text": selected.get("janus_raw_text", ""), | |
| "findings": [selected.get("findings")] if selected.get("findings") else [], | |
| "impression": [selected.get("impression")] if selected.get("impression") else [], | |
| "recommendations": selected.get("recommendations", []), | |
| "raw_text": "\n".join( | |
| str(x) | |
| for x in [ | |
| selected.get("title"), | |
| selected.get("findings"), | |
| selected.get("impression"), | |
| ] | |
| if x | |
| ), | |
| "structured": selected, | |
| "report_ar": reports_ar, | |
| "report_en": reports_en, | |
| }, | |
| "quality": quality, | |
| "gradcam": { | |
| "enabled": True, | |
| "heatmap_stats": result.get("heatmap_stats", {}), | |
| "visualizations": { | |
| **((result.get("database") or {}).get("visualization_urls") or {}), | |
| }, | |
| "visualization_keys": list((result.get("visualizations") or {}).keys()), | |
| }, | |
| "discrepancies": {"flags": [], "count": 0}, | |
| "safety": { | |
| "wording_flags": confidence_flags, | |
| "manual_review_required": bool(safety_flags) | |
| or bool((result.get("report_integrity") or {}).get("manual_review_required")), | |
| "reasons": safety_flags, | |
| }, | |
| "database": result.get("database", {}), | |
| "storage": { | |
| "original_image_url": (result.get("database") or {}).get("original_image_url"), | |
| "visualization_urls": (result.get("database") or {}).get("visualization_urls", {}), | |
| }, | |
| "janus": result.get("janus", {}), | |
| "report_integrity": result.get("report_integrity", {}), | |
| } | |
| def analyze( | |
| image_path: str, | |
| language: str = "ar", | |
| patient_name: str = "", | |
| birth_date: str = "", | |
| gender: str = "", | |
| age: str = "", | |
| user_id: str = "", | |
| username: str = "", | |
| department: str = "", | |
| role: str = "doctor", | |
| supervisor_id: str = "", | |
| firebase_token: str = "", | |
| ) -> str: | |
| request_id = str(uuid.uuid4()) | |
| try: | |
| identity = _identity(firebase_token, user_id) | |
| raw, filename = _read_uploaded_image(image_path) | |
| language = (language or "ar").strip().lower() | |
| if language not in {"ar", "en"}: | |
| language = "ar" | |
| patient, _client_user = _metadata( | |
| patient_name, | |
| birth_date, | |
| gender, | |
| age, | |
| user_id, | |
| username, | |
| department, | |
| role, | |
| supervisor_id, | |
| ) | |
| user = { | |
| "user_id": identity["uid"], | |
| "username": str(username or "")[:160], | |
| "department": identity.get("department", ""), | |
| "role": identity["role"], | |
| "supervisor_id": identity.get("supervisor_id", ""), | |
| } | |
| logger.info( | |
| "ANALYZE START request_id=%s bytes=%d user_id=%s patient_name_has_arabic=%s", | |
| request_id, | |
| len(raw), | |
| user.get("user_id"), | |
| bool(re.search(r"[\u0600-\u06FF]", patient.get("patient_name", ""))), | |
| ) | |
| result = predict_bytes(raw, filename, language) | |
| if not isinstance(result, dict) or result.get("status") != "success": | |
| raise RuntimeError("Inference did not return success.") | |
| result["analysis_id"] = str(uuid.uuid4()) | |
| result["request_id"] = request_id | |
| result["server_version"] = SERVER_VERSION | |
| db_info = {"saved": False, "analysis_id": result["analysis_id"]} | |
| try: | |
| saved = database.save_analysis( | |
| result=result, | |
| original_bytes=raw, | |
| original_filename=filename, | |
| patient=patient, | |
| user=user, | |
| ) | |
| db_info.update({ | |
| "saved": bool(saved.get("saved", False)), | |
| "analysis_id": saved.get("analysis_id", result["analysis_id"]), | |
| "original_image_url": saved.get("original_image_url"), | |
| "visualization_urls": saved.get("visualization_urls", {}), | |
| "storage_saved": bool( | |
| saved.get("storage_saved", saved.get("original_image_url")) | |
| ), | |
| }) | |
| if saved.get("reason"): | |
| db_info["reason"] = saved["reason"] | |
| if saved.get("error"): | |
| db_info["error"] = saved["error"] | |
| db_info["error_type"] = saved.get("error_type") | |
| except Exception as exc: | |
| logger.exception("SUPABASE STORAGE/DATABASE SAVE FAILED request_id=%s", request_id) | |
| db_info.update({ | |
| "saved": False, | |
| "storage_saved": False, | |
| "error": str(exc), | |
| "error_type": type(exc).__name__, | |
| }) | |
| result["database"] = db_info | |
| if identity["role"] == "patient": | |
| return _json({ | |
| "success": True, | |
| "status": "submitted", | |
| "analysis_id": result["analysis_id"], | |
| "workflow_status": "pending_clinician_review", | |
| "visible_to_patient": False, | |
| "message": "تم استلام الطلب. ستظهر النتيجة الطبية بعد مراجعتها واعتمادها.", | |
| "database_saved": bool(db_info.get("saved")), | |
| "request_id": request_id, | |
| }) | |
| result["visualizations"] = {} | |
| return _json(_build_unified(result)) | |
| except PermissionError as exc: | |
| logger.warning("ANALYZE AUTH REJECTED request_id=%s reason=%s", request_id, str(exc)) | |
| return _json({ | |
| "success": False, | |
| "status": "unauthorized", | |
| "request_id": request_id, | |
| "error": "Authentication required", | |
| "message": "يرجى تسجيل الدخول مجددًا ثم المحاولة.", | |
| }) | |
| except ValueError as exc: | |
| logger.warning("ANALYZE REJECTED request_id=%s reason=%s", request_id, str(exc)) | |
| return _json({ | |
| "success": False, | |
| "status": "rejected", | |
| "request_id": request_id, | |
| "error": "Image validation failed", | |
| "message": str(exc), | |
| }) | |
| except Exception as exc: | |
| logger.exception("ANALYZE ERROR request_id=%s", request_id) | |
| return _json({ | |
| "success": False, | |
| "status": "error", | |
| "request_id": request_id, | |
| "error": "Internal server error", | |
| "message": "حدث خطأ داخلي أثناء التحليل. يرجى المحاولة لاحقًا.", | |
| }) | |
| def gpu_health() -> str: | |
| info = health_info() | |
| info.update({ | |
| "server_version": SERVER_VERSION, | |
| "transport": "Gradio Server + ZeroGPU", | |
| "database": database.config_info(), | |
| "runtime": _runtime_info(), | |
| "api": {"analyze": True, "health": True, "classes": True}, | |
| }) | |
| return _json(info) | |
| def health() -> str: | |
| info = root_info() | |
| info.update({ | |
| "server_version": SERVER_VERSION, | |
| "transport": "Gradio Server + ZeroGPU", | |
| "zero_gpu": True, | |
| "database": database.config_info(), | |
| "runtime": _runtime_info(), | |
| }) | |
| return _json(info) | |
| def classes() -> str: | |
| return _json(classes_info()) | |
| def db_health() -> str: | |
| try: | |
| result = database.health_check() | |
| except Exception as exc: | |
| logger.exception("DB HEALTH ENDPOINT FAILED") | |
| result = { | |
| "ok": False, | |
| "error": str(exc), | |
| "error_type": type(exc).__name__, | |
| "database": database.config_info(), | |
| } | |
| return _json(result) | |
| def _health_endpoint() -> str: | |
| info = root_info() | |
| info.update({ | |
| "server_version": SERVER_VERSION, | |
| "transport": "Gradio Blocks + ZeroGPU", | |
| "zero_gpu": True, | |
| "database": database.config_info(), | |
| "runtime": _runtime_info(), | |
| "api": { | |
| "analyze": True, | |
| "health": True, | |
| "gpu_health": True, | |
| "classes": True, | |
| "list_reports": True, | |
| "get_report": True, | |
| "review_report": True, | |
| "update_report_content": True, | |
| }, | |
| }) | |
| return _json(info) | |
| def _classes_endpoint() -> str: | |
| return _json(classes_info()) | |
| with gr.Blocks( | |
| title="FeatureX Clinical AI — ZeroGPU", | |
| analytics_enabled=False, | |
| ) as demo: | |
| gr.Markdown( | |
| "# FeatureX Clinical AI\n" | |
| "EfficientNet-B4 + Grad-CAM + Janus-Pro-CXR + Hybrid Bilingual Report Checks" | |
| ) | |
| with gr.Row(): | |
| image = gr.Image( | |
| label="Upload Chest X-Ray", | |
| type="filepath", | |
| sources=["upload", "webcam"], | |
| format="png", | |
| height=360, | |
| ) | |
| language = gr.Radio(["ar", "en"], value="ar", label="Report Language") | |
| with gr.Row(): | |
| patient_name = gr.Textbox(label="patient_name") | |
| birth_date = gr.Textbox(label="birth_date") | |
| gender = gr.Textbox(label="gender") | |
| age = gr.Textbox(label="age") | |
| with gr.Row(): | |
| user_id = gr.Textbox(label="user_id") | |
| username = gr.Textbox(label="username") | |
| department = gr.Textbox(label="department") | |
| role = gr.Textbox(label="role", value="doctor") | |
| supervisor_id = gr.Textbox(label="supervisor_id") | |
| firebase_token = gr.Textbox(label="firebase_token", visible=False) | |
| analyze_button = gr.Button("Analyze X-Ray", variant="primary") | |
| analyze_output = gr.Textbox(label="JSON response", lines=20) | |
| analyze_button.click( | |
| analyze, | |
| inputs=[ | |
| image, | |
| language, | |
| patient_name, | |
| birth_date, | |
| gender, | |
| age, | |
| user_id, | |
| username, | |
| department, | |
| role, | |
| supervisor_id, | |
| firebase_token, | |
| ], | |
| outputs=analyze_output, | |
| api_name="analyze", | |
| concurrency_limit=1, | |
| ) | |
| health_button = gr.Button("Health") | |
| health_output = gr.Textbox(label="Health JSON") | |
| health_button.click( | |
| _health_endpoint, inputs=[], outputs=health_output, api_name="health" | |
| ) | |
| classes_button = gr.Button("Classes") | |
| classes_output = gr.Textbox(label="Classes JSON") | |
| classes_button.click( | |
| _classes_endpoint, inputs=[], outputs=classes_output, api_name="classes" | |
| ) | |
| gpu_health_button = gr.Button("GPU Health") | |
| gpu_health_output = gr.Textbox(label="GPU Health JSON") | |
| gpu_health_button.click( | |
| gpu_health, inputs=[], outputs=gpu_health_output, api_name="gpu_health" | |
| ) | |
| db_health_button = gr.Button("Database / Storage Health") | |
| db_health_output = gr.Textbox(label="Database / Storage Health JSON", lines=12) | |
| db_health_button.click( | |
| db_health, inputs=[], outputs=db_health_output, api_name="db_health" | |
| ) | |
| # ---------- Home screen endpoints ---------- | |
| home_token = gr.Textbox(visible=False) | |
| home_queue_output = gr.Textbox(visible=False) | |
| home_queue_btn = gr.Button("Home Queue API", visible=False) | |
| home_queue_btn.click( | |
| home_queue_api, | |
| inputs=[home_token], | |
| outputs=home_queue_output, | |
| api_name="home_queue", | |
| ) | |
| stats_token = gr.Textbox(visible=False) | |
| stats_output = gr.Textbox(visible=False) | |
| stats_btn = gr.Button("Department Stats API", visible=False) | |
| stats_btn.click( | |
| department_stats_api, | |
| inputs=[stats_token], | |
| outputs=stats_output, | |
| api_name="department_stats", | |
| ) | |
| # ---------- Protected report workflow APIs ---------- | |
| report_user_id = gr.Textbox(visible=False) | |
| report_token = gr.Textbox(visible=False) | |
| report_id = gr.Textbox(visible=False) | |
| review_action = gr.Textbox(visible=False) | |
| review_notes = gr.Textbox(visible=False) | |
| list_reports_button = gr.Button("List Reports API", visible=False) | |
| list_reports_output = gr.Textbox(visible=False) | |
| list_reports_button.click( | |
| list_reports_api, | |
| inputs=[report_user_id, report_token], | |
| outputs=list_reports_output, | |
| api_name="list_reports", | |
| ) | |
| get_report_button = gr.Button("Get Report API", visible=False) | |
| get_report_output = gr.Textbox(visible=False) | |
| get_report_button.click( | |
| get_report_api, | |
| inputs=[report_id, report_token], | |
| outputs=get_report_output, | |
| api_name="get_report", | |
| ) | |
| review_report_button = gr.Button("Review Report API", visible=False) | |
| review_report_output = gr.Textbox(visible=False) | |
| review_report_button.click( | |
| review_report_api, | |
| inputs=[report_id, review_action, review_notes, report_token], | |
| outputs=review_report_output, | |
| api_name="review_report", | |
| ) | |
| # ---------- Report content editing (before approval) ---------- | |
| edit_id = gr.Textbox(visible=False) | |
| edit_findings = gr.Textbox(visible=False) | |
| edit_impression = gr.Textbox(visible=False) | |
| edit_recs = gr.Textbox(visible=False) | |
| edit_token = gr.Textbox(visible=False) | |
| edit_output = gr.Textbox(visible=False) | |
| edit_button = gr.Button("Update Report Content API", visible=False) | |
| edit_button.click( | |
| update_report_content_api, | |
| inputs=[edit_id, edit_findings, edit_impression, edit_recs, edit_token], | |
| outputs=edit_output, | |
| api_name="update_report_content", | |
| ) | |
| edit_ar = gr.Textbox(visible=False) | |
| edit_en = gr.Textbox(visible=False) | |
| edit_notes = gr.Textbox(visible=False) | |
| edit_bilingual_token = gr.Textbox(visible=False) | |
| edit_bilingual_output = gr.Textbox(visible=False) | |
| edit_bilingual_btn = gr.Button("Update Report Bilingual API", visible=False) | |
| edit_bilingual_btn.click( | |
| update_report_bilingual_api, | |
| inputs=[report_id, edit_ar, edit_en, edit_notes, edit_bilingual_token], | |
| outputs=edit_bilingual_output, | |
| api_name="update_report_bilingual", | |
| ) | |
| team_stats_token = gr.Textbox(visible=False) | |
| team_stats_output = gr.Textbox(visible=False) | |
| team_stats_btn = gr.Button("Team Stats API", visible=False) | |
| team_stats_btn.click( | |
| team_stats_api, | |
| inputs=[team_stats_token], | |
| outputs=team_stats_output, | |
| api_name="team_stats", | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch( | |
| show_error=True, | |
| max_file_size="15mb", | |
| ssr_mode=False, | |
| ) |