diff --git a/.streamlit/config.toml b/.streamlit/config.toml new file mode 100644 index 0000000..611290b --- /dev/null +++ b/.streamlit/config.toml @@ -0,0 +1,19 @@ +# Streamlit configuration. +# +# This must live at the repository root: when the app entrypoint is in a +# subdirectory (dashboard/app.py), Streamlit Community Cloud only reads the +# config file from the repo root. + +[theme] +base = "light" +primaryColor = "#4C6EF5" +backgroundColor = "#FFFFFF" +secondaryBackgroundColor = "#F1F3F9" +textColor = "#1A1C23" +font = "sans serif" + +[server] +headless = true + +[browser] +gatherUsageStats = false diff --git a/README.md b/README.md index 0d78704..d931672 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,54 @@ dvc repro ls data/exports/ ``` +## ๐Ÿ“Š Results Dashboard + +An interactive Streamlit dashboard visualizes the pipeline's recorded outputs โ€” +evaluation metrics, per-stage benchmarks (runtime & memory), build-vs-buy cost +analysis, distribution drift, and the analysis reports. It reads the committed +JSON/markdown outputs and does **not** re-run the extraction pipeline, so it is +lightweight and independent of the heavy extraction stack (Docling, layout +models, OCR). + +### Run locally + +```bash +# Dashboard-only dependencies (no torch/docling needed) +pip install -r dashboard/requirements.txt + +# Launch +streamlit run dashboard/app.py +``` + +Then open http://localhost:8501. + +### Deploy to Streamlit Community Cloud (free) + +The dashboard is set up to deploy as-is โ€” no extra work needed to keep the heavy +pipeline dependencies out of the hosted build: + +1. Push this repo to GitHub (already done). +2. Go to [share.streamlit.io](https://share.streamlit.io) โ†’ **Create app** โ†’ + **Deploy a public app from GitHub**. +3. Set: + - **Repository:** `Effyrt/Docuparse` + - **Branch:** `main` + - **Main file path:** `dashboard/app.py` + - **Python version** (Advanced settings): `3.11` +4. Click **Deploy**. + +**Why this works cleanly:** Community Cloud searches the entrypoint's directory +*before* the repo root and uses the first dependency file it finds, so it installs +[`dashboard/requirements.txt`](dashboard/requirements.txt) (streamlit + plotly + +pandas only) instead of the heavy root `requirements.txt`. The Streamlit config +lives at [`.streamlit/config.toml`](.streamlit/config.toml), which is where +Community Cloud reads it from when the entrypoint is in a subdirectory. + +> A live "upload a PDF and parse it" demo is intentionally out of scope: the +> extraction stack is heavy and takes minutes per document, which is a poor fit +> for an always-on hosted demo. The dashboard focuses on the results the +> pipeline produces. + ## Project Structure ``` diff --git a/dashboard/__init__.py b/dashboard/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dashboard/app.py b/dashboard/app.py new file mode 100644 index 0000000..4521ebe --- /dev/null +++ b/dashboard/app.py @@ -0,0 +1,234 @@ +"""DocuParse results dashboard (Streamlit). + +Visualizes the pipeline's *recorded* outputs โ€” evaluation metrics, per-stage +benchmarks, build-vs-buy cost analysis, distribution drift, and the markdown +reports. It does NOT run the extraction pipeline (that needs the heavy stack and +minutes per document); it reads the JSON/markdown the pipeline already produced. + +Run locally: + pip install -r requirements-dashboard.txt + streamlit run dashboard/app.py + +All data access lives in dashboard/data_loader.py (pure stdlib, unit-tested). +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pandas as pd +import plotly.express as px +import streamlit as st + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from dashboard import data_loader as dl # noqa: E402 + +st.set_page_config(page_title="DocuParse Dashboard", page_icon="๐Ÿ“„", layout="wide") + + +def _fmt(value, digits: int = 4) -> str: + return "โ€”" if value is None else f"{value:.{digits}f}" + + +# --------------------------------------------------------------------------- # +# Sidebar +# --------------------------------------------------------------------------- # +st.sidebar.title("๐Ÿ“„ DocuParse") +st.sidebar.caption( + "Results dashboard for the SEC-filing extraction pipeline. " + "Figures are read from committed pipeline outputs, not generated live." +) +page = st.sidebar.radio( + "View", + ["Overview", "Benchmarks", "Cost (Build vs Buy)", "Distribution Drift", "Reports"], +) +st.sidebar.info( + "This dashboard visualizes recorded runs. A live 'upload a PDF and parse' " + "demo is intentionally out of scope โ€” the extraction stack (Docling, layout " + "models, OCR) is heavy and slow to host." +) + + +# --------------------------------------------------------------------------- # +# Overview +# --------------------------------------------------------------------------- # +def render_overview() -> None: + st.title("Evaluation Overview") + metrics = dl.load_metrics() + if not metrics: + st.warning("No metrics.json found.") + return + + cards = dl.metric_cards(metrics) + cols = st.columns(3) + for i, card in enumerate(cards): + with cols[i % 3]: + target = card["target"] + help_txt = None + if target is not None: + direction = "โ‰ค" if card["lower_is_better"] else "โ‰ฅ" + help_txt = f"Target {direction} {target}" + badge = "" + if card["passed"] is True: + badge = " โœ…" + elif card["passed"] is False: + badge = " โŒ" + digits = 3 if card["label"] == "Overall Pass Rate" else 4 + st.metric(card["label"] + badge, _fmt(card["value"], digits), help=help_txt) + + st.caption( + f"Evaluated on {metrics.get('total_evaluations', 'โ€”')} ground-truth items. " + "Note: the current ground-truth set is small and hand-built โ€” treat these " + "as directional, not production-grade accuracy." + ) + + history = dl.load_metrics_history() + if len(history) > 1: + st.subheader("Metric history") + df = pd.DataFrame(history) + keys = [k for k in ["text_avg_wer", "text_avg_cer", "table_avg_f1", + "overall_pass_rate"] if k in df.columns] + fig = px.line(df, x="timestamp", y=keys, markers=True) + fig.update_layout(legend_title_text="metric", yaxis_title="value") + st.plotly_chart(fig, width="stretch") + + +# --------------------------------------------------------------------------- # +# Benchmarks +# --------------------------------------------------------------------------- # +def render_benchmarks() -> None: + st.title("Pipeline Benchmarks") + benchmark = dl.load_benchmark() + if not benchmark: + st.warning("No benchmark results found.") + return + + info = benchmark.get("benchmark_info", {}) + sysinfo = info.get("system_info", {}) + c1, c2, c3, c4 = st.columns(4) + c1.metric("Platform", sysinfo.get("platform", "โ€”")) + c2.metric("CPU cores", sysinfo.get("cpu_count_logical", "โ€”")) + c3.metric("Total memory (GB)", sysinfo.get("total_memory_gb", "โ€”")) + c4.metric("Max pages/stage", info.get("max_pages_per_stage", "โ€”")) + + rows = dl.stage_performance(benchmark) + if rows: + df = pd.DataFrame(rows) + left, right = st.columns(2) + with left: + st.subheader("Runtime by stage (s)") + st.plotly_chart( + px.bar(df, x="stage", y="runtime_seconds", text="runtime_seconds"), + width="stretch", + ) + with right: + st.subheader("Peak memory by stage (MB)") + st.plotly_chart( + px.bar(df, x="stage", y="peak_memory_mb", text="peak_memory_mb"), + width="stretch", + ) + st.dataframe(df, width="stretch") + + failures = dl.benchmark_failures(benchmark) + if failures: + st.subheader("โš ๏ธ Recorded failures in this run") + st.caption("Surfaced honestly from the benchmark โ€” not hidden.") + st.dataframe(pd.DataFrame(failures), width="stretch") + + +# --------------------------------------------------------------------------- # +# Cost +# --------------------------------------------------------------------------- # +def render_cost() -> None: + st.title("Cost: Build vs Buy") + cost = dl.load_cost_analysis() + if not cost: + st.warning("No cost analysis found.") + return + + info = cost.get("analysis_info", {}) + st.caption( + f"Based on {info.get('pages_analyzed', 'โ€”')} pages ยท " + f"pricing snapshot {info.get('pricing_date', 'โ€”')}. " + "Static public-pricing estimate; excludes engineering/maintenance cost." + ) + + be = dl.break_even(cost) + if be: + c1, c2, c3 = st.columns(3) + c1.metric("Cheapest cloud (per 1k pages)", f"${be.get('cloud_cost_usd', 0):.2f}") + c2.metric("Cheapest self-hosted", f"${be.get('cheapest_infrastructure_cost_usd', 0):.2f}") + c3.metric("Break-even volume (pages)", f"{be.get('break_even_volume_pages', 0):,}") + + cloud = dl.cloud_cost_rows(cost) + if cloud: + st.subheader("Cloud service cost per processor") + df = pd.DataFrame(cloud) + fig = px.bar(df, x="processor", y="total_cost_usd", color="service", barmode="group") + fig.update_layout(yaxis_title="USD per analyzed volume") + st.plotly_chart(fig, width="stretch") + + infra = dl.infrastructure_rows(cost) + if infra: + st.subheader("Self-hosted infrastructure options") + st.plotly_chart( + px.bar(pd.DataFrame(infra), x="option", y="total_cost_usd", text="total_cost_usd"), + width="stretch", + ) + + +# --------------------------------------------------------------------------- # +# Drift +# --------------------------------------------------------------------------- # +def render_drift() -> None: + st.title("Distribution Drift") + drift = dl.load_drift() + if not drift: + st.warning("No drift analysis found.") + return + st.caption("Distributions of extracted-text and table shapes across analyzed pages.") + + fields = [ + ("text_analysis", "word_counts", "Words per chunk"), + ("text_analysis", "sentence_counts", "Sentences per chunk"), + ("text_analysis", "numeric_token_ratios", "Numeric-token ratio"), + ] + cols = st.columns(len(fields)) + for col, (section, field, label) in zip(cols, fields): + series = dl.drift_series(drift, section, field) + with col: + st.subheader(label) + if series: + st.plotly_chart( + px.histogram(pd.DataFrame({label: series}), x=label, nbins=10), + width="stretch", + ) + st.caption(f"n={len(series)}") + else: + st.info("No data") + + +# --------------------------------------------------------------------------- # +# Reports +# --------------------------------------------------------------------------- # +def render_reports() -> None: + st.title("Analysis Reports") + reports = dl.list_reports() + if not reports: + st.warning("No reports found.") + return + names = [p.stem.replace("_", " ").title() for p in reports] + choice = st.selectbox("Report", options=list(range(len(reports))), + format_func=lambda i: names[i]) + st.markdown(dl.read_report(reports[choice])) + + +PAGES = { + "Overview": render_overview, + "Benchmarks": render_benchmarks, + "Cost (Build vs Buy)": render_cost, + "Distribution Drift": render_drift, + "Reports": render_reports, +} +PAGES[page]() diff --git a/dashboard/data_loader.py b/dashboard/data_loader.py new file mode 100644 index 0000000..c7ae3a2 --- /dev/null +++ b/dashboard/data_loader.py @@ -0,0 +1,233 @@ +"""Data access + transforms for the DocuParse results dashboard. + +This module deliberately depends only on the Python standard library so it can +be unit-tested without Streamlit, pandas, or the heavy extraction stack. The +Streamlit app (``dashboard/app.py``) imports these functions and only handles +rendering. + +All data comes from files the pipeline already produced and committed to the +repo (evaluation metrics, benchmarks, cost analysis, drift analysis, reports). +The dashboard does not run the pipeline; it visualizes its recorded outputs. +""" + +from __future__ import annotations + +import glob +import json +import os +from pathlib import Path +from typing import Any, Dict, List, Optional + +# Repo root = parent of the dashboard/ package. +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def _read_json(path: Path) -> Optional[Any]: + """Load JSON, returning None if the file is missing or malformed.""" + try: + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + except (OSError, json.JSONDecodeError): + return None + + +def _latest(pattern: str) -> Optional[Path]: + """Return the most recently modified file matching a glob under the repo.""" + matches = glob.glob(str(REPO_ROOT / pattern)) + if not matches: + return None + return Path(max(matches, key=os.path.getmtime)) + + +# --------------------------------------------------------------------------- # +# Evaluation metrics +# --------------------------------------------------------------------------- # +def load_metrics(root: Path = REPO_ROOT) -> Dict[str, Any]: + """Load the headline evaluation metrics (metrics.json).""" + return _read_json(root / "metrics.json") or {} + + +def load_metrics_history(root: Path = REPO_ROOT) -> List[Dict[str, Any]]: + """Load the time series of evaluation runs, sorted by timestamp.""" + data = _read_json(root / "evaluation" / "metrics" / "metrics_history.json") or [] + if not isinstance(data, list): + return [] + return sorted(data, key=lambda r: r.get("timestamp", "")) + + +def metric_cards(metrics: Dict[str, Any]) -> List[Dict[str, Any]]: + """Shape the headline metrics into display cards with pass/fail vs thresholds. + + Each card: {label, value, target, unit, passed, lower_is_better}. + """ + thresholds = metrics.get("thresholds", {}) + cards = [ + { + "label": "Text WER", + "value": metrics.get("text_avg_wer"), + "target": thresholds.get("wer_threshold"), + "lower_is_better": True, + }, + { + "label": "Text CER", + "value": metrics.get("text_avg_cer"), + "target": thresholds.get("cer_threshold"), + "lower_is_better": True, + }, + { + "label": "Table Precision", + "value": metrics.get("table_avg_precision"), + "target": thresholds.get("precision_threshold"), + "lower_is_better": False, + }, + { + "label": "Table Recall", + "value": metrics.get("table_avg_recall"), + "target": thresholds.get("recall_threshold"), + "lower_is_better": False, + }, + { + "label": "Table F1", + "value": metrics.get("table_avg_f1"), + "target": thresholds.get("f1_threshold"), + "lower_is_better": False, + }, + { + "label": "Overall Pass Rate", + "value": metrics.get("overall_pass_rate"), + "target": None, + "lower_is_better": False, + }, + ] + for card in cards: + value, target = card["value"], card["target"] + if value is None or target is None: + card["passed"] = None + elif card["lower_is_better"]: + card["passed"] = value <= target + else: + card["passed"] = value >= target + return cards + + +# --------------------------------------------------------------------------- # +# Benchmarks (per-stage runtime + memory) +# --------------------------------------------------------------------------- # +def load_benchmark(root: Path = REPO_ROOT) -> Dict[str, Any]: + """Load the most recent corrected pipeline benchmark.""" + path = _latest("benchmarks/results/CORRECTED_pipeline_benchmark_*.json") + return _read_json(path) if path else {} + + +def stage_performance(benchmark: Dict[str, Any]) -> List[Dict[str, Any]]: + """Flatten per-stage runtime/memory/failure counts for charting.""" + rows = [] + for stage, data in (benchmark.get("stage_benchmarks") or {}).items(): + mem = data.get("memory_usage", {}) or {} + rows.append( + { + "stage": stage, + "runtime_seconds": data.get("total_runtime_seconds"), + "pages_processed": data.get("pages_processed"), + "pages_failed": data.get("pages_failed", 0), + "peak_memory_mb": mem.get("peak_memory_mb"), + "failure_count": len(data.get("failures", []) or []), + "avg_time_per_page": data.get("avg_time_per_page"), + } + ) + return rows + + +def benchmark_failures(benchmark: Dict[str, Any]) -> List[Dict[str, Any]]: + """Collect every recorded failure across stages (honest surfacing).""" + failures: List[Dict[str, Any]] = [] + for stage, data in (benchmark.get("stage_benchmarks") or {}).items(): + for fail in data.get("failures", []) or []: + failures.append({"stage": stage, **fail}) + return failures + + +# --------------------------------------------------------------------------- # +# Cost analysis (build vs buy) +# --------------------------------------------------------------------------- # +def load_cost_analysis(root: Path = REPO_ROOT) -> Dict[str, Any]: + """Load the most recent cloud vs infrastructure cost analysis.""" + path = _latest("benchmarks/results/cost_analysis_*.json") + return _read_json(path) if path else {} + + +def cloud_cost_rows(cost: Dict[str, Any]) -> List[Dict[str, Any]]: + """Flatten cloud service processor costs into {service, processor, total_usd}.""" + rows = [] + services = (cost.get("cloud_cost_estimates", {}) or {}).get("service_costs", {}) or {} + for service, processors in services.items(): + for processor, vals in processors.items(): + rows.append( + { + "service": service, + "processor": processor, + "total_cost_usd": vals.get("total_cost_usd"), + "cost_per_page": vals.get("cost_per_page"), + } + ) + return rows + + +def infrastructure_rows(cost: Dict[str, Any]) -> List[Dict[str, Any]]: + """Flatten self-hosted infrastructure options into {option, total_usd, ...}.""" + rows = [] + infra = (cost.get("infrastructure_comparison", {}) or {}).get( + "infrastructure_costs", {} + ) or {} + for option, vals in infra.items(): + rows.append( + { + "option": option, + "total_cost_usd": vals.get("total_cost_usd"), + "description": vals.get("description", option), + } + ) + return rows + + +def break_even(cost: Dict[str, Any]) -> Dict[str, Any]: + """Return the break-even summary block.""" + return (cost.get("infrastructure_comparison", {}) or {}).get( + "break_even_analysis", {} + ) or {} + + +# --------------------------------------------------------------------------- # +# Distribution drift +# --------------------------------------------------------------------------- # +def load_drift(root: Path = REPO_ROOT) -> Dict[str, Any]: + """Load drift analysis raw distributions.""" + return _read_json( + root / "evaluation" / "visualizations" / "drift_analysis_results.json" + ) or {} + + +def drift_series(drift: Dict[str, Any], section: str, field: str) -> List[float]: + """Pull a numeric series (e.g. text word_counts) out of the drift data.""" + raw = (drift.get(section, {}) or {}).get("raw_data", {}) or {} + values = raw.get(field, []) or [] + return [v for v in values if isinstance(v, (int, float))] + + +# --------------------------------------------------------------------------- # +# Reports +# --------------------------------------------------------------------------- # +def list_reports(root: Path = REPO_ROOT) -> List[Path]: + """Return committed markdown reports, excluding placeholders.""" + reports_dir = root / "reports" + if not reports_dir.exists(): + return [] + return sorted(p for p in reports_dir.glob("*.md") if p.stat().st_size > 0) + + +def read_report(path: Path) -> str: + """Read a report's markdown text.""" + try: + return path.read_text(encoding="utf-8") + except OSError: + return "" diff --git a/dashboard/requirements.txt b/dashboard/requirements.txt new file mode 100644 index 0000000..dc45ef8 --- /dev/null +++ b/dashboard/requirements.txt @@ -0,0 +1,12 @@ +# Dependencies for the results dashboard (dashboard/app.py). +# +# This file lives next to the entrypoint on purpose: Streamlit Community Cloud +# searches the entrypoint's directory FIRST and a dependency file here takes +# precedence over the repo-root requirements.txt. That keeps the heavy +# extraction stack (torch, docling, ...) out of the hosted dashboard build. +# +# Lightweight and independent of the extraction pipeline, so the dashboard runs +# and deploys on its own. +streamlit>=1.49.0 # width="stretch" API used by the app +plotly>=5.22.0 +pandas>=2.1.0 diff --git a/tests/unit/test_dashboard_data.py b/tests/unit/test_dashboard_data.py new file mode 100644 index 0000000..cdfcdb3 --- /dev/null +++ b/tests/unit/test_dashboard_data.py @@ -0,0 +1,101 @@ +"""Unit tests for the dashboard data loader. + +These exercise the pure-stdlib transforms against both synthetic fixtures and +the real committed pipeline outputs, without importing Streamlit/pandas. +""" + +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +from dashboard import data_loader as dl # noqa: E402 + + +# --- transforms on synthetic fixtures --------------------------------------- # +def test_metric_cards_pass_fail_direction(): + metrics = { + "text_avg_wer": 0.02, + "table_avg_f1": 0.95, + "thresholds": {"wer_threshold": 0.05, "f1_threshold": 0.87}, + } + cards = {c["label"]: c for c in dl.metric_cards(metrics)} + # Lower-is-better metric under its threshold passes. + assert cards["Text WER"]["passed"] is True + # Higher-is-better metric above its threshold passes. + assert cards["Table F1"]["passed"] is True + + +def test_metric_cards_fail_when_worse_than_threshold(): + metrics = { + "text_avg_wer": 0.10, + "table_avg_f1": 0.50, + "thresholds": {"wer_threshold": 0.05, "f1_threshold": 0.87}, + } + cards = {c["label"]: c for c in dl.metric_cards(metrics)} + assert cards["Text WER"]["passed"] is False + assert cards["Table F1"]["passed"] is False + + +def test_metric_cards_none_when_missing(): + cards = {c["label"]: c for c in dl.metric_cards({})} + assert cards["Text WER"]["passed"] is None + + +def test_stage_performance_and_failures(): + benchmark = { + "stage_benchmarks": { + "text_extraction": { + "total_runtime_seconds": 1.5, + "pages_processed": 0, + "memory_usage": {"peak_memory_mb": 70.0}, + "failures": [{"file": "x.pdf", "error": "boom"}], + } + } + } + rows = dl.stage_performance(benchmark) + assert rows[0]["stage"] == "text_extraction" + assert rows[0]["peak_memory_mb"] == 70.0 + assert rows[0]["failure_count"] == 1 + failures = dl.benchmark_failures(benchmark) + assert failures == [{"stage": "text_extraction", "file": "x.pdf", "error": "boom"}] + + +def test_drift_series_filters_non_numeric(): + drift = {"text_analysis": {"raw_data": {"word_counts": [1, 2, None, "x", 3]}}} + assert dl.drift_series(drift, "text_analysis", "word_counts") == [1, 2, 3] + + +def test_loaders_tolerate_missing_files(tmp_path): + # An empty repo root yields empty structures, never exceptions. + assert dl.load_metrics(tmp_path) == {} + assert dl.load_metrics_history(tmp_path) == [] + assert dl.list_reports(tmp_path) == [] + + +# --- transforms on the real committed data ---------------------------------- # +def test_real_metrics_present_and_cards_shaped(): + metrics = dl.load_metrics() + assert metrics, "metrics.json should be committed and non-empty" + cards = dl.metric_cards(metrics) + assert len(cards) == 6 + assert all("label" in c and "passed" in c for c in cards) + + +def test_real_benchmark_has_stages(): + rows = dl.stage_performance(dl.load_benchmark()) + stages = {r["stage"] for r in rows} + assert {"text_extraction", "table_extraction", "docling"} <= stages + + +def test_real_cost_analysis_break_even(): + be = dl.break_even(dl.load_cost_analysis()) + assert "break_even_volume_pages" in be + assert dl.infrastructure_rows(dl.load_cost_analysis()), "infra options expected" + + +def test_real_reports_listed(): + names = {p.name for p in dl.list_reports()} + assert "benchmarks.md" in names + assert "xbrl_cross_verification_report.md" in names