""" MedASR + Gemini LLM Pipeline — HuggingFace Space ================================================= Required secret: GEMINI_API_KEY """ import os, re, json import gradio as gr import google.generativeai as genai import onnxruntime as ort import numpy as np import soundfile as sf # Safe API configuration api_key = os.environ.get("GEMINI_API_KEY") if api_key: genai.configure(api_key=api_key) SAMPLES = { "GP Consultation — Fever & headache (Eka)": "i have been having fever and headache for the past three days and also some body ache " "antigen test for dengue was done and it came negative so i am prescribing you dolo 650 " "for fever and pantop dsr 40 for your stomach please take them after food", "GP Consultation — Thyroid symptoms (Eka)": "patient feels sad not able to wake up and energy less blood test was done thyroid levels " "came low so starting thyroxine 25 mcg once daily in the morning on empty stomach " "follow up after 15 days", "Radiology Report — CT Chest PE (MedASR official)": "ct chest pe protocol indication 54 year old female shortness of breath evaluate for pe " "technique standard protocol findings pulmonary vasculature the main pa is patent there " "are filling defects in the segmental branches of the right lower lobe compatible with " "acute pe no saddle embolus lungs no pneumothorax small bilateral effusions right greater " "than left impression acute segmental pe right lower lobe", "Patient Symptoms — Back & leg pain (Hani89)": "i am experiencing severe back pain and leg pain since last week i was given dolo 650 " "tablet thrice a day for 6 days and pantop dsr for 7 days and also thyroxine 25 mcg " "for 15 days", } MODEL_PATH = "model_quantized.onnx" # adjust if different session = None def load_model(): global session if session is None: session = ort.InferenceSession(MODEL_PATH) return session def preprocess_audio(audio_path): audio, sr = sf.read(audio_path) if len(audio.shape) > 1: audio = np.mean(audio, axis=1) # Resample if needed if sr != 16000: from scipy.signal import resample audio = resample(audio, int(len(audio) * 16000 / sr)) return audio.astype(np.float32) def transcribe_audio(audio_path): try: session = load_model() audio = preprocess_audio(audio_path) inputs = {session.get_inputs()[0].name: audio} outputs = session.run(None, inputs) # This part depends on your model decoding transcript = str(outputs[0]) return transcript except Exception as e: return f"ASR Error: {e}" def run_full_pipeline(audio): if not audio: return "No audio", "", "", "" transcript = transcribe_audio(audio) if transcript.startswith("ASR Error"): return transcript, "", "", "" correction = run_correction(transcript) soap = run_soap(transcript) entities = run_entities(transcript) return transcript, correction, soap, entities def prompt_correction(t): return f"""You are a medical transcript editor. Fix transcription errors, add punctuation, expand abbreviations, and preserve all clinical content exactly. ASR transcript: {t} """ def prompt_soap(t): return f"""Convert this into a structured SOAP note. SUBJECTIVE: OBJECTIVE: ASSESSMENT: PLAN: Transcript: {t} """ def prompt_entities(t): return ( "Extract medical entities and return ONLY valid JSON:\n" '{"diagnoses":[],"medications":[],"dosages":[],"symptoms":[],"procedures":[],"instructions":[]}\n\n' f"Transcript:\n{t}" ) def call_gemini(prompt: str) -> str: api_key = os.environ.get("GEMINI_API_KEY") if not api_key: return "❌ GEMINI_API_KEY not set in Hugging Face Space." try: genai.configure(api_key=api_key) # safer model model = genai.GenerativeModel("gemini-2.5-flash") response = model.generate_content(prompt) return response.text if hasattr(response, "text") else str(response) except Exception as e: return f"❌ Gemini API error: {e}" def load_sample(name): return SAMPLES.get(name, "") def run_correction(transcript): if not transcript.strip(): return "Please enter or select a transcript first." return call_gemini(prompt_correction(transcript)) def run_soap(transcript): if not transcript.strip(): return "Please enter or select a transcript first." return call_gemini(prompt_soap(transcript)) def run_entities(transcript): if not transcript.strip(): return "Please enter or select a transcript first." raw = call_gemini(prompt_entities(transcript)) try: clean = re.sub(r"```json|```", "", raw).strip() obj = json.loads(clean) icons = { "diagnoses": "🔴 Diagnoses", "medications": "💊 Medications", "dosages": "📏 Dosages", "symptoms": "🤒 Symptoms", "procedures": "🔬 Procedures", "instructions": "📋 Instructions", } lines = [] for key, label in icons.items(): vals = obj.get(key, []) if vals: lines.append(f"**{label}**") for v in vals: lines.append(f" • {v}") lines.append("") return "\n".join(lines) if lines else "No entities found." except Exception: return raw def run_all(transcript): if not transcript.strip(): empty = "No transcript provided." return empty, empty, empty return ( run_correction(transcript), run_soap(transcript), run_entities(transcript), ) with gr.Blocks( title="MedASR + Gemini Pipeline", theme=gr.themes.Soft(), ) as demo: audio_input = gr.Audio(type="filepath", label="Upload Medical Audio") run_audio_btn = gr.Button("🎤 Run Full Pipeline") raw_transcript_out = gr.Textbox(label="Raw Transcript", lines=6) gr.Markdown(""" # ⚕ MedASR + Gemini Pipeline Phase 2 — Transcript → Clinical NLP (Correction, SOAP, Entities) """) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Input Transcript") sample_dd = gr.Dropdown( choices=list(SAMPLES.keys()), value=list(SAMPLES.keys())[0], label="Load sample" ) transcript_box = gr.Textbox( value=SAMPLES[list(SAMPLES.keys())[0]], lines=9, label="Transcript" ) sample_dd.change(load_sample, inputs=sample_dd, outputs=transcript_box) run_all_btn = gr.Button("⚡ Run All") with gr.Column(scale=1): gr.Markdown("### Output") with gr.Tab("Correction"): correction_out = gr.Textbox(lines=12, interactive=False) run_a_btn = gr.Button("Run A") with gr.Tab("SOAP"): soap_out = gr.Textbox(lines=14, interactive=False) run_b_btn = gr.Button("Run B") with gr.Tab("Entities"): entities_out = gr.Markdown() run_c_btn = gr.Button("Run C") run_a_btn.click(run_correction, inputs=transcript_box, outputs=correction_out) run_b_btn.click(run_soap, inputs=transcript_box, outputs=soap_out) run_c_btn.click(run_entities, inputs=transcript_box, outputs=entities_out) run_all_btn.click( run_all, inputs=transcript_box, outputs=[correction_out, soap_out, entities_out] ) run_audio_btn.click( fn=run_full_pipeline, inputs=audio_input, outputs=[raw_transcript_out, correction_out, soap_out, entities_out] ) if __name__ == "__main__": demo.launch()