from __future__ import annotations import json import os from pathlib import Path from typing import Any from .demo_pack import ingest_demo_pack, list_demo_packs, load_index, store_uploaded_manual from .storage import DbPaths, connect, init_db, reset_db, DEFAULT_DATA_DIR, DEFAULT_DB_PATH, DEFAULT_ARTIFACTS_DIR, DEFAULT_DEMO_PACKS_DIR from .reasoning import build_response as shared_build_response from .tracing import utc_now, write_trace_artifact APP_TITLE = "P3 Off-Grid Field Repair Logbook" PACK_ROOT = DEFAULT_DEMO_PACKS_DIR DATA_DIR = DEFAULT_DATA_DIR DB_PATH = DEFAULT_DB_PATH THEME_CSS_PATH = Path(__file__).resolve().parents[1] / "assets" / "theme.css" def _pack_update(*, choices, value=None): try: import gradio as gr except ModuleNotFoundError: return {'choices': choices, 'value': value} return gr.update(choices=choices, value=value) def _ensure_bootstrap() -> None: init_db(DB_PATH) def _pack_choices() -> list[str]: packs = list_demo_packs(PACK_ROOT) return [p.name for p in packs] def _load_index(): return load_index(DB_PATH) def _format_manuals_table() -> list[list[str]]: with connect(DB_PATH) as conn: rows = conn.execute( """ SELECT m.id, m.title, m.source_url, m.license_name, COUNT(ms.id) AS sections FROM manuals m LEFT JOIN manual_sections ms ON ms.manual_id = m.id GROUP BY m.id ORDER BY m.id """ ).fetchall() return [[row["id"], row["title"], row["license_name"], row["sections"], row["source_url"]] for row in rows] def _format_sections_table(limit: int = 30) -> list[list[str]]: with connect(DB_PATH) as conn: rows = conn.execute( """ SELECT ms.id, m.title AS manual_title, ms.section_title, ms.section_slug, LENGTH(ms.content) AS chars FROM manual_sections ms JOIN manuals m ON m.id = ms.manual_id ORDER BY ms.id LIMIT ? """, (limit,), ).fetchall() return [[row["id"], row["manual_title"], row["section_title"], row["section_slug"], row["chars"]] for row in rows] def _format_jobs_table(limit: int = 40, query: str = "") -> list[list[str]]: with connect(DB_PATH) as conn: if query.strip(): like = f"%{query.strip()}%" rows = conn.execute( """ SELECT id, created_at, job_title, equipment_type, location, severity, symptom, resolution_status, photo_caption FROM jobs WHERE job_title LIKE ? OR equipment_type LIKE ? OR location LIKE ? OR symptom LIKE ? OR notes LIKE ? ORDER BY id DESC LIMIT ? """, (like, like, like, like, like, limit), ).fetchall() else: rows = conn.execute( """ SELECT id, created_at, job_title, equipment_type, location, severity, symptom, resolution_status, photo_caption FROM jobs ORDER BY id DESC LIMIT ? """, (limit,), ).fetchall() return [[row["id"], row["job_title"], row["equipment_type"], row["location"], row["severity"], row["resolution_status"], row["symptom"][:120]] for row in rows] def _extract_bullets(text: str, max_items: int = 3) -> list[str]: bullets: list[str] = [] for para in text.split("\n\n"): para = para.strip() if not para: continue for line in para.splitlines(): line = line.strip().lstrip("-•*").strip() if len(line) < 24: continue bullets.append(line) if len(bullets) >= max_items: return bullets return bullets def _build_response(symptom: str, equipment_type: str, location: str, notes: str, photo_path: str | None) -> tuple[str, list[list[Any]], dict[str, Any]]: return shared_build_response(symptom, equipment_type, location, notes, photo_path, _load_index()) def load_pack_with_trace( pack_name: str, db_path: str | Path = DB_PATH, artifact_dir: str | Path = DEFAULT_ARTIFACTS_DIR, ) -> tuple[str, list[list[str]], list[list[str]], list[list[str]], Any, dict[str, Any], Path]: global DB_PATH pack_name = pack_name or "" pack_dir = PACK_ROOT / pack_name if pack_name else None choice_list = _pack_choices() original_db_path = DB_PATH DB_PATH = Path(db_path) try: _ensure_bootstrap() if not pack_dir or not pack_dir.exists(): trace_path = write_trace_artifact( artifact_dir, { 'kind': 'app-load', 'status': 'missing_pack', 'pack_name': pack_name, 'db_path': str(DB_PATH), }, ) info = {'pack_name': pack_name, 'trace_path': str(trace_path), 'status': 'missing_pack'} return ( f"⚠️ Example Data not found: **{pack_name}**. Please select valid example data from the dropdown.", _format_manuals_table(), _format_sections_table(), _format_jobs_table(), _pack_update(choices=choice_list, value=pack_name or None), info, trace_path, ) started_at = utc_now() info = ingest_demo_pack(pack_dir, db_path=DB_PATH, reset=True) finished_at = utc_now() trace_path = write_trace_artifact( artifact_dir, { 'kind': 'app-load', 'pack_name': pack_name, 'pack_dir': str(pack_dir), 'db_path': str(DB_PATH), 'started_at': started_at, 'finished_at': finished_at, 'status': 'loaded', 'info': info, }, ) status = ( f"✅ Loaded Example Data: **{info['pack_name']}** — {info['manual_count']} manual(s), " f"{info['job_count']} job(s), {info['photo_count']} photo(s)." ) info = dict(info) info['trace_path'] = str(trace_path) return status, _format_manuals_table(), _format_sections_table(), _format_jobs_table(), _pack_update(choices=choice_list, value=pack_name), info, trace_path finally: DB_PATH = original_db_path def load_pack_ui(pack_name: str) -> tuple[str, list[list[str]], list[list[str]], list[list[str]], Any, dict[str, Any]]: status, manuals, sections, jobs, pack_update, info, _ = load_pack_with_trace(pack_name) return status, manuals, sections, jobs, pack_update, info def submit_job_ui(job_title: str, technician: str, location: str, equipment_type: str, severity: str, symptom: str, notes: str, photo_path: str | None) -> tuple[str, list[list[Any]], list[list[str]], list[list[str]]]: _ensure_bootstrap() with connect(DB_PATH) as conn: from .demo_pack import store_job job_id = store_job( conn, { "title": job_title, "technician": technician, "location": location, "equipment_type": equipment_type, "severity": severity, "symptom": symptom, "notes": notes, "photo": photo_path, "expected_section_ids": [], "expected_section_titles": [], }, pack_dir=None, is_demo=False, ) conn.commit() body, citations_rows, payload = _build_response(symptom, equipment_type, location, notes, photo_path) with connect(DB_PATH) as conn: conn.execute( "UPDATE jobs SET response_json = ?, linked_section_ids = ? WHERE id = ?", (json.dumps(payload), json.dumps(payload.get("retrieved_sections", [])), job_id), ) conn.commit() history_rows = _format_jobs_table() return body + f"\n\nSaved as job #{job_id}.", citations_rows, history_rows, _format_sections_table() def search_history_ui(query: str) -> list[list[str]]: return _format_jobs_table(query=query) def inspect_job_ui(job_id: int | str) -> str: if not str(job_id).strip(): return "Select a job id to inspect." with connect(DB_PATH) as conn: row = conn.execute( """ SELECT * FROM jobs WHERE id = ? """, (int(job_id),), ).fetchone() if not row: return f"Job #{job_id} not found." try: payload = json.loads(row["response_json"]) except Exception: payload = {} lines = [ f"# Job {row['id']}: {row['job_title']}", f"Technician: {row['technician']}", f"Location: {row['location']}", f"Equipment: {row['equipment_type']} ({row['severity']})", f"Symptom: {row['symptom']}", f"Notes: {row['notes']}", "", f"Status: {row['resolution_status']}", f"Photo: {row['photo_caption'] or 'none'}", "", "Retrieved references:", ] for ref in payload.get("retrieved_sections", []): lines.append(f"- {ref}") if row["linked_section_ids"]: lines.append("") lines.append(f"Linked section ids: {row['linked_section_ids']}") return "\n".join(lines) def run_eval_ui() -> str: from .eval import evaluate_pack report = evaluate_pack(PACK_ROOT / "p3_field_repair_logbook", db_path=DB_PATH) return json.dumps(report, indent=2) def _format_pack_info(info: dict) -> str: """Format pack metadata as readable Markdown instead of raw JSON.""" if not info or info.get('status') == 'missing_pack': return "*No example data loaded.*" pack_name = info.get('pack_name', info.get('pack_root', 'Unknown')) manual_count = info.get('manual_count', '?') job_count = info.get('job_count', '?') photo_count = info.get('photo_count', 0) return f"📦 **Example Data:** {pack_name} — {manual_count} manuals · {job_count} jobs · {photo_count} photos" def build_app() -> gr.Blocks: import gradio as gr _ensure_bootstrap() packs = _pack_choices() default_pack = packs[0] if packs else "" with gr.Blocks(title="Field Repair Logbook", css_paths=THEME_CSS_PATH) as demo: gr.Markdown("""# 🔧 Field Repair Logbook Safety-first manual RAG for off-grid diagnostics, job logging, and searchable history.""") status = gr.Markdown("*Loading sample data…*", elem_classes=["status-bar"]) with gr.Row(): refresh_button = gr.Button("🔄 Refresh Views", variant="secondary") pack_info_display = gr.Markdown("*Initialising…*", elem_classes=["pack-info"], visible=False) with gr.Tabs(): with gr.Tab("🔧 New Job"): with gr.Row(): with gr.Column(scale=1, min_width=400): gr.Markdown("### Job Details") job_title = gr.Textbox(label="Job Title", value="Generator won't start", elem_classes=["field-input"]) with gr.Row(): technician = gr.Textbox(label="Technician", value="Operator") location = gr.Textbox(label="Location", value="Solar shed") with gr.Row(): equipment_type = gr.Textbox(label="Equipment Type", value="off-grid inverter") severity = gr.Dropdown(["low", "medium", "high"], value="medium", label="⚠️ Severity") symptom = gr.Textbox(label="Symptom Description", lines=4, value="Inverter flashes a fault light after a cloudy morning and battery bank seems low.") notes = gr.Textbox(label="Field Notes", lines=3, value="Measured a low battery voltage; unsure if controller is limiting charge.") photo = gr.Image(type="filepath", label="📷 Upload Photo (optional)") submit = gr.Button("🔍 Analyze & Save Job", variant="primary", size="lg") with gr.Column(scale=1, min_width=400): gr.Markdown("### AI Diagnosis") response = gr.Markdown(label="RAG assistance", elem_classes=["diagnosis-card"]) gr.Markdown("### 📚 References") citations = gr.Dataframe(headers=["score", "kind", "id", "title", "citation"], datatype=["str", "str", "number", "str", "str"], label="Top Citations") sections_preview = gr.Dataframe(headers=["id", "manual", "section", "slug", "chars"], datatype=["number", "str", "str", "str", "number"], label="Indexed Sections") with gr.Tab("📋 History"): with gr.Row(): history_query = gr.Textbox(label="🔍 Search Jobs", value="battery", scale=3) history_search = gr.Button("Search", variant="secondary", scale=1) history_table = gr.Dataframe(headers=["id", "title", "equipment", "location", "severity", "status", "symptom"], datatype=["number", "str", "str", "str", "str", "str", "str"], label="Past Jobs") with gr.Row(): job_id_box = gr.Number(label="Inspect Job ID", value=0, precision=0, scale=1) job_details = gr.Markdown(label="Job Details", elem_classes=["diagnosis-card"]) with gr.Tab("📖 Manuals"): gr.Markdown("### Import & Browse Manuals") gr.Markdown("*Upload repair manuals (.pdf, .txt, or .md) to expand the knowledge base, or browse already-imported manuals below.*") manual_upload = gr.File( label="📤 Upload Repair Manuals", file_count="multiple", file_types=[".pdf", ".txt", ".md"], type="filepath", elem_classes=["upload-area"], ) upload_manual_btn = gr.Button("📥 Import Uploaded Manuals", variant="primary") upload_manual_status = gr.Markdown("", elem_classes=["status-bar"]) gr.Markdown("---") manuals_table = gr.Dataframe(headers=["id", "title", "license", "sections", "source"], datatype=["number", "str", "str", "number", "str"], label="Imported Manuals") manual_sections_table = gr.Dataframe(headers=["id", "manual", "section", "slug", "chars"], datatype=["number", "str", "str", "str", "number"], label="Manual Sections") with gr.Tab("✅ Evaluation"): gr.Markdown("### Golden-Scenario Evaluation") gr.Markdown("*Run the automated evaluation suite to test the RAG pipeline against known-good scenarios.*") eval_button = gr.Button("▶️ Run Evaluation", variant="primary") eval_output = gr.Code(language="json", label="Eval Report") with gr.Tab("📖 How It Works"): gr.Markdown( """ ### How to use the Off-Grid Field Repair Logbook 1. **Submit a New Job:** Under the **New Job** tab, fill out the Job Title, Technician name, Equipment Type, Severity dropdown, and describe the Symptoms. You can upload an optional equipment photo. Click **Analyze & Save Job** to run offline AI diagnostics. 2. **Consult AI Diagnosis & References:** Review the generated RAG advice and look at the **Top Citations** list to read exact excerpts from matching technical manuals. 3. **Browse History:** Go to the **History** tab to search past diagnostics or review previously completed repair cases. 4. **Manage Manuals:** Under the **Manuals** tab, drag and drop new manuals and click **Import Uploaded Manuals** to parse, index, and load them into the system knowledge base. 5. **Verify Suite Performance:** Execute standard test vectors in the **Evaluation** tab to test query answering accuracy. *Maintains all data locally for remote field operability where internet connection is absent.* """ ) def _load_pack_formatted(pack_name): status_text, manuals, sections, jobs, pack_update, info = load_pack_ui(pack_name) info_text = _format_pack_info(info) return status_text, manuals, sections, jobs, pack_update, info_text def _refresh_formatted(): return ( "✅ Views refreshed.", _format_manuals_table(), _format_sections_table(), _format_jobs_table(), _format_pack_info({"pack_root": str(PACK_ROOT)}), ) def _on_load(): # Auto-load the first available demo pack on startup if packs: status_text, manuals, sections, jobs, _, info = load_pack_ui(packs[0]) info_text = _format_pack_info(info) else: status_text = "✅ App ready. Submit a job to get AI diagnosis." manuals = _format_manuals_table() sections = _format_sections_table() jobs = _format_jobs_table() info_text = "*No sample data available.*" return status_text, manuals, sections, jobs, info_text def _import_manuals(files): if not files: return "⚠️ No files selected.", _format_manuals_table(), _format_sections_table() imported = 0 failures: list[str] = [] for f in files: try: with connect(DB_PATH) as conn: store_uploaded_manual(conn, f) conn.commit() imported += 1 except Exception as e: label = getattr(f, "name", None) or getattr(f, "path", None) or str(f) failures.append(f"{Path(label).name}: {e}") if failures: status = f"⚠️ Imported {imported} manual(s); {len(failures)} failed: " + "; ".join(failures) else: status = f"✅ Imported {imported} manual(s) successfully!" return status, _format_manuals_table(), _format_sections_table() refresh_button.click( _refresh_formatted, inputs=[], outputs=[status, manuals_table, sections_preview, history_table, pack_info_display], ) submit.click( submit_job_ui, inputs=[job_title, technician, location, equipment_type, severity, symptom, notes, photo], outputs=[response, citations, history_table, sections_preview], ) history_search.click(search_history_ui, inputs=[history_query], outputs=[history_table]) job_id_box.change(inspect_job_ui, inputs=[job_id_box], outputs=[job_details]) eval_button.click(run_eval_ui, inputs=[], outputs=[eval_output]) upload_manual_btn.click( _import_manuals, inputs=[manual_upload], outputs=[upload_manual_status, manuals_table, manual_sections_table], ) demo.load( _on_load, inputs=[], outputs=[status, manuals_table, sections_preview, history_table, pack_info_display], ) return demo def main() -> None: app = build_app() app.launch( server_name=os.environ.get("SERVER_NAME", "0.0.0.0"), show_error=True, share=False, ) if __name__ == "__main__": main()