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", {}), } @spaces.GPU(duration=45) 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": "حدث خطأ داخلي أثناء التحليل. يرجى المحاولة لاحقًا.", }) @spaces.GPU(duration=30) 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, )