From fe95cb572cf141436e44e4b2fa02fdb6f2b00426 Mon Sep 17 00:00:00 2001 From: dschlueter Date: Wed, 7 Oct 2026 05:01:27 +0200 Subject: [PATCH] feat: Konsens zweier Bildmodelle als Entscheider, elastischer Bildvergleich MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Entscheidung (Standard: decide=consensus): - Zwei lokale Vision-Modelle (local: Qwen3-VL, local2: Gemma 4) lesen jedes Formelbild; angenommen wird nur bei inhaltlicher Übereinstimmung (Normalform oder Render-Gleichheit), Pandoc-Prüfung und grobem Bild-Veto - Optional dritte Lesart als Gegenprobe (require_third) bzw. Stichwahl - Fettdruck standardmäßig irrelevant (bold_relevant=0) - Token-Wahrscheinlichkeiten aus Logprobs (p_min, p_mean) statt Selbsteinschätzung; "Nachdenken" lokaler Modelle abgeschaltet (Gemma 4: ~10x schneller) - consensus.py: Normalform auf Baumbasis (Abstände, \left/\right, kursiv/aufrecht, Längenvarianten, Matrix-Umgebungen, TeX-Umlaute, \atop), Render-Vergleich Bildvergleich (Veto und Pixel-Modus decide=pixel): - Elastische Ausrichtung per DTW über Spalten und Zeilen; Dehnen kostet proportional zur Tinte, Varianten nutzen die Verzerrung des Kandidaten - Layouttreue Prüfvarianten, Äquivalenzklassen, rauschnormierter Kontrast Werkzeuge: calibrate --consensus, tools/vergleich.py (50er-Stichprobe mit Bildern), tools/konsens_bericht.py (Übersichtsseite); Ausgaben in vergleich/ (nicht versioniert) Messung Göllmann (2 x 150 Formeln, Fettdruck egal): Qwen = Gemma = GLM angenommen 61 % / 57 %, davon falsch 0; Qwen = Gemma allein: 1 bzw. 2 Fehler. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 3 + formula_img2latex/calibrate.py | 139 ++++++++++++++++++++- formula_img2latex/cli.py | 7 ++ formula_img2latex/compare.py | 168 +++++++++++++++++++++----- formula_img2latex/confusables.py | 29 ++++- formula_img2latex/consensus.py | 145 ++++++++++++++++++++++ formula_img2latex/decide.py | 99 ++++++++++++++- formula_img2latex/engines/__init__.py | 20 +-- formula_img2latex/engines/vlm.py | 76 +++++++++++- formula_img2latex/pipeline.py | 88 ++++++++++++-- formula_img2latex/verify.py | 12 +- tests/test_core.py | 8 +- tests/test_pipeline.py | 81 ++++++++++++- tools/konsens_bericht.py | 125 +++++++++++++++++++ tools/vergleich.py | 160 ++++++++++++++++++++++++ 15 files changed, 1087 insertions(+), 73 deletions(-) create mode 100644 formula_img2latex/consensus.py create mode 100644 tools/konsens_bericht.py create mode 100644 tools/vergleich.py diff --git a/.gitignore b/.gitignore index 352c1ac..fac35a5 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,6 @@ output/ .vscode/ *.swp .DS_Store + +# Vergleichsbilder zum Einstellen (Ausschnitte aus Büchern, nicht versionieren) +vergleich/ diff --git a/formula_img2latex/calibrate.py b/formula_img2latex/calibrate.py index dcc6def..416f717 100644 --- a/formula_img2latex/calibrate.py +++ b/formula_img2latex/calibrate.py @@ -18,8 +18,11 @@ import os import posixpath import random import re +import shutil import statistics +import subprocess import sys +import tempfile import zipfile from concurrent.futures import ProcessPoolExecutor, as_completed from pathlib import Path @@ -64,6 +67,9 @@ def corpus(epub: Path) -> list[tuple[str, bytes, str, bool]]: def synth_jpeg(tex: str, display: bool, rng: random.Random) -> bytes | None: + """Small JPEG like the Gehrke books: LaTeX + dvipng with bitmap fonts at ~102 dpi, quality 72.""" + if shutil.which("dvipng") and shutil.which("latex"): + return _dvipng_jpeg(tex, display, int(round(SYNTH_PX_PER_PT * 72.27 * rng.uniform(0.97, 1.03)))) r = render_many([(tex, display)], "cm")[0] if not r.ok: return None @@ -79,13 +85,32 @@ def synth_jpeg(tex: str, display: bool, rng: random.Random) -> bytes | None: return buf.getvalue() +def _dvipng_jpeg(tex: str, display: bool, dpi: int) -> bytes | None: + with tempfile.TemporaryDirectory(prefix="f2l-dvipng-") as d: + body = ("$\\displaystyle " if display else "$") + tex + "$" + Path(d, "t.tex").write_text("\\documentclass[10pt]{article}\\usepackage{amsmath,amssymb,mathtools,bm,mathrsfs}" + "\\pagestyle{empty}\\begin{document}" + body + "\\end{document}", encoding="utf-8") + env = dict(os.environ, openout_any="p", openin_any="p") + subprocess.run(["latex", "-interaction=nonstopmode", "-no-shell-escape", "t.tex"], cwd=d, env=env, + capture_output=True, timeout=60, check=False) + if not Path(d, "t.dvi").exists(): + return None + subprocess.run(["dvipng", "-q", "-D", str(dpi), "-T", "tight", "-bg", "White", "-o", "t.png", "t.dvi"], + cwd=d, capture_output=True, timeout=60, check=False) + if not Path(d, "t.png").exists(): + return None + buf = io.BytesIO() + Image.open(Path(d, "t.png")).convert("L").save(buf, "JPEG", quality=72) + return buf.getvalue() + + def _equivalent(orig: np.ndarray, a: str, b: str, display: bool, preset: str) -> bool: """True if a and b render identically at the original's scale (e.g. \\le vs \\leq).""" ra, rb = render_many([(a, display), (b, display)], preset) if not (ra.ok and rb.ok): return False ca = compare(orig, ra) - cb = compare(orig, rb, px_per_pt=ca.metrics.get("px_per_pt")) + cb = compare(orig, rb, warp_from=ca) return ca.ok and cb.ok and contrast(ca, cb) is None @@ -99,7 +124,7 @@ def _job(args: tuple[str, bytes, str, bool, str, int]) -> dict[str, Any]: vt = verify(orig, truth, display, preset) res["truth_ver"] = vt.to_json() res["truth_status"] = decide(res["truth_ver"], True, None, th)[0] - starts = [v.tex for v in variants(truth)] + starts = [v.tex for v in variants(truth) if not v.equivalent] rng.shuffle(starts) res["errors"] = [] for wrong in starts[:3]: @@ -135,7 +160,7 @@ def calibrate(epubs: list[Path], samples: int = 200, preset: str | None = None, img = synth_jpeg(tex, display, rng) if img: syn.append((name + "#synth.jpg", img, tex, display)) - sets.append((f"{epub.stem[:40]} [synthetisch, cm]", syn, "cm")) + sets.append((f"{epub.stem[:40]} [synthetisch dvipng, cm]", syn, "cm")) summary: dict[str, Any] = {"sets": []} for label, items, p in sets: log(f"prüfe {label}: {len(items)} Formeln …") @@ -194,3 +219,111 @@ def _print(s: dict[str, Any], log: Any) -> None: f"Review/Bild {s['errors_to_review_or_rejected']}, FALSCH ANGENOMMEN {s['errors_wrongly_accepted']}") for ex in s["wrongly_accepted_examples"]: log(f" falsch: {ex['final']} (Start {ex['start']})") + + +# ------------------------------------------------------------------------- consensus calibration + +def _consensus_eval(args: tuple) -> dict[str, Any]: + """Worker: compare both readings with each other and with the truth; rough image check.""" + from .consensus import same_reading + name, data, truth, display, preset, ra, rb = args + os.environ.setdefault("OMP_NUM_THREADS", "1") + ta, tb = ra.get("tex") or "", rb.get("tex") or "" + res: dict[str, Any] = {"name": name, "truth": truth, "display": display, "a": ra, "b": rb, + "a_ok": bool(same_reading(ta, truth, display, preset)), + "b_ok": bool(same_reading(tb, truth, display, preset)), + "agree": same_reading(ta, tb, display, preset)} + res["veto"] = None + if res["agree"]: + v = verify(load_ink(data, name), ta, display, preset, check_variants=False) + res["veto"] = {"ok": v.ok, "reason": v.reason, "metrics": v.metrics} + return res + + +def calibrate_consensus(epubs: list[Path], samples: int = 150, preset: str | None = None, synthetic: bool = True, + report: Path | None = None, workers: int = 0, quiet: bool = False, + engines: tuple[str, str] = ("local", "local2")) -> dict[str, Any]: + """Measure the two-model consensus rule on images whose TeX is known.""" + from concurrent.futures import ThreadPoolExecutor + + from .decide import consensus_thresholds, decide_consensus + from .engines import make_engine + + def log(m: str) -> None: + if not quiet: + print(m, file=sys.stderr, flush=True) + + eng = [make_engine(e) for e in engines] + for e in eng: + e.check() + log("Modelle: " + ", ".join(f"{e.name}={e.model}" for e in eng)) + rng = random.Random(42) + sets: list[tuple[str, list[tuple[str, bytes, str, bool]], str]] = [] + for epub in epubs: + items = corpus(epub) + if not items: + continue + smp = rng.sample(items, min(samples, len(items))) + sets.append((f"{epub.stem[:40]} [echt]", smp, preset or _best_preset(smp[:12]))) + if synthetic: + syn = [(n + "#synth.jpg", img, t, d) for n, _d, t, d in smp for img in [synth_jpeg(t, d, rng)] if img] + sets.append((f"{epub.stem[:40]} [dvipng, Gehrke-Stil]", syn, "cm")) + summary: dict[str, Any] = {"engines": [f"{e.name}={e.model}" for e in eng], "sets": []} + for label, items, p in sets: + log(f"{label}: {len(items)} Formeln lesen …") + readings: list[list[dict[str, Any]]] = [[{}] * len(items) for _ in eng] + + def ask(k: int, i: int) -> None: + n, d, _t, disp = items[i] + readings[k][i] = eng[k].recognize(d, n, "", disp).to_json() + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(lambda ki: ask(*ki), [(k, i) for k in range(len(eng)) for i in range(len(items))])) + jobs = [(n, d, t, disp, p, readings[0][i], readings[1][i]) for i, (n, d, t, disp) in enumerate(items)] + with ProcessPoolExecutor(max_workers=workers or max(1, (os.cpu_count() or 2) - 2)) as ex: + results = list(ex.map(_consensus_eval, jobs)) + s = _summarize_consensus(label, results, consensus_thresholds, decide_consensus) + summary["sets"].append(s) + s["results"] = results + _print_consensus(s, log) + if report: + report.write_text(json.dumps(summary, ensure_ascii=False, indent=1, default=str), encoding="utf-8") + log(f"Bericht: {report}") + return summary + + +def _summarize_consensus(label: str, results: list[dict[str, Any]], consensus_thresholds: Any, + decide_consensus: Any) -> dict[str, Any]: + n = len(results) + out: dict[str, Any] = {"label": label, "formulas": n, + "a_correct": sum(r["a_ok"] for r in results), "b_correct": sum(r["b_ok"] for r in results), + "agree": sum(bool(r["agree"]) for r in results), + "agree_wrong": sum(bool(r["agree"]) and not r["a_ok"] for r in results), + "sweep": []} + out["agree_wrong_examples"] = [{"truth": r["truth"], "a": r["a"].get("tex"), "b": r["b"].get("tex")} + for r in results if r["agree"] and not r["a_ok"]][:10] + for pm in (0.0, 0.8, 0.9, 0.95, 0.98): + for pn in (0.0, 0.2, 0.3, 0.5): + th = consensus_thresholds({"min_p_mean": pm, "min_p_min": pn}) + acc = wrong = 0 + for r in results: + agree = [(0, 1, r["agree"])] if r["agree"] else [] + st, _reason, _use = decide_consensus([r["a"], r["b"]], agree, r["veto"], True, None, th) + if st == "auto_accepted": + acc += 1 + wrong += not r["a_ok"] + out["sweep"].append({"min_p_mean": pm, "min_p_min": pn, "accepted": acc, "wrong": wrong}) + return out + + +def _print_consensus(s: dict[str, Any], log: Any) -> None: + n = max(1, s["formulas"]) + log(f" {s['label']}: {s['formulas']} Formeln") + log(f" Modell A richtig {s['a_correct'] / n:.0%}, Modell B richtig {s['b_correct'] / n:.0%}, " + f"einig {s['agree'] / n:.0%}, davon FALSCH {s['agree_wrong']}") + for ex in s["agree_wrong_examples"][:5]: + log(f" einig, aber falsch: {ex['a']} (richtig: {ex['truth']})") + log(" Schwellen (Mittel/Minimum) → automatisch angenommen / davon falsch:") + for row in s["sweep"]: + log(f" p_mean ≥ {row['min_p_mean']:.2f}, p_min ≥ {row['min_p_min']:.2f}: " + f"{row['accepted']:4d} ({row['accepted'] / n:.0%}) / {row['wrong']}") diff --git a/formula_img2latex/cli.py b/formula_img2latex/cli.py index fd11edb..548bfc1 100644 --- a/formula_img2latex/cli.py +++ b/formula_img2latex/cli.py @@ -121,6 +121,8 @@ def build_parser() -> argparse.ArgumentParser: help="zusätzlich kleine JPEG-Nachbildungen (wie Gehrke) aus denselben Formeln erzeugen") sp.add_argument("--report", type=Path, default=None, help="Bericht als JSON speichern") sp.add_argument("--workers", type=int, default=0) + sp.add_argument("--consensus", action="store_true", + help="Konsensregel messen: beide lokalen Modelle (local, local2) lesen die Formeln") sp.add_argument("--quiet", "-q", action="store_true") return p @@ -160,6 +162,11 @@ def main(argv: list[str] | None = None) -> int: def _dispatch(args: argparse.Namespace) -> int: from . import pipeline + if args.command == "calibrate" and args.consensus: + from .calibrate import calibrate_consensus + calibrate_consensus(args.epubs, samples=args.samples, preset=None if args.preset == "auto" else args.preset, + synthetic=args.synthetic, report=args.report, workers=args.workers, quiet=args.quiet) + return EXIT_OK if args.command == "calibrate": from .calibrate import calibrate calibrate(args.epubs, samples=args.samples, preset=None if args.preset == "auto" else args.preset, diff --git a/formula_img2latex/compare.py b/formula_img2latex/compare.py index 526aa58..1bdebae 100644 --- a/formula_img2latex/compare.py +++ b/formula_img2latex/compare.py @@ -34,6 +34,10 @@ BIN_THRESHOLD = 0.4 MAX_ASPECT_DEVIATION = 0.25 TARGET_STROKE = 2.0 # strokes thicker than this (px) are scaled down before comparing OVERSAMPLE = 4 +ELASTIC_BAND = 0.4 # max. drift of the column/row mapping, share of the formula size +ELASTIC_STEP = 0.05 # cost of every stretched or compressed step (ink² units) +ELASTIC_INK = 0.5 # extra stretch cost per unit of ink in the stretched column +CONTRAST_BLUR = 0.7 # px; damps hinting/antialiasing differences in the contrast test SUBPIXEL_MAX_HEIGHT = 150 # quarter-pixel alignment only for small images (px) @@ -45,6 +49,10 @@ class Comparison: original: np.ndarray | None = None # aligned canvases, for diff images candidate: np.ndarray | None = None mismatch: np.ndarray | None = None + raw: np.ndarray | None = None # rigidly aligned candidate before elastic warping + maps: tuple[np.ndarray, np.ndarray] | None = None # (columns, rows) of the elastic warp + margin: int = 0 + noise: float | None = None # cached noise level, see noise_level() def stroke_width(ink: np.ndarray) -> float: @@ -121,12 +129,77 @@ def _aligned_candidate(orig: np.ndarray, hi: np.ndarray, scale: float, margin: i return best, best_score +def _dtw_map(o: np.ndarray, c: np.ndarray, band_frac: float = ELASTIC_BAND, step_cost: float = ELASTIC_STEP, + ink_cost: float = ELASTIC_INK) -> tuple[np.ndarray, float]: + """Monotone mapping orig column i -> candidate column map[i] (dynamic time warping). + + Each original column advances the candidate by 0 (stretch), 1, 2 or 3 columns. Blank or + uniform columns (spacing, arrow shafts, fraction bars) can therefore grow or shrink, while + every candidate column that is skipped costs its full ink, so no symbol can be dropped and + no order can change. Stretching costs in proportion to the column's ink, so warping happens + in spacing and thin strokes, not inside glyphs. Returns the map and the share of non-diagonal + steps. + """ + n, m = o.shape[1], c.shape[1] + oc, cc = o.T.astype(np.float64), c.T.astype(np.float64) + cost = (oc * oc).sum(1)[:, None] + (cc * cc).sum(1)[None, :] - 2.0 * oc @ cc.T + np.maximum(cost, 0.0, out=cost) + ink = (cc * cc).sum(1) + band = max(4, int(band_frac * max(n, m))) + jj = np.arange(m) + inf = np.inf + back = np.zeros((n, m), dtype=np.int8) + # start: any column, skipping the candidate ink before it + prev = cost[0] + np.concatenate([[0.0], np.cumsum(ink)[:-1]]) + prev[jj > band] = inf + for i in range(1, n): + diag = np.concatenate([[inf], prev[:-1]]) + rep = prev + step_cost + ink_cost * ink + sk2 = np.concatenate([[inf, inf], prev[:-2]]) + np.concatenate([[0.0], ink[:-1]]) + step_cost + sk3 = (np.concatenate([[inf, inf, inf], prev[:-3]]) + np.concatenate([[0.0], ink[:-1]]) + + np.concatenate([[0.0, 0.0], ink[:-2]]) + 2 * step_cost) + opts = np.stack([diag, rep, sk2, sk3]) + k = np.argmin(opts, axis=0) + cur = opts[k, jj] + cost[i] + cur[np.abs(jj - i * m / n) > band] = inf + back[i] = k + prev = cur + tail = np.concatenate([np.cumsum(ink[::-1])[::-1][1:], [0.0]]) # candidate ink after the end + j = int(np.argmin(prev + tail)) + path = np.empty(n, dtype=np.int64) + nondiag = 0 + for i in range(n - 1, -1, -1): + path[i] = j + if i == 0: + break + k = int(back[i, j]) + nondiag += k != 0 + j -= (1, 0, 2, 3)[k] + j = max(j, 0) + return path, nondiag / max(1, n - 1) + + +def elastic_warp(o: np.ndarray, c: np.ndarray) -> tuple[np.ndarray, float, tuple[np.ndarray, np.ndarray]]: + """Warp candidate canvas *c* onto original canvas *o*: columns first, then rows.""" + cols, sc = _dtw_map(o, c) + w = c[:, cols] + rows, sr = _dtw_map(o.T, w.T) + return w[rows, :], max(sc, sr), (cols, rows) + + def compare(orig_ink: np.ndarray, rendered: Rendered, tolerance: float | None = None, - px_per_pt: float | None = None) -> Comparison: + px_per_pt: float | None = None, elastic: bool = True, warp_from: Comparison | None = None + ) -> Comparison: """Compare a trimmed original ink image with a rendered candidate. *px_per_pt* fixes the scale (e.g. for confusable variants of an already aligned candidate: - same font size, so only the position is searched). + same font size, so only the position is searched). With *elastic* (default) spacing, arrow + and bar lengths may differ: the candidate is warped onto the original before measuring, so + only the symbols and their relative positions count. + + *warp_from*: reuse the scale, canvas and elastic warp of an already compared candidate. Used + for its confusable variants, which must not get their own warp – a free warp would let a + wrong symbol bend itself into the right one. """ if not rendered.ok: return Comparison(False, rendered.error or "nicht gerendert") @@ -153,6 +226,8 @@ def compare(orig_ink: np.ndarray, rendered: Rendered, tolerance: float | None = sx = sy margin = int(np.ceil(max(h, w) * 0.06)) + 3 + if warp_from is not None: + px_per_pt = warp_from.metrics["px_per_pt"] if px_per_pt: lo = up = px_per_pt * norm / zoom rounds, step = 2, lo * 0.004 @@ -163,6 +238,8 @@ def compare(orig_ink: np.ndarray, rendered: Rendered, tolerance: float | None = grid = list(np.linspace(lo, up, 9)) # the scaled candidate must always fit onto the padded original margin = max(margin, int(np.ceil(max(hi.shape[1] * up - w, hi.shape[0] * up - h, 0))) // 2 + 3) + if warp_from is not None: + margin = warp_from.margin # same canvas as the candidate best_scale, best_score, best_cand = lo, -np.inf, None for _round in range(rounds): # coarse grid, then finer grids around the best scale for s in grid: @@ -174,6 +251,27 @@ def compare(orig_ink: np.ndarray, rendered: Rendered, tolerance: float | None = if best_cand is None: return Comparison(False, "Kandidat passt nicht auf das Original", metrics) opad = _pad(orig, margin) + rigid_ncc = _ncc(opad, best_cand) + warp_share = 0.0 + raw, maps = best_cand, None + if warp_from is not None and warp_from.maps is not None and warp_from.raw is not None: + raw = _register(best_cand, warp_from.raw) + cols, rows = warp_from.maps + best_cand = raw[:, cols][rows, :] + maps = warp_from.maps + elif elastic: + # The rigid scale is a compromise when the layout differs; glyph height (sy) is the better + # guide. Try both and keep what explains the original best after warping. + scales = [best_scale] if px_per_pt else sorted({best_scale, sy * 0.98, sy, sy * 1.02}) + best_e = None + for s in scales: + cand = best_cand if s == best_scale else _aligned_candidate(orig, hi, float(s), margin)[0] + warped, share, mp = elastic_warp(opad, cand) + score = _ncc(opad, warped) + if best_e is None or score > best_e[0]: + best_e = (score, s, warped, share, cand, mp) + assert best_e is not None + _, best_scale, best_cand, warp_share, raw, maps = best_e tol = tolerance if tolerance is not None else max(1.0, 0.6 * sw) ob, rb = opad > BIN_THRESHOLD, best_cand > BIN_THRESHOLD @@ -221,46 +319,57 @@ def compare(orig_ink: np.ndarray, rendered: Rendered, tolerance: float | None = + max(0.0, abs(hi.shape[0] * best_scale - h) - 1.5) / h, "px_per_pt": best_scale * zoom / norm, "tolerance": tol, + "rigid_ncc": rigid_ncc, + "warp_share": warp_share, + "elastic": 1.0 if elastic else 0.0, }) - return Comparison(True, None, metrics, opad, best_cand, mismatch) + return Comparison(True, None, metrics, opad, best_cand, mismatch, raw, maps, margin) def contrast(a: Comparison, b: Comparison, max_shift: float = 0.5) -> float | None: - """How much better candidate *a* explains the original than candidate *b*, in [-1, 1]. + """Evidence that candidate *a* explains the original better than variant *b*. - Only the pixels where the two renders differ are looked at (a likelihood-ratio test on the - discriminating region), so a wrong subscript digit is not drowned by the rest of the formula. - Returns None if the two renders are identical at this resolution (equivalent spellings such as - \\le and \\leq). + Both are warped with the same elastic mapping (warp_from), so they differ only where the + symbols differ. Inside a glyph-sized window around that region the squared errors of both + are compared. The difference is measured in units of the candidate's typical residual (the + noise level of the whole formula: JPEG, hinting, antialiasing) per differing pixel: + + evidence = (err_b - err_a) / (noise² · differing pixels) + + Positive: the original shows *a*; negative: it shows *b*; near 0: indistinguishable at this + resolution. Returns None if the two renders are identical (equivalent spellings). """ if not (a.ok and b.ok) or a.original is None or b.original is None: - return 1.0 if a.ok and not b.ok else (-1.0 if b.ok and not a.ok else 0.0) + return 99.0 if a.ok and not b.ok else (-99.0 if b.ok and not a.ok else 0.0) assert a.candidate is not None and b.candidate is not None - a_orig, a_cand, b_cand = a.original, a.candidate, b.candidate if a.original.shape != b.original.shape: - # Both canvases are the same original with different symmetric margins: crop to the smaller. - hh = min(a.original.shape[0], b.original.shape[0]) - ww = min(a.original.shape[1], b.original.shape[1]) - - def crop(x: np.ndarray) -> np.ndarray: - dy, dx = (x.shape[0] - hh) // 2, (x.shape[1] - ww) // 2 - return x[dy:dy + hh, dx:dx + ww] - - a_orig, a_cand, b_cand = crop(a.original), crop(a.candidate), crop(b.candidate) - b_cand = _register(b_cand, a_cand) - region = np.abs(a_cand - b_cand) > 0.2 + return 99.0 # the variant does not even fit onto the candidate's canvas + ao, ac = a.original, a.candidate + bc = _register(b.candidate, ac) + region = np.abs(ac - bc) > 0.2 if not region.any(): return None - # Window around the differing pixels; the variants are rendered layout-preserving (see - # confusables.probe), so the difference stays local to the changed symbol. - g = glyph_size(a_orig > BIN_THRESHOLD) + g = glyph_size(ao > BIN_THRESHOLD) ys, xs = np.nonzero(region) y0, y1 = max(0, ys.min() - g), min(region.shape[0], ys.max() + g + 1) x0, x1 = max(0, xs.min() - g), min(region.shape[1], xs.max() + g + 1) - o = a_orig[y0:y1, x0:x1] - ea = _best_shift_error(o, a_cand[y0:y1, x0:x1], max_shift) - eb = _best_shift_error(o, b_cand[y0:y1, x0:x1], max_shift) - return (eb - ea) / (eb + ea) if eb + ea > 0 else 0.0 + o = ao[y0:y1, x0:x1] + ea = _best_shift_error(o, ac[y0:y1, x0:x1], max_shift) + eb = _best_shift_error(o, bc[y0:y1, x0:x1], max_shift) + return (eb - ea) / (noise_level(a) * float(region.sum())) + + +def noise_level(c: Comparison) -> float: + """Mean squared residual (blurred) of a candidate on the ink support of the whole formula.""" + if c.noise is None: + assert c.original is not None and c.candidate is not None + o, r = c.original, c.candidate + if CONTRAST_BLUR: + o = cv2.GaussianBlur(o, (0, 0), CONTRAST_BLUR) + r = cv2.GaussianBlur(r, (0, 0), CONTRAST_BLUR) + support = cv2.dilate(((c.original > 0.1) | (c.candidate > 0.1)).astype(np.uint8), np.ones((3, 3), np.uint8)) > 0 + c.noise = float(((o - r) ** 2)[support].mean()) + 1e-4 if support.any() else 1.0 + return c.noise def _shift(img: np.ndarray, dy: float, dx: float) -> np.ndarray: @@ -297,6 +406,9 @@ _LOCAL_SHIFTS = [(dy * 0.25, dx * 0.25) for dy in range(-4, 5) for dx in range(- def _best_shift_error(o: np.ndarray, win: np.ndarray, max_shift: float) -> float: """Smallest squared error of window *win* against *o* over small sub-pixel shifts.""" + if CONTRAST_BLUR: + o = cv2.GaussianBlur(o, (0, 0), CONTRAST_BLUR) + win = cv2.GaussianBlur(win, (0, 0), CONTRAST_BLUR) pad = 2 big = np.pad(win, pad) best = float("inf") diff --git a/formula_img2latex/confusables.py b/formula_img2latex/confusables.py index cea3878..6401fa0 100644 --- a/formula_img2latex/confusables.py +++ b/formula_img2latex/confusables.py @@ -71,6 +71,23 @@ GROUPS: list[tuple[str, ...]] = [ ("'", "\\prime", ","), ] +# Same symbol in another length, size or upright/italic shape. Per the project's acceptance rule +# ("every symbol right and in the right place; spacing and lengths may differ") such variants +# never count as doubt; they may still replace the candidate if they match the image better. +# Bold vs. normal and \\mathbb/\\mathcal/\\mathfrak vs. normal stay distinct (vector vs. scalar, ℝ vs. R). +EQUIVALENT: list[set[str]] = [ + {"\\vec", "\\overrightarrow"}, {"\\bar", "\\overline"}, {"\\hat", "\\widehat"}, {"\\tilde", "\\widetilde"}, + {"\\to", "\\rightarrow", "\\longrightarrow"}, {"\\frac", "\\dfrac", "\\tfrac"}, {"\\le", "\\leq"}, + {"\\ge", "\\geq"}, {"\\ne", "\\neq"}, {"\\dots", "\\ldots"}, {"'", "\\prime"}, + {"\\boldsymbol", "\\mathbf", "\\bm"}, {"\\mathrm", "\\mathit"}, +] +_UPRIGHT = {"\\mathrm", "\\mathit"} + + +def _equivalent(t: str, alt: str) -> bool: + return any(t in g and alt in g for g in EQUIVALENT) + + _INDEX: dict[str, list[str]] = {} for _g in GROUPS: for _t in _g: @@ -97,6 +114,7 @@ class Variant: kind: str # e.g. "subst 3->8", "sub<->sup", "delete -", "insert -" priority: int # lower = tried first when capping probe: str = "" # layout-preserving TeX for the contrast test ("" = render tex itself) + equivalent: bool = False # same symbols, only length/size/upright-italic differs (see EQUIVALENT) @property def render_tex(self) -> str: @@ -172,10 +190,11 @@ def variants(tex: str, limit: int = MAX_VARIANTS) -> list[Variant]: toks = tokenize(tex) out: dict[str, Variant] = {} - def add(new_toks: list[str], kind: str, prio: int, probe_toks: list[str] | None = None) -> None: + def add(new_toks: list[str], kind: str, prio: int, probe_toks: list[str] | None = None, + equivalent: bool = False) -> None: s = "".join(new_toks).strip() if s and s != tex.strip() and s not in out: - out[s] = Variant(s, kind, prio, "".join(probe_toks).strip() if probe_toks else "") + out[s] = Variant(s, kind, prio, "".join(probe_toks).strip() if probe_toks else "", equivalent) def probe(i: int, wrap: str | None) -> list[str] | None: if wrap is None: @@ -192,16 +211,16 @@ def variants(tex: str, limit: int = MAX_VARIANTS) -> list[Variant]: for alt in _INDEX.get(t, []): prio = 1 if t.isdigit() or not t.isalpha() else 2 add(toks[:i] + [_needs_space(prev, alt, nxt)] + toks[i + 1:], f"subst {t}->{alt}", prio, - probe(i, _probe_subst(t, alt))) + probe(i, _probe_subst(t, alt)), _equivalent(t, alt)) if len(t) == 1 and t.isalpha() and not _in_font_arg(toks, i): for font in (_FONTS_UPPER if t.isupper() else _FONTS_LOWER): add(toks[:i] + [font + "{" + t + "}"] + toks[i + 1:], f"font {font}{{{t}}}", 3, - probe(i, _probe_subst(t, font + "{" + t + "}"))) + probe(i, _probe_subst(t, font + "{" + t + "}")), font in _UPRIGHT) if len(t) == 1 and t.isalpha() and t.lower() not in "xvnioepkwatbszg": add(toks[:i] + [t.swapcase()] + toks[i + 1:], f"case {t}", 3, probe(i, _probe_subst(t, t.swapcase()))) if t in _DELETABLE: add(toks[:i] + toks[i + 1:], f"delete {t}", 1, - probe(i, None if t in _TAKES_ARG or t.startswith("\\\\,") else _phantom(t))) + probe(i, None if t in _TAKES_ARG or t.startswith("\\\\,") else _phantom(t)), t in _UPRIGHT) if t in _MINUS_AFTER and nxt not in ("-", "+"): add(toks[:i + 1] + ["-"] + toks[i + 1:], f"insert - after {t}", 2) if len(t) == 1 and t.isalpha() and nxt != "'": diff --git a/formula_img2latex/consensus.py b/formula_img2latex/consensus.py new file mode 100644 index 0000000..010098a --- /dev/null +++ b/formula_img2latex/consensus.py @@ -0,0 +1,145 @@ +"""Do two LaTeX readings of the same image say the same thing? + +Acceptance rule of the project: every symbol must be right and in the right place relative to the +others; spacing and lengths may differ. Two readings therefore count as the same if + +1. their canonical forms are equal (spacing, \\left/\\right, braces, upright vs. italic, length + variants such as \\vec/\\overrightarrow removed or unified), or +2. they render to the same picture (render against render: exact, no noisy original involved), + e.g. \\sin x vs. \\operatorname{sin} x. +""" + +from __future__ import annotations + +import re + +import numpy as np + +from .compare import _register +from .confusables import tokenize +from .imageio import trim +from .render import render_many +from .tex import prepare_for_render, split_tag, strip_delimiters + +# tokens that carry no symbol +_DROP = {"\\,", "\\;", "\\:", "\\!", "\\ ", "~", "\\quad", "\\qquad", "\\left", "\\right", "\\big", "\\Big", "\\bigg", + "\\Bigg", "\\bigl", "\\bigr", "\\Bigl", "\\Bigr", "\\biggl", "\\biggr", "\\displaystyle", "\\textstyle", + "\\scriptstyle", "\\limits", "\\nolimits", "\\middle", "\\hfill", "\\enspace", "\\thinspace", "\\medspace"} +# wrappers whose argument is kept (upright vs. italic is the same symbol) +_UNWRAP = {"\\mathrm", "\\mathit", "\\operatorname", "\\textrm", "\\textit", "\\mathnormal", "\\mbox", "\\text", + "\\textnormal", "\\textup"} +_SYNONYM = { + "\\overrightarrow": "\\vec", "\\overline": "\\bar", "\\widehat": "\\hat", "\\widetilde": "\\tilde", + "\\rightarrow": "\\to", "\\longrightarrow": "\\to", "\\dfrac": "\\frac", "\\tfrac": "\\frac", + "\\leq": "\\le", "\\geq": "\\ge", "\\neq": "\\ne", "\\ldots": "\\dots", "\\prime": "'", + "\\mathbf": "\\boldsymbol", "\\bm": "\\boldsymbol", "\\lbrace": "\\{", "\\rbrace": "\\}", "\\vert": "|", + "\\lvert": "|", "\\rvert": "|", "\\Vert": "\\|", "\\lVert": "\\|", "\\rVert": "\\|", "\\gets": "\\leftarrow", + "\\land": "\\wedge", "\\lor": "\\vee", "\\lnot": "\\neg", "\\leqslant": "\\le", "\\geqslant": "\\ge", + "\\varnothing": "\\emptyset", "\\colon": ":", "\\cdotp": "\\cdot", "\\ast": "*", + "\\iff": "\\Leftrightarrow", "\\Longleftrightarrow": "\\Leftrightarrow", "\\implies": "\\Rightarrow", + "\\Longrightarrow": "\\Rightarrow", "\\impliedby": "\\Leftarrow", "\\Longleftarrow": "\\Leftarrow", + "\\longleftarrow": "\\leftarrow", "\\longmapsto": "\\mapsto", "\\longleftrightarrow": "\\leftrightarrow", + "\\cdots": "\\dots", "\\dotsc": "\\dots", "\\dotsb": "\\dots", "\\mid": "|", +} +# matrix environments with built-in delimiters -> delimiters + plain matrix +_ENV_DELIMS = {"pmatrix": ("(", ")"), "bmatrix": ("[", "]"), "Bmatrix": ("\\{", "\\}"), "vmatrix": ("|", "|"), + "Vmatrix": ("\\|", "\\|"), "matrix": ("", "")} + + +_UMLAUT = {"a": "ä", "o": "ö", "u": "ü", "A": "Ä", "O": "Ö", "U": "Ü"} +_BOLD = {"\\boldsymbol", "\\mathbf", "\\bm", "\\pmb"} +# commands that never take an argument: a group right after them is free-standing +_NOARG = {"\\cdot", "\\times", "\\pm", "\\mp", "\\le", "\\ge", "\\ne", "\\in", "\\to", "\\Rightarrow", + "\\Leftrightarrow", "\\approx", "\\equiv", "\\sim", "\\quad", "\\qquad", "\\cup", "\\cap", "\\setminus"} + + +def canonical(tex: str, bold: bool = False) -> str: + """Canonical token string of a formula (see module docstring). + + *bold*: whether bold vs. normal counts (False: \\boldsymbol{v} and v are the same reading). + """ + body, tag = split_tag(strip_delimiters(tex)) + body = prepare_for_render(body) + body = re.sub(r'\{?\\"\{?([aouAOU])\}?\}?', lambda m: _UMLAUT[m.group(1)], body) # f{\"u}r -> für + body = re.sub(r"\{?\\ss\{?\}?\}?", "ß", body) + body = re.sub(r"\{([^{}]*)\}\s*\\atop\s*\{([^{}]*)\}", r"\\substack{\1\\\\\2}", body) + body = re.sub(r"\\begin\{array\}\{[^{}]*\}", r"\\begin{matrix}", body).replace("\\end{array}", "\\end{matrix}") + for env, (lft, rgt) in _ENV_DELIMS.items(): + body = body.replace(f"\\begin{{{env}}}", f"{lft} \\begin{{MATRIX}}").replace(f"\\end{{{env}}}", f"\\end{{MATRIX}} {rgt}") + body = re.sub(r"\\not\s*=", r"\\ne ", body) + body = re.sub(r"\\not\s*\\in(?![A-Za-z])", r"\\notin ", body) + toks = [t for t in tokenize(body) if not t.isspace()] + toks = [_SYNONYM.get(t, t) for t in toks if t not in _DROP] + unwrap = _UNWRAP if bold else _UNWRAP | _BOLD + toks = [t for t in toks if t not in unwrap] # keep the argument group, unwrap below + toks = _flatten_groups(toks) + s = " ".join(toks) + return s + (f"#tag({tag})" if tag else "") + + +def _parse(toks: list[str], i: int = 0) -> tuple[list, int]: + """Token list -> nested lists for {...} groups.""" + out: list = [] + while i < len(toks): + t = toks[i] + if t == "{": + sub, i = _parse(toks, i + 1) + out.append(sub) + continue + if t == "}": + return out, i + 1 + out.append(t) + i += 1 + return out, i + + +def _emit(nodes: list) -> list[str]: + """Serialize: argument groups keep braces (if longer than one token), free groups dissolve.""" + out: list[str] = [] + prev: object = None + for n in nodes: + if isinstance(n, list): + inner = _emit(n) + is_arg = isinstance(prev, list) or prev in ("^", "_", "]") or ( + isinstance(prev, str) and prev.startswith("\\") and prev[1:].isalpha() and prev not in _NOARG) + if len(inner) <= 1 or not is_arg: + out.extend(inner) + else: + out.extend(["{", *inner, "}"]) + else: + out.append(n) + prev = n + return out + + +def _flatten_groups(toks: list[str]) -> list[str]: + return _emit(_parse(toks)[0]) + + +def render_equal(a: str, b: str, display: bool, preset: str = "cm", zoom: float = 4.0) -> bool: + """True if both readings render to the same picture (render vs. render).""" + ra, rb = render_many([(a, display), (b, display)], preset) + if not (ra.ok and rb.ok): + return False + ia, ib = trim(ra.rasterize(zoom)), trim(rb.rasterize(zoom)) + if ia.size == 0 or ib.size == 0 or abs(ia.shape[0] - ib.shape[0]) > 2 or abs(ia.shape[1] - ib.shape[1]) > 2: + return False + h, w = max(ia.shape[0], ib.shape[0]) + 6, max(ia.shape[1], ib.shape[1]) + 6 + ca = np.zeros((h, w), np.float32) + cb = np.zeros((h, w), np.float32) + ca[3:3 + ia.shape[0], 3:3 + ia.shape[1]] = ia + cb[3:3 + ib.shape[0], 3:3 + ib.shape[1]] = ib + cb = _register(cb, ca, max_px=2) + diff = np.abs(ca - cb) > 0.5 + return int(diff.sum()) <= max(2, int(0.002 * (ca > 0.5).sum())) + + +def same_reading(a: str, b: str, display: bool, preset: str = "cm", bold: bool = False) -> str | None: + """'normalform' or 'render' if both readings mean the same, else None.""" + if not a or not b: + return None + if canonical(a, bold) == canonical(b, bold): + return "normalform" + if render_equal(strip_delimiters(a), strip_delimiters(b), display, preset): + return "render" + return None diff --git a/formula_img2latex/decide.py b/formula_img2latex/decide.py index 31082b3..a58e888 100644 --- a/formula_img2latex/decide.py +++ b/formula_img2latex/decide.py @@ -12,8 +12,9 @@ from __future__ import annotations from typing import Any DEFAULTS: dict[str, float] = { - "min_contrast": 0.40, # accept: worst win over a confusable variant (calibrated: worst rival 0.35) - "review_contrast": -0.10, # below this a variant explains the image better -> reject + "min_contrast": 2.5, # accept: weakest evidence against any confusable variant, one-line formulas + "min_contrast_multiline": 4.5, # the same for formulas with several lines (aligned, matrices) + "review_contrast": -1.0, # below this a variant explains the image better -> reject "min_ncc": 0.85, # accept "review_ncc": 0.15, # below: nothing in common (wrong image/figure) -> keep image "max_aspect_dev": 0.08, # accept: proportions within 8 % @@ -43,14 +44,17 @@ def decide(ver: dict[str, Any], pandoc_ok: bool, book_px_per_pt: float | None, problems.append("Pandoc kann die Formel nicht nach MathML umsetzen") if ncc < th["min_ncc"]: problems.append(f"Ähnlichkeit {ncc:.3f} < {th['min_ncc']}") - if aspect_dev > th["max_aspect_dev"]: + # With elastic comparison, lengths and spacing may differ by design; only symbols and their + # relative positions count. + if aspect_dev > th["max_aspect_dev"] and not m.get("elastic"): problems.append(f"Proportionen weichen um {aspect_dev:.0%} ab") if book_px_per_pt: dev = abs(float(m.get("px_per_pt", book_px_per_pt)) / book_px_per_pt - 1.0) if dev > th["max_scale_dev"]: problems.append(f"Schriftgröße weicht um {dev:.0%} vom Buch ab (Stil?)") - if mc is not None and mc < th["min_contrast"]: - problems.append(f"nicht eindeutig gegenüber „{ver.get('worst_variant')}“ (Kontrast {mc:.2f})") + need = th["min_contrast_multiline"] if "\\\\" in (ver.get("tex") or "") else th["min_contrast"] + if mc is not None and mc < need: + problems.append(f"nicht eindeutig gegenüber „{ver.get('worst_variant')}“ (Beleg {mc:.2f} < {need})") if not problems: why = "alle Prüfungen bestanden" if mc is None: @@ -58,3 +62,88 @@ def decide(ver: dict[str, Any], pandoc_ok: bool, book_px_per_pt: float | None, return "auto_accepted", why hopeless = ncc < th["review_ncc"] or (mc is not None and mc < th["review_contrast"]) return ("rejected_keep_image" if hopeless else "needs_review"), "; ".join(problems) + + +# --------------------------------------------------------------------------- consensus mode + +CONSENSUS_DEFAULTS: dict[str, float] = { + "min_p_mean": 0.90, # geometric mean of the LaTeX token probabilities, each reading + "min_p_min": 0.30, # least certain LaTeX token, each reading + "veto_ncc": 0.75, # rough image check: the agreed reading must resemble the original + "veto_scale_dev": 0.15, # and have the book's font size (px/pt) within 15 % + "bold_relevant": 0, # 1: bold vs. normal counts as a different symbol + "require_third": 0, # 1: a third reading (e.g. cloud) must agree as well +} + + +def consensus_thresholds(overrides: dict[str, Any] | None = None) -> dict[str, float]: + t = dict(CONSENSUS_DEFAULTS) + for k, v in (overrides or {}).items(): + if k in t and v is not None: + t[k] = float(v) + return t + + +def _sure(r: dict[str, Any], th: dict[str, float]) -> str | None: + """None if the reading is confident enough, else a German reason.""" + pm, pn = r.get("p_mean"), r.get("p_min") + if pm is None or pn is None: + return f"{r.get('model', '?')}: keine Token-Wahrscheinlichkeiten" + if pm < th["min_p_mean"] or pn < th["min_p_min"]: + return f"{r.get('model', '?')} unsicher (Mittel {pm:.2f}, Minimum {pn:.2f})" + return None + + +def decide_consensus(readings: list[dict[str, Any]], agree: list[tuple[int, int, str]], veto: dict[str, Any] | None, + pandoc_ok: bool, book_px_per_pt: float | None, th: dict[str, float] + ) -> tuple[str, str, int | None]: + """Decide from several model readings of one image. + + *readings*: latest candidate per model ({model, tex, is_formula, p_min, p_mean}), the two + primary models first. *agree*: pairs (i, j, how) of readings that say the same. *veto*: + verification of the chosen reading (rough image check). Returns (status, reason, index of + the reading to use or None). + """ + if len(readings) < 2: + return "needs_review", "nur eine Lesart (zweites Modell fehlt)", 0 if readings and readings[0].get("tex") else None + flags = [r.get("is_formula") for r in readings[:2]] + if all(f is False for f in flags): + return "not_formula", "beide Modelle: keine Formel", None + pair = None + for i, j, how in agree: + if {i, j} == {0, 1}: + pair = (i, j, how) + break + tie = "" + if pair is None: + for i, j, how in agree: # tie-break: a third reading agrees with one of the two + if 2 in (i, j) or i >= 2 or j >= 2: + pair, tie = (i, j, how), " (Stichwahl)" + break + if pair is None: + texts = " | ".join(f"{r.get('model', '?')}: {r.get('tex') or '–'}" for r in readings[:3]) + return "needs_review", f"Modelle uneinig: {texts}", 0 if readings[0].get("tex") else 1 + i, j, how = pair + if th.get("require_third") and not tie: + third = [k for k in range(2, len(readings))] + if not third: + return "needs_review", f"einig ({how}), aber dritte Lesart fehlt", i + if not any({a, b} in ({0, k}, {1, k}) for a, b, _ in agree for k in third): + texts = " | ".join(f"{r.get('model', '?')}: {r.get('tex') or '–'}" for r in readings[2:3]) + return "needs_review", f"zwei Modelle einig ({how}), dritte Lesart weicht ab: {texts}", i + tie = " (dritte Lesart bestätigt)" + use = i if (readings[i].get("p_mean") or 0) >= (readings[j].get("p_mean") or 0) else j + problems = [p for p in (_sure(readings[i], th), _sure(readings[j], th)) if p] + if not pandoc_ok: + problems.append("Pandoc kann die Formel nicht nach MathML umsetzen") + if veto is None or not veto.get("ok"): + problems.append("Bildabgleich nicht möglich" + (f": {veto.get('reason')}" if veto and veto.get("reason") else "")) + else: + m = veto.get("metrics", {}) + if m.get("ncc", 0.0) < th["veto_ncc"]: + problems.append(f"Bild-Veto: Ähnlichkeit {m.get('ncc', 0.0):.2f} < {th['veto_ncc']}") + if book_px_per_pt and abs(m.get("px_per_pt", book_px_per_pt) / book_px_per_pt - 1) > th["veto_scale_dev"]: + problems.append(f"Bild-Veto: Schriftgröße weicht um {abs(m['px_per_pt'] / book_px_per_pt - 1):.0%} ab") + if problems: + return "needs_review", f"einig ({how}){tie}; " + "; ".join(problems), use + return "auto_accepted", f"zwei Modelle einig ({how}){tie}, sicher, Bild-Veto bestanden", use diff --git a/formula_img2latex/engines/__init__.py b/formula_img2latex/engines/__init__.py index 9d0dc1f..4b04a26 100644 --- a/formula_img2latex/engines/__init__.py +++ b/formula_img2latex/engines/__init__.py @@ -12,18 +12,24 @@ import os from .vlm import VisionEngine DEFAULT_LOCAL_URL = "http://127.0.0.1:8011/v1" +DEFAULT_LOCAL2_URL = "http://127.0.0.1:8012/v1" DEFAULT_CLOUD_URL = "https://openrouter.ai/api/v1" DEFAULT_CLOUD_MODEL = "google/gemini-2.5-flash" def make_engine(name: str, model: str | None = None, url: str | None = None) -> VisionEngine: - """'local' (llama.cpp server) or 'cloud' (OpenRouter).""" - if name == "local": + """'local'/'local2' (two llama.cpp servers with different models) or 'cloud' (OpenRouter).""" + if name in ("local", "local2"): + suffix = "" if name == "local" else "2" return VisionEngine( - name="local", - base_url=url or os.environ.get("FORMULA_LOCAL_URL", DEFAULT_LOCAL_URL), - model=model or os.environ.get("FORMULA_LOCAL_MODEL", ""), - api_key=os.environ.get("FORMULA_LOCAL_API_KEY", ""), + name=name, + base_url=url or os.environ.get(f"FORMULA_LOCAL{suffix}_URL", + DEFAULT_LOCAL_URL if name == "local" else DEFAULT_LOCAL2_URL), + model=model or os.environ.get(f"FORMULA_LOCAL{suffix}_MODEL", ""), + api_key=os.environ.get(f"FORMULA_LOCAL{suffix}_API_KEY", ""), + # Transcribing needs no chain of thought; it costs ~10x time (Gemma 4) and flattens the + # token probabilities of the answer. FORMULA_LOCAL_THINKING=1 switches it back on. + thinking=os.environ.get(f"FORMULA_LOCAL{suffix}_THINKING", "0") == "1", ) if name == "cloud": return VisionEngine( @@ -33,4 +39,4 @@ def make_engine(name: str, model: str | None = None, url: str | None = None) -> api_key=os.environ.get("OPENROUTER_API_KEY", ""), needs_key=True, ) - raise ValueError(f"unbekannte Engine {name!r} (local, cloud)") + raise ValueError(f"unbekannte Engine {name!r} (local, local2, cloud)") diff --git a/formula_img2latex/engines/vlm.py b/formula_img2latex/engines/vlm.py index 1e44721..8bd986e 100644 --- a/formula_img2latex/engines/vlm.py +++ b/formula_img2latex/engines/vlm.py @@ -25,14 +25,17 @@ PROMPT = """Transcribe the mathematical formula in the image into LaTeX (amsmath Rules: - Reproduce exactly what is printed: every symbol, digit, sign, sub- and superscript, prime, accent and bracket. - Do not simplify, complete, reorder or correct anything. Do not add a trailing period unless one is printed. -- Use \\mathbb, \\mathcal, \\mathrm, \\boldsymbol etc. where the print uses such fonts. +- Check every letter and digit for BOLD type. Bold symbols (vectors, matrices, the zero vector) are common + and easy to miss: write them as \\boldsymbol{v}, \\boldsymbol{0}. Non-bold symbols stay plain. +- Use \\mathbb, \\mathcal, \\mathrm etc. where the print uses such fonts. - Use \\frac for fractions, \\sqrt for roots, \\begin{pmatrix}/\\begin{array}/\\begin{cases} for matrices and cases, \\begin{aligned} for several aligned lines. - Write an equation number such as (3.7) at the right margin as \\tag{3.7}. - Do not wrap the result in $ or \\[ \\]. - If the image is not a formula (photo, diagram, plot, plain text), set is_formula to false and latex to "". +- confidence: how sure you are (0-100) that every symbol of your LaTeX matches the image. {context} -Answer with JSON only: {"is_formula": true, "latex": "..."}""" +Answer with JSON only: {"is_formula": true, "latex": "...", "confidence": 0}""" CONTEXT_LINE = "- The formula is {where} this text from the book (for orientation only, do not copy it): \"{text}\"\n" @@ -48,6 +51,9 @@ class Recognition: seconds: float = 0.0 tokens_in: int = 0 tokens_out: int = 0 + p_min: float | None = None # smallest token probability inside the LaTeX string + p_mean: float | None = None # geometric mean of those token probabilities + self_conf: float | None = None # the model's own 0-100 estimate (stored, not trusted) def to_json(self) -> dict: return {k: v for k, v in self.__dict__.items()} @@ -101,9 +107,63 @@ def parse_answer(text: str) -> tuple[str | None, bool | None]: return None, None +def parse_confidence(text: str) -> float | None: + m = re.search(r'"confidence"\s*:\s*"?(\d+(?:\.\d+)?)', text) + if not m: + return None + v = float(m.group(1)) + return v * 100 if v <= 1.0 and "." in m.group(1) else v + + +def _latex_span(content: str) -> tuple[int, int] | None: + """Character span of the JSON string value of "latex" in the model answer.""" + m = re.search(r'"latex"\s*:\s*"', content) + if not m: + return None + i = m.end() + j = i + while j < len(content): + if content[j] == "\\": + j += 2 + continue + if content[j] == '"': + return i, j + j += 1 + return i, len(content) + + +def latex_token_probs(content: str, items: list[dict]) -> tuple[float | None, float | None]: + """(min, geometric mean) of the probabilities of the tokens that make up the LaTeX value. + + *items* is the OpenAI `logprobs.content` list ({"token", "logprob"}). Token texts are laid end + to end to find which tokens overlap the LaTeX string; if they do not reproduce the answer + (some servers report bytes), all tokens are used. + """ + lps = [(it.get("token", ""), it.get("logprob")) for it in items if it.get("logprob") is not None] + if not lps: + return None, None + span = _latex_span(content) + joined = "".join(t for t, _ in lps) + sel: list[float] = [] + if span and joined == content: + pos = 0 + for tok, lp in lps: + a, b = pos, pos + len(tok) + pos = b + if b > span[0] and a < span[1] and tok.strip(): + sel.append(float(lp)) + if not sel: + sel = [float(lp) for t, lp in lps if t.strip()] + if not sel: + return None, None + return float(np.exp(min(sel))), float(np.exp(np.mean(sel))) + + class VisionEngine: - def __init__(self, name: str, base_url: str, model: str, api_key: str = "", needs_key: bool = False) -> None: + def __init__(self, name: str, base_url: str, model: str, api_key: str = "", needs_key: bool = False, + thinking: bool | None = None) -> None: self.name = name + self.thinking = thinking # None: server default; False: ask llama.cpp chat templates not to "think" self.base_url = base_url.rstrip("/") self.model = model self.api_key = api_key @@ -142,6 +202,9 @@ class VisionEngine: "model": self.model, "temperature": 0, "max_tokens": 1500, + "logprobs": True, + "top_logprobs": 1, + **({"chat_template_kwargs": {"enable_thinking": self.thinking}} if self.thinking is not None else {}), "messages": [{ "role": "user", "content": [ @@ -162,12 +225,15 @@ class VisionEngine: continue r.raise_for_status() js = r.json() - text = (js["choices"][0]["message"].get("content") or "").strip() + choice = js["choices"][0] + content = choice["message"].get("content") or "" + text = content.strip() usage = js.get("usage") or {} tex, isf = parse_answer(text) + p_min, p_mean = latex_token_probs(content, ((choice.get("logprobs") or {}).get("content") or [])) return Recognition(self.name, js.get("model", self.model), tex, isf, text[:2000], None, time.time() - t0, int(usage.get("prompt_tokens") or 0), - int(usage.get("completion_tokens") or 0)) + int(usage.get("completion_tokens") or 0), p_min, p_mean, parse_confidence(text)) except (requests.RequestException, KeyError, IndexError, ValueError) as e: last_err = str(e)[:300] time.sleep(1 + attempt) diff --git a/formula_img2latex/pipeline.py b/formula_img2latex/pipeline.py index 61da9d4..7a9306b 100644 --- a/formula_img2latex/pipeline.py +++ b/formula_img2latex/pipeline.py @@ -18,7 +18,7 @@ from PIL import Image from .checks import pandoc_mathml_ok from .classify import classify from .compare import diff_image -from .decide import decide, thresholds +from .decide import consensus_thresholds, decide, decide_consensus, thresholds from .engines import make_engine from .imageio import load_ink from .render import PRESETS @@ -136,6 +136,7 @@ def recognize(source: Source, ws: Workspace, engine_name: str, model: str | None stats["tokens_in"] += rec.tokens_in stats["tokens_out"] += rec.tokens_out cand = {"engine": engine.name, "model": engine.model, "tex": rec.tex, "is_formula": rec.is_formula, + "p_min": rec.p_min, "p_mean": rec.p_mean, "self_conf": rec.self_conf, "error": rec.error, "seconds": round(rec.seconds, 2), "created_at": now_iso()} f.candidates.append(cand) if rec.error: @@ -155,13 +156,15 @@ def recognize(source: Source, ws: Workspace, engine_name: str, model: str | None # ------------------------------------------------------------------------------------ verify def comparison_pngs(c: Any) -> dict[str, bytes]: - """PNG images of an aligned comparison: original, candidate render and colored overlay.""" + """PNG images of an aligned comparison: original, candidate render (elastic and rigid), overlay.""" out: dict[str, bytes] = {} if c is None or c.original is None: return out - for key, arr in (("orig", c.original), ("render", c.candidate)): + for key, arr in (("orig", c.original), ("render", c.candidate), ("rigid", c.raw)): + if arr is None: + continue buf = io.BytesIO() - Image.fromarray(((1 - np.clip(arr, 0, 1)) * 255).astype(np.uint8), "L").save(buf, "PNG") + Image.fromarray(((1 - np.clip(arr, 0, 1)) * 255).astype(np.uint8)).save(buf, "PNG") out[key] = buf.getvalue() buf = io.BytesIO() Image.fromarray(diff_image(c, upscale=1)).save(buf, "PNG") @@ -169,15 +172,20 @@ def comparison_pngs(c: Any) -> dict[str, bytes]: return out -def _verify_job(args: tuple[str, bytes, str, list[str], bool, str]) -> tuple[str, list[dict], dict[str, bytes]]: - """Worker process: verify all distinct candidates of one formula; return records and images.""" - fid, data, name, texes, display, preset = args +def _verify_job(args: tuple) -> tuple[str, list[dict], dict[str, bytes]]: + """Worker process: verify all distinct candidates of one formula; return records and images. + + With variants=False (consensus mode) this is only the rough image check: render, elastic + comparison, scale – no confusable variants, no repair. + """ + fid, data, name, texes, display, preset = args[:6] + variants = args[6] if len(args) > 6 else True os.environ.setdefault("OMP_NUM_THREADS", "1") orig = load_ink(data, name) results = [] best = None for tex in texes: - v = verify(orig, tex, display, preset) + v = verify(orig, tex, display, preset, check_variants=variants) results.append(v.to_json()) key = (v.ok, v.min_contrast if v.min_contrast is not None else 1.0, v.metrics.get("ncc", 0.0)) if best is None or key > best[0]: @@ -229,7 +237,8 @@ def verify_all(source: Source, ws: Workspace, preset: str | None = None, include jobs = [] for f in formulas: ref = _ref_for(source, f) - jobs.append((f.id, source.original_bytes(ref), f.asset, _distinct_candidates(f), f.display, preset)) + jobs.append((f.id, source.original_bytes(ref), f.asset, _distinct_candidates(f), f.display, preset, + decide_mode(ws) == "pixel")) log(f"verify: {len(jobs)} Formeln, Schrift {preset}", quiet) by_id = {f.id: f for f in formulas} ws.review_dir.mkdir(parents=True, exist_ok=True) @@ -251,7 +260,7 @@ def verify_all(source: Source, ws: Workspace, preset: str | None = None, include progress(done, len(jobs)) elif not quiet and (done % 25 == 0 or done == len(jobs)): log(f" {done}/{len(jobs)}") - return decide_all(ws, quiet=quiet) + return decide_all(ws, quiet=quiet, preset=preset) def _cands_key(f: Formula) -> str: @@ -263,7 +272,24 @@ def _verified_current(f: Formula) -> bool: return v.get("candidates_key") == _cands_key(f) -def decide_all(ws: Workspace, quiet: bool = False) -> dict[str, int]: +def decide_mode(ws: Workspace) -> str: + return ws.settings.get("decide") or "consensus" + + +def readings_of(f: Formula) -> list[dict]: + """Latest reading per model; local engines first (primary pair), then the rest (tie-break).""" + latest: dict[str, dict] = {} + for c in f.candidates: + if c.get("error"): + continue + latest[f"{c.get('engine')}|{c.get('model')}"] = c + order = {"local": 0, "local2": 1, "cloud": 2} + return sorted(latest.values(), key=lambda c: order.get(c.get("engine", ""), 3)) + + +def decide_all(ws: Workspace, quiet: bool = False, preset: str | None = None) -> dict[str, int]: + if decide_mode(ws) == "consensus": + return decide_all_consensus(ws, quiet, preset or ws.settings.get("preset") or "cm") th = thresholds(ws.settings.get("thresholds")) formulas = [f for f in ws.all() if f.verification and f.status not in HUMAN_STATUSES] # Book-wide scale: all formula images of a book share one resolution, so px/pt is constant. @@ -294,6 +320,46 @@ def decide_all(ws: Workspace, quiet: bool = False) -> dict[str, int]: return counts +def decide_all_consensus(ws: Workspace, quiet: bool, preset: str) -> dict[str, int]: + from .consensus import same_reading + th = consensus_thresholds(ws.settings.get("consensus_thresholds")) + formulas = [f for f in ws.all() if f.candidates and f.status not in HUMAN_STATUSES] + plan = [] + for f in formulas: + rs = readings_of(f) + bold = bool(th.get("bold_relevant")) + agree = [(i, j, how) for i in range(len(rs)) for j in range(i + 1, len(rs)) + for how in [same_reading(rs[i].get("tex") or "", rs[j].get("tex") or "", f.display, preset, bold)] if how] + plan.append((f, rs, agree)) + # Book-wide scale from agreed readings that resemble their original well. + by_tex = {f.id: {r["tex"]: r for r in (f.verification or {}).get("all", [])} for f, _, _ in plan} + good = [] + for f, rs, agree in plan: + for i, j, _ in agree[:1]: + v = by_tex[f.id].get(rs[i].get("tex")) + if v and v.get("ok") and v["metrics"].get("ncc", 0) >= 0.85: + good.append(v["metrics"]["px_per_pt"]) + book_ppt = float(statistics.median(good)) if len(good) >= 5 else ws.settings.get("book_px_per_pt") + ws.settings["book_px_per_pt"] = book_ppt + chosen = [] + for f, rs, agree in plan: + status, reason, use = decide_consensus(rs, agree, None, True, book_ppt, th) # preliminary: which reading + chosen.append(rs[use]["tex"] if use is not None and rs[use].get("tex") else None) + pandoc = pandoc_mathml_ok([(t or "x", f.display) for (f, _, _), t in zip(plan, chosen)]) if plan else [] + counts: dict[str, int] = {} + for (f, rs, agree), tex, pok in zip(plan, chosen, pandoc): + veto = by_tex[f.id].get(tex) if tex else None + status, reason, use = decide_consensus(rs, agree, veto, pok, book_ppt, th) + f.status, f.reason = status, reason + f.final_tex = _with_tag(f, strip_delimiters(tex)) if status == "auto_accepted" and tex else None + f.decided_at = now_iso() + ws.save(f) + counts[status] = counts.get(status, 0) + 1 + ws.save_meta() + log("Entscheidung (Konsens): " + ", ".join(f"{k} {v}" for k, v in sorted(counts.items())), quiet) + return counts + + def _with_tag(f: Formula, tex: str) -> str: """Re-attach an equation number recognized as \\tag{...} – as visible text, like epub2md does.""" for c in f.candidates: diff --git a/formula_img2latex/verify.py b/formula_img2latex/verify.py index 3d9fb4b..758fcf1 100644 --- a/formula_img2latex/verify.py +++ b/formula_img2latex/verify.py @@ -16,7 +16,7 @@ from .compare import Comparison, compare, contrast from .confusables import variants from .render import render_many -SWITCH_MARGIN = 0.25 # a variant must beat the candidate by this much to replace it +SWITCH_MARGIN = 2.0 # evidence (see compare.contrast) by which a variant must win to replace the candidate MAX_CLIMB = 3 MIN_NCC_FOR_VARIANTS = 0.6 # hopeless candidates are not worth the variant test @@ -53,21 +53,23 @@ def _check_variants(orig: np.ndarray, cand: Comparison, tex: str, display: bool, if not vs: return None, None, None, 0, 0, None rendered = render_many([(v.render_tex, display) for v in vs], preset) - ppt = cand.metrics["px_per_pt"] min_c, worst, worst_kind, n, eq = None, None, None, 0, 0 rival: tuple[float, str] | None = None for v, r in zip(vs, rendered): if not r.ok: continue - c = contrast(cand, compare(orig, r, px_per_pt=ppt)) + c = contrast(cand, compare(orig, r, warp_from=cand)) if c is None: eq += 1 continue + if c < -SWITCH_MARGIN and (rival is None or c < rival[0]): + rival = (c, v.tex) + if v.equivalent: + eq += 1 + continue n += 1 if min_c is None or c < min_c: min_c, worst, worst_kind = c, v.tex, v.kind - if c < -SWITCH_MARGIN and (rival is None or c < rival[0]): - rival = (c, v.tex) return min_c, worst, worst_kind, n, eq, rival diff --git a/tests/test_core.py b/tests/test_core.py index 65cf46e..6276e93 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -95,12 +95,14 @@ def test_verify_accepts_truth_and_repairs_single_errors() -> None: @needs_latex def test_decision_rules() -> None: th = thresholds() - ok = {"ok": True, "tex": "x", "metrics": {"ncc": 0.95, "aspect_dev": 0.0, "px_per_pt": 1.4}, "min_contrast": 0.6} + ok = {"ok": True, "tex": "x", "metrics": {"ncc": 0.95, "aspect_dev": 0.0, "px_per_pt": 1.4}, "min_contrast": 3.0} assert decide(ok, True, 1.4, th)[0] == "auto_accepted" assert decide(ok, False, 1.4, th)[0] == "needs_review" # Pandoc fails assert decide(ok, True, 1.7, th)[0] == "needs_review" # wrong size/style - assert decide({**ok, "min_contrast": 0.2}, True, 1.4, th)[0] == "needs_review" - assert decide({**ok, "min_contrast": -0.5}, True, 1.4, th)[0] == "rejected_keep_image" + assert decide({**ok, "min_contrast": 1.0}, True, 1.4, th)[0] == "needs_review" + assert decide({**ok, "min_contrast": -2.0}, True, 1.4, th)[0] == "rejected_keep_image" + multi = {**ok, "tex": r"\begin{aligned}a&=b\\c&=d\end{aligned}"} + assert decide(multi, True, 1.4, th)[0] == "needs_review" # stricter for several lines assert decide({"ok": False, "reason": "x"}, True, None, th)[0] == "render_failed" diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 80c3436..1fdd672 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -18,7 +18,8 @@ def _inject(ws: Workspace, by_asset: dict[str, str]) -> None: The mechanics are tested here, not the calibrated thresholds (test_core.test_decision_rules), so the contrast threshold is set low enough for the synthetic images. """ - ws.settings["thresholds"] = {"min_contrast": 0.1} + ws.settings["thresholds"] = {"min_contrast": 0.5} + ws.settings["decide"] = "pixel" ws.save_meta() for f in ws.all(): for name, tex in by_asset.items(): @@ -106,3 +107,81 @@ def test_prepare_image_enlarges_small_formulas(epub2md_dir: Path) -> None: data = (epub2md_dir / "media" / "images" / "0001-inline.jpg").read_bytes() with Image.open(io.BytesIO(prepare_image(data, "x.jpg"))) as im: assert im.height >= 96 and im.mode == "L" + + +# ------------------------------------------------------------------------------- consensus + +def _reading(engine: str, model: str, tex: str | None, p_mean: float = 0.99, p_min: float = 0.8, + is_formula: bool = True) -> dict: + return {"engine": engine, "model": model, "tex": tex, "is_formula": is_formula, "p_mean": p_mean, "p_min": p_min} + + +@needs_latex +@needs_pandoc +def test_consensus_mode(epub2md_dir: Path) -> None: + source, ws = pipeline.open_all(epub2md_dir, None, create=True) + pipeline.scan(source, ws, quiet=True) + by = {f.asset.rsplit("/", 1)[1]: f for f in ws.all()} + inline, display = by["0001-inline.jpg"], by["0002-display.jpg"] + inline.candidates = [_reading("local", "A", r"L=\{-\sqrt{2};\sqrt{2}\}"), + _reading("local2", "B", r"L = \{ - \sqrt2 ; \sqrt2 \}")] # same, other spelling + display.candidates = [_reading("local", "A", r"x_{2,k}=\arcsin\left(-\frac{3}{5}\right)+2k\pi"), + _reading("local2", "B", r"x_{3,k}=\arcsin\left(-\frac{3}{5}\right)+2k\pi")] # differ + ws.save(inline) + ws.save(display) + pipeline.verify_all(source, ws, preset="cm", workers=2, quiet=True) + st = {f.asset.rsplit("/", 1)[1]: f for f in ws.all()} + assert st["0001-inline.jpg"].status == "auto_accepted", st["0001-inline.jpg"].reason + assert st["0002-display.jpg"].status == "needs_review" + assert "uneinig" in st["0002-display.jpg"].reason + + # a third reading breaks the tie + d = st["0002-display.jpg"] + d.candidates.append(_reading("cloud", "C", r"x_{2,k} = \arcsin(-\frac35) + 2k\pi")) + ws.save(d) + pipeline.verify_all(source, ws, preset="cm", workers=2, quiet=True) + d = ws.load(d.id) + assert d.status == "auto_accepted" and "Stichwahl" in d.reason, d.reason + + +def test_consensus_rules() -> None: + from formula_img2latex.decide import consensus_thresholds, decide_consensus + th = consensus_thresholds() + a, b = _reading("local", "A", "x^2"), _reading("local2", "B", "x^{2}") + veto = {"ok": True, "metrics": {"ncc": 0.9, "px_per_pt": 1.4}} + assert decide_consensus([a, b], [(0, 1, "normalform")], veto, True, 1.4, th)[0] == "auto_accepted" + assert decide_consensus([a, b], [], veto, True, 1.4, th)[0] == "needs_review" # disagree + low = _reading("local2", "B", "x^{2}", p_mean=0.7) + assert decide_consensus([a, low], [(0, 1, "normalform")], veto, True, 1.4, th)[0] == "needs_review" # unsure + bad = {"ok": True, "metrics": {"ncc": 0.5, "px_per_pt": 1.4}} + assert decide_consensus([a, b], [(0, 1, "normalform")], bad, True, 1.4, th)[0] == "needs_review" # veto + big = {"ok": True, "metrics": {"ncc": 0.9, "px_per_pt": 1.8}} + assert decide_consensus([a, b], [(0, 1, "normalform")], big, True, 1.4, th)[0] == "needs_review" # size + assert decide_consensus([a, b], [(0, 1, "normalform")], veto, False, 1.4, th)[0] == "needs_review" # pandoc + nf = [_reading("local", "A", None, is_formula=False), _reading("local2", "B", None, is_formula=False)] + assert decide_consensus(nf, [], None, True, 1.4, th)[0] == "not_formula" + assert decide_consensus([a], [], veto, True, 1.4, th)[0] == "needs_review" # one model + + +def test_canonical_form() -> None: + from formula_img2latex.consensus import canonical + same = [(r"x^{2}+y^{2}", "x^2 + y^2"), (r"\vec{b}", r"\overrightarrow{b}"), (r"\left(a\right)", "(a)"), + (r"\mathrm{d}x", "dx"), (r"\frac{1}{2}", r"\dfrac12"), (r"{\mathrm{e}}^{x}", "e^x"), (r"x\le1", r"x \leq 1"), + (r"\mathbf{v}", r"\boldsymbol{v}")] + differ = [(r"x_{2,k}", r"x_{2;k}"), ("x^{2}", "x_{2}"), (r"\mathbb{R}", "R"), ("a_{n+1}", "a_n+1"), + (r"\frac{a}{b+c}", r"\frac{a}{b}+c"), (r"x \in t", r"x \int"), (r"\sqrt[3]{x}", r"\sqrt{x}")] + assert all(canonical(a) == canonical(b) for a, b in same) + assert all(canonical(a) != canonical(b) for a, b in differ) + + +def test_logprob_confidence() -> None: + import math + + from formula_img2latex.engines.vlm import latex_token_probs, parse_confidence + content = '{"is_formula": true, "latex": "x^{2}", "confidence": 97}' + toks = ['{"', "is", "_formula", '": ', "true", ', "', "latex", '": "', "x", "^{", "2", "}", '", "', "confidence", + '": ', "97", "}"] + items = [{"token": t, "logprob": math.log(0.5 if t == "2" else 0.99)} for t in toks] + p_min, p_mean = latex_token_probs(content, items) + assert p_min == 0.5 and 0.8 < p_mean < 0.9 # only the four LaTeX tokens count + assert parse_confidence(content) == 97 diff --git a/tools/konsens_bericht.py b/tools/konsens_bericht.py new file mode 100644 index 0000000..f7e10d5 --- /dev/null +++ b/tools/konsens_bericht.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""Visual report of a consensus calibration (calibrate --consensus --report X.json). + + python3 tools/konsens_bericht.py BERICHT.json EPUB [--name konsens-goellmann] + +Writes vergleich//index.html (not versioned): per formula the original image, the rendered +truth, the rendered readings of both models and the verdict (agree+right / agree+WRONG / disagree). +""" + +from __future__ import annotations + +import argparse +import base64 +import html +import io +import json +import random +import sys +import zipfile +from pathlib import Path + +import numpy as np +from PIL import Image + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from formula_img2latex.calibrate import _dvipng_jpeg # noqa: E402 +from formula_img2latex.render import render_many # noqa: E402 +from formula_img2latex.tex import strip_delimiters # noqa: E402 + + +def png_uri(img: Image.Image) -> str: + buf = io.BytesIO() + img.save(buf, "PNG") + return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode() + + +def orig_img(data: bytes, scale: int) -> Image.Image: + im = Image.open(io.BytesIO(data)) + if im.mode in ("P", "LA", "RGBA") or "transparency" in im.info: + im = im.convert("RGBA") + bg = Image.new("RGBA", im.size, "white") + im = Image.alpha_composite(bg, im) + im = im.convert("RGB") + return im.resize((im.width * scale, im.height * scale), Image.NEAREST) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("report", type=Path) + ap.add_argument("epub", type=Path) + ap.add_argument("--name", default=None) + args = ap.parse_args() + rep = json.loads(args.report.read_text(encoding="utf-8")) + name = args.name or "konsens-" + args.epub.name.split("_")[0].lower() + out = Path(__file__).resolve().parents[1] / "vergleich" / name + out.mkdir(parents=True, exist_ok=True) + z = zipfile.ZipFile(args.epub) + models = rep.get("engines", ["A", "B"]) + parts = [f"

