""" Gradio web interface for the Thyroid Nodule Analysis Pipeline. Supports two analysis modes: Single image - upload one ultrasound image and run the full pipeline. Patient mode - upload multiple images of the same patient; results are shown per-image and then aggregated using WMV + TI-RADS feature aggregation. File validation: Only images are accepted (.png, .jpg, .jpeg). Invalid formats produce a clear error message before any model inference. Partial-failure handling: Each pipeline step reports its status; if an upstream step fails, the interface still shows results from all steps that succeeded. """ import os # ZeroGPU _ON_HF = os.environ.get("SPACE_ID") is not None if _ON_HF: import spaces # type: ignore import time import numpy as np from PIL import Image import gradio as gr from config import ( DEVICE, ACCEPTED_EXTENSIONS, DISPLAY_FEATURES, FEATURE_LABELS, FEATURE_CATEGORY, TIRADS_RISK, TIRADS_FNAB, ) from pipeline import ThyroidPipeline, ImageResult, PatientResult # Initialise pipeline (models loaded once at startup) print(f"[App] Starting Thyroid Pipeline — device: {DEVICE}") pipeline = ThyroidPipeline() print("[App] Pipeline ready.") # Input validation ACCEPTED_EXT_STR = ", ".join(sorted(ACCEPTED_EXTENSIONS)) def validate_images(files: list) -> tuple[bool, str]: """ Check that all uploaded files are valid image formats. Parameters files : list - Gradio file objects (have a .name attribute with the path) Returns (is_valid, error_message) """ if not files: return False, "No files uploaded. Please upload at least one ultrasound image." for f in files: path = f.name if hasattr(f, "name") else str(f) ext = os.path.splitext(path)[-1].lower() if ext not in ACCEPTED_EXTENSIONS: return False, ( f"Unsupported file format: '{os.path.basename(path)}' ({ext}). " f"Accepted formats: {ACCEPTED_EXT_STR}. " "Please upload a valid ultrasound image." ) # Try to open with PIL to catch corrupted files early try: img = Image.open(path) img.verify() except Exception as e: return False, ( f"File '{os.path.basename(path)}' could not be read as an image. " f"It may be corrupted or have an incorrect extension. ({e})" ) return True, "" # HTML result builders def _error_html(title: str, message: str) -> str: return f"""

{title}

{message}

""" def _warning_html(title: str, message: str) -> str: return f"""

{title}

{message}

""" def _status_section_html(result: ImageResult) -> str: """Generate a warning/error banner if the pipeline did not complete fully.""" if result.status == "OK": return "" if result.status == "NO_DETECTION": return _error_html("No Nodule Detected", result.status_message) if result.status == "SEG_FAILED": return _warning_html("Segmentation Failed", result.status_message) if result.status == "CLASSIF_FAILED": return _warning_html("Classification Failed", result.status_message) return _warning_html("Pipeline Warning", result.status_message) def _image_result_html(result: ImageResult, idx: int | None = None) -> str: """Build the full HTML summary for a single ImageResult.""" header = f"Image {idx}" if idx is not None else "Result" status_block = _status_section_html(result) if result.status == "NO_DETECTION": return status_block # Detection section (always present if boxes were found) detection_html = "" if result.n_boxes > 0: detection_html = f"""

Detection (YOLOv8s)

""" if result.status in ("SEG_FAILED", "CLASSIF_FAILED"): seg_html = "" if result.status == "CLASSIF_FAILED" and result.nodule_area_px is not None: seg_html = f"""

Segmentation (UNet++ · ResNet50 encoder · SCSE)

""" return f"""

{header}

{status_block} {detection_html} {seg_html}

Elapsed: {result.elapsed_s:.2f}s

""" # Full result (status == "OK") # NOTE: ACR TI-RADS score is intentionally NOT shown here. # TI-RADS is a patient-level score computed after aggregating radiomic # features across ALL images. It is displayed only in the Patient Result # section above. Showing a per-image TR score would be clinically incorrect. label_color = "#c0392b" if result.label == "MALIGNANT" else "#27ae60" return f"""

{result.label}


{detection_html}

Segmentation (UNet++ · ResNet50 encoder · SCSE)

Classification (ResNet50)

Processing time: {result.elapsed_s:.2f}s  |  TI-RADS score is shown in the Patient Result above (aggregated across all images).

