Coverage for src / quber / core / parsers / result.py: 49%
106 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-09-23 22:14 -0400
1"""The output contract a `Parser` hands back: the document plus the signals
2docling produces alongside it that quber used to discard.
4`result.document` was the only thing the parser returned. Three more pieces ride
5with it now, each needed to fuse docling's view of a page with the
6Set-of-Mark/Camelot extraction:
8- `confidence` — docling's own per-page parse/layout/table/ocr scores, taken
9 straight from `ConversionResult.confidence`.
10- `pages` — the parsed-page cells (`text`, top-left box, `from_ocr`,
11 `confidence`), one list per page. docling frees these after assembling the
12 document unless `generate_parsed_pages=True`; capturing them is what makes the
13 OCR-vs-native signal available downstream.
14- `table_provenance` — derived once per table from the cells under its box: the
15 fraction read by OCR, the mean OCR confidence, and a native/ocr/empty verdict.
16 A table whose region carries no native text layer (verdict `ocr`, or `empty`
17 when Camelot also finds nothing) is a table rendered as an image.
19Boxes are normalized to one frame here: cells and table boxes are both stored
20top-left in PDF points, so a table box and the cells under it compare directly.
21"""
23from __future__ import annotations
25import json
26from dataclasses import dataclass, field
27from pathlib import Path
29# Full docling appears in annotations only (`confidence`, `from_conversion`'s
30# parameter) — never constructed here. Keeping it out of the runtime imports
31# is what lets the CPU fuse job load artifacts without the GPU stack; the
32# runtime needs docling-core alone.
33from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Tuple
35if TYPE_CHECKING:
36 from docling.datamodel.base_models import ConfidenceReport
37 from docling.datamodel.document import ConversionResult
39from docling_core.types.doc.base import CoordOrigin
40from docling_core.types.doc.document import DoclingDocument, TableItem
41from docling_core.types.doc.page import BoundingRectangle
42from pydantic import BaseModel, Field
44Box = Tuple[float, float, float, float]
47class ParsedCell(BaseModel):
48 """One parsed-page text cell, box in top-left PDF points."""
50 text: str
51 box: Box = Field(description="(x0, y0, x1, y1), top-left origin, PDF points")
52 from_ocr: bool = Field(description="True if docling read this cell off the page image")
53 confidence: float = Field(default=0.0, description="docling's OCR confidence for the cell")
56class PageParse(BaseModel):
57 """The parsed-page cells for one page, with the page's point dimensions."""
59 page_no: int
60 width: float
61 height: float
62 cells: List[ParsedCell] = Field(default_factory=list)
65class PageScore(BaseModel):
66 """docling's per-page confidence scores (NaN where docling did not score)."""
68 page_no: int
69 parse_score: float
70 layout_score: float
71 table_score: float
72 ocr_score: float
75class TableProvenance(BaseModel):
76 """OCR-vs-native verdict for one docling table, from the cells under its box."""
78 self_ref: str = Field(description="The TableItem.self_ref this provenance describes")
79 page_no: int
80 cell_count: int = Field(description="Parsed-page cells whose center falls in the table box")
81 from_ocr_fraction: float = Field(description="Fraction of those cells docling read by OCR")
82 mean_ocr_confidence: float = Field(description="Mean confidence over the OCR-read cells")
83 verdict: Literal["native", "ocr", "empty"] = Field(
84 description=(
85 "native: the table box has a real text layer. ocr: docling read it off "
86 "the page image (a table rendered as an image). empty: no cells fall in "
87 "the box at all."
88 )
89 )
92@dataclass
93class ParseResult:
94 """Everything one parse produced: the document and docling's side signals.
96 `document` is the canonical IR, unchanged from before. The rest is what quber
97 used to free: per-page confidence, the parsed-page cells, and the per-table
98 OCR/native verdict derived from them.
99 """
101 document: DoclingDocument
102 confidence: Optional[ConfidenceReport] = None
103 page_scores: List[PageScore] = field(default_factory=list)
104 pages: List[PageParse] = field(default_factory=list)
105 table_provenance: List[TableProvenance] = field(default_factory=list)
107 def provenance_by_ref(self) -> Dict[str, TableProvenance]:
108 """Per-table provenance keyed by `TableItem.self_ref`."""
109 return {p.self_ref: p for p in self.table_provenance}
111 def save(self, directory: Path, base: str) -> Dict[str, Path]:
112 """Write the parse artifacts under `directory`, named from `base`.
114 Three files form the document step's artifact contract — the seam the
115 fusion step (and a later cloud job) reloads:
117 - `<base>.docling.json` — the DoclingDocument.
118 - `<base>.confidence.json` — per-page scores + per-table provenance.
119 - `<base>.cells.json` — the parsed-page cells.
121 Returns the written paths keyed by `document` / `confidence` / `cells`.
122 """
123 directory.mkdir(parents=True, exist_ok=True)
124 document_path = directory / f"{base}.docling.json"
125 confidence_path = directory / f"{base}.confidence.json"
126 cells_path = directory / f"{base}.cells.json"
128 self.document.save_as_json(document_path)
129 confidence_path.write_text(
130 json.dumps(
131 {
132 "page_scores": [s.model_dump() for s in self.page_scores],
133 "table_provenance": [p.model_dump() for p in self.table_provenance],
134 },
135 indent=2,
136 default=str,
137 ),
138 encoding="utf-8",
139 )
140 cells_path.write_text(
141 json.dumps([p.model_dump() for p in self.pages], indent=2, default=str),
142 encoding="utf-8",
143 )
144 return {"document": document_path, "confidence": confidence_path, "cells": cells_path}
146 @classmethod
147 def load(cls, directory: Path, base: str) -> "ParseResult":
148 """Reconstruct a ParseResult from artifacts written by `save`.
150 The `confidence` object is not restored (only the derived `page_scores`
151 are, which is all the fusion step reads); the document, page cells, and
152 per-table provenance round-trip exactly.
153 """
154 document = DoclingDocument.load_from_json(directory / f"{base}.docling.json")
155 confidence_blob = json.loads((directory / f"{base}.confidence.json").read_text())
156 page_scores = [PageScore(**s) for s in confidence_blob.get("page_scores", [])]
157 table_provenance = [TableProvenance(**p) for p in confidence_blob.get("table_provenance", [])]
158 cells_blob = json.loads((directory / f"{base}.cells.json").read_text())
159 pages = [PageParse(**p) for p in cells_blob]
160 return cls(
161 document=document,
162 confidence=None,
163 page_scores=page_scores,
164 pages=pages,
165 table_provenance=table_provenance,
166 )
168 @classmethod
169 def from_conversion(cls, result: ConversionResult) -> "ParseResult":
170 """Build a ParseResult from a docling ConversionResult.
172 Reads the per-page scores and the parsed-page cells off the result, then
173 derives the per-table provenance from the cells under each table box.
174 Requires the converter to have run with `generate_parsed_pages=True`;
175 without it `result.pages[i].cells` is empty and every table reads `empty`.
176 """
177 document = result.document
179 page_scores: List[PageScore] = []
180 if result.confidence is not None:
181 for page_no, scores in sorted(result.confidence.pages.items()):
182 page_scores.append(
183 PageScore(
184 page_no=page_no,
185 parse_score=scores.parse_score,
186 layout_score=scores.layout_score,
187 table_score=scores.table_score,
188 ocr_score=scores.ocr_score,
189 )
190 )
192 pages: List[PageParse] = []
193 for page in result.pages:
194 width = page.size.width if page.size else 0.0
195 height = page.size.height if page.size else 0.0
196 cells = [
197 ParsedCell(
198 text=cell.text,
199 box=_cell_box_top_left(cell.rect),
200 from_ocr=cell.from_ocr,
201 confidence=cell.confidence or 0.0,
202 )
203 for cell in page.cells
204 ]
205 pages.append(PageParse(page_no=page.page_no, width=width, height=height, cells=cells))
207 provenance = derive_table_provenance(document, pages)
209 return cls(
210 document=document,
211 confidence=result.confidence,
212 page_scores=page_scores,
213 pages=pages,
214 table_provenance=provenance,
215 )
218def derive_table_provenance(document: DoclingDocument, pages: List[PageParse]) -> List[TableProvenance]:
219 """The OCR-versus-native verdict for every table in `document`.
221 Purely geometric: each table's box is compared against the parsed-page cells
222 it encloses. A document whose tables were renumbered after the parse (fusion
223 grafts and splits tables) therefore reads correctly, because nothing here
224 depends on a reference recorded earlier.
225 """
226 pages_by_no = {p.page_no: p for p in pages}
227 return [_table_provenance(table, pages_by_no) for table in document.tables if table.prov]
230def _cell_box_top_left(rect: BoundingRectangle) -> Box:
231 """A parsed-cell rectangle (four corner points) -> a top-left (x0,y0,x1,y1) box."""
232 xs = [rect.r_x0, rect.r_x1, rect.r_x2, rect.r_x3]
233 ys = [rect.r_y0, rect.r_y1, rect.r_y2, rect.r_y3]
234 return (min(xs), min(ys), max(xs), max(ys))
237def _table_box_top_left(table: TableItem, page_height: float) -> Box:
238 """A table's provenance bbox -> a top-left (x0,y0,x1,y1) box in PDF points.
240 docling table boxes are bottom-left origin; flip y against the page height so
241 they share the parsed cells' top-left frame.
242 """
243 bbox = table.prov[0].bbox
244 if bbox.coord_origin == CoordOrigin.TOPLEFT:
245 return (bbox.l, bbox.t, bbox.r, bbox.b)
246 return (bbox.l, page_height - bbox.t, bbox.r, page_height - bbox.b)
249def _center_in(box: Box, region: Box) -> bool:
250 cx, cy = (box[0] + box[2]) / 2, (box[1] + box[3]) / 2
251 x0, y0, x1, y1 = (
252 min(region[0], region[2]),
253 min(region[1], region[3]),
254 max(region[0], region[2]),
255 max(region[1], region[3]),
256 )
257 return x0 <= cx <= x1 and y0 <= cy <= y1
260def _table_provenance(table: TableItem, pages_by_no: Dict[int, PageParse]) -> TableProvenance:
261 """Derive the OCR/native verdict for one table from the cells under its box."""
262 page_no = table.prov[0].page_no
263 page = pages_by_no.get(page_no)
264 self_ref = table.self_ref
266 if page is None:
267 return TableProvenance(
268 self_ref=self_ref,
269 page_no=page_no,
270 cell_count=0,
271 from_ocr_fraction=0.0,
272 mean_ocr_confidence=0.0,
273 verdict="empty",
274 )
276 region = _table_box_top_left(table, page.height)
277 under = [c for c in page.cells if _center_in(c.box, region)]
278 if not under:
279 return TableProvenance(
280 self_ref=self_ref,
281 page_no=page_no,
282 cell_count=0,
283 from_ocr_fraction=0.0,
284 mean_ocr_confidence=0.0,
285 verdict="empty",
286 )
288 ocr_cells = [c for c in under if c.from_ocr]
289 from_ocr_fraction = len(ocr_cells) / len(under)
290 mean_conf = (sum(c.confidence for c in ocr_cells) / len(ocr_cells)) if ocr_cells else 0.0
291 verdict: Literal["native", "ocr", "empty"] = "ocr" if from_ocr_fraction > 0.5 else "native"
292 return TableProvenance(
293 self_ref=self_ref,
294 page_no=page_no,
295 cell_count=len(under),
296 from_ocr_fraction=from_ocr_fraction,
297 mean_ocr_confidence=mean_conf,
298 verdict=verdict,
299 )