Konsensmessung: {html.escape(args.epub.name[:70])}

", + f"

Modell A = {html.escape(models[0])}, Modell B = {html.escape(models[1])}. " + "Grün = beide einig und richtig, Rot = beide einig, aber falsch, " + "Grau = uneinig. Gerendert mit pdflatex; „richtig“ heißt: gleiche Zeichen an gleicher Stelle " + "(Abstände, Pfeillängen, kursiv/aufrecht egal).

"] + for s in rep["sets"]: + res = s["results"] + n = max(1, len(res)) + parts.append(f"

{html.escape(s['label'])}

{len(res)} Formeln · A richtig {s['a_correct'] / n:.0%} · " + f"B richtig {s['b_correct'] / n:.0%} · einig {s['agree'] / n:.0%} · " + f"einig, aber falsch: {s['agree_wrong']}

") + order = sorted(res, key=lambda r: (0 if r["agree"] and not r["a_ok"] else 1 if not r["agree"] else 2)) + texes = [] + for r in order: + texes += [(strip_delimiters(r["truth"]), r["display"]), (r["a"].get("tex") or "", r["display"]), + (r["b"].get("tex") or "", r["display"])] + rendered = render_many([(t or "\\text{–}", d) for t, d in texes], "cm") + rng = random.Random(1) + for k, r in enumerate(order): + if r["name"].endswith("#synth.jpg"): + data = _dvipng_jpeg(strip_delimiters(r["truth"]), r["display"], 102) or b"" + else: + data = z.read(r["name"]) + try: + o = orig_img(data, 2) + except Exception: # noqa: BLE001 + continue + imgs = [] + for j in range(3): + rr = rendered[3 * k + j] + if rr.ok: + g = (rr.rasterize(2.0) * 255).astype(np.uint8) + imgs.append(png_uri(Image.fromarray(255 - g))) + else: + imgs.append("") + if r["agree"] and not r["a_ok"]: + cls, verdict = "bad", "einig, aber FALSCH" + elif r["agree"]: + cls, verdict = "good", "einig und richtig" + else: + cls, verdict = "dis", "uneinig" + mark = lambda ok: "✓" if ok else "✗" # noqa: E731 + pa, pb = r["a"].get("p_mean"), r["b"].get("p_mean") + parts.append( + f"
{verdict}
" + f"" + f"" + f"" + f"
Original
richtig{html.escape(r['truth'])}
A {mark(r['a_ok'])}{html.escape(r['a'].get('tex') or '–')}" + f" p={pa and round(pa, 3)}
B {mark(r['b_ok'])}{html.escape(r['b'].get('tex') or '–')}" + f" p={pb and round(pb, 3)}
") + del rng + rng = random.Random(1) + (out / "index.html").write_text( + "Konsens" + "".join(parts), encoding="utf-8") + print(out / "index.html") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/vergleich.py b/tools/vergleich.py new file mode 100644 index 0000000..2eac2dd --- /dev/null +++ b/tools/vergleich.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +"""Tuning aid: verify a fixed sample of formulas and write comparison images + an HTML overview. + + python3 tools/vergleich.py AUSGABE [--n 50] [--seed 1] [--name gehrke-50] + +The sample (formula ids) is stored in vergleich//sample.json and reused on later runs, so +changes to the procedure are compared on the same formulas. Output (not versioned, it shows +excerpts of the book): + + vergleich// + ├── index.html all formulas: image panel, LaTeX, measurements, decision, reason + ├── summary.json counts per status, all numbers + ├── sample.json the formula ids of the sample + └── NN-.png Original / LaTeX rigid / LaTeX elastic / difference +""" + +from __future__ import annotations + +import argparse +import html +import io +import json +import random +import sys +from concurrent.futures import ProcessPoolExecutor, as_completed +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from formula_img2latex.checks import pandoc_mathml_ok # noqa: E402 +from formula_img2latex.decide import decide, thresholds # noqa: E402 +from formula_img2latex.pipeline import _distinct_candidates, _ref_for, _verify_job, open_all # noqa: E402 + +FONT = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf" +LABELS = (("orig", "Original"), ("rigid", "LaTeX starr"), ("render", "LaTeX elastisch"), ("diff", "Differenz")) + + +def panel(pngs: dict[str, bytes], target_h: int = 70) -> Image.Image: + """Stack the four images, enlarged with visible pixels, each with a label.""" + imgs = {k: Image.open(io.BytesIO(v)).convert("RGB") for k, v in pngs.items()} + if not imgs: + return Image.new("RGB", (300, 40), "white") + h0 = next(iter(imgs.values())).height + f = max(1, min(8, round(target_h / max(h0, 1)))) + try: + font = ImageFont.truetype(FONT, 14) + except OSError: + font = ImageFont.load_default() + rows = [] + for key, label in LABELS: + if key not in imgs: + continue + im = imgs[key] + im = im.resize((im.width * f, im.height * f), Image.NEAREST) + row = Image.new("RGB", (im.width + 140, im.height + 8), "white") + ImageDraw.Draw(row).text((4, 4), label, fill=(80, 80, 80), font=font) + row.paste(im, (136, 4)) + rows.append(row) + w = max(r.width for r in rows) + out = Image.new("RGB", (w, sum(r.height for r in rows) + 4), "white") + y = 0 + for r in rows: + out.paste(r, (0, y)) + y += r.height + ImageDraw.Draw(out).rectangle([0, 0, out.width - 1, out.height - 1], outline=(200, 200, 200)) + return out + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("source", type=Path) + ap.add_argument("--n", type=int, default=50) + ap.add_argument("--seed", type=int, default=1) + ap.add_argument("--name", default=None, help="Unterordner in vergleich/ (Standard: -)") + ap.add_argument("--preset", default=None) + ap.add_argument("--workers", type=int, default=16) + ap.add_argument("--new-sample", action="store_true", help="neue Stichprobe ziehen") + args = ap.parse_args() + + source, ws = open_all(args.source, None) + name = args.name or f"{args.source.name.split('_')[0].lower()}-{args.n}" + out = Path(__file__).resolve().parents[1] / "vergleich" / name + out.mkdir(parents=True, exist_ok=True) + sample_file = out / "sample.json" + pool = [f for f in ws.all() if _distinct_candidates(f) and (f.classify or {}).get("label") != "figure"] + if sample_file.exists() and not args.new_sample: + ids = json.loads(sample_file.read_text()) + else: + ids = sorted(f.id for f in random.Random(args.seed).sample(pool, min(args.n, len(pool)))) + sample_file.write_text(json.dumps(ids, indent=1)) + by_id = {f.id: f for f in pool} + formulas = [by_id[i] for i in ids if i in by_id] + preset = args.preset or ws.settings.get("preset") or "cm" + book_ppt = ws.settings.get("book_px_per_pt") + th = thresholds(ws.settings.get("thresholds")) + + jobs = [(f.id, source.original_bytes(_ref_for(source, f)), f.asset, _distinct_candidates(f), f.display, preset) + for f in formulas] + results: dict[str, tuple[dict, dict[str, bytes]]] = {} + with ProcessPoolExecutor(max_workers=args.workers) as ex: + for fut in as_completed([ex.submit(_verify_job, j) for j in jobs]): + fid, res, pngs = fut.result() + best = max(res, key=lambda r: (r["ok"], r["min_contrast"] if r["min_contrast"] is not None else 1.0, + r["metrics"].get("ncc", 0.0))) + results[fid] = (best, pngs) + print(f" {len(results)}/{len(jobs)}", file=sys.stderr, end="\r") + print(file=sys.stderr) + pandoc = dict(zip([f.id for f in formulas], pandoc_mathml_ok([(results[f.id][0]["tex"], f.display) for f in formulas]))) + + for old in out.glob("*.png"): + old.unlink() + counts: dict[str, int] = {} + rows_html, summary = [], [] + for n, f in enumerate(formulas, 1): + best, pngs = results[f.id] + status, reason = decide(best, pandoc[f.id], book_ppt, th) + counts[status] = counts.get(status, 0) + 1 + img = f"{n:02d}-{f.id}.png" + panel(pngs).save(out / img) + m = best.get("metrics", {}) + mc = best.get("min_contrast") + summary.append({"n": n, "id": f.id, "status": status, "reason": reason, "tex": best.get("tex"), + "ocr": [c.get("tex") for c in f.candidates if c.get("tex")], "display": f.display, + "min_contrast": mc, "worst_variant": best.get("worst_variant"), + "worst_kind": best.get("worst_kind"), "climbed_from": best.get("climbed_from"), + "metrics": m}) + climbed = "" + if best.get("climbed_from"): + climbed = "
korrigiert aus: " + html.escape(best["climbed_from"][0]) + "
" + rows_html.append( + f"