""" def _patient_summary_html(patient: PatientResult) -> str: """Build the aggregated patient-level summary HTML block.""" has_classification = patient.final_label in ("MALIGNANT", "BENIGN") if patient.final_label == "MALIGNANT": label_color = "#c0392b" result_title = "Patient Result: MALIGNANT" elif patient.final_label == "BENIGN": label_color = "#27ae60" result_title = "Patient Result: BENIGN" else: label_color = "#6b7280" result_title = "Patient Result: Classification unavailable" prob_text = ( f"{patient.aggregated_prob:.1%}" if has_classification and patient.aggregated_prob is not None else "N/A" ) tr_class = patient.tirads_class tirads_block = f""" Class TR{tr_class} Risk interpretation {patient.tirads_risk} FNA biopsy recommendation {patient.fnab_recommendation} """ if tr_class is not None else """ Not enough data for TI-RADS scoring """ return f"""

{result_title}

Aggregated malignancy probability: {prob_text}  |  {patient.n_images_processed} / {patient.n_images_submitted} images classified  |  {patient.elapsed_total_s:.2f}s total


ACR TI-RADS Score (patient-level, aggregated features)

{tirads_block}
""" def _features_html(features: dict) -> str: """ Build the top-10 radiomic features HTML table. Parameters features : dict - feature values (aggregated patient-level vector) """ rows = "" for i, feat in enumerate(DISPLAY_FEATURES): val = features.get(feat, float("nan")) val_str = f"{val:.4f}" if not (isinstance(val, float) and np.isnan(val)) else "N/A" bg = "#f5f5f5" if i % 2 == 0 else "#ffffff" rows += f""" {FEATURE_LABELS[feat]} {FEATURE_CATEGORY[feat]} {val_str} """ return f""" {rows}
Feature TI-RADS Category Value
""" # Main pipeline callback (called by Gradio) def analyze_images(files: list) -> tuple: """ Entry point called by the Gradio Analyze button. Parameters files : list of Gradio file objects Returns 7-tuple matching Gradio output components: (patient_summary_html, image_gallery, # list of PIL images for gr.Gallery per_image_html, # per-image result accordion HTML features_html, # radiomic features table (best image or aggregated) status_html, # top-level status / error banner timer_html) # processing time summary """ EMPTY = (None, [], "", "", "", "") is_valid, err_msg = validate_images(files) if not is_valid: error_block = _error_html("Invalid Input", err_msg) return error_block, [], "", "", error_block, "" pil_images = [] for f in files: path = f.name if hasattr(f, "name") else str(f) try: img = Image.open(path).convert("RGB") pil_images.append(img) except Exception as e: error_block = _error_html( "Image Load Error", f"Could not load '{os.path.basename(path)}': {e}" ) return error_block, [], "", "", error_block, "" # Run patient pipeline patient: PatientResult = pipeline.run_patient(pil_images) # Build gallery images # Gallery contains: for each image - original, detection, segmentation, crop, gradcam (None slots are skipped) gallery_images = [] for idx, res in enumerate(patient.per_image_results, start=1): label = f"Img {idx}" if res.image_original is not None: gallery_images.append((res.image_original, f"{label} - Original")) if res.image_detection is not None: gallery_images.append((res.image_detection, f"{label} - Detection (YOLO)")) if res.image_segmentation is not None: gallery_images.append((res.image_segmentation, f"{label} - Segmentation (UNet++)")) if res.image_crop is not None: gallery_images.append((res.image_crop, f"{label} - Nodule ROI (ResNet50)")) if res.image_gradcam is not None: gallery_images.append((res.image_gradcam, f"{label} - Grad-CAM")) # Per-image results HTML per_img_parts = [] for idx, res in enumerate(patient.per_image_results, start=1): per_img_parts.append( f"

Image {idx}

" + _image_result_html(res) ) per_image_html = "".join(per_img_parts) if per_img_parts else "" # Patient summary HTML all_statuses = [res.status for res in patient.per_image_results] if patient.n_images_processed == 0 and not patient.aggregated_features: has_classif_failed = any(s == "CLASSIF_FAILED" for s in all_statuses) has_seg_failed = any(s == "SEG_FAILED" for s in all_statuses) has_any_detection = any(s != "NO_DETECTION" for s in all_statuses) if has_classif_failed: patient_summary = _error_html( "Classification and TI-RADS Scoring Unavailable", "The nodule was detected and segmented, but the segmented mask is too " "small to produce a valid ResNet50 classification or TI-RADS score. " "This typically occurs when the nodule occupies less than 0.5% of the " "image area. Detection and segmentation outputs are shown in the gallery below." ) elif has_seg_failed: patient_summary = _error_html( "Segmentation Failed", "A nodule was detected by YOLO but UNet++ could not produce a valid " "segmentation mask. Classification and TI-RADS scoring were not performed. " "The detection result is shown in the gallery below." ) elif has_any_detection: patient_summary = _error_html( "Pipeline Incomplete", "Detection succeeded but the pipeline could not complete classification " "or TI-RADS scoring. See the per-image details below for the specific error." ) else: patient_summary = _error_html( "No Nodule Detected", "None of the uploaded images produced a valid detection. " "Please verify that the images are thyroid ultrasound scans." ) else: patient_summary = _patient_summary_html(patient) # Radiomic features table # Show the AGGREGATED feature vector that was passed to the Random Forest, # not raw features from a single image. This is the patient-level vector # produced by PatientAggregator (MAX/MIN/MEAN across all images), which # is exactly what determined the TI-RADS score shown above. if patient.aggregated_features: feats_html = _features_html(patient.aggregated_features) else: feats_html = "" # Timer banner timer_html = ( f"

Total processing time: " f"{patient.elapsed_total_s:.2f}s " f"| Device: {str(DEVICE).upper()} " f"| Images classified: {patient.n_images_processed}" f"/{patient.n_images_submitted}

" ) return ( patient_summary, gallery_images, per_image_html, feats_html, "", timer_html, ) # Gradio interface css = """ .main-title { text-align: center !important; } h1 { text-align: center !important; } footer { display: none !important; } .thumbnails button[aria-label="Share"], .gallery button[aria-label="Share"], .icon-button-wrapper button[aria-label="Share"], button[aria-label="Share"] { display: none !important; } #file-upload-box .wrap.svelte-1vmd51o > .svelte-8prmba:not(svg):not(button) { display: none !important; } .gallery caption, .gallery .caption, [class*="caption"], [class*="thumbnail"] span, .thumbnails span, .gallery-item span, figure figcaption { font-family: ui-sans-serif, system-ui, sans-serif !important; font-style: normal !important; font-weight: 400 !important; font-size: 13px !important; border-top: none !important; padding-top: 4px !important; } #gallery-no-scroll, #gallery-no-scroll .grid-wrap, #gallery-no-scroll .grid-container { max-height: none !important; height: auto !important; overflow: visible !important; } """ with gr.Blocks(title="Thyroid Nodule Analysis Pipeline") as demo: # Header gr.Markdown("""
# Thyroid Nodule Analysis Pipeline **Automated detection · segmentation · classification · ACR TI-RADS scoring**
""") # Input row with gr.Row(): with gr.Column(scale=1): gr.Markdown( "Upload one or more ultrasound images of the same patient, " "then click **Analyze**." ) input_files = gr.File( label = "Upload ultrasound image(s) - single image or multiple views of the same patient", file_count = "multiple", file_types = [".png", ".jpg", ".jpeg"], elem_id = "file-upload-box", ) run_btn = gr.Button("Analyze", variant="primary", size="lg") timer_out = gr.HTML(label="") with gr.Column(scale=1): status_out = gr.HTML(label="") patient_out = gr.HTML(label="Patient Result") gr.Markdown("---") # Image gallery gr.Markdown("### Visual Pipeline Outputs") gr.HTML( "

" "Gallery shows: Original → Detection → Segmentation → " "Nodule ROI → Grad-CAM, for each uploaded image.

" ) gallery_out = gr.Gallery( label = "Pipeline outputs", show_label = True, columns = 5, height = "auto", object_fit = "contain", elem_id = "gallery-no-scroll", ) gr.Markdown("---") gr.Markdown("#### Per-image pipeline details") per_image_out = gr.HTML(label="") gr.Markdown("---") # TI-RADS features table gr.Markdown("### TI-RADS Radiomic Features") features_out = gr.HTML(label="") # Button wiring run_btn.click( fn = analyze_images, inputs = [input_files], outputs = [ patient_out, gallery_out, per_image_out, features_out, status_out, timer_out, ], ) # Launch if __name__ == "__main__": demo.launch( theme = gr.themes.Soft(), css = css, server_name = "0.0.0.0", share = False, )