"""
def _build_generated_document(state: dict[str, Any]) -> dict[str, Any]:
"""Create an NHS-format clinic letter for S5/S6 review."""
selected_patient = (state or {}).get("selected_patient") or {}
patient_context = (state or {}).get("patient_context") or {}
demographics = patient_context.get("demographics", {})
labs = patient_context.get("recent_labs", [])
problems = patient_context.get("problem_list", [])
meds = patient_context.get("medications", [])
patient_name = str(demographics.get("name") or selected_patient.get("name") or "Patient")
dob = str(demographics.get("dob") or "Unknown")
nhs = str(demographics.get("nhs_number") or "Unknown")
today = datetime.now().strftime("%d %B %Y")
gp_name = "Andrew Wilson"
address = "Riverside Medical Practice\n14 Harcourt Street\nLondon"
investigations = "\n".join(
f"- {lab.get('name', 'Test')}: {lab.get('value', '')} {lab.get('unit', '')} ({lab.get('date', '')})".strip()
for lab in labs
) or "- No recent investigations available"
medication_line = ", ".join(m.get("name", "") for m in meds if m.get("name"))
main_problem = problems[0] if problems else "ongoing clinical concerns"
if "Margaret Thompson" in patient_name:
history = "She attended for diabetes and cardiovascular risk review with persistent hyperglycaemia despite current therapy. She reports reduced activity tolerance and occasional post-prandial fatigue over recent weeks."
assessment = "Suboptimal glycaemic control with HbA1c 8.2% in the context of known type 2 diabetes, hypertension, and hyperlipidaemia. Renal function remains acceptable but trending down."
plan_lines = [
"Increase metformin optimisation counselling and initiate structured diabetic diet support.",
"Arrange repeat HbA1c, renal profile, and urine ACR in 8 weeks.",
"Continue lisinopril/atorvastatin/aspirin and monitor blood pressure weekly.",
]
else:
history = f"I reviewed {patient_name} regarding {main_problem.lower()} and ongoing symptom burden in clinic. The patient reports variable day-to-day control and is keen for treatment optimisation."
assessment = f"Current presentation is consistent with {main_problem.lower()}, requiring continued medication review and follow-up."
plan_lines = [
"Continue current treatment with safety-netting advice.",
"Repeat key blood tests prior to next follow-up.",
"Review in specialist clinic to reassess response and escalation needs.",
]
letter_text = (
f"{today}\n\n"
f"Dr {gp_name}\n"
f"{address}\n\n"
f"Dear Dr {gp_name},\n\n"
f"Re: {patient_name} (DOB: {dob}, NHS: {nhs})\n"
f" {address}\n\n"
f"Thank you for referring / I reviewed {patient_name} in General Practice Clinic on {today}.\n\n"
"History of Presenting Complaint\n"
f"{history}\n\n"
"Examination\n"
"The patient was comfortable at rest, haemodynamically stable, and clinically euvolaemic on examination. No acute red-flag findings were identified today.\n\n"
"Investigations\n"
f"{investigations}\n\n"
"Assessment\n"
f"{assessment}\n\n"
"Plan\n"
+ "\n".join(f"{i + 1}. {line}" for i, line in enumerate(plan_lines))
+ f"\n\nI will review {patient_name} in 8 weeks. Please do not hesitate to contact us if there are any concerns in the interim.\n\n"
"Yours sincerely,\n\n"
"Dr Sarah Chen\n"
"Consultant, General Practice\n"
"Clarke NHS Trust"
)
sections = [
{"heading": "NHS Clinic Letter", "content": letter_text},
{"heading": "Clinical Issues", "content": "\n".join(f"- {item}" for item in problems) or "- None listed"},
{"heading": "Current Medications", "content": medication_line or "None documented"},
{"heading": "Follow-up", "content": "Review in 8 weeks with repeat investigations."},
]
return {
"title": "NHS Clinic Letter",
"status": "ready_for_review",
"sections": sections,
"patient_name": patient_name,
"nhs_number": nhs,
}
def _render_letter_sections(letter_sections: list[dict[str, str]]) -> tuple[str, str, str, str]:
"""Map generated letter sections onto fixed textbox outputs.
Combines all sections into section 1 as the full editable letter.
Sections 2-4 are hidden and returned empty.
Args:
letter_sections (list[dict[str, str]]): Ordered letter sections.
Returns:
tuple[str, str, str, str]: Combined letter in slot 1, empty slots 2-4.
"""
combined = "\n\n".join(
f"{s.get('heading', '')}\n{s.get('content', '').strip()}".strip()
for s in letter_sections
if s.get('content', '').strip()
)
return (combined, "", "", "")
def _handle_patient_selection(state: dict[str, Any], patient_index: int):
"""Update state and call backend context endpoint when a patient index is selected.
Args:
state (dict[str, Any]): Current UI session state.
patient_index (int): Selected patient index from the dashboard.
Returns:
tuple[...]: Updated state, feedback, context HTML, context shell HTML, and visibility updates.
"""
clinic_payload = load_clinic_list()
patients = clinic_payload.get("patients", [])
if patient_index < 0 or patient_index >= len(patients):
return state, "Patient selection failed: patient index out of range.", _context_screen_html({}, {}), "", "", "", "", "", gr.update(), *show_screen("s1")
patient = patients[patient_index]
updated_state = select_patient(state, patient)
updated_state['current_patient_index'] = patient_index
patient_id = str(patient.get("id", ""))
if os.getenv("USE_MOCK_FHIR", "").lower() == "true":
context = _mock_context_for_index(patient_index)
feedback = f"Loaded mock patient context for {patient['name']} ({patient_id})."
else:
try:
context = _api_request("POST", f"/patients/{patient_id}/context")
except Exception as exc:
context = _mock_context_for_index(patient_index)
feedback = f"Patient selected with frontend mock context fallback: {exc}"
else:
feedback = f"Loaded patient context for {patient['name']} ({patient_id})."
updated_state["patient_context"] = context
completed_patients = set(updated_state.get("completed_patients", []))
if patient_index in completed_patients:
generated = updated_state.get("signed_letters", {}).get(str(patient_index), "")
if generated:
updated_state["signed_document_text"] = generated
updated_state["screen"] = "s5"
return updated_state, f"Opened completed patient {patient['name']} in Document Review.", _context_screen_html(patient, context), generated, "", "", "", f"
{escape(generated)}
", "Previously signed letter loaded for review.", *show_screen("s5")
return updated_state, feedback, _context_screen_html(patient, context), "", "", "", "", "", gr.update(), *show_screen("s2")
def _handle_back_to_dashboard(state):
"""Navigate from context screen back to dashboard.
Args:
state (dict[str, Any]): Current application state.
Returns:
tuple[...]: Updated state, feedback text, and visibility updates.
"""
updated_state = dict(state or initial_consultation_state())
updated_state["screen"] = "s1"
return updated_state, "Returned to dashboard.", *show_screen("s1")
def _handle_start_consultation(state):
"""Start consultation by calling backend and storing consultation ID.
Args:
state (dict[str, Any]): Current application state.
Returns:
tuple[...]: Updated state, feedback, recording HTML, timer tick update, and visibility updates.
"""
updated_state = dict(state or initial_consultation_state())
patient_id = str((updated_state.get("selected_patient") or {}).get("id", ""))
if not patient_id:
return updated_state, "Please select a patient first.", _recording_screen_html("00:00"), gr.update(active=False), *show_screen("s1")
try:
payload = _api_request("POST", "/consultations/start", json={"patient_id": patient_id})
except Exception as exc:
return updated_state, f"Failed to start consultation: {exc}", _recording_screen_html("00:00"), gr.update(active=False), *show_screen("s2")
updated_state["consultation"] = {"id": payload.get("consultation_id"), "status": payload.get("status", "recording")}
updated_state["recording_started_at"] = datetime.now(tz=timezone.utc).isoformat()
updated_state["screen"] = "s3"
return updated_state, "Consultation recording started.", _recording_screen_html("00:00"), gr.update(active=True), *show_screen("s3")
def _update_recording_timer(state):
"""Compute MM:SS elapsed timer value for the active consultation recording.
Args:
state (dict[str, Any]): Current UI state containing recording start metadata.
Returns:
str: HTML with elapsed timer.
"""
started_at = str((state or {}).get("recording_started_at", "")).strip()
if not started_at:
return _recording_screen_html("00:00")
elapsed_s = max(int((datetime.now(tz=timezone.utc) - _safe_datetime_from_iso(started_at)).total_seconds()), 0)
minutes, seconds = divmod(elapsed_s, 60)
return _recording_screen_html(f"{minutes:02d}:{seconds:02d}")
def _stage_from_pipeline(stage: str) -> tuple[int, str, str]:
"""Map backend pipeline stage to display values.
Args:
stage (str): Backend pipeline stage value.
Returns:
tuple[int, str, str]: Stage number, label, and description.
"""
mapping = {
"transcribing": (1, "Finalising transcript…", "MedASR processing audio"),
"retrieving_context": (2, "Synthesising patient context…", "MedGemma 4B querying records"),
"generating_document": (3, "Generating clinical letter…", "MedGemma 27B composing document"),
"complete": (3, "Generating clinical letter…", "MedGemma 27B composing document"),
}
return mapping.get(stage, mapping["transcribing"])
def _ensure_mock_audio_file(audio_path: str | None) -> str | None:
"""Create a short silent WAV when running in mock mode and no audio was captured."""
if audio_path:
return audio_path
if os.getenv("MEDASR_MODEL_ID", "").lower() != "mock":
return None
upload_dir = Path("data/uploads/mock")
upload_dir.mkdir(parents=True, exist_ok=True)
silent_path = upload_dir / "silent.wav"
with wave.open(str(silent_path), "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(16000)
wav_file.writeframes(b"\x00\x00" * (16000 * 6))
return str(silent_path)
def _start_processing(state, audio_path):
"""Upload audio and end consultation, then transition to processing screen.
Args:
state (dict[str, Any]): Current UI state.
audio_path (str | None): File path returned by Gradio audio component.
Returns:
tuple[...]: Updated state, feedback, processing HTML, timer update, and visibility updates.
"""
updated_state = dict(state or initial_consultation_state())
raw_consultation_id = (updated_state.get("consultation") or {}).get("id")
consultation_id = str(raw_consultation_id) if raw_consultation_id is not None else ""
if not consultation_id:
return updated_state, "Consultation session is missing. Start consultation again.", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"), gr.update(active=False), *show_screen("s3")
resolved_audio_path = _ensure_mock_audio_file(audio_path)
if not resolved_audio_path:
return updated_state, "Please capture audio before ending consultation.", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"), gr.update(active=False), *show_screen("s3")
if os.getenv("MEDASR_MODEL_ID", "").lower() == "mock":
updated_state["captured_audio_path"] = resolved_audio_path
updated_state["processing_started_at"] = datetime.now(tz=timezone.utc).isoformat()
updated_state["consultation"] = updated_state.get("consultation") or {}
updated_state["consultation"]["id"] = ""
updated_state["consultation"]["status"] = "processing"
updated_state["screen"] = "s4"
return updated_state, "Consultation ended. Processing audio and generating document.", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"), gr.update(active=True), *show_screen("s4")
try:
with Path(resolved_audio_path).open("rb") as stream:
_api_request("POST", f"/consultations/{consultation_id}/audio", files={"audio_file": (Path(resolved_audio_path).name, stream, "audio/wav")}, data={"is_final": "true"}, timeout=120.0)
_api_request("POST", f"/consultations/{consultation_id}/end", timeout=180.0)
except Exception as exc:
return updated_state, f"Failed to end consultation: {exc}", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"), gr.update(active=False), *show_screen("s3")
updated_state["captured_audio_path"] = resolved_audio_path
updated_state["processing_started_at"] = datetime.now(tz=timezone.utc).isoformat()
updated_state["consultation"]["status"] = "processing"
updated_state["screen"] = "s4"
return updated_state, "Consultation ended. Processing audio and generating document.", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"), gr.update(active=True), *show_screen("s4")
def _poll_processing_progress(state):
"""Poll backend consultation progress and transition to review when complete.
Args:
state (dict[str, Any]): Current state containing consultation metadata.
Returns:
tuple[...]: Updated state and UI updates for processing/review screens.
"""
updated_state = dict(state or initial_consultation_state())
consultation_id = str((updated_state.get("consultation") or {}).get("id", ""))
started_at = str(updated_state.get("processing_started_at", "") or "")
elapsed = "Elapsed: 00:00"
if started_at:
elapsed_s = max(int((datetime.now(tz=timezone.utc) - _safe_datetime_from_iso(started_at)).total_seconds()), 0)
minutes, seconds = divmod(elapsed_s, 60)
elapsed = f"Elapsed: {minutes:02d}:{seconds:02d}"
if not consultation_id:
doc = _build_generated_document(updated_state)
updated_state["generated_document"] = doc
updated_state["consultation"] = updated_state.get("consultation") or {"id": None, "status": "review"}
updated_state["consultation"]["status"] = "review"
updated_state["screen"] = "s5"
s1, s2, s3, s4 = _render_letter_sections(doc.get("sections", []))
fhir = ""
return updated_state, "Processing complete. Review the generated clinic letter.", _processing_screen_html(3, "Generating clinical letter…", "MedGemma 27B composing document", elapsed), gr.update(active=False), s1, s2, s3, s4, fhir, *show_screen("s5")
try:
progress = _api_request("GET", f"/consultations/{consultation_id}/progress")
except Exception as exc:
return updated_state, f"Progress polling failed: {exc}", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", elapsed), gr.update(active=False), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), *show_screen("s4")
stage_number, stage_label, stage_description = _stage_from_pipeline(str(progress.get("stage", "transcribing")))
if str(progress.get("stage", "")) != "complete":
return updated_state, f"Processing in progress: {stage_label}", _processing_screen_html(stage_number, stage_label, stage_description, elapsed), gr.update(active=True), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), *show_screen("s4")
document_payload = _api_request("GET", f"/consultations/{consultation_id}/document").get("document") or _build_generated_document(updated_state)
updated_state["generated_document"] = document_payload
updated_state["consultation"]["status"] = "review"
updated_state["screen"] = "s5"
s1, s2, s3, s4 = _render_letter_sections(document_payload.get("sections", []))
fhir = " ".join(
[
f"NHS: {escape(str(document_payload.get('nhs_number', 'N/A')))}",
f"Patient: {escape(str(document_payload.get('patient_name', 'N/A')))}",
]
)
return updated_state, "Processing complete. Review the generated clinic letter.", _processing_screen_html(3, "Generating clinical letter…", "MedGemma 27B composing document", elapsed), gr.update(active=False), s1, s2, s3, s4, fhir, *show_screen("s5")
def _regenerate_document(state):
"""Restart processing view before polling backend completion status again.
Args:
state (dict[str, Any]): Current UI state.
Returns:
tuple[...]: Updated state and screen visibility updates.
"""
updated_state = dict(state or initial_consultation_state())
updated_state["processing_started_at"] = datetime.now(tz=timezone.utc).isoformat()
updated_state["consultation"]["status"] = "processing"
updated_state["screen"] = "s4"
return updated_state, "Regenerating entire clinic letter.", _processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"), gr.update(active=True), *show_screen("s4")
def _cancel_processing(state):
"""Cancel processing workflow and return to live consultation screen.
Args:
state (dict[str, Any]): Current UI state.
Returns:
tuple[...]: Updated state, feedback, and visibility updates.
"""
updated_state = dict(state or initial_consultation_state())
updated_state["screen"] = "s3"
updated_state["consultation"]["status"] = "recording"
return updated_state, "Processing cancelled. Returned to consultation.", gr.update(active=False), *show_screen("s3")
def _sign_off_document(state, section_1, section_2, section_3, section_4):
"""Persist edited sections to backend sign-off endpoint and show final letter.
Args:
state (dict[str, Any]): Current UI state.
section_1 (str): Edited section one text.
section_2 (str): Edited section two text.
section_3 (str): Edited section three text.
section_4 (str): Edited section four text.
Returns:
tuple[...]: Updated state and signed-off UI content.
"""
updated_state = dict(state or initial_consultation_state())
raw_consultation_id = (updated_state.get("consultation") or {}).get("id")
consultation_id = str(raw_consultation_id) if raw_consultation_id is not None else ""
edited_sections = [section_1, section_2, section_3, section_4]
payload_sections: list[dict[str, str]] = []
for index, section_text in enumerate(edited_sections):
if not section_text.strip():
continue
lines = section_text.splitlines()
heading = lines[0].strip() if lines else f"Section {index + 1}"
content = "\n".join(lines[1:]).strip() if len(lines) > 1 else ""
payload_sections.append({"heading": heading, "content": content})
if consultation_id:
try:
_api_request("POST", f"/consultations/{consultation_id}/document/sign-off", json={"sections": payload_sections})
except Exception as exc:
return updated_state, f"Sign-off failed: {exc}", gr.update(), "", gr.update(), *show_screen("s5")
signed_letter = section_1.strip() if section_1 and section_1.strip() else "\n\n".join(part.strip() for part in edited_sections if part and part.strip())
updated_state["signed_document_text"] = signed_letter
selected_index = int(updated_state.get('current_patient_index', 0))
signed_letters = dict(updated_state.get('signed_letters', {}))
signed_letters[str(selected_index)] = signed_letter
updated_state['signed_letters'] = signed_letters
if "consultation" not in updated_state or not isinstance(updated_state.get("consultation"), dict):
updated_state["consultation"] = {"id": None, "status": "idle"}
updated_state["consultation"]["status"] = "signed_off"
updated_state["screen"] = "s6"
export_path = Path("data") / "demo" / "latest_signed_letter.txt"
export_path.write_text(signed_letter + "\n", encoding="utf-8")
signed_html = f"
{escape(signed_letter)}
"
return updated_state, "Document signed off. You can now copy or download the letter.", signed_html, signed_letter, gr.update(value=str(export_path)), *show_screen("s6")
def _copy_signed_document(state):
"""Refresh copy textbox payload for signed letter actions.
Args:
state (dict[str, Any]): Current UI state.
Returns:
tuple[dict[str, Any], str, str]: State, status message, and copy text payload.
"""
updated_state = dict(state or initial_consultation_state())
signed_text = str(updated_state.get("signed_document_text") or "")
if not signed_text:
return updated_state, "No signed letter available to copy yet.", ""
return updated_state, "Letter content refreshed for copy.", signed_text
def _prepare_signed_download(state):
"""Refresh downloadable signed-letter text artifact.
Args:
state (dict[str, Any]): Current UI state.
Returns:
tuple[dict[str, Any], str, Any]: State, status message, and file update.
"""
updated_state = dict(state or initial_consultation_state())
signed_text = str(updated_state.get("signed_document_text") or "")
if not signed_text:
return updated_state, "No signed letter available to download yet.", gr.update(value=None)
export_path = Path("data") / "demo" / "latest_signed_letter.txt"
export_path.write_text(signed_text + "\n", encoding="utf-8")
return updated_state, "Download file refreshed.", gr.update(value=str(export_path))
def _next_patient(state):
"""Reset consultation workflow and return to dashboard after sign-off.
Args:
state (dict[str, Any]): Current UI state.
Returns:
tuple[...]: Reset state and cleared UI content updates.
"""
updated_state = dict(state or initial_consultation_state())
completed = set(updated_state.get('completed_patients', []))
current_index = int(updated_state.get('current_patient_index', 0))
completed.add(current_index)
updated_state['completed_patients'] = sorted(completed)
dashboard = build_dashboard_html(load_clinic_list(), completed_patients=updated_state['completed_patients'])
refreshed_state = initial_consultation_state()
refreshed_state['completed_patients'] = updated_state['completed_patients']
refreshed_state['signed_letters'] = dict(updated_state.get('signed_letters', {}))
return refreshed_state, "Ready for next patient. Please select a patient card.", "", "", "", "", "", "", dashboard, *show_screen("s1")
def build_ui() -> gr.Blocks:
"""Build the primary Clarke UI blocks for the dashboard flow.
Args:
None: Function reads local static assets and clinic JSON data.
Returns:
gr.Blocks: Configured Gradio Blocks application.
"""
clinic_payload = load_clinic_list()
with gr.Blocks(theme=clarke_theme, css=Path("frontend/assets/style.css").read_text(encoding="utf-8"), title="Clarke", head=CLARKE_HEAD) as demo:
app_state = gr.State(initial_consultation_state())
gr.HTML(build_global_style_block())
feedback_text = gr.Markdown("", visible=False)
with gr.Column(visible=False) as screen_s2:
context_screen_html = gr.HTML(_context_screen_html({}, {}))
hidden_start_button = gr.Button("hidden-start-consultation", visible=True, elem_id="hidden-start-consultation")
hidden_back_button = gr.Button("hidden-back", visible=True, elem_id="hidden-back")
with gr.Column(visible=False) as screen_s3:
recording_html = gr.HTML(_recording_screen_html("00:00"))
consultation_audio = gr.Audio(sources=["microphone"], streaming=False, type="filepath", label="Consultation Audio", elem_id="clarke-audio-input")
recording_tick = gr.Timer(value=1.0, active=False)
gr.HTML("""""")
hidden_end_btn = gr.Button("hidden-end-consultation", visible=True, elem_id="hidden-end-consultation")
with gr.Column(visible=False) as screen_s4:
processing_html = gr.HTML(_processing_screen_html(1, "Finalising transcript…", "MedASR processing audio", "Elapsed: 00:00"))
processing_tick = gr.Timer(value=1.0, active=False)
hidden_cancel_button = gr.Button("hidden-cancel", visible=True, elem_id="hidden-cancel")
with gr.Column(visible=False) as screen_s5:
gr.HTML("