{n:02d} · {status} · " + f"{'abgesetzt' if f.display else 'im Text'} · {html.escape(f.occurrences[0]['file'])}:{f.occurrences[0]['line']}

" + f"
{html.escape(best.get('tex') or '')}
{climbed}" + f"
NCC elastisch {m.get('ncc', 0):.3f} · starr {m.get('rigid_ncc', 0):.3f} · " + f"Verzerrung {m.get('warp_share', 0):.2f} · px/pt {m.get('px_per_pt', 0):.3f} · " + f"Kontrast {('–' if mc is None else f'{mc:.3f}')} (knappste Variante: {html.escape(best.get('worst_kind') or '–')}) · " + f"Pandoc {'ok' if pandoc[f.id] else 'FEHLER'}
{html.escape(reason)}
") + head = " · ".join(f"{k}: {v}" for k, v in sorted(counts.items())) + (out / "index.html").write_text( + "Vergleich " + html.escape(name) + "" + f"

Vergleich {html.escape(name)}

{len(formulas)} Formeln, Schrift {preset} · {head}

" + "

Zeilen je Bild: Original · LaTeX starr ausgerichtet · LaTeX elastisch ausgerichtet · Differenz " + "(schwarz = beide, rot = nur Original, blau = nur LaTeX, gelb = Abweichungsfleck)

" + + "".join(rows_html), encoding="utf-8") + (out / "summary.json").write_text(json.dumps({"preset": preset, "counts": counts, "formulas": summary}, + ensure_ascii=False, indent=1), encoding="utf-8") + print(f"{len(formulas)} Formeln geprüft – {head}") + print(f"Übersicht: {out / 'index.html'}") + return 0 + + +if __name__ == "__main__": + sys.exit(main())