| """Data access for the NayanaOCRBench explorer. Everything is read from DATA_DIR (the mounted bucket).""" |
| import difflib, html, json, os |
| from functools import lru_cache |
|
|
| import duckdb |
| import polars as pl |
|
|
| DATA_DIR = os.environ.get("NAYANA_DATA", "/data") |
| if not os.path.exists(os.path.join(DATA_DIR, "index", "samples.parquet")): |
| |
| _local = os.path.join(os.path.dirname(__file__), "..", "nayanaocr_corpus_2026_preview", ".work", "bucket_mirror") |
| if os.path.exists(os.path.join(_local, "index", "samples.parquet")): |
| DATA_DIR = os.path.abspath(_local) |
|
|
| IDX = os.path.join(DATA_DIR, "index") |
| IMG = os.path.join(DATA_DIR, "images") |
|
|
| LANG_NAME = {"en": "English", "zh": "Chinese", "hi": "Hindi", "bn": "Bengali", "mr": "Marathi", "gu": "Gujarati", "pa": "Punjabi", "or": "Odia", "kn": "Kannada", |
| "ta": "Tamil", "te": "Telugu", "ml": "Malayalam", "sa": "Sanskrit", "ja": "Japanese", "ko": "Korean", "de": "German", "fr": "French", "es": "Spanish", |
| "it": "Italian", "ru": "Russian", "ar": "Arabic", "th": "Thai"} |
| LANG_ORDER = ["en", "hi", "bn", "mr", "gu", "pa", "or", "kn", "ta", "te", "ml", "sa", "zh", "ja", "ko", "de", "fr", "es", "it", "ru", "ar", "th"] |
| GROUPS = [("Indic", ["hi", "bn", "mr", "gu", "pa", "or", "kn", "ta", "te", "ml", "sa"]), ("East Asian", ["zh", "ja", "ko"]), |
| ("European", ["en", "de", "fr", "es", "it", "ru"]), ("Other", ["ar", "th"])] |
| INK, INK2, RULE, EMPTY = "var(--body-text-color)", "var(--body-text-color-subdued)", "var(--border-color-primary)", "var(--background-fill-secondary)" |
| METRICS = { |
| "accuracy": ("Accuracy % (100 − CER)", True, 100), "composite": ("Composite %", True, 100), "f1": ("F1 %", True, 100), |
| "rougeL": ("ROUGE-L", True, 1), "cer": ("CER (lower is better)", False, 100), "wer": ("WER (lower is better)", False, 3), |
| } |
| RAMP = ["#e3ecf9", "#c7d9f3", "#a9c5ec", "#8bb0e4", "#6d9adb", "#4f83d0", "#3a6bbf", "#2a539f", "#1a3b78"] |
| TYPE_COLOR = {"title": "#eb6834", "text": "#2a78d6", "text_block": "#2a78d6", "list": "#1baf7a", "table": "#eda100", "table_body": "#eda100", |
| "table_caption": "#eda100", "table_footnote": "#eda100", "figure": "#e87ba4", "image": "#e87ba4", "image_body": "#e87ba4", |
| "image_caption": "#e87ba4", "index": "#4a3aa7", "equation": "#e34948", "header": "#008300", "footer": "#008300", "page_number": "#008300", |
| "abandon": "#8c8a93", "code": "#e34948"} |
|
|
| _con = duckdb.connect(database=":memory:") |
| _con.execute(f"CREATE VIEW samples AS SELECT * FROM read_parquet('{os.path.join(IDX, 'samples.parquet')}')") |
| _con.execute(f"CREATE VIEW runs AS SELECT * FROM read_parquet('{os.path.join(IDX, 'runs.parquet')}')") |
| _con.execute(f"CREATE VIEW gt AS SELECT * FROM read_parquet('{os.path.join(IDX, 'gt.parquet')}')") |
|
|
|
|
| def q(sql, params=None) -> pl.DataFrame: |
| |
| cur = _con.cursor() |
| try: |
| return cur.execute(sql, params or []).pl() |
| finally: |
| cur.close() |
|
|
|
|
| @lru_cache(maxsize=None) |
| def models(split: str) -> list[str]: |
| return q("SELECT DISTINCT model_short FROM runs WHERE split=? ORDER BY model_short", [split])["model_short"].to_list() |
|
|
|
|
| @lru_cache(maxsize=None) |
| def langs(split: str) -> list[str]: |
| ls = set(q("SELECT DISTINCT lang FROM runs WHERE split=?", [split])["lang"].to_list()) |
| return [l for l in LANG_ORDER if l in ls] |
|
|
|
|
| def stats() -> dict: |
| r = q("SELECT count(*) AS runs, sum(n) AS preds, count(DISTINCT model) AS models, count(DISTINCT lang) AS langs, sum(spend_usd) AS spend FROM runs").row(0, named=True) |
| g = q("SELECT split, count(*) AS n FROM gt GROUP BY split").rows() |
| r["gt"] = dict(g) |
| return r |
|
|
|
|
| |
| def matrix(split: str, metric: str) -> pl.DataFrame: |
| """rows = models, columns = langs, values = run-level mean of metric.""" |
| df = q(f"SELECT model_short, family, lang, {metric} AS v, n FROM runs WHERE split=?", [split]) |
| return df |
|
|
|
|
| def leaderboard_table(split: str, metric: str, sort: str) -> pl.DataFrame: |
| df = matrix(split, metric) |
| ls = langs(split) |
| higher = METRICS[metric][1] |
| wide = df.pivot(on="lang", index=["model_short", "family"], values="v") |
| for l in ls: |
| if l not in wide.columns: |
| wide = wide.with_columns(pl.lit(None).cast(pl.Float64).alias(l)) |
| wide = wide.select(["model_short", "family"] + ls) |
| fill = 0.0 if higher else None |
| vals = wide.select(ls) |
| wide = wide.with_columns([ |
| pl.mean_horizontal([pl.col(l).fill_null(fill) for l in ls]).alias("avg_all") if higher else pl.mean_horizontal([pl.col(l) for l in ls]).alias("avg_all"), |
| pl.mean_horizontal([pl.col(l) for l in ls]).alias("avg_eval"), |
| pl.sum_horizontal([pl.col(l).is_not_null().cast(pl.Int32) for l in ls]).alias("n_langs"), |
| ]) |
| key = {"Average over all 22 languages": "avg_all", "Average over evaluated languages": "avg_eval", "Indic average": None, "Family": "family"}[sort] |
| if sort == "Indic average": |
| indic = [l for l in GROUPS[1][1] if l in ls] |
| wide = wide.with_columns(pl.mean_horizontal([pl.col(l) for l in indic]).alias("_k")) |
| key = "_k" |
| wide = wide.sort(key, descending=higher if key != "family" else False, nulls_last=True) |
| return wide.drop([c for c in ["_k"] if c in wide.columns]) |
|
|
|
|
| def heatmap_svg(split: str, metric: str, sort: str) -> str: |
| wide = leaderboard_table(split, metric, sort) |
| ls = langs(split) |
| label, higher, vmax = METRICS[metric] |
| groups = [] |
| small = [] |
| for g, gl in GROUPS: |
| kept = [l for l in gl if l in ls] |
| if g == "Other": |
| small.extend(kept) |
| elif len(kept) >= 2: |
| groups.append((g, kept)) |
| else: |
| small.extend(kept) |
| if small: |
| groups.append(("Other", small)) |
| order = [l for g in groups for l in g[1]] |
| left, top, cw, ch = 150, 50, 44, 28 |
| W, H = left + len(order) * cw + 130, top + wide.height * ch + 6 |
| s = [f'<div style="display:flex;justify-content:center;overflow-x:auto"><svg viewBox="0 0 {W} {H}" width="100%" style="max-width:{W}px;min-width:{int(W*0.7)}px;display:block;font-family:ui-sans-serif,system-ui,sans-serif">'] |
| x = left |
| for g, gl in groups: |
| w = len(gl) * cw |
| s.append(f'<text x="{x + w / 2}" y="14" text-anchor="middle" font-size="10" font-weight="700" fill="{INK2}" letter-spacing=".06em">{g.upper()}</text><line x1="{x + 3}" x2="{x + w - 3}" y1="20" y2="20" stroke="{RULE}"/>') |
| x += w |
| for j, l in enumerate(order): |
| s.append(f'<text x="{left + j * cw + cw / 2}" y="40" text-anchor="middle" font-size="12" font-family="ui-monospace,monospace" fill="{INK}">{l}</text>') |
| s.append(f'<text x="{left + len(order) * cw + 10}" y="40" font-size="10" fill="{INK2}">avg/{len(ls)} · eval-only</text>') |
| for i, row in enumerate(wide.iter_rows(named=True)): |
| y = top + i * ch |
| s.append(f'<text x="{left - 8}" y="{y + ch / 2 + 4}" text-anchor="end" font-size="12.5" fill="{INK}">{html.escape(row["model_short"])}</text>') |
| for j, l in enumerate(order): |
| v = row.get(l) |
| cx = left + j * cw |
| if v is None: |
| s.append(f'<rect x="{cx + 1}" y="{y + 1}" width="{cw - 2}" height="{ch - 2}" rx="3" fill="{EMPTY}"/><text x="{cx + cw / 2}" y="{y + ch / 2 + 4}" text-anchor="middle" font-size="10" fill="{INK2}">—</text>') |
| else: |
| t = (v / vmax) if higher else 1 - min(v, vmax) / vmax |
| step = max(0, min(8, int(t * 9))) |
| txt = f"{v:.0f}" if vmax >= 10 else f"{v:.2f}" |
| s.append(f'<rect class="cell" data-model="{html.escape(row["model_short"])}" data-lang="{l}" x="{cx + 1}" y="{y + 1}" width="{cw - 2}" height="{ch - 2}" rx="3" fill="{RAMP[step]}" style="cursor:pointer"><title>{html.escape(row["model_short"])} · {LANG_NAME.get(l, l)} · {label}: {v:.2f}</title></rect>' |
| f'<text x="{cx + cw / 2}" y="{y + ch / 2 + 4}" text-anchor="middle" font-size="10.5" font-family="ui-monospace,monospace" fill="{"#ffffff" if step >= 5 else "#15140f"}" style="pointer-events:none">{txt}</text>') |
| ax = left + len(order) * cw + 10 |
| a1, a2 = row["avg_all"], row["avg_eval"] |
| f = (lambda v: "—" if v is None else (f"{v:.1f}" if vmax >= 10 else f"{v:.3f}")) |
| s.append(f'<text x="{ax}" y="{y + ch / 2 + 4}" font-size="12" font-family="ui-monospace,monospace" fill="{INK}">{f(a1)}</text><text x="{ax + 58}" y="{y + ch / 2 + 4}" font-size="12" font-family="ui-monospace,monospace" fill="{INK2}">{f(a2)}</text>') |
| s.append("</svg></div>") |
| return "".join(s) |
|
|
|
|
| |
| def sample_list(split: str, lang: str, model: str, sort: str, lo: float, hi: float, search: str, limit: int = 120) -> pl.DataFrame: |
| order = {"Worst first": "accuracy ASC", "Best first": "accuracy DESC", "Longest output": "pred_chars DESC", "Sample id": "sample_id"}[sort] |
| where = "s.split=? AND s.lang=? AND s.model_short=? AND s.accuracy BETWEEN ? AND ?" |
| params = [split, lang, model, lo, hi] |
| if search: |
| where += " AND s.sample_id ILIKE ?" |
| params.append(f"%{search}%") |
| return q(f"""SELECT s.sample_id, s.accuracy, s.cer, s.wer, s.f1, s.composite, s.pred_chars, s.gt_chars, g.has_image |
| FROM samples s LEFT JOIN gt g USING (split, lang, sample_id) |
| WHERE {where} ORDER BY {order} NULLS LAST LIMIT {limit}""", params) |
|
|
|
|
| def image_path(split: str, lang: str, sid: str, thumb=False) -> str | None: |
| p = os.path.join(IMG, "thumbs", split, lang, sid + ".webp") if thumb else os.path.join(IMG, split, lang, sid + ".webp") |
| return p if os.path.exists(p) else None |
|
|
|
|
| def sample_detail(split: str, lang: str, sid: str) -> dict: |
| g = q("SELECT * FROM gt WHERE split=? AND lang=? AND sample_id=?", [split, lang, sid]) |
| preds = q("SELECT model_short, family, prediction, gt_trace, accuracy, cer, wer, f1, precision, recall, composite, rougeL, pred_chars, gt_chars, in_tok, out_tok FROM samples WHERE split=? AND lang=? AND sample_id=? ORDER BY accuracy DESC NULLS LAST", [split, lang, sid]) |
| return {"gt": g.row(0, named=True) if g.height else None, "preds": preds} |
|
|
|
|
| def overlay_html(img_path: str, layout_json: str, show_boxes: bool, width: int | None, height: int | None) -> str: |
| """Image with SVG layout overlay (boxes are in a 1000x1000 normalised space).""" |
| if not img_path: |
| return "<div style='padding:40px;text-align:center;color:var(--body-text-color-subdued)'>No image for this sample in the bucket.</div>" |
| import base64 |
| with open(img_path, "rb") as f: |
| b64 = base64.b64encode(f.read()).decode() |
| boxes = "" |
| legend = "" |
| if show_boxes: |
| try: |
| layout = json.loads(layout_json or "[]") |
| except Exception: |
| layout = [] |
| seen = {} |
| for r in layout: |
| bb = r.get("bbox") or [] |
| if len(bb) != 4: |
| continue |
| c = TYPE_COLOR.get(r.get("type"), "#8c8a93") |
| seen[r.get("type")] = c |
| boxes += f'<rect x="{bb[0]}" y="{bb[1]}" width="{bb[2] - bb[0]}" height="{bb[3] - bb[1]}" fill="{c}" fill-opacity=".14" stroke="{c}" stroke-width="2" vector-effect="non-scaling-stroke"><title>{html.escape(str(r.get("type")))}</title></rect>' |
| legend = " ".join(f'<span style="font-size:11px;color:var(--body-text-color-subdued)"><span style="display:inline-block;width:10px;height:10px;background:{c};border-radius:2px;margin-right:4px"></span>{html.escape(str(t))}</span>' for t, c in seen.items()) |
| return f"""<div style="position:relative;background:#fff;border:1px solid var(--border-color-primary);border-radius:8px;overflow:hidden"> |
| <img src="data:image/webp;base64,{b64}" style="display:block;width:100%;height:auto"> |
| <svg viewBox="0 0 1000 1000" preserveAspectRatio="none" style="position:absolute;inset:0;width:100%;height:100%">{boxes}</svg> |
| </div><div style="margin-top:6px;display:flex;gap:12px;flex-wrap:wrap">{legend}</div>""" |
|
|
|
|
| def diff_html(gt: str, pred: str, max_chars: int = 6000, rtl: bool = False) -> str: |
| """Side-by-side GT vs prediction with character-level diff highlighting.""" |
| gt = (gt or "")[:max_chars] |
| pred = (pred or "")[:max_chars] |
| sm = difflib.SequenceMatcher(None, gt, pred, autojunk=False) |
| L, R = [], [] |
| for op, i1, i2, j1, j2 in sm.get_opcodes(): |
| a, b = html.escape(gt[i1:i2]), html.escape(pred[j1:j2]) |
| if op == "equal": |
| L.append(a); R.append(b) |
| elif op == "delete": |
| L.append(f'<mark class="del">{a}</mark>') |
| elif op == "insert": |
| R.append(f'<mark class="ins">{b}</mark>') |
| else: |
| L.append(f'<mark class="del">{a}</mark>'); R.append(f'<mark class="ins">{b}</mark>') |
| ratio = sm.ratio() |
| css = ("<style>.dcol{white-space:pre-wrap;word-break:break-word;font:12.5px/1.6 ui-monospace,Menlo,monospace;background:var(--background-fill-secondary);color:var(--body-text-color);border:1px solid var(--border-color-primary);border-radius:8px;padding:12px;max-height:560px;overflow:auto}" |
| ".dcol mark.del{background:rgba(227,73,72,.28);color:inherit;border-radius:2px;text-decoration:line-through;text-decoration-color:rgba(227,73,72,.8)}.dcol mark.ins{background:rgba(27,175,122,.28);color:inherit;border-radius:2px}" |
| ".dh{font-size:11px;letter-spacing:.08em;text-transform:uppercase;color:var(--body-text-color-subdued);font-weight:700;margin:0 0 6px}</style>") |
| d = ' dir="rtl" style="text-align:right"' if rtl else "" |
| return (f'{css}<div style="display:grid;grid-template-columns:1fr 1fr;gap:12px">' |
| f'<div><p class="dh">Ground truth · {len(gt):,} chars · <span style="color:#e34948">red = missing in prediction</span></p><div class="dcol"{d}>{"".join(L)}</div></div>' |
| f'<div><p class="dh">Prediction · {len(pred):,} chars · similarity {ratio:.2f} · <span style="color:#1baf7a">green = extra / changed</span></p><div class="dcol"{d}>{"".join(R)}</div></div></div>') |
|
|
|
|
| |
| def compare(split: str, lang: str, a: str, b: str) -> pl.DataFrame: |
| return q("""SELECT x.sample_id, x.accuracy AS acc_a, y.accuracy AS acc_b, x.accuracy - y.accuracy AS delta, x.pred_chars AS chars_a, y.pred_chars AS chars_b |
| FROM samples x JOIN samples y USING (split, lang, sample_id) |
| WHERE x.split=? AND x.lang=? AND x.model_short=? AND y.model_short=? ORDER BY abs(delta) DESC""", [split, lang, a, b]) |
|
|
|
|
| def language_view(split: str, lang: str) -> pl.DataFrame: |
| return q("""SELECT model_short AS model, family, count(*) AS n, round(avg(accuracy),1) AS acc_mean, round(median(accuracy),1) AS acc_median, |
| round(quantile_cont(accuracy, 0.1),1) AS acc_p10, round(avg(cer),1) AS cer, round(avg(wer),2) AS wer, round(avg(f1),1) AS f1, |
| sum(CASE WHEN pred_chars IS NULL OR pred_chars < 5 THEN 1 ELSE 0 END) AS empty_out, |
| sum(CASE WHEN accuracy < 10 THEN 1 ELSE 0 END) AS below_10, |
| sum(CASE WHEN pred_chars > 3*gt_chars THEN 1 ELSE 0 END) AS runaway_out |
| FROM samples WHERE split=? AND lang=? GROUP BY 1,2 ORDER BY acc_mean DESC""", [split, lang]) |
|
|
|
|
| def hardest_pages(split: str, lang: str, k: int = 24) -> pl.DataFrame: |
| return q("""SELECT sample_id, round(avg(accuracy),1) AS mean_acc, round(max(accuracy),1) AS best_acc, count(*) AS n_models |
| FROM samples WHERE split=? AND lang=? GROUP BY sample_id HAVING count(*) >= 3 ORDER BY mean_acc ASC LIMIT ?""", [split, lang, k]) |
|
|
