"""Gradio app. Main UI definition and layout.""" from __future__ import annotations import html import logging from typing import Any import gradio as gr from config import get_app_config from data.preprocessing import ( draw_defects, image_to_data_uri, image_to_png_bytes, load_image, ) from pipeline.pipeline import run_diagnosis from storage.cache import get_cache from storage.database import get_diagnosis, init_db, list_recent, record_diagnosis from ui.components import ( EMPTY_STATE, HEADER_HTML, LIGHTTABLE_EMPTY_STATE, REPORT_EMPTY_STATE, comparison_viewer_html, confidence_notice_html, defect_table_rows, defect_pills_html, diagnosis_html, history_choices, history_detail_html, history_table_rows, metadata_html, raw_json_text, review_frame_html, run_state_html, stats_html, ) logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) MAX_ZEROGPU_DURATION_SECONDS = 120 def _gpu_duration_seconds() -> int: return min( max(1, int(get_app_config().gpu_duration_seconds)), MAX_ZEROGPU_DURATION_SECONDS, ) def _gpu_decorator(): try: import spaces except ImportError: return lambda fn: fn return spaces.GPU(duration=_gpu_duration_seconds()) DEFAULT_FILM_TYPES = [ "Unknown / Not specified", "Kodak Portra 400 (35mm)", "Kodak Tri-X 400 (35mm)", "Kodak Ektar 100 (35mm)", "Ilford HP5 Plus (35mm)", "Ilford Delta 100 (35mm)", "Ilford FP4 Plus (120)", "CineStill 800T (35mm)", "Fujifilm Pro 400H (35mm)", "Fomapan 400 (35mm)", "Other / Unknown", ] STORAGE_OPTIONS = [ "unknown", "fridge, sealed", "freezer, sealed", "room temp, sealed", "room temp, loose", "shoe box, attic", "shoe box, basement", ] RESOLUTION_OPTIONS = [2000, 3000, 4000, 5000, 6000, 8000] METADATA_CONFIDENCE_OPTIONS = [ "Low, rough guess", "Medium, partly verified", "High, verified from notes or edge marks", ] PipelineOutputs = tuple[ Any, Any, Any, Any, str, str, str, str, str, str, str, list[list[str]], str, Any, list[list[str]], ] def normalize_metadata_confidence(value: str | None) -> str: text = (value or "low").strip().lower() if text.startswith("high"): return "high" if text.startswith("medium"): return "medium" return "low" def _history_state( selected_id: str | None = None, ) -> tuple[dict | None, Any, list[list[str]]]: entries = list_recent(limit=get_app_config().max_history_items) choices = history_choices(entries) ids = [value for _label, value in choices] value = selected_id if selected_id in ids else (ids[0] if ids else None) selected = next((entry for entry in entries if entry.get("id") == value), None) return selected, gr.update(choices=choices, value=value), history_table_rows(entries) def _empty_outputs( message: str = "Awaiting scan.", ) -> PipelineOutputs: selected_entry, selector_update, history_rows = _history_state() empty = f'

{html.escape(message)}

' hidden_html = gr.update(value="", visible=False) return ( gr.update(value=LIGHTTABLE_EMPTY_STATE, visible=True), gr.update(value=None, visible=False), gr.update(value=[], visible=False), hidden_html, run_state_html(None), empty, empty, hidden_html, hidden_html, "", "{}", [], history_detail_html(selected_entry), selector_update, history_rows, ) def _review_gallery(pil_image: Any, annotated: Any) -> list[tuple[Any, str]]: return [ (pil_image, "Original scan"), (annotated, "Validated overlay"), ] def _attach_preview(result: dict, pil_image: Any, annotated: Any) -> dict: result = dict(result) if not isinstance(result.get("preview"), dict): result["preview"] = { "original": image_to_data_uri(pil_image, max_side=720, quality=86), "overlay": image_to_data_uri(annotated, max_side=720, quality=86), } return result def pipeline_error_html(exc: Exception) -> str: text = str(exc) lower = text.lower() if "no cuda gpu" in lower or "cuda" in lower or "gpu" in lower: title = "GPU unavailable" body = ( "The diagnosis needs a live GPU slot. Please retry in a moment, " "or run the app on a GPU-backed Space." ) else: title = "Pipeline error" body = text or "The diagnostic pipeline stopped unexpectedly." return ( '
' f'
' f"{html.escape(title)}
" f"

{html.escape(body)}

" ) @_gpu_decorator() def run_pipeline( image: Any, film_type: str, film_age_years: int, storage: str, scan_dpi: int, metadata_confidence: str = "low", progress: gr.Progress = gr.Progress(), ) -> PipelineOutputs: """Gradio handler for the diagnose button.""" if image is None: return _empty_outputs("No image provided.") try: progress(0.0, "Hashing image for cache lookup...") pil_image = load_image(image) cache = get_cache() image_bytes = image_to_png_bytes(pil_image) metadata = { "film_type": film_type or "Unknown / Not specified", "film_age_years": int(film_age_years or 0), "storage": storage or "unknown", "scan_resolution_dpi": int(scan_dpi or 4000), "metadata_confidence": normalize_metadata_confidence(metadata_confidence), } cached = cache.get(image_bytes, metadata=metadata) was_cached = cached is not None if cached is not None: logger.info("Returning cached diagnosis") result = cached else: progress(0.05, "Loading GPU models if needed...") progress(0.1, "Stage 1/2: running vision defect extraction...") result = run_diagnosis( image=pil_image, film_type=metadata["film_type"], film_age_years=metadata["film_age_years"], storage=metadata["storage"], scan_resolution_dpi=metadata["scan_resolution_dpi"], metadata_confidence=metadata["metadata_confidence"], ) progress(0.85, "Stage 2/2: persisting diagnosis...") progress(1.0, "Done.") counts = result.get("defects", {}).get("label_counts", {}) or {} defects = result.get("defects", {}).get("defects", []) or [] annotated = draw_defects(pil_image, defects) result = _attach_preview(result, pil_image, annotated) if not was_cached: try: diagnosis_id = record_diagnosis(result) result["diagnosis_id"] = diagnosis_id except Exception as exc: # pragma: no cover logger.warning("Failed to record diagnosis: %s", exc) cache.put(image_bytes, result, metadata=metadata) elif not result.get("diagnosis_id"): cache.put(image_bytes, result, metadata=metadata) compare = gr.update( value=comparison_viewer_html(pil_image, annotated), visible=True, ) gallery = gr.update(value=_review_gallery(pil_image, annotated), visible=True) review_links = gr.update( value=review_frame_html(pil_image, annotated), visible=True, ) run_state = run_state_html(result) stats = stats_html(result) notice = confidence_notice_html(result) pills = gr.update(value=defect_pills_html(counts), visible=True) diag = gr.update( value=diagnosis_html(result.get("diagnosis", {}).get("diagnosis_text", "")), visible=True, ) meta = metadata_html(result) raw_json = raw_json_text(result) table_rows = defect_table_rows(result) selected_entry, selector_update, history_rows = _history_state(result.get("diagnosis_id")) return ( gr.update(value="", visible=False), compare, gallery, review_links, run_state, stats, notice, pills, diag, meta, raw_json, table_rows, history_detail_html(selected_entry), selector_update, history_rows, ) except Exception as exc: # pragma: no cover logger.exception("Pipeline failed") err = pipeline_error_html(exc) selected_entry, selector_update, history_rows = _history_state() hidden_html = gr.update(value="", visible=False) return ( gr.update(value=LIGHTTABLE_EMPTY_STATE, visible=True), gr.update(value=None, visible=False), gr.update(value=[], visible=False), hidden_html, err, err, "", hidden_html, hidden_html, "", "{}", [], history_detail_html(selected_entry), selector_update, history_rows, ) def refresh_history(selected_id: str | None = None) -> tuple[Any, str, str, list[list[str]]]: selected_entry, selector_update, history_rows = _history_state(selected_id) return ( selector_update, history_detail_html(selected_entry), raw_json_text(selected_entry), history_rows, ) def open_history(diagnosis_id: str | None) -> tuple[str, str]: entry = get_diagnosis(diagnosis_id or "") if diagnosis_id else None return history_detail_html(entry), raw_json_text(entry) def _history_id_from_selection(rows: list[list[str]] | None, index: Any) -> str | None: row_index: int | None = None if isinstance(index, (list, tuple)) and index: try: row_index = int(index[0]) except (TypeError, ValueError): row_index = None elif isinstance(index, int): row_index = index if row_index is None or row_index < 0: return None entries = list_recent(limit=get_app_config().max_history_items) if row_index >= len(entries): return None diagnosis_id = str(entries[row_index].get("id") or "").strip() return diagnosis_id or None def open_history_from_table(rows: list[list[str]] | None, evt: gr.SelectData) -> tuple[Any, str, str]: diagnosis_id = _history_id_from_selection(rows, getattr(evt, "index", None)) entry = get_diagnosis(diagnosis_id or "") if diagnosis_id else None return gr.update(value=diagnosis_id), history_detail_html(entry), raw_json_text(entry) def build_app() -> gr.Blocks: init_db() selected_entry, selector_update, initial_history_rows = _history_state() initial_choices = selector_update["choices"] if isinstance(selector_update, dict) else [] initial_value = selector_update["value"] if isinstance(selector_update, dict) else None with gr.Blocks( title="Project Halide", fill_width=True, fill_height=True, elem_id="halide-app", ) as app: gr.HTML(HEADER_HTML) with gr.Row(elem_classes="halide-workbench", equal_height=False): with gr.Column(scale=2, min_width=300, elem_classes="halide-intake-panel"): gr.HTML( '
Scan intake
' '

Metadata is context, visible evidence is primary.

' ) image_input = gr.Image( label="Film scan", type="pil", height=330, sources=["upload", "clipboard"], buttons=["download", "fullscreen"], elem_classes="halide-upload", ) film_type = gr.Dropdown( choices=DEFAULT_FILM_TYPES, value=DEFAULT_FILM_TYPES[0], label="Film stock", allow_custom_value=True, ) film_age = gr.Slider( minimum=0, maximum=80, step=1, value=0, label="Age (years)", buttons=["reset"], ) scan_dpi = gr.Dropdown( choices=RESOLUTION_OPTIONS, value=4000, label="DPI", allow_custom_value=True, ) storage = gr.Radio( choices=STORAGE_OPTIONS, value=STORAGE_OPTIONS[0], label="Storage", ) metadata_confidence = gr.Dropdown( choices=METADATA_CONFIDENCE_OPTIONS, value=METADATA_CONFIDENCE_OPTIONS[0], label="Metadata confidence", interactive=True, ) run_btn = gr.Button( "Diagnose scan", variant="primary", size="lg", elem_id="halide-run-button", ) gr.HTML( '
' 'Runtime' 'Open weights, GPU only' '

MiniCPM-V extracts evidence. Nemotron writes the lab report.

' "
" ) with gr.Column(scale=6, min_width=560, elem_classes="halide-main-stage"): run_state_output = gr.HTML(value=run_state_html(None)) with gr.Group(elem_classes="halide-lighttable"): gr.HTML( '
' '
Light tableOriginal versus validated overlay
' 'Review' "
" ) lighttable_empty = gr.HTML(value=LIGHTTABLE_EMPTY_STATE) compare_output = gr.HTML( value="", elem_id="halide-compare", visible=False, ) review_gallery = gr.Gallery( value=[], label="Review frames", columns=2, rows=1, height=220, allow_preview=True, object_fit="contain", buttons=["download", "fullscreen"], elem_classes="halide-review-gallery", visible=False, ) review_links_output = gr.HTML(value="", visible=False) with gr.Column(scale=3, min_width=390, elem_classes="halide-inspector"): with gr.Tabs(selected="report", elem_classes="halide-inspector-tabs"): with gr.Tab("Report", id="report"): notice_output = gr.HTML(value=REPORT_EMPTY_STATE) defect_summary = gr.HTML(value="", visible=False) diagnosis_output = gr.HTML(value="", visible=False) with gr.Tab("Evidence", id="evidence"): stats_output = gr.HTML(value=EMPTY_STATE) metadata_output = gr.HTML(value=EMPTY_STATE) defect_table = gr.Dataframe( value=[], headers=["#", "Label", "Confidence", "Box"], datatype=["str", "str", "str", "str"], type="array", label="Validated boxes", interactive=False, wrap=True, max_height=300, ) with gr.Tab("History", id="history"): gr.HTML( '
Select a row or choose a saved run.
' ) history_select = gr.Dropdown( choices=initial_choices, value=initial_value, label="Saved diagnosis", interactive=True, ) with gr.Row(elem_classes="halide-history-actions"): open_history_btn = gr.Button("Open selected", size="sm") refresh_btn = gr.Button("Refresh", size="sm") history_table = gr.Dataframe( value=initial_history_rows, headers=["Saved", "Film stock", "Defects", "Labels"], datatype=["str", "str", "str", "str"], type="array", label="Recent diagnoses", interactive=False, wrap=True, max_height=260, elem_classes="halide-history-table", ) history_detail = gr.HTML( value=history_detail_html(selected_entry) ) with gr.Tab("JSON", id="json"): raw_output = gr.Code( value="{}", language="json", label="Pipeline JSON", lines=18, max_lines=30, wrap_lines=True, buttons=["copy", "download"], ) gr.HTML( '' ) run_btn.click( fn=run_pipeline, inputs=[ image_input, film_type, film_age, storage, scan_dpi, metadata_confidence, ], outputs=[ lighttable_empty, compare_output, review_gallery, review_links_output, run_state_output, stats_output, notice_output, defect_summary, diagnosis_output, metadata_output, raw_output, defect_table, history_detail, history_select, history_table, ], api_name="run_pipeline", ) refresh_btn.click( fn=refresh_history, inputs=[history_select], outputs=[history_select, history_detail, raw_output, history_table], ) open_history_btn.click( fn=open_history, inputs=[history_select], outputs=[history_detail, raw_output], ) history_select.change( fn=open_history, inputs=[history_select], outputs=[history_detail, raw_output], ) history_table.select( fn=open_history_from_table, inputs=[history_table], outputs=[history_select, history_detail, raw_output], ) return app