Files
fuelish/tooling-container/src/compare_ground_truth.py
T
matthias aae57d7a2b Self-hosted, verifiable pipeline for extracting structured data (station,
date/time, price per litre, litres, discount, total) from German fuel
receipt photos: grayscale + mild bilateral filter + contrast stretch,
Tesseract OCR, then LLM structuring via a small local model (Qwen 2.5,
served by Ollama). Each step is independently runnable and inspectable
through numbered scripts against a shared tooling container.
2026-09-28 19:17:47 +02:00

95 lines
3.1 KiB
Python

#!/usr/bin/env python3
"""Test-only tool: compare a directory of extracted.json results (produced
by running 1_preprocess_image.sh/2_ocr_image.sh/3_extract_data.sh over a
set of receipts) against the ground-truth spreadsheet, matching rows by
date and reporting per-field accuracy."""
import argparse
import glob
import json
import os
from odf.opendocument import load
from odf.table import Table, TableRow, TableCell
from odf.text import P
def load_ground_truth(ods_path: str) -> dict:
doc = load(ods_path)
table = doc.spreadsheet.getElementsByType(Table)[0]
rows = table.getElementsByType(TableRow)
gt = {}
for r in rows[1:]:
cells = r.getElementsByType(TableCell)
vals = []
for c in cells:
ps = c.getElementsByType(P)
vals.append(" ".join(str(p) for p in ps))
if not vals or not vals[0]:
continue
date = vals[0]
try:
gt[date] = {
"time": vals[1],
"price_per_litre": float(vals[2].replace(",", ".")) if vals[2] else None,
"litres": float(vals[3].replace(",", ".")) if vals[3] else None,
"discount": float(vals[4].replace("€", "").replace(",", ".").strip()) if vals[4] else None,
"total_price": float(vals[5].replace("€", "").replace(",", ".").strip()) if vals[5] else None,
}
except (ValueError, IndexError):
continue
return gt
def close(a, b, tol=0.05):
if a is None or b is None:
return a is None and b is None
return abs(abs(a) - abs(b)) <= tol
def main():
parser = argparse.ArgumentParser()
parser.add_argument("-gt", dest="ods_path", required=True, help="path to ground-truth .ods")
parser.add_argument("-results", dest="results_dir", required=True,
help="directory containing one subdirectory per receipt, each with an extracted.json")
args = parser.parse_args()
gt = load_ground_truth(args.ods_path)
fields = ["price_per_litre", "litres", "discount", "total_price"]
field_ok = {k: 0 for k in fields}
field_total = {k: 0 for k in fields}
matched = 0
total = 0
for json_path in sorted(glob.glob(os.path.join(args.results_dir, "*", "extracted.json"))):
receipt = os.path.basename(os.path.dirname(json_path))
total += 1
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
truth = gt.get(data.get("date"))
if not truth:
print(f"{receipt}: NO DATE MATCH (extracted date={data.get('date')!r})")
continue
matched += 1
bad = []
for k in fields:
field_total[k] += 1
ok = close(data.get(k), truth[k])
if ok:
field_ok[k] += 1
else:
bad.append(k)
status = "OK" if not bad else f"WRONG FIELDS {bad}"
print(f"{receipt}: {status}")
print("\n=== SUMMARY ===")
print(f"Receipts matched to ground truth: {matched}/{total}")
for k in fields:
print(f" {k}: {field_ok[k]}/{field_total[k]} correct")
if __name__ == "__main__":
main()