dschlueter/formula_img2latex
Watch
1
0
Fork
You've already forked formula_img2latex
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat: Konsens zweier Bildmodelle als Entscheider, elastischer Bildvergleich

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-10-07 05:01:27 +02:00
commit fe95cb572c
15 changed files with 1087 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -24,3 +24,6 @@ output/
.vscode/
*.swp
.DS_Store
# Vergleichsbilder zum Einstellen (Ausschnitte aus Büchern, nicht versionieren)
vergleich/

139
formula_img2latex/calibrate.py
View file

@ -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']}")

7
formula_img2latex/cli.py
View file

@ -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,

168
formula_img2latex/compare.py
View file

@ -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")

29
formula_img2latex/confusables.py
View file

@ -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 != "'":

145
formula_img2latex/consensus.py Normal file
View file

@ -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

99
formula_img2latex/decide.py
View file

@ -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

20
formula_img2latex/engines/__init__.py
View file

@ -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)")

76
formula_img2latex/engines/vlm.py
View file

@ -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)

88
formula_img2latex/pipeline.py
View file

@ -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:

12
formula_img2latex/verify.py
View file

@ -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

8
tests/test_core.py
View file

@ -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"

81
tests/test_pipeline.py
View file

@ -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

125
tools/konsens_bericht.py Normal file
View file

@ -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/<name>/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"<h1>Konsensmessung: {html.escape(args.epub.name[:70])}</h1>",
f"<p>Modell A = {html.escape(models[0])}, Modell B = {html.escape(models[1])}. "
"Grün = beide einig und richtig, <b style='color:#a32020'>Rot = beide einig, aber falsch</b>, "
"Grau = uneinig. Gerendert mit pdflatex; „richtig“ heißt: gleiche Zeichen an gleicher Stelle "
"(Abstände, Pfeillängen, kursiv/aufrecht egal).</p>"]
for s in rep["sets"]:
res = s["results"]
n = max(1, len(res))
parts.append(f"<h2>{html.escape(s['label'])}</h2><p>{len(res)} Formeln · A richtig {s['a_correct'] / n:.0%} · "
f"B richtig {s['b_correct'] / n:.0%} · einig {s['agree'] / n:.0%} · "
f"<b>einig, aber falsch: {s['agree_wrong']}</b></p>")
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"<section class={cls}><div class=v>{verdict}</div><table>"
f"<tr><th>Original</th><td><img src='{png_uri(o)}'></td><td></td></tr>"
f"<tr><th>richtig</th><td><img src='{imgs[0]}'></td><td><code>{html.escape(r['truth'])}</code></td></tr>"
f"<tr><th>A {mark(r['a_ok'])}</th><td><img src='{imgs[1]}'></td><td><code>{html.escape(r['a'].get('tex') or '–')}</code>"
f" <small>p={pa and round(pa, 3)}</small></td></tr>"
f"<tr><th>B {mark(r['b_ok'])}</th><td><img src='{imgs[2]}'></td><td><code>{html.escape(r['b'].get('tex') or '–')}</code>"
f" <small>p={pb and round(pb, 3)}</small></td></tr></table></section>")
del rng
rng = random.Random(1)
(out / "index.html").write_text(
"<!doctype html><meta charset=utf-8><title>Konsens</title><style>"
"body{font:14px system-ui,sans-serif;margin:16px;background:#f6f6f3}section{background:#fff;border:1px solid #ddd;"
"border-left:6px solid #999;border-radius:6px;padding:6px 10px;margin:8px 0}section.good{border-left-color:#1f7a3a}"
"section.bad{border-left-color:#a32020;background:#fff5f5}.v{font-weight:600;font-size:13px}"
"th{text-align:left;font-weight:400;color:#666;width:80px;vertical-align:middle}td{padding:2px 8px;vertical-align:middle}"
"code{font-size:12px}img{max-width:700px}</style>" + "".join(parts), encoding="utf-8")
print(out / "index.html")
return 0
if __name__ == "__main__":
sys.exit(main())

160
tools/vergleich.py Normal file
View file

@ -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/<name>/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/<name>/
├── 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-<id>.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: <buch>-<n>)")
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 = "<div class=fix>korrigiert aus: <code>" + html.escape(best["climbed_from"][0]) + "</code></div>"
rows_html.append(
f"<section class='{status}'><h2>{n:02d} · <span class=st>{status}</span> · "
f"{'abgesetzt' if f.display else 'im Text'} · {html.escape(f.occurrences[0]['file'])}:{f.occurrences[0]['line']}</h2>"
f"<img src='{img}'><div><code>{html.escape(best.get('tex') or '')}</code></div>{climbed}"
f"<div class=m>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'}</div><div class=r>{html.escape(reason)}</div></section>")
head = " · ".join(f"{k}: {v}" for k, v in sorted(counts.items()))
(out / "index.html").write_text(
"<!doctype html><meta charset=utf-8><title>Vergleich " + html.escape(name) + "</title><style>"
"body{font:14px system-ui,sans-serif;margin:16px;background:#f6f6f3}section{background:#fff;border:1px solid #ddd;"
"border-radius:6px;padding:8px 12px;margin:10px 0}h2{font-size:14px;margin:0 0 6px}img{max-width:100%}"
".auto_accepted .st{color:#1f7a3a}.needs_review .st{color:#a15c00}.rejected_keep_image .st,.render_failed .st{color:#a32020}"
".m{font:12px monospace;color:#555;margin-top:4px}.r{font-size:13px}.fix{font-size:12px;color:#2a5db0}"
"code{font-size:12px}</style>"
f"<h1>Vergleich {html.escape(name)}</h1><p>{len(formulas)} Formeln, Schrift {preset} · {head}</p>"
"<p>Zeilen je Bild: Original · LaTeX starr ausgerichtet · LaTeX elastisch ausgerichtet · Differenz "
"(schwarz = beide, rot = nur Original, blau = nur LaTeX, gelb = Abweichungsfleck)</p>"
+ "".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())