#!/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 @_fastapi.exception_handler(Exception) 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 # ═══════════════════════════════════════════════════════════════════════════ @_fastapi.get("/api/health") 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 @_fastapi.post("/api/diagnose", response_model=DiagnoseResponse) 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}) @_fastapi.post("/api/diagnose/upload") 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('

IndiDermaX

AI Dermatology - Neo4j Graph, 245 Diseases, 5 Agents

') 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('
IndiDermaX CMADD Pipeline - 6 stages, 5 agents, Neo4j + NVIDIA
AI decision support tool. Always consult a dermatologist.
') 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")