Coverage for src / quber / core / figures / values.py: 84%
159 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-09-23 22:14 -0400
1"""Reconcile the values printed in a document's figures, with per-value provenance.
3The scan reads each figure into prose, so a chart's numbers reach the document
4with no per-value geometry and nothing checking them. This module gives every
5plotted value a second, independent read and a trace to the page:
7- The parse's positioned text cells are the fragment source. A born-digital
8 page carries every printed chart value as a native cell with a box; a page
9 docling read from its image carries the same cells from OCR, marked so.
10- A local reader looks at the page image with the figure's fragments and
11 reports each printed value with the fragment ids that ground it.
12- A parser turns the scan's prose reading into the same value shape.
13- A deterministic tie matches the two within each page and assigns statuses
14 from the cell-status registry: agreement grounded in a fragment is
15 `reconciled` with the fragment's box; a positional disagreement is
16 `value_misread`; a one-sided value is `value_unreconciled` with the
17 direction in the note.
19Matching is value-first with label confirmation, consuming pairs so repeated
20values resolve by count. Values are compared in a normalized form; the stored
21value keeps the printed form. Boxes are normalized 0..1, top-left origin, the
22same frame every other figure box in this package uses.
23"""
25from __future__ import annotations
27import re
28from pathlib import Path
29from typing import Dict, List, Optional, Protocol, Sequence, Tuple
31from loguru import logger
32from pydantic import BaseModel, Field
34from quber.core.figures.models import Box, FigureRecord, FigureValue, FigureValueRun, PageScan
36#: Minimum fraction of a fragment's area inside a figure's box for the
37#: fragment to belong to that figure, measured against the fragment.
38SCOPE_OVERLAP = 0.5
40#: Common words carrying no identity, dropped before label comparison.
41STOP_WORDS = frozenset({"the", "of", "per", "to", "by", "and", "in", "a"})
44class PageFragment(BaseModel):
45 """One positioned text cell from the parse, in the page's normalized frame."""
47 id: str
48 text: str
49 box: Box
50 from_ocr: bool = False
51 confidence: float = 1.0
54class ParsedValue(BaseModel):
55 """One labeled value read out of the scan's prose description of a figure."""
57 label: str = Field(default="")
58 series: str = Field(default="")
59 value: str
62class ReadValue(BaseModel):
63 """One printed value the local reader saw on the page image."""
65 chart_title: str = Field(default="")
66 label: str = Field(default="")
67 series: str = Field(default="")
68 value: str
69 fragment_ids: List[str] = Field(default_factory=list)
72class ScanValueParser(Protocol):
73 """Turns a figure's prose reading into labeled values."""
75 async def parse(self, figure_texts: Sequence[str], page: int) -> List[ParsedValue]: ...
78class LocalValueReader(Protocol):
79 """Reads a page image's figure values, citing the fragments that ground them."""
81 async def read(
82 self, image_path: Path, fragments: Sequence[PageFragment], page: int
83 ) -> List[ReadValue]: ...
86def page_fragments(cells_page: dict) -> List[PageFragment]:
87 """The parse's cells for one page, normalized to the 0..1 top-left frame.
89 `cells_page` is one entry of the document's cells artifact: page_no,
90 width, height, and cells each holding text, box in top-left PDF points,
91 from_ocr, and confidence.
92 """
93 width = float(cells_page["width"]) or 1.0
94 height = float(cells_page["height"]) or 1.0
95 fragments: List[PageFragment] = []
96 page_no = cells_page["page_no"]
97 for i, cell in enumerate(cells_page.get("cells", [])):
98 text = str(cell.get("text", "")).strip()
99 if not text:
100 continue
101 left, top, right, bottom = cell["box"]
102 fragments.append(
103 PageFragment(
104 id=f"p{page_no}.c{i}",
105 text=text,
106 box={
107 "left": left / width,
108 "top": top / height,
109 "right": right / width,
110 "bottom": bottom / height,
111 },
112 from_ocr=bool(cell.get("from_ocr", False)),
113 confidence=float(cell.get("confidence", 1.0)),
114 )
115 )
116 return fragments
119def fragments_in_box(fragments: Sequence[PageFragment], box: Box) -> List[PageFragment]:
120 """Fragments whose area sits mostly inside the box."""
121 hits = []
122 for frag in fragments:
123 f = frag.box
124 inter_w = max(0.0, min(f["right"], box["right"]) - max(f["left"], box["left"]))
125 inter_h = max(0.0, min(f["bottom"], box["bottom"]) - max(f["top"], box["top"]))
126 area = max((f["right"] - f["left"]) * (f["bottom"] - f["top"]), 1e-9)
127 if (inter_w * inter_h) / area >= SCOPE_OVERLAP:
128 hits.append(frag)
129 return hits
132def norm_value(raw: str) -> str:
133 """A value reduced to its comparable core: sign, digits, decimal point."""
134 s = str(raw).strip().replace("$", "").replace(",", "")
135 s = s.replace("–", "-").replace("—", "-")
136 negative = s.startswith("(") and s.endswith(")")
137 if negative:
138 s = s[1:-1]
139 s = s.replace("(", "").replace(")", "").rstrip("%").strip()
140 if s in {"-", "--", ""}:
141 return "DASH"
142 try:
143 number = float(s)
144 except ValueError:
145 return s.lower()
146 if negative and number > 0:
147 number = -number
148 return f"{number:g}"
151def label_words(*parts: str) -> frozenset:
152 words = set()
153 for part in parts:
154 words.update(re.findall(r"[a-z0-9/+.]+", str(part).lower()))
155 return frozenset(words - STOP_WORDS)
158def value_fragment(
159 read: ReadValue, fragments_by_id: Dict[str, PageFragment]
160) -> Tuple[Optional[PageFragment], bool]:
161 """The cited fragment printing the value, and whether any cited one does.
163 Returns the first cited fragment whose text contains the value in either
164 printed or normalized form. When citations exist but none contains the
165 value, the first cited fragment anchors the position of the disagreement.
166 """
167 cited = [fragments_by_id[i] for i in read.fragment_ids if i in fragments_by_id]
168 target = norm_value(read.value)
169 for frag in cited:
170 if read.value in frag.text or (target != "DASH" and target in norm_value(frag.text)):
171 return frag, True
172 return (cited[0] if cited else None), False
175def picture_for_value(figures: Sequence[FigureRecord], box: Optional[Box], raw_value: str) -> Optional[str]:
176 """The picture a value belongs to, or None when no figure claims it.
178 A boxed value belongs to the figure whose box contains the value's
179 center, the smallest such figure when they nest. An unboxed value
180 belongs to the one figure whose scan text prints it, and to no figure
181 when the value appears in several or in none — a wrong anchor misleads
182 where an absent one just falls back to the page.
183 """
184 if box is not None:
185 cx = (box["left"] + box["right"]) / 2
186 cy = (box["top"] + box["bottom"]) / 2
187 best: Optional[Tuple[float, Optional[str]]] = None
188 for fig in figures:
189 fb = fig.box
190 if fb is None or fig.picture_ref is None:
191 continue
192 if fb["left"] <= cx <= fb["right"] and fb["top"] <= cy <= fb["bottom"]:
193 area = (fb["right"] - fb["left"]) * (fb["bottom"] - fb["top"])
194 if best is None or area < best[0]:
195 best = (area, fig.picture_ref)
196 if best is not None:
197 return best[1]
198 if raw_value:
199 holders = {fig.picture_ref for fig in figures if fig.picture_ref and raw_value in fig.text}
200 if len(holders) == 1:
201 return next(iter(holders))
202 return None
205def tie_page(
206 page: int,
207 figures: Sequence[FigureRecord],
208 scan_values: Sequence[ParsedValue],
209 local_values: Sequence[ReadValue],
210 fragments: Sequence[PageFragment],
211) -> List[FigureValue]:
212 """Match the two readings of one page's figures and assign statuses."""
213 fragments_by_id = {f.id: f for f in fragments}
214 remaining = list(scan_values)
215 out: List[FigureValue] = []
217 def take_match(read: ReadValue, require_label: bool) -> Optional[ParsedValue]:
218 target = norm_value(read.value)
219 read_words = label_words(read.label, read.series)
220 for candidate in remaining:
221 if norm_value(candidate.value) != target:
222 continue
223 candidate_words = label_words(candidate.label, candidate.series)
224 overlap = bool(read_words & candidate_words)
225 if require_label and not overlap:
226 continue
227 if not require_label and candidate_words and read_words and not overlap:
228 continue
229 remaining.remove(candidate)
230 return candidate
231 return None
233 for read in local_values:
234 matched = take_match(read, require_label=True) or take_match(read, require_label=False)
235 fragment, contains = value_fragment(read, fragments_by_id)
236 if matched and contains and fragment is not None:
237 status, note = "reconciled", None
238 elif matched and fragment is not None:
239 status = "value_misread"
240 note = f"the page prints {fragment.text!r} where this value should appear - possible misread"
241 elif matched:
242 status = "value_unreconciled"
243 note = "corroborated by both readings, but not anchored to printed text on the page"
244 else:
245 status = "value_unreconciled"
246 note = "one measurement, read from the page; uncontradicted, not independently corroborated"
247 box = fragment.box if fragment is not None else None
248 out.append(
249 FigureValue(
250 page=page,
251 picture_ref=picture_for_value(figures, box, read.value),
252 chart_title=read.chart_title,
253 label=read.label,
254 series=read.series,
255 value=read.value,
256 status=status,
257 note=note,
258 fragment_ids=list(read.fragment_ids),
259 box=box,
260 )
261 )
263 for leftover in remaining:
264 out.append(
265 FigureValue(
266 page=page,
267 picture_ref=picture_for_value(figures, None, leftover.value),
268 label=leftover.label,
269 series=leftover.series,
270 value=leftover.value,
271 status="value_unreconciled",
272 note="one measurement, from the figure description; uncontradicted, not corroborated on the printed page",
273 )
274 )
275 return out
278async def read_figure_values(
279 scans: Sequence[PageScan],
280 cells_pages: Dict[int, dict],
281 page_images: Dict[int, Path],
282 parser: ScanValueParser,
283 reader: LocalValueReader,
284 document: str,
285) -> FigureValueRun:
286 """Reconcile every scanned figure page's values.
288 In: the run's page scans, the parse's cells keyed by page, a rendered
289 image per scanned page, and the two readers. Out: one FigureValueRun with
290 a FigureValue per value either reader produced.
291 """
292 run = FigureValueRun(document=document)
293 for scan in scans:
294 if not scan.figures:
295 continue
296 page = scan.page
297 cells_page = cells_pages.get(page)
298 image = page_images.get(page)
299 if cells_page is None or image is None:
300 run.errors.append(f"page {page}: missing cells or render; figure values skipped")
301 continue
302 fragments = page_fragments(cells_page)
303 try:
304 scan_values = await parser.parse([f.text for f in scan.figures], page)
305 local_values = await reader.read(image, fragments, page)
306 except Exception as exc:
307 run.errors.append(f"page {page}: figure-value read failed: {exc}")
308 logger.warning("figure values page {}: {}", page, exc)
309 continue
310 run.values.extend(tie_page(page, scan.figures, scan_values, local_values, fragments))
311 counts = {"reconciled": run.reconciled, "flagged": run.flagged}
312 logger.info("figure values: {} pages -> {}", sum(1 for s in scans if s.figures), counts)
313 return run