from __future__ import annotations
import json
import os
import re
import statistics
import subprocess
import sys
import time
from pathlib import Path
CORPUS_ROOT = Path(
os.environ.get(
"ICDAR2013_CORPUS",
"/tmp/spectre-bench/ICDAR_2013_table_evaluate-master/icdar2013-competition-dataset-with-gt",
)
)
PARITY_BIN = Path(__file__).resolve().parents[1] / "target" / "release" / "examples" / "parity"
ITERS = 2
if not CORPUS_ROOT.is_dir():
sys.exit(
f"ICDAR 2013 corpus not at {CORPUS_ROOT}\n"
f"Set $ICDAR2013_CORPUS or fetch via the recipe in this script's docstring."
)
if not PARITY_BIN.is_file():
sys.exit(
f"parity binary not at {PARITY_BIN}\n"
f"Build with: cargo build --release --example parity"
)
def time_call(fn) -> tuple[float, int]:
times: list[float] = []
out = ""
for _ in range(ITERS):
t0 = time.perf_counter()
out = fn()
times.append((time.perf_counter() - t0) * 1000.0)
chars = len(out) if isinstance(out, str) else 0
return statistics.median(times), chars
def spectre_call(path: Path) -> tuple[float, float, int, bool]:
strict_times: list[float] = []
lenient_times: list[float] = []
chars = 0
strict_ok = False
for _ in range(ITERS):
proc = subprocess.run(
[str(PARITY_BIN), str(path)],
capture_output=True, text=True, check=False, timeout=60,
)
if proc.returncode != 0:
return -1.0, -1.0, -1, False
for line in proc.stdout.splitlines():
m_strict = re.search(
r"strict-warm-2\s+chars=\d+\s+time_ms=([\d.]+)\s+ok=(true|false)", line
)
if m_strict:
strict_times.append(float(m_strict.group(1)))
strict_ok = (m_strict.group(2) == "true")
m_warm = re.search(r"text-warm-2\s+chars=(\d+)\s+time_ms=([\d.]+)", line)
if m_warm:
chars = int(m_warm.group(1))
lenient_times.append(float(m_warm.group(2)))
if not lenient_times or not strict_times:
return -1.0, -1.0, -1, False
return (
statistics.median(strict_times),
statistics.median(lenient_times),
chars,
strict_ok,
)
def pymupdf_call(path: Path) -> tuple[float, int]:
import pymupdf
def run():
with pymupdf.open(str(path)) as doc:
return "\n\n".join(p.get_text() for p in doc)
try:
return time_call(run)
except Exception:
return -1.0, -1
def pdfminer_call(path: Path) -> tuple[float, int]:
from pdfminer.high_level import extract_text
try:
return time_call(lambda: extract_text(str(path)))
except Exception:
return -1.0, -1
def pdfplumber_call(path: Path) -> tuple[float, int]:
import pdfplumber
def run():
with pdfplumber.open(str(path)) as pdf:
return "\n\n".join((p.extract_text() or "") for p in pdf.pages)
try:
return time_call(run)
except Exception:
return -1.0, -1
TOOLS = [
("spectre_rs", spectre_call),
("pymupdf", pymupdf_call),
("pdfminer.six", pdfminer_call),
("pdfplumber", pdfplumber_call),
]
def main() -> None:
import hashlib
seen: dict[str, Path] = {}
for p in sorted(CORPUS_ROOT.glob("**/*.pdf")):
digest = hashlib.sha256(p.read_bytes()).hexdigest()
seen.setdefault(digest, p)
pdfs = sorted(seen.values())
print(f"corpus: {len(pdfs)} unique PDFs from {CORPUS_ROOT.name}, {ITERS} iters per tool", file=sys.stderr)
spectre_rows: list[tuple[Path, float, float, int, bool]] = []
other_results: dict[str, list[tuple[Path, float, int]]] = {
n: [] for n, _ in TOOLS if n != "spectre_rs"
}
for i, pdf in enumerate(pdfs, 1):
line = f"[{i:>3}/{len(pdfs)}] {pdf.name}: "
s_ms, l_ms, chars, strict_ok = spectre_call(pdf)
spectre_rows.append((pdf, s_ms, l_ms, chars, strict_ok))
line += (
f"spectre[strict={'Y' if strict_ok else 'N'}={s_ms:6.1f}ms,len={l_ms:6.1f}ms,ch={chars:>7}] "
)
for name, fn in TOOLS:
if name == "spectre_rs":
continue
o_ms, o_chars = fn(pdf)
other_results[name].append((pdf, o_ms, o_chars))
line += f"{name}={o_ms:6.1f}ms({o_chars:>7}) "
print(line, file=sys.stderr)
strict_ok_pdfs = {p for p, _, _, _, ok in spectre_rows if ok}
n_strict_fail = len(pdfs) - len(strict_ok_pdfs)
print(f"\n## Headline — Full ICDAR 2013 corpus, strict mode ({len(pdfs)} unique PDFs)\n")
print(
f"spectre runs `extract_text` (strict): the {n_strict_fail} PDFs whose CID fonts omit `/ToUnicode` raise `ExtractError::PageExtractFailed`. The fail-time is included in spectre's total. pymupdf / pdfminer.six / pdfplumber silently produce output on the same documents (without surfacing the underlying parse failure) — see Correctness section below.\n"
)
print("| Tool | Total time | Successful PDFs | vs spectre |")
print("|---|---:|---:|---:|")
spectre_strict_ms = sum(ms for _, ms, _, _, _ in spectre_rows if ms > 0)
print(
f"| **spectre_rs** (`extract_text`, strict) | **{spectre_strict_ms / 1000:6.2f} s** | "
f"{len(strict_ok_pdfs)}/{len(pdfs)} (3 raise, time-to-failure included) | **1.0x (baseline)** |"
)
for name in (n for n, _ in TOOLS if n != "spectre_rs"):
ok_runs = [(p, ms, ch) for p, ms, ch in other_results[name] if ms > 0]
total_ms = sum(ms for _, ms, _ in ok_runs)
mult = f"{total_ms / spectre_strict_ms:5.2f}x" if spectre_strict_ms > 0 else "n/a"
print(f"| {name} | {total_ms / 1000:6.2f} s | {len(ok_runs)}/{len(pdfs)} | {mult} |")
print(f"\n### Lenient-mode reference (silent-skip behaviour, matches Python tools)\n")
print(
"spectre running `extract_text_lenient` (drops failing pages silently, matching pymupdf/pdfminer/pdfplumber's silent-skip behaviour). Provided so callers who explicitly want the established libraries' failure mode can see the comparison.\n"
)
print("| Tool | Total time | vs spectre (lenient) |")
print("|---|---:|---:|")
spectre_lenient_ms = sum(ms for _, _, ms, _, _ in spectre_rows if ms > 0)
print(f"| spectre_rs (lenient) | {spectre_lenient_ms / 1000:6.2f} s | 1.0x (baseline) |")
for name in (n for n, _ in TOOLS if n != "spectre_rs"):
total_ms = sum(ms for _, ms, _ in other_results[name] if ms > 0)
mult = f"{total_ms / spectre_lenient_ms:5.2f}x" if spectre_lenient_ms > 0 else "n/a"
print(f"| {name} | {total_ms / 1000:6.2f} s | {mult} |")
print(f"\n## Char-count parity vs pymupdf\n")
spectre_chars_map = {p: ch for p, _, _, ch, _ in spectre_rows}
pymupdf_chars_map = {p: ch for p, _, ch in other_results["pymupdf"]}
deltas = []
for p, sc in spectre_chars_map.items():
pc = pymupdf_chars_map.get(p, -1)
if sc > 0 and pc > 0:
deltas.append((sc - pc) / pc)
if deltas:
print(f"PDFs compared (both non-empty): {len(deltas)}/{len(pdfs)}")
print(f"Mean delta (spectre / pymupdf - 1): {statistics.mean(deltas) * 100:+.1f}%")
print(f"Median delta: {statistics.median(deltas) * 100:+.1f}%")
print(f"PDFs within +/-10%: {sum(1 for d in deltas if abs(d) < 0.10)}/{len(deltas)}")
json_out = Path("/tmp/spectre-bench/icdar2013_results.json")
json_out.write_text(json.dumps({
"spectre_rs": [
{"pdf": str(p.relative_to(CORPUS_ROOT)), "strict_ms": s_ms,
"lenient_ms": l_ms, "chars": ch, "strict_ok": ok}
for p, s_ms, l_ms, ch, ok in spectre_rows
],
**{n: [{"pdf": str(p.relative_to(CORPUS_ROOT)), "ms": ms, "chars": ch} for p, ms, ch in rs]
for n, rs in other_results.items()},
"strict_failures": [
str(p.relative_to(CORPUS_ROOT))
for p, _, _, _, ok in spectre_rows if not ok
],
}, indent=2))
csv_out = Path(__file__).resolve().parents[1] / "data" / "icdar2013-results.csv"
csv_out.parent.mkdir(parents=True, exist_ok=True)
pymupdf_by_pdf = {p: (ms, ch) for p, ms, ch in other_results["pymupdf"]}
pdfminer_by_pdf = {p: (ms, ch) for p, ms, ch in other_results["pdfminer.six"]}
pdfplumber_by_pdf = {p: (ms, ch) for p, ms, ch in other_results["pdfplumber"]}
rows_for_csv: list[list[str]] = []
for p, s_ms, l_ms, ch, ok in spectre_rows:
py_ms, py_ch = pymupdf_by_pdf.get(p, (-1, -1))
pm_ms, _ = pdfminer_by_pdf.get(p, (-1, -1))
pp_ms, _ = pdfplumber_by_pdf.get(p, (-1, -1))
delta_pct = ""
if ch > 0 and py_ch > 0:
delta_pct = f"{(ch - py_ch) / py_ch * 100:+.1f}"
status = "OK" if ok else "STRICT_FAIL_CID"
rows_for_csv.append([
str(p.relative_to(CORPUS_ROOT)),
f"{s_ms:.2f}", f"{l_ms:.2f}", f"{py_ms:.2f}", f"{pm_ms:.2f}", f"{pp_ms:.2f}",
str(ch), str(py_ch), delta_pct, status,
])
sum_strict = sum(s_ms for _, s_ms, _, _, _ in spectre_rows if s_ms > 0)
sum_lenient = sum(l_ms for _, _, l_ms, _, _ in spectre_rows if l_ms > 0)
sum_py = sum(ms for _, ms, _ in other_results["pymupdf"] if ms > 0)
sum_pm = sum(ms for _, ms, _ in other_results["pdfminer.six"] if ms > 0)
sum_pp = sum(ms for _, ms, _ in other_results["pdfplumber"] if ms > 0)
sum_ch = sum(ch for _, _, _, ch, _ in spectre_rows if ch > 0)
sum_py_ch = sum(ch for _, _, ch in other_results["pymupdf"] if ch > 0)
fail_count = sum(1 for _, _, _, _, ok in spectre_rows if not ok)
rows_for_csv.append([
f"TOTAL ({len(spectre_rows)} PDFs)",
f"{sum_strict:.2f}", f"{sum_lenient:.2f}", f"{sum_py:.2f}",
f"{sum_pm:.2f}", f"{sum_pp:.2f}",
str(sum_ch), str(sum_py_ch),
f"mean {statistics.mean(d for d in [(ch - pymupdf_by_pdf[p][1])/pymupdf_by_pdf[p][1] for p, _, _, ch, _ in spectre_rows if ch > 0 and pymupdf_by_pdf.get(p, (0,0))[1] > 0]) * 100:+.1f} median {statistics.median(d for d in [(ch - pymupdf_by_pdf[p][1])/pymupdf_by_pdf[p][1] for p, _, _, ch, _ in spectre_rows if ch > 0 and pymupdf_by_pdf.get(p, (0,0))[1] > 0]) * 100:+.1f}",
f"{len(spectre_rows) - fail_count}_OK_{fail_count}_FAIL",
])
import csv as _csv
with csv_out.open("w", newline="") as f:
writer = _csv.writer(f)
writer.writerow([
"pdf", "spectre_strict_ms", "spectre_lenient_ms",
"pymupdf_ms", "pdfminer_ms", "pdfplumber_ms",
"spectre_chars", "pymupdf_chars", "char_delta_pct_vs_pymupdf",
"spectre_status",
])
for row in rows_for_csv:
writer.writerow(row)
print(f"\nfull per-PDF JSON: {json_out}")
print(f"per-PDF CSV (committed in repo): {csv_out}")
if __name__ == "__main__":
main()