Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """ | |
| indidermax — Production FastAPI + Gradio App for HuggingFace Spaces | |
| ==================================================================== | |
| - FastAPI backend with REST endpoints for Android integration | |
| - Gradio chatbot UI for web demo | |
| - Real Neo4j graph queries + NVIDIA NIM vision API | |
| - Full CMADD 5-agent debate pipeline with transparent logging | |
| HF Secrets expected: | |
| NEO4J_URI, NEO4J_USERNAME, NEO4J_PASSWORD, NEO4J_DATABASE | |
| NVIDIA_API_KEY | |
| """ | |
| from __future__ import annotations | |
| import os, sys, json, time, base64, re, threading | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| from datetime import datetime | |
| import httpx | |
| import gradio as gr | |
| from fastapi import FastAPI, File, UploadFile, Form, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import JSONResponse | |
| from pydantic import BaseModel | |
| from PIL import Image | |
| import io | |
| import tempfile | |
| from fpdf import FPDF | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # CONFIG | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| NEO4J_URI = os.getenv("NEO4J_URI", "") | |
| NEO4J_USER = os.getenv("NEO4J_USERNAME", "") | |
| NEO4J_PASS = os.getenv("NEO4J_PASSWORD", "") | |
| NEO4J_DB = os.getenv("NEO4J_DATABASE", "neo4j") | |
| NVIDIA_KEY = os.getenv("NVIDIA_API_KEY", "") | |
| NVIDIA_URL = "https://integrate.api.nvidia.com/v1/chat/completions" | |
| NVIDIA_MODEL = "google/gemma-3n-e4b-it" | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # NEO4J CLIENT | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| _neo4j_driver = None | |
| def get_neo4j(): | |
| global _neo4j_driver | |
| if _neo4j_driver is None and NEO4J_URI: | |
| try: | |
| from neo4j import GraphDatabase | |
| _neo4j_driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASS)) | |
| except Exception: | |
| pass | |
| return _neo4j_driver | |
| def neo4j_fetch_candidates(descriptors: list[str], body_part: str, symptoms: list[str], limit: int = 10) -> list[dict]: | |
| """Query Neo4j with composite scoring across morphology, body, symptom, effect, visual layers.""" | |
| driver = get_neo4j() | |
| if not driver: | |
| return _fallback_candidates(descriptors, body_part) | |
| desc_terms = [d.lower().strip() for d in descriptors if d.strip()] | |
| bp = body_part.lower().strip() if body_part else "" | |
| query = """ | |
| MATCH (d:Disease) | |
| OPTIONAL MATCH (d)-[:PRESENTS_WITH]->(m:Morphology) | |
| OPTIONAL MATCH (d)-[:COMMON_AT]->(b:BodyRegion) | |
| OPTIONAL MATCH (d)-[:HAS_SYMPTOM]->(s:Symptom) | |
| OPTIONAL MATCH (d)-[:MAY_CAUSE]->(e:Effect) | |
| OPTIONAL MATCH (d)-[:HAS_VISUAL_ATOM]->(va:VisualAtom) | |
| OPTIONAL MATCH (mc:MainClass)-[:HAS_SUB_CLASS]->(sc:SubClass)-[:HAS_DISEASE]->(d) | |
| WITH d, mc, sc, | |
| collect(DISTINCT toLower(m.name)) AS morphs, | |
| collect(DISTINCT toLower(b.name)) AS bodies, | |
| collect(DISTINCT toLower(s.name)) AS syms, | |
| collect(DISTINCT toLower(e.name)) AS effs, | |
| collect(DISTINCT toLower(va.name)) AS atoms | |
| WITH d, mc, sc, morphs, bodies, syms, effs, atoms, | |
| size([t IN $desc_terms WHERE any(m IN morphs WHERE m CONTAINS t OR t CONTAINS m)]) AS morph_score, | |
| size([t IN $desc_terms WHERE any(s IN syms WHERE s CONTAINS t OR t CONTAINS s)]) AS sym_score, | |
| size([t IN $desc_terms WHERE any(e IN effs WHERE e CONTAINS t OR t CONTAINS e)]) AS eff_score, | |
| size([t IN $desc_terms WHERE any(v IN atoms WHERE v CONTAINS t OR t CONTAINS v)]) AS vis_score, | |
| CASE WHEN $bp <> '' AND any(b IN bodies WHERE b CONTAINS $bp OR $bp CONTAINS b) THEN 1 ELSE 0 END AS body_score | |
| WITH d.name AS disease, mc.name AS main_class, sc.name AS sub_class, | |
| morph_score, sym_score, eff_score, vis_score, body_score, | |
| (morph_score * 5 + body_score * 3 + sym_score * 2 + eff_score * 0.5 + vis_score * 2) AS score | |
| WHERE score > 0 | |
| RETURN disease, main_class, sub_class, score, morph_score, sym_score, body_score | |
| ORDER BY score DESC | |
| LIMIT $limit | |
| """ | |
| try: | |
| with driver.session(database=NEO4J_DB) as sess: | |
| result = sess.run(query, {"desc_terms": desc_terms, "bp": bp, "limit": limit}) | |
| candidates = [] | |
| for r in result: | |
| candidates.append({ | |
| "disease": r["disease"], | |
| "main_class": r.get("main_class", ""), | |
| "sub_class": r.get("sub_class", ""), | |
| "score": r["score"], | |
| "morph_match": r.get("morph_score", 0), | |
| "sym_match": r.get("sym_score", 0), | |
| "body_match": r.get("body_score", 0), | |
| }) | |
| return candidates | |
| except Exception as e: | |
| print(f"[Neo4j] Error: {e}") | |
| return _fallback_candidates(descriptors, body_part) | |
| # Fallback: use bundled CLINICAL_KB + neo4j_cache.json | |
| _FALLBACK_KB = None | |
| def _load_fallback(): | |
| global _FALLBACK_KB | |
| if _FALLBACK_KB is not None: | |
| return _FALLBACK_KB | |
| cp = Path(__file__).parent / "neo4j_cache.json" | |
| if cp.exists(): | |
| with open(cp) as f: | |
| _FALLBACK_KB = json.load(f) | |
| return _FALLBACK_KB | |
| def _fallback_candidates(descriptors, body_part): | |
| from pipeline import CLINICAL_KB as KB | |
| cache = _load_fallback() | |
| desc_lower = [d.lower().strip() for d in descriptors if d.strip()] | |
| bp = body_part.lower().strip() if body_part else "" | |
| results = [] | |
| disease_source = KB | |
| if cache and "diseases" in cache: | |
| disease_source = {d["disease"].lower(): d for d in cache["diseases"]} | |
| for key, val in disease_source.items(): | |
| if isinstance(val, dict) and "disease" in val: | |
| n = val | |
| morphs = [m.lower() for m in n.get("morphologies", [])] | |
| brs = [b.lower() for b in n.get("body_regions", [])] | |
| morph_match = sum(1 for d in desc_lower for m in morphs if d in m or m in d) | |
| body_match = 1.0 if bp and any(bp in b or b in bp for b in brs) else 0.0 | |
| score = morph_match * 5 + body_match * 3 | |
| elif isinstance(val, dict) and "descriptors" in val: | |
| kb_desc = set(d.lower() for d in val.get("descriptors", [])) | |
| kb_body = " ".join(val.get("body_locations", [])).lower() | |
| morph_match = sum(1 for d in desc_lower if d in kb_desc) | |
| body_match = 1.0 if bp and bp in kb_body else 0.0 | |
| score = morph_match * 3 + body_match * 2 | |
| else: | |
| score = 0 | |
| name = val.get("disease", key) if isinstance(val, dict) else str(key) | |
| if score > 0: | |
| results.append({"disease": name, "score": score}) | |
| results.sort(key=lambda x: x["score"], reverse=True) | |
| return results[:10] | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # NVIDIA NIM VISION | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| VISION_PROMPT = """You are a dermatologist. Analyze this skin image and return ONLY a JSON object: | |
| { | |
| "primary_lesion": "specific lesion type", | |
| "color": "primary color", | |
| "body_location": "specific body region", | |
| "morphology_detail": "5-8 word morphological description", | |
| "key_descriptors": ["5-8", "specific", "clinical", "descriptors"], | |
| "pattern": "distribution pattern" | |
| }""" | |
| _nvidia_lock = threading.Lock() | |
| _nvidia_last = 0.0 | |
| _NVIDIA_RATE = 60.0 / 40 # 40 req/min | |
| def analyze_image_nvidia(image_data: bytes) -> dict: | |
| """Call NVIDIA NIM vision API to analyze skin image.""" | |
| global _nvidia_last | |
| if not NVIDIA_KEY: | |
| return {"error": "NVIDIA_API_KEY not set"} | |
| with _nvidia_lock: | |
| now = time.monotonic() | |
| wait = _nvidia_last + _NVIDIA_RATE - now | |
| if wait > 0: | |
| time.sleep(wait) | |
| _nvidia_last = time.monotonic() | |
| img_b64 = base64.b64encode(image_data).decode() | |
| try: | |
| resp = httpx.post( | |
| NVIDIA_URL, | |
| json={ | |
| "model": NVIDIA_MODEL, | |
| "messages": [{"role": "user", "content": [ | |
| {"type": "text", "text": VISION_PROMPT}, | |
| {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}}, | |
| ]}], | |
| "max_tokens": 300, "temperature": 0.1, | |
| }, | |
| headers={"Authorization": f"Bearer {NVIDIA_KEY}", "Content-Type": "application/json"}, | |
| timeout=45, | |
| ) | |
| resp.raise_for_status() | |
| text = resp.json()["choices"][0]["message"]["content"].strip() | |
| if text.startswith("```"): | |
| text = "\n".join(text.split("\n")[1:]).rstrip("`").strip() | |
| return json.loads(text) | |
| except Exception as e: | |
| return {"error": str(e)} | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # PIPELINE (imports from self-contained pipeline.py) | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| from pipeline import ( | |
| _extract_features, _tokenize, SYNONYMS, | |
| _agent_symptom_analyst, _agent_temporal_matcher, | |
| _agent_differential_debater, _agent_evidence_synthesizer, | |
| _agent_visual_concept, dtw_distance, PipelineLogger, | |
| _format_answer, _age_factor, DTW_PATTERNS, CLINICAL_KB, | |
| ) | |
| def run_full_pipeline( | |
| user_message: str = "", | |
| image_data: bytes | None = None, | |
| patient_age: int | None = None, | |
| session_id: str = "api", | |
| use_neo4j: bool = True, | |
| use_vision: bool = True, | |
| ) -> dict: | |
| logger = PipelineLogger() | |
| # Stage 1: Input | |
| logger.log("1_input", "parser", {"status": f"Msg: '{user_message[:60]}'", "image": image_data is not None, "age": patient_age}) | |
| if patient_age is None and user_message: | |
| for pat in [r"(?:i\s*(?:am|'?m))\s*(\d{1,3})\s*(?:years?\s*old|yo)", r"age\s*(?:is\s*)?(\d{1,3})", r"(\d{1,3})\s*(?:years?\s*old|yo)"]: | |
| m = re.search(pat, user_message, re.IGNORECASE) | |
| if m: patient_age = int(m.group(1)); break | |
| # Stage 2: Features | |
| descriptors, body_part, symptoms, effects = _extract_features(user_message) | |
| logger.log("2_features", "extractor", {"status": "Done", "features": descriptors, "body_part": body_part, "symptoms": symptoms}) | |
| # Stage 2b: Image analysis via NVIDIA | |
| visual_features = {} | |
| if image_data and use_vision and NVIDIA_KEY: | |
| vf = analyze_image_nvidia(image_data) | |
| if "error" not in vf: | |
| visual_features = vf | |
| if vf.get("key_descriptors"): | |
| for d in vf["key_descriptors"]: | |
| if d.lower() not in descriptors: | |
| descriptors.append(d.lower()) | |
| if vf.get("body_location") and not body_part: | |
| body_part = vf["body_location"] | |
| logger.log("2b_vision", "nvidia_nim", {"status": f"Analyzed: {vf.get('primary_lesion','?')} @ {vf.get('body_location','?')}", "visual": vf}) | |
| else: | |
| logger.log("2b_vision", "nvidia_nim", {"status": f"Failed: {vf['error'][:60]}"}) | |
| # Stage 3: Candidates | |
| if use_neo4j and NEO4J_URI: | |
| candidates = neo4j_fetch_candidates(descriptors, body_part, symptoms) | |
| logger.log("3_retrieval", "neo4j_graph", {"status": f"Neo4j: {len(candidates)} candidates"}) | |
| else: | |
| candidates = _fallback_candidates(descriptors, body_part) | |
| logger.log("3_retrieval", "local_cache", {"status": f"Local: {len(candidates)} candidates"}) | |
| if not candidates: | |
| return _empty_result(logger) | |
| logger.log("3_retrieval", "results", {"candidates": [{"disease": c["disease"], "score": c["score"]} for c in candidates[:10]]}) | |
| top5 = candidates[:5] | |
| # Stage 4a-e: CMADD Agents | |
| va = _agent_visual_concept(descriptors, top5) | |
| logger.log("4a_visual", "visual_agent", {"status": f"Top: {va['top_disease']}", "matched": va.get("descriptors_matched", [])}) | |
| sym = _agent_symptom_analyst(symptoms, top5) | |
| logger.log("4b_symptom", "symptom_analyst", {"status": "Done", "symptom_overlap": {c["disease"]: sym.get(c["disease"], 0) for c in top5}}) | |
| dtw_s = _agent_temporal_matcher(symptoms, top5) | |
| logger.log("4c_temporal", "temporal_matcher", {"status": "Done", "dtw_match": {c["disease"]: round(dtw_s.get(c["disease"], 1.0), 3) for c in top5}}) | |
| debated = _agent_differential_debater(top5, sym, dtw_s, descriptors, patient_age, visual_features) | |
| logger.log("4d_debater", "differential_debater", {"status": "Fused", "candidates": [{"disease": c["disease"], "score": c["combined_score"]} for c in debated[:5]]}) | |
| evidence = _agent_evidence_synthesizer(debated[0]["disease"]) | |
| logger.log("4e_evidence", "evidence_synth", {"status": f"{len(evidence)} sources"}) | |
| # Stage 5: Output | |
| top = debated[0] | |
| answer = _format_answer(top, debated, evidence, descriptors, body_part, patient_age) | |
| logger.log("5_output", "finalizer", {"status": "Complete"}) | |
| return { | |
| "answer": answer, | |
| "top_disease": top["disease"], | |
| "top_score": top.get("combined_score", top.get("score", 0)), | |
| "candidates": [{"disease": c["disease"], "score": c.get("combined_score", c.get("score", 0))} for c in debated[:5]], | |
| "differentials": [{"disease": c["disease"], "score": c.get("combined_score", c.get("score", 0))} for c in debated[1:4]], | |
| "evidence": [{"title": e.get("title", ""), "source": e.get("source", "")} for e in evidence[:5]], | |
| "logs": logger.logs, | |
| "log_text": logger.get_log_text(), | |
| "pipeline": {"stages": 6, "agents": 5, "neo4j": bool(NEO4J_URI), "vision": bool(NVIDIA_KEY and image_data)}, | |
| } | |
| def _empty_result(logger): | |
| return {"answer": "⚠️ Insufficient evidence for diagnosis. Please provide more details.", "top_disease": "N/A", "top_score": 0, "candidates": [], "differentials": [], "evidence": [], "logs": logger.logs, "log_text": logger.get_log_text(), "pipeline": {"stages": 1, "agents": 0}} | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # FASTAPI APP (for API routes + local dev) | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| _fastapi = FastAPI(title="indidermax API", version="2.0.0") | |
| _fastapi.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) | |
| from fastapi import Request | |
| from fastapi.responses import JSONResponse as _JSONResponse | |
| async def _global_exception_handler(request: Request, exc: Exception): | |
| return _JSONResponse( | |
| status_code=500, | |
| content={"success": False, "error": str(exc), "endpoint": str(request.url)}, | |
| ) | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # API Endpoints | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| async def health(): | |
| neo4j_ok = False | |
| try: | |
| driver = get_neo4j() | |
| if driver: | |
| with driver.session(database=NEO4J_DB) as s: | |
| s.run("RETURN 1") | |
| neo4j_ok = True | |
| except Exception: | |
| pass | |
| fallback = _load_fallback() | |
| return { | |
| "status": "healthy", | |
| "mode": "neo4j_live" if neo4j_ok else ("cache_fallback" if fallback else "kb_only"), | |
| "neo4j": neo4j_ok, | |
| "nvidia": bool(NVIDIA_KEY), | |
| "timestamp": datetime.utcnow().isoformat(), | |
| } | |
| class DiagnoseRequest(BaseModel): | |
| message: str = "" | |
| patient_age: Optional[int] = None | |
| image_base64: Optional[str] = None | |
| session_id: str = "api" | |
| class DiagnoseResponse(BaseModel): | |
| success: bool | |
| top_disease: str | |
| top_score: float | |
| candidates: list[dict] | |
| differentials: list[dict] | |
| evidence: list[dict] | |
| answer: str | |
| log_text: str | |
| pipeline: dict | |
| async def api_diagnose(req: DiagnoseRequest): | |
| img_bytes = None | |
| if req.image_base64: | |
| try: | |
| img_bytes = base64.b64decode(req.image_base64) | |
| except Exception: | |
| raise HTTPException(400, "Invalid base64 image") | |
| result = run_full_pipeline( | |
| user_message=req.message, image_data=img_bytes, | |
| patient_age=req.patient_age, session_id=req.session_id, | |
| ) | |
| return DiagnoseResponse(success=True, **{k: v for k, v in result.items() if k in DiagnoseResponse.model_fields}) | |
| async def api_diagnose_upload( | |
| message: str = Form(""), | |
| patient_age: Optional[int] = Form(None), | |
| image: UploadFile = File(None), | |
| ): | |
| img_bytes = None | |
| if image: | |
| img_bytes = await image.read() | |
| result = run_full_pipeline(user_message=message, image_data=img_bytes, patient_age=patient_age) | |
| return JSONResponse({"success": True, **result}) | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # GRADIO UI | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # Doctor Conversation Protocol | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| DOCTOR_FLOW = [ | |
| {"key": "duration", "question": "How long have you had this condition? (e.g., a few days, weeks, months, or years)"}, | |
| {"key": "sensation", "question": "Is the affected area itchy, painful, or burning? Please describe what you feel."}, | |
| {"key": "progression", "question": "Has the rash or lesion been spreading, changing shape, or changing color over time?"}, | |
| {"key": "systemic", "question": "Do you have any other symptoms unrelated to the skin, like fever, fatigue, or joint pain?"}, | |
| {"key": "prior_treatment", "question": "Have you tried any creams, medications, or home remedies for this? If yes, what did you use?"}, | |
| {"key": "recurrence", "question": "Is this the first time you have this condition, or has it occurred before? Anyone in your family with similar skin issues?"}, | |
| {"key": "triggers", "question": "Have you recently used any new products (soaps, cosmetics, detergents), foods, or been outdoors a lot?"}, | |
| ] | |
| MAX_FOLLOWUPS = 4 | |
| def _init_chat_state() -> dict: | |
| return { | |
| "turn": 0, | |
| "phase": "greeting", | |
| "image_path": None, | |
| "accumulated_descriptors": [], | |
| "accumulated_symptoms": [], | |
| "accumulated_effects": [], | |
| "body_part": "", | |
| "age": None, | |
| "followup_index": 0, | |
| "patient_responses": {}, | |
| "asked_questions": [], | |
| "_last_logs": "", | |
| } | |
| def _get_next_question(state: dict) -> str | None: | |
| idx = state.get("followup_index", 0) | |
| if idx < len(DOCTOR_FLOW) and idx < MAX_FOLLOWUPS: | |
| return DOCTOR_FLOW[idx]["question"] | |
| return None | |
| def _process_gathering_response(state: dict, message: str) -> dict: | |
| idx = state["followup_index"] | |
| if idx < len(DOCTOR_FLOW): | |
| key = DOCTOR_FLOW[idx]["key"] | |
| state["patient_responses"][key] = message | |
| state["asked_questions"].append(DOCTOR_FLOW[idx]["question"]) | |
| desc, bp, sym, eff = _extract_features(message) | |
| state["accumulated_descriptors"].extend(desc) | |
| state["accumulated_symptoms"].extend(sym) | |
| state["accumulated_effects"].extend(eff) | |
| if bp and not state.get("body_part"): | |
| state["body_part"] = bp | |
| state["followup_index"] += 1 | |
| return state | |
| def _build_combined_message(state: dict, latest_message: str = "") -> str: | |
| all_desc = list(dict.fromkeys(state.get("accumulated_descriptors", []))) | |
| all_sym = list(dict.fromkeys(state.get("accumulated_symptoms", []))) | |
| all_eff = list(dict.fromkeys(state.get("accumulated_effects", []))) | |
| body = state.get("body_part", "") | |
| parts = [] | |
| if latest_message: | |
| parts.append(latest_message) | |
| if all_desc: | |
| parts.append("Descriptors: " + ", ".join(all_desc)) | |
| if body: | |
| parts.append("Body part: " + body) | |
| if all_sym: | |
| parts.append("Symptoms: " + ", ".join(all_sym)) | |
| if all_eff: | |
| parts.append("Effects: " + ", ".join(all_eff)) | |
| if state.get("patient_responses"): | |
| parts.append("Patient history: " + "; ".join( | |
| f"{k}={v}" for k, v in state["patient_responses"].items() | |
| )) | |
| return " | ".join(parts) | |
| def _load_test_images(): | |
| td = Path(__file__).parent / "test_images" | |
| if not td.exists(): return {} | |
| g = {} | |
| for d in sorted(td.iterdir()): | |
| if d.is_dir(): | |
| imgs = [str(p) for p in d.glob("*") if p.suffix.lower() in (".jpg",".jpeg",".png",".webp")] | |
| if imgs: g[d.name.replace("_"," ")] = imgs[:5] | |
| return g | |
| def generate_pdf_report(history: list, logs: str) -> str | None: | |
| if not history and not logs: | |
| return None | |
| pdf = FPDF() | |
| pdf.set_auto_page_break(auto=True, margin=15) | |
| pdf.add_page() | |
| # ═══════════════ HEADER ═══════════════ | |
| pdf.set_fill_color(0, 102, 204) | |
| pdf.rect(0, 0, 210, 40, "F") | |
| pdf.set_font("Helvetica", style="B", size=22) | |
| pdf.set_text_color(255, 255, 255) | |
| pdf.set_y(8) | |
| pdf.cell(0, 10, "IndiDermaX Diagnosis Report", ln=True, align="C") | |
| pdf.set_font("Helvetica", size=10) | |
| pdf.cell(0, 8, "AI-Assisted Dermatology Consultation - CMADD Pipeline", ln=True, align="C") | |
| pdf.set_text_color(0, 0, 0) | |
| pdf.ln(10) | |
| # ═══════════════ TRANSCRIPT ═══════════════ | |
| pdf.set_font("Helvetica", style="B", size=14) | |
| pdf.set_text_color(0, 51, 102) | |
| pdf.cell(0, 10, "Consultation Transcript", ln=True) | |
| pdf.set_draw_color(0, 102, 204) | |
| pdf.line(10, pdf.get_y(), 200, pdf.get_y()) | |
| pdf.ln(5) | |
| pdf.set_text_color(0, 0, 0) | |
| if history: | |
| turn = 0 | |
| for msg in history: | |
| if isinstance(msg, list): | |
| parts = [(m, None) for m in msg if m] | |
| elif isinstance(msg, dict): | |
| role = msg.get("role", "user") | |
| content = msg.get("content", "") | |
| parts = [(content, role)] | |
| else: | |
| continue | |
| for content, forced_role in parts: | |
| content = str(content).strip() | |
| content = content.encode('latin-1', 'replace').decode('latin-1') | |
| if not content: | |
| continue | |
| if forced_role == "user" or ("user" in str(forced_role or "").lower()): | |
| role_label = "Patient" | |
| bg = (235, 245, 255) | |
| elif forced_role == "assistant" or ("assistant" in str(forced_role or "").lower()): | |
| role_label = "IndiDermaX" | |
| bg = (240, 255, 240) | |
| else: | |
| role_label = "User" if "user" in str(forced_role or "") else "Assistant" | |
| bg = (245, 245, 245) | |
| pdf.set_fill_color(*bg) | |
| pdf.set_font("Helvetica", style="B", size=10) | |
| y_before = pdf.get_y() | |
| pdf.cell(30, 6, f"{role_label}:", ln=False) | |
| pdf.set_font("Helvetica", size=10) | |
| w = 150 | |
| if pdf.get_string_width(f"{role_label}:") > 28: | |
| pdf.set_x(42) | |
| w = 148 | |
| pdf.multi_cell(w, 6, content) | |
| pdf.set_fill_color(255, 255, 255) | |
| pdf.ln(2) | |
| # ═══════════════ LOGS ═══════════════ | |
| pdf.add_page() | |
| pdf.set_font("Helvetica", style="B", size=14) | |
| pdf.set_text_color(0, 51, 102) | |
| pdf.cell(0, 10, "CMADD Pipeline Logs & Evidence", ln=True) | |
| pdf.set_draw_color(0, 102, 204) | |
| pdf.line(10, pdf.get_y(), 200, pdf.get_y()) | |
| pdf.ln(5) | |
| pdf.set_text_color(0, 0, 0) | |
| pdf.set_fill_color(248, 250, 252) | |
| pdf.set_font("Courier", size=8) | |
| if logs: | |
| for line in logs.split('\n'): | |
| line = line.encode('latin-1', 'replace').decode('latin-1').rstrip() | |
| if not line.strip(): | |
| pdf.ln(2) | |
| continue | |
| # Style different log types | |
| if line.startswith("##"): | |
| pdf.set_font("Courier", style="B", size=9) | |
| pdf.set_text_color(0, 51, 102) | |
| pdf.ln(3) | |
| pdf.cell(0, 5, line[:120], ln=True) | |
| pdf.set_text_color(0, 0, 0) | |
| pdf.set_font("Courier", size=8) | |
| elif line.startswith("###"): | |
| pdf.set_font("Courier", style="B", size=9) | |
| pdf.ln(2) | |
| pdf.cell(0, 5, line[:120], ln=True) | |
| pdf.set_font("Courier", size=8) | |
| elif line.startswith(" #"): | |
| pdf.set_font("Courier", style="B", size=8) | |
| pdf.cell(0, 5, line[:120], ln=True) | |
| pdf.set_font("Courier", size=8) | |
| else: | |
| for i in range(0, len(line), 110): | |
| chunk = line[i:i + 110] | |
| pdf.cell(0, 4.5, chunk, ln=True) | |
| else: | |
| pdf.set_font("Helvetica", size=10) | |
| pdf.cell(0, 6, "No pipeline logs available.", ln=True) | |
| # ═══════════════ FOOTER ═══════════════ | |
| pdf.set_y(-25) | |
| pdf.set_font("Helvetica", style="I", size=8) | |
| pdf.set_text_color(128, 128, 128) | |
| pdf.cell(0, 5, "IndiDermaX CMADD Pipeline - 6 stages, 5 agents, Neo4j + NVIDIA", ln=True, align="C") | |
| pdf.cell(0, 5, "AI-assisted decision support only. Always consult a dermatologist.", ln=True, align="C") | |
| tf = tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") | |
| pdf.output(tf.name) | |
| return tf.name | |
| custom_theme = gr.themes.Default( | |
| primary_hue="cyan", | |
| secondary_hue="indigo", | |
| neutral_hue="slate", | |
| font=[gr.themes.GoogleFont("Outfit"), "sans-serif"], | |
| ).set( | |
| body_background_fill="#f8fafc", | |
| body_text_color="#0f172a", | |
| block_background_fill="rgba(255, 255, 255, 0.7)", | |
| block_border_width="1px", | |
| block_border_color="rgba(0, 0, 0, 0.1)", | |
| block_radius="16px", | |
| button_primary_background_fill="#0ea5e9", | |
| button_primary_text_color="white", | |
| button_primary_border_color="transparent", | |
| button_secondary_background_fill="rgba(0, 0, 0, 0.05)", | |
| button_secondary_border_color="rgba(0, 0, 0, 0.1)", | |
| button_secondary_text_color="#0f172a", | |
| input_background_fill="#ffffff", | |
| input_border_color="rgba(0, 0, 0, 0.1)", | |
| ) | |
| CSS = """ | |
| body { background: #f8fafc !important; min-height: 100vh; color: #0f172a !important; } | |
| .gradio-container { max-width: 1200px !important; margin: 0 auto !important; background: transparent !important; } | |
| .main-header { text-align: center; padding: 30px 0 10px; animation: fadeInDown 0.8s ease-out; } | |
| .main-header h1 { background: -webkit-linear-gradient(45deg, #0ea5e9, #818cf8); -webkit-background-clip: text; -webkit-text-fill-color: transparent; font-size: 3.5rem; font-weight: 800; margin: 0; letter-spacing: -1px; } | |
| .main-header p { color: #64748b; font-size: 1.15rem; margin: 8px 0 0; font-weight: 300; } | |
| .upload-box, .output-panel { backdrop-filter: blur(16px); -webkit-backdrop-filter: blur(16px); box-shadow: 0 4px 20px 0 rgba(0, 0, 0, 0.05); } | |
| .upload-box { border: 2px dashed rgba(0, 0, 0, 0.15) !important; padding: 25px; text-align: center; min-height: 240px; transition: all 0.3s ease; background: rgba(255, 255, 255, 0.6) !important; } | |
| .upload-box:hover { border-color: #0ea5e9 !important; background: rgba(255, 255, 255, 1) !important; } | |
| .chatbot { border: none !important; background: transparent !important; } | |
| .message.user { background: linear-gradient(135deg, rgba(14, 165, 233, 0.1), rgba(99, 102, 241, 0.1)) !important; border: 1px solid rgba(14, 165, 233, 0.2) !important; color: #0f172a !important; } | |
| .message.bot { background: #ffffff !important; border: 1px solid rgba(0, 0, 0, 0.05) !important; box-shadow: 0 2px 10px rgba(0,0,0,0.02); color: #0f172a !important; } | |
| .log-panel { background: #f1f5f9 !important; border: 1px solid #e2e8f0 !important; border-radius: 12px; padding: 16px; font-size: 12.5px; font-family: 'JetBrains Mono', 'Courier New', monospace; color: #334155 !important; box-shadow: inset 0 2px 10px rgba(0,0,0,0.02); line-height: 1.6; max-height: 300px; overflow-y: auto; } | |
| .diagnose-btn { background: linear-gradient(135deg, #0ea5e9, #6366f1) !important; color: white !important; border: none !important; box-shadow: 0 4px 15px rgba(14, 165, 233, 0.3) !important; transition: all 0.3s cubic-bezier(0.4, 0, 0.2, 1) !important; } | |
| .diagnose-btn:hover { transform: translateY(-3px) scale(1.02) !important; box-shadow: 0 8px 20px rgba(14, 165, 233, 0.4) !important; } | |
| .test-gallery img { border-radius: 12px; border: 2px solid transparent; transition: all 0.4s ease; cursor: pointer; box-shadow: 0 2px 8px rgba(0,0,0,0.05); } | |
| .test-gallery img:hover { border-color: #0ea5e9; transform: scale(1.08) translateY(-4px); box-shadow: 0 10px 20px rgba(0,0,0,0.15); } | |
| footer { visibility: hidden; } | |
| @keyframes fadeInDown { from { opacity: 0; transform: translateY(-20px); } to { opacity: 1; transform: translateY(0); } } | |
| """ | |
| def create_gradio_ui(): | |
| gallery = _load_test_images() | |
| diseases = sorted(gallery.keys()) | |
| default_disease = diseases[0] if diseases else None | |
| def get_imgs(d): return [(img, os.path.basename(img)) for img in gallery.get(d, [])] | |
| def select_img(evt: gr.SelectData, imgs): | |
| if imgs and 0 <= evt.index < len(imgs): return imgs[evt.index][0] | |
| return None | |
| def chat_fn(msg, hist, img, age, state): | |
| """Doctor-like multi-turn conversational flow with phases: | |
| greeting → gathering (4 follow-ups) → diagnosis → done | |
| """ | |
| if not state or "turn" not in state: | |
| state = _init_chat_state() | |
| # Store image and age from this turn | |
| if img and img != state.get("image_path"): | |
| state["image_path"] = img | |
| if age is not None: | |
| state["age"] = int(age) | |
| state["turn"] += 1 | |
| turn = state["turn"] | |
| # ---- PHASE: greeting (first interaction) ---- | |
| if state["phase"] == "greeting": | |
| if msg: | |
| desc, bp, sym, eff = _extract_features(msg) | |
| state["accumulated_descriptors"].extend(desc) | |
| state["accumulated_symptoms"].extend(sym) | |
| state["accumulated_effects"].extend(eff) | |
| if bp: | |
| state["body_part"] = bp | |
| state["phase"] = "gathering" | |
| if not state.get("image_path") and not state["accumulated_descriptors"]: | |
| response = ( | |
| "👋 Thank you for reaching out. I'm here to help assess your skin condition.\n\n" | |
| "To give you the most accurate analysis, please **upload a clear photo** of the affected skin area. " | |
| "Meanwhile, I have a few questions:\n\n" | |
| ) | |
| else: | |
| response = "👋 Thank you for sharing that. I'll analyze your information carefully.\n\n" | |
| next_q = _get_next_question(state) | |
| if next_q: | |
| response += f"📋 **{next_q}**" | |
| state["_last_logs"] = "" | |
| hist = (hist or []) + [ | |
| {"role": "user", "content": "👤 " + (msg or "[Image uploaded]")}, | |
| {"role": "assistant", "content": "🤖 " + response}, | |
| ] | |
| return hist, "", state, "" | |
| # ---- PHASE: gathering (follow-up questions) ---- | |
| if state["phase"] == "gathering": | |
| state = _process_gathering_response(state, msg) | |
| next_q = _get_next_question(state) | |
| if next_q: | |
| response = f"Thank you for sharing that.\n\n📋 **{next_q}**" | |
| hist = (hist or []) + [ | |
| {"role": "user", "content": "👤 " + msg}, | |
| {"role": "assistant", "content": "🤖 " + response}, | |
| ] | |
| return hist, "", state, "" | |
| else: | |
| # Gathering complete — prep image bytes and run diagnosis | |
| img_bytes = None | |
| if state.get("image_path"): | |
| try: | |
| pil = Image.open(state["image_path"]) | |
| buf = io.BytesIO() | |
| pil.save(buf, format="JPEG") | |
| img_bytes = buf.getvalue() | |
| except Exception: | |
| pass | |
| combined = _build_combined_message(state, msg) | |
| result = run_full_pipeline( | |
| user_message=combined, | |
| image_data=img_bytes, | |
| patient_age=state.get("age"), | |
| session_id=f"chat_turn{turn}", | |
| ) | |
| answer = result["answer"] | |
| logs = result.get("log_text", "") | |
| state["_last_logs"] = logs | |
| state["phase"] = "done" | |
| response = "I've gathered enough information. Let me analyze everything now...\n\n" + answer | |
| hist = (hist or []) + [ | |
| {"role": "user", "content": "👤 " + msg}, | |
| {"role": "assistant", "content": "🤖 " + response}, | |
| ] | |
| return hist, "", state, logs | |
| # ---- PHASE: done (restart on new message) ---- | |
| if state["phase"] == "done": | |
| state = _init_chat_state() | |
| state["turn"] = turn | |
| if msg: | |
| desc, bp, sym, eff = _extract_features(msg) | |
| state["accumulated_descriptors"].extend(desc) | |
| state["accumulated_symptoms"].extend(sym) | |
| state["accumulated_effects"].extend(eff) | |
| if bp: | |
| state["body_part"] = bp | |
| if img: | |
| state["image_path"] = img | |
| if age is not None: | |
| state["age"] = int(age) | |
| response = "🔄 Starting a new assessment for you.\n\n" | |
| state["phase"] = "gathering" | |
| next_q = _get_next_question(state) | |
| if next_q: | |
| response += f"📋 **{next_q}**" | |
| state["_last_logs"] = "" | |
| hist = (hist or []) + [ | |
| {"role": "user", "content": "👤 " + (msg or "[Image uploaded]")}, | |
| {"role": "assistant", "content": "🤖 " + response}, | |
| ] | |
| return hist, "", state, "" | |
| # Fallback | |
| hist = (hist or []) + [ | |
| {"role": "user", "content": "👤 " + (msg or "")}, | |
| {"role": "assistant", "content": "🤖 I'm not sure how to proceed. Please describe your skin concern."}, | |
| ] | |
| return hist, "", state, "" | |
| with gr.Blocks(title="IndiDermaX - AI Dermatology") as gradio_ui_blocks: | |
| gr.HTML('<div class="main-header"><h1>IndiDermaX</h1><p>AI Dermatology - Neo4j Graph, 245 Diseases, 5 Agents</p></div>') | |
| with gr.Row(equal_height=True): | |
| with gr.Column(scale=1): | |
| with gr.Group(elem_classes=["upload-box"]): | |
| img_in = gr.Image(label="📷 Upload Skin Image", type="filepath", height=240) | |
| age_in = gr.Number(label="Age", minimum=0, maximum=120, step=1, value=None) | |
| msg_in = gr.Textbox(label="Describe Symptoms", placeholder="e.g. Red scaly ring-shaped patch on inner thigh, very itchy for 2 weeks...", lines=3, max_lines=5) | |
| with gr.Row(): | |
| btn = gr.Button("🔍 Start Consultation", variant="primary", elem_classes=["diagnose-btn"]) | |
| clr = gr.Button("🗑 Clear", variant="secondary") | |
| if diseases: | |
| gr.Markdown("#### 🖼 Quick Test Images") | |
| dd = gr.Dropdown(choices=diseases, label="Select Disease", value=default_disease) | |
| gal = gr.Gallery(label="Click image to load it for diagnosis", columns=3, height=160, object_fit="cover") | |
| dd.change(fn=get_imgs, inputs=[dd], outputs=[gal]) | |
| gal.select(fn=select_img, inputs=[gal], outputs=[img_in]) | |
| with gr.Column(scale=2): | |
| with gr.Group(elem_classes=["output-panel"]): | |
| gr.Markdown("### 💬 Diagnosis Chat") | |
| status_text = gr.Markdown("🟢 **Ready** — Upload an image and describe your symptoms to begin.") | |
| chat = gr.Chatbot(label="", height=380, show_label=False) | |
| with gr.Row(): | |
| pdf_btn = gr.Button("📄 Generate PDF Report", size="sm") | |
| pdf_out = gr.File(label="Download Report") | |
| with gr.Accordion("📊 Pipeline Logs & Evidence", open=False): | |
| log_out = gr.Textbox(label="", lines=12, max_lines=20, show_label=False, elem_classes=["log-panel"]) | |
| state = gr.State(_init_chat_state()) | |
| acc_logs = gr.State("") | |
| def submit(msg, hist, img, age, st, logs): | |
| if not msg and not img: | |
| return hist, "", st, logs, status_text.value | |
| hist, empty, st, new_logs = chat_fn(msg, hist, img, age, st) | |
| full_logs = (logs + "\n" + new_logs).strip() if new_logs else logs | |
| phase = st.get("phase", "greeting") | |
| phase_map = { | |
| "greeting": "🟡 **Greeting** — Starting consultation...", | |
| "gathering": f"🟡 **Gathering History** — Question {st.get('followup_index', 0)} of {MAX_FOLLOWUPS}", | |
| "done": "🟢 **Diagnosis Complete** — You can send a new message to start another assessment.", | |
| } | |
| status = phase_map.get(phase, "🟢 **Ready**") | |
| return hist, empty, st, full_logs, status | |
| btn.click( | |
| fn=submit, | |
| inputs=[msg_in, chat, img_in, age_in, state, acc_logs], | |
| outputs=[chat, msg_in, state, log_out, status_text], | |
| ) | |
| msg_in.submit( | |
| fn=submit, | |
| inputs=[msg_in, chat, img_in, age_in, state, acc_logs], | |
| outputs=[chat, msg_in, state, log_out, status_text], | |
| ) | |
| pdf_btn.click(fn=generate_pdf_report, inputs=[chat, log_out], outputs=[pdf_out]) | |
| clr.click( | |
| fn=lambda: ( | |
| [], None, None, "", _init_chat_state(), "", | |
| "🟢 **Ready** — Upload an image and describe your symptoms to begin.", | |
| None, | |
| ), | |
| inputs=[], | |
| outputs=[chat, img_in, age_in, msg_in, state, log_out, status_text, pdf_out], | |
| ) | |
| if default_disease: gradio_ui_blocks.load(fn=get_imgs, inputs=[dd], outputs=[gal]) | |
| gr.HTML('<div style="text-align:center;color:#64748b;font-size:11px;padding:16px"><b>IndiDermaX CMADD Pipeline</b> - 6 stages, 5 agents, Neo4j + NVIDIA<br>AI decision support tool. Always consult a dermatologist.</div>') | |
| return gradio_ui_blocks | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| # MOUNT FASTAPI ON GRADIO & START | |
| # ═══════════════════════════════════════════════════════════════════════════ | |
| import sys | |
| gradio_ui = create_gradio_ui() | |
| print(f"[indidermax] Starting...", flush=True) | |
| print(f"[indidermax] Neo4j: {'CONFIGURED' if NEO4J_URI else 'MISSING (cache fallback)'}", flush=True) | |
| print(f"[indidermax] NVIDIA: {'CONFIGURED' if NVIDIA_KEY else 'MISSING (vision disabled)'}", flush=True) | |
| # Mount FastAPI routes onto Gradio — /api/* → FastAPI, / → Gradio | |
| try: | |
| app = gr.mount_gradio_app(_fastapi, gradio_ui, path="/", theme=custom_theme, css=CSS) | |
| print("[indidermax] FastAPI routes mounted on Gradio", flush=True) | |
| except Exception as e: | |
| print(f"[indidermax] WARNING: mount failed: {e}", flush=True) | |
| app = gradio_ui | |
| # For local dev only | |
| if __name__ == "__main__": | |
| print("[indidermax] Local mode — starting uvicorn", flush=True) | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860, log_level="info") | |