from dotenv import load_dotenv load_dotenv() # noqa: E402 — must precede all LangChain/LangSmith imports from pathlib import Path # noqa: E402 import gradio as gr # noqa: E402 from adapters.modal_chat_model import ModalChatModel # noqa: E402 from core.bind import ( # noqa: E402 list_kobo_sheets, parse_kobo, kobo_summary, run_bind_step, suggest_sheets, ) from core.route import resolve_route, step1_trace_line # noqa: E402 from core.schemas import BindResponse # noqa: E402 from core.select import select_step # noqa: E402 from core.session import load_state, new_session, write_state # noqa: E402 from core.spec import assemble_spec_yaml, render_spec_md # noqa: E402 from ui.components import ( # noqa: E402 PATH_MESSAGE, make_lens_group, make_path_radio, make_sector_group, make_sector_group_no_catalog, make_type_group, ) _DEMO_XLSX = Path(__file__).parent / "demo_data" / "vasyr_2023.xlsx" def _trace_md(trace: list[str]) -> str: """Join trace beats with Markdown hard breaks so each renders on its own line. Gradio's Markdown collapses a single '\\n' into a space; two trailing spaces before the newline force a hard break. """ return " \n".join(trace) def run_pipeline( path_value, question_v, sectors_v, sectors_no_cat_v, lens_v, type_v, session_id, kobo_file_path, demo_mode, survey_sheet, choices_sheet, ): """Page-2 generator: stream route -> select -> mapping, then advance to page 3. Yields a 7-tuple: (page1_update, page2_update, page3_update, trace_md, spec_md, session_id, spec_download) """ sectors_v = (sectors_v or []) + (sectors_no_cat_v or []) if not sectors_v and path_value != PATH_MESSAGE: yield (gr.update(), gr.update(), gr.update(), "⚠ Select at least one sector.", gr.update(), session_id, gr.update()) return trace: list[str] = [] # Switch to page 2 and start the route beat. trace.append("● Routing…") yield ( gr.update(visible=False), gr.update(visible=True), gr.update(visible=False), _trace_md(trace), gr.update(), session_id, gr.update(), ) # Beat 1 — routing (REAL: Phase C's resolve_route). input_mode = "message" if path_value == PATH_MESSAGE else "chips" model = ModalChatModel() if input_mode == "message" else None try: res = resolve_route( input_mode, question_v or "", sectors_v or [], lens_v or [], type_v or [], model, ) except (ValueError, Exception) as exc: trace[-1] = f"● Routing… ✗ {exc}" yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) return in_scope = res.in_scope session_id = new_session(input_mode, question_v or "") write_state(session_id, { "current_step": 1, "route": res.route.model_dump(), "route_scope": {"in_scope": in_scope, "out_of_scope": res.out_of_scope}, "trace": [step1_trace_line(res)], }) trace[-1] = f"● Routing… ✓ {', '.join(in_scope) or '(none in scope)'}" yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) if not in_scope: trace.append("⚠ No in-scope sectors — nothing to select.") yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) return # Beat 2 — select (REAL: Phase D's select_step). trace.append("● Selecting indicators…") yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) try: result = select_step(in_scope) except Exception as exc: # noqa: BLE001 trace[-1] = f"● Selecting indicators… ✗ {exc}" yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) return trace[-1] = f"● Selecting indicators… ✓ {len(result.selected_ids)} found" write_state(session_id, { "current_step": 2, "selected_ids": result.selected_ids, "indicator_defs": result.indicator_defs, }) yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) # Beat 3 — parse Kobo instrument if demo_mode: xlsx_path = _DEMO_XLSX elif kobo_file_path is None: xlsx_path = None elif hasattr(kobo_file_path, "path"): xlsx_path = Path(kobo_file_path.path) elif isinstance(kobo_file_path, dict): xlsx_path = Path(kobo_file_path.get("path") or kobo_file_path.get("name", "")) else: xlsx_path = Path(str(kobo_file_path)) if xlsx_path is None: trace.append("⚠ No XLSForm uploaded — bind loop skipped.") stub_md = "## Analysis spec\n\n*Upload a Kobo XLSForm and re-run to generate the spec.*" yield ( gr.update(visible=False), gr.update(visible=False), gr.update(visible=True), _trace_md(trace), stub_md, session_id, gr.update(), ) return slug = xlsx_path.stem.lower().replace(" ", "_") session_dir = Path("sessions") / session_id trace.append("● Parsing instrument…") yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) try: cache_path = parse_kobo( xlsx_path, slug, session_dir / f"kobo_{slug}.json", survey_sheet=survey_sheet, choices_sheet=choices_sheet, ) except Exception as exc: # noqa: BLE001 trace[-1] = f"● Parsing instrument… ✗ {exc}" yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) return trace[-1] = f"● Parsing instrument… ✓ {slug}" summary_map = kobo_summary(cache_path) # Beat 4 — per-indicator bind loop (streams one line per verdict) trace.append("Step 3 — Mapping indicators to the dataset:") yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) bind_results: dict = {} model = ModalChatModel() verdict_labels = { "MEASURABLE": "Measurable", "PROXY": "Proxy", "NOT_MEASURABLE": "Not measurable", } for ind_id in result.selected_ids: ind_def = result.indicator_defs.get(ind_id, {}) try: bind_res = run_bind_step(ind_id, ind_def, cache_path, summary_map, model) except Exception as exc: # noqa: BLE001 bind_res = BindResponse( indicator_id=ind_id, variables=[], measurable="NOT_MEASURABLE", reasons=f"Bind error: {exc}", result_ids=[], ) bind_results[ind_id] = bind_res.model_dump() write_state(session_id, {"bind_results": {ind_id: bind_res.model_dump()}}) vars_str = ", ".join(bind_res.variables) or "(no matching variable)" verdict = verdict_labels[bind_res.measurable] trace.append(f" {ind_def.get('label', ind_id)} → {vars_str} … {verdict}") yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) # Beat 5 — assemble YAML spec + render MD trace.append("● Assembling analysis spec…") yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) try: state = load_state(session_id) spec_yaml = assemble_spec_yaml( session_dir, slug, state["route"], bind_results, result.indicator_defs, ) spec_md_path = render_spec_md(spec_yaml, session_dir) write_state(session_id, { "current_step": 3, "artifacts": {"spec_yaml": str(spec_yaml), "spec_md": str(spec_md_path)}, }) trace[-1] = "● Assembling analysis spec… ✓" spec_md_content = spec_md_path.read_text() except Exception as exc: # noqa: BLE001 trace[-1] = f"● Assembling analysis spec… ✗ {exc}" yield (gr.update(), gr.update(), gr.update(), _trace_md(trace), gr.update(), session_id, gr.update()) return yield ( gr.update(visible=False), gr.update(visible=False), gr.update(visible=True), _trace_md(trace), spec_md_content, session_id, gr.update(value=str(spec_md_path), visible=True), ) _SPEC_TABLE_CSS = """ """ def build_demo() -> gr.Blocks: with gr.Blocks(title="Analyste") as demo: gr.HTML(_SPEC_TABLE_CSS) session_id = gr.State("") # ---- Page 1: input ---- with gr.Group(visible=True) as page1: gr.Markdown( "## Analyste\n" "*It reads a Kobo survey and writes the spec for its own analysis — " "telling you what it can and can't claim.*" ) path = make_path_radio() with gr.Group(visible=True) as message_group: question = gr.Textbox( label="Your question", placeholder="e.g. What does the food security data tell us about Aleppo?", lines=2, ) with gr.Group(visible=False) as chips_group: sectors = make_sector_group() sectors_no_cat = make_sector_group_no_catalog() lens = make_lens_group() type_ = make_type_group() gr.Markdown("---") gr.Markdown("**Choose a dataset**") with gr.Row(equal_height=True): with gr.Column(): kobo_file = gr.File( label="Upload Kobo XLSForm (.xlsx)", file_types=[".xlsx"], height=140, ) with gr.Column(): demo_btn = gr.Button( "📂 Use demo dataset (VASyR 2023)", size="lg", variant="secondary", ) demo_note = gr.Markdown(visible=False) with gr.Group(visible=False) as sheet_group: with gr.Row(): survey_sheet_dd = gr.Dropdown(label="Survey sheet", interactive=True) choices_sheet_dd = gr.Dropdown(label="Choices sheet", interactive=True) run_btn = gr.Button("Run", variant="primary") demo_mode = gr.State(False) # ---- Page 2: work in progress (Phase D — streaming pipeline) ---- with gr.Group(visible=False) as page2: gr.Markdown("## 2. Work in progress") trace_md = gr.Markdown("") # ---- Page 3: spec (STUB — Phase E renders render_spec.py output) ---- with gr.Group(visible=False) as page3: gr.Markdown("## 3. Analysis spec") spec_md = gr.Markdown("", elem_id="spec-viewer") spec_download = gr.File(label="Download spec (.md)", visible=False) gr.Markdown("---") gr.Button( "Validate → start analysis", interactive=False, variant="secondary", ) gr.Markdown( "### 4. Analysis — *not built in this demo*\n" "*This demo stops at the certified analysis spec. Running the numbers " "is the next stage (out of scope here).*", ) def _toggle(path_value): is_msg = path_value == PATH_MESSAGE return gr.update(visible=is_msg), gr.update(visible=not is_msg) path.change(_toggle, inputs=path, outputs=[message_group, chips_group]) def _detect_sheets(xlsx_path_val, is_demo: bool = False): try: if is_demo: path_to_check = _DEMO_XLSX elif hasattr(xlsx_path_val, "path"): # Gradio 5+/6+ SSR FileData object path_to_check = Path(xlsx_path_val.path) elif isinstance(xlsx_path_val, dict): path_to_check = Path(xlsx_path_val.get("path") or xlsx_path_val.get("name", "")) else: path_to_check = Path(str(xlsx_path_val)) sheets = list_kobo_sheets(path_to_check) survey, choices = suggest_sheets(sheets) return ( is_demo, gr.update(value="✓ VASyR 2023 loaded" if is_demo else "", visible=is_demo), gr.update(visible=True), gr.update(choices=sheets, value=survey), gr.update(choices=sheets, value=choices), ) except Exception: # noqa: BLE001 return ( is_demo, gr.update(visible=False), gr.update(visible=False), gr.update(choices=[], value=None), gr.update(choices=[], value=None), ) _sheet_outputs = [demo_mode, demo_note, sheet_group, survey_sheet_dd, choices_sheet_dd] kobo_file.upload( lambda f: _detect_sheets(f, is_demo=False), inputs=[kobo_file], outputs=_sheet_outputs, ) demo_btn.click( lambda: _detect_sheets(None, is_demo=True), outputs=_sheet_outputs, ) run_btn.click( run_pipeline, inputs=[path, question, sectors, sectors_no_cat, lens, type_, session_id, kobo_file, demo_mode, survey_sheet_dd, choices_sheet_dd], outputs=[page1, page2, page3, trace_md, spec_md, session_id, spec_download], ) return demo demo = build_demo()