"""Tests for the figure workflow: nomination, reading a scan, and grafting.
The documents are synthetic but built with docling-core's own API, so the box
frame conversions and the mutation calls run for real. The scan responses carry
the shape a live dpt-2 scan returns — chunks of `figure`, `text` and
`marginalia`, each with a 0-based page and a left/top/right/bottom box, and a
metadata block reporting the job, the version and the credits.
"""
from __future__ import annotations
import asyncio
import tempfile
from pathlib import Path
from typing import Any, Dict, List, Sequence, Tuple, cast
import pytest
from docling_core.types.doc.base import BoundingBox, CoordOrigin, Size
from docling_core.types.doc.common.meta import DescriptionMetaField
from docling_core.types.doc.common.reference import ImageRef
from docling_core.types.doc.document import (
DoclingDocument,
PictureClassificationClass,
PictureClassificationData,
PictureClassificationMetaField,
PictureClassificationPrediction,
PictureMeta,
ProvenanceItem,
TableCell,
TableData,
TableItem,
)
from docling_core.types.doc.labels import DocItemLabel
from PIL import Image as PILImage
from quber.agents.cell_reader import MockCellReader
from quber.agents.figure_correction import FigureCorrection, Furniture, Marker, MockFigureCorrector, Note
from quber.core.extractors.base import ExtractedTable
from quber.core.extractors.set_of_mark.assemble import TableAssembly
from quber.core.figures.capture import capture_tables
from quber.core.figures.correct import candidates, correct_figures
from quber.core.figures.crosscheck import crosscheck_cells
from quber.core.figures.geometry import prov_box
from quber.core.figures.graft import (
FOOTNOTE_MARKS_FIELD,
FOOTNOTES_FIELD,
graft_figures,
graft_tables,
unread_pictures,
)
from quber.core.figures.grid import scanned_tables
from quber.core.figures.models import FigureRecord, FigureRun, PageScan, ScannedTable
from quber.core.figures.nominate import nominate_pages
from quber.core.figures.orchestrator import resolve_parse
from quber.core.figures.scan import page_scan
PAGE_W, PAGE_H = 720.0, 540.0
NormBox = Tuple[float, float, float, float]
def _bottomleft(box: NormBox) -> BoundingBox:
x1, y1, x2, y2 = box
return BoundingBox(
l=x1 * PAGE_W,
r=x2 * PAGE_W,
t=(1.0 - y1) * PAGE_H,
b=(1.0 - y2) * PAGE_H,
coord_origin=CoordOrigin.BOTTOMLEFT,
)
def _document(
pictures: Sequence[Tuple[int, NormBox, str]],
legacy_classes: bool = False,
tables: Sequence[Tuple[int, NormBox]] = (),
) -> DoclingDocument:
"""A document with pictures given as (page, normalized box, class name).
An empty class name leaves the picture unclassified. `legacy_classes` writes
the class into the deprecated annotations list instead of the meta field, the
shape older parse output carries. `tables` adds empty tables at (page,
normalized box), which is all nomination and matching read off them.
"""
doc = DoclingDocument(name="synthetic")
pages = {p for p, _box, _cls in pictures} | {p for p, _box in tables}
for page in sorted(pages):
doc.add_page(page_no=page, size=Size(width=PAGE_W, height=PAGE_H))
for page, box in tables:
doc.add_table(
data=TableData(table_cells=[], num_rows=0, num_cols=0),
prov=ProvenanceItem(page_no=page, bbox=_bottomleft(box), charspan=(0, 0)),
)
for page, box, cls in pictures:
prov = ProvenanceItem(page_no=page, bbox=_bottomleft(box), charspan=(0, 0))
picture = doc.add_picture(prov=prov)
if not cls:
continue
if legacy_classes:
picture.annotations.append(
PictureClassificationData(
provenance="test",
predicted_classes=[PictureClassificationClass(class_name=cls, confidence=0.9)],
)
)
else:
picture.meta = PictureMeta(
classification=PictureClassificationMetaField(
predictions=[PictureClassificationPrediction(class_name=cls, confidence=0.53)]
)
)
return doc
def _page_dims(pages: Sequence[int]) -> Dict[int, Tuple[float, float]]:
return dict.fromkeys(pages, (PAGE_W, PAGE_H))
def _chunk(kind: str, box: NormBox, markdown: str, chunk_id: str) -> Dict[str, Any]:
x1, y1, x2, y2 = box
return {
"id": chunk_id,
"type": kind,
"markdown": markdown,
"grounding": {"box": {"left": x1, "top": y1, "right": x2, "bottom": y2}, "page": 0},
}
def _response(chunks: List[Dict[str, Any]], credits: float = 3.0) -> Dict[str, Any]:
return {
"markdown": "\n\n".join(c["markdown"] for c in chunks),
"chunks": chunks,
"metadata": {
"credit_usage": credits,
"job_id": "cmrm81gqe0566ej01mtn47e4n",
"page_count": 1,
"version": "dpt-2-20260410",
"failed_pages": [],
},
}
BAR_CHART_TEXT = (
"<::bar chart\nY-axis: $ in millions, scaled from 0 to 600.\nX-axis: Years\n"
"Bars:\n2023: 310\n2024: 425\n2025: 560\n: bar chart::>"
)
# --- nomination ------------------------------------------------------------
def test_nominates_only_pages_holding_a_chart_class():
doc = _document(
[
(1, (0.1, 0.1, 0.9, 0.4), "logo"),
(2, (0.1, 0.1, 0.4, 0.5), "bar_chart"),
(3, (0.1, 0.1, 0.9, 0.4), "icon"),
]
)
nominated = nominate_pages(doc)
assert [n.page for n in nominated] == [2]
assert nominated[0].picture_classes == ["bar_chart"]
def test_nominates_a_page_once_carrying_every_chart_on_it():
doc = _document(
[
(7, (0.05, 0.15, 0.45, 0.65), "bar_chart"),
(7, (0.55, 0.15, 0.95, 0.65), "pie_chart"),
(7, (0.02, 0.02, 0.12, 0.06), "logo"),
]
)
nominated = nominate_pages(doc)
assert len(nominated) == 1
assert nominated[0].page == 7
assert nominated[0].picture_classes == ["bar_chart", "pie_chart"]
assert len(nominated[0].picture_refs) == 2
def test_nomination_covers_every_picture_that_is_not_page_furniture():
doc = _document(
[
(1, (0.1, 0.1, 0.4, 0.4), "line_chart"),
(2, (0.1, 0.1, 0.4, 0.4), "flow_chart"),
(3, (0.1, 0.1, 0.4, 0.4), "map"),
(4, (0.1, 0.1, 0.4, 0.4), "screenshot"),
(5, (0.1, 0.1, 0.4, 0.4), "logo"),
]
)
assert [n.page for n in nominate_pages(doc)] == [1, 2, 3, 4]
def test_every_nominated_page_is_scanned_whatever_its_pictures_are_called():
"""A diagram is content. Nothing decides on a page's behalf that it holds
nothing worth reading, so no class name keeps a nominated page from a scan."""
doc = _document(
[
(1, (0.1, 0.1, 0.4, 0.4), "line_chart"),
(2, (0.1, 0.1, 0.4, 0.4), "map"),
(3, (0.1, 0.1, 0.4, 0.4), "flow_chart"),
(4, (0.1, 0.1, 0.4, 0.4), "screenshot"),
]
)
assert [n.page for n in nominate_pages(doc)] == [1, 2, 3, 4]
def test_nomination_reads_the_deprecated_annotations_shape():
doc = _document([(4, (0.1, 0.1, 0.4, 0.4), "pie_chart")], legacy_classes=True)
assert [n.page for n in nominate_pages(doc)] == [4]
def test_unclassified_picture_is_nominated():
doc = _document([(1, (0.1, 0.1, 0.4, 0.4), "")])
nominated = nominate_pages(doc)
assert [n.page for n in nominated] == [1]
def test_a_table_read_off_the_page_image_nominates_its_page():
doc = _document([], tables=[(16, (0.05, 0.1, 0.95, 0.8)), (17, (0.05, 0.1, 0.95, 0.8))])
verdicts = {"#/tables/0": "ocr", "#/tables/1": "native"}
nominated = nominate_pages(doc, verdicts)
assert [n.page for n in nominated] == [16]
assert nominated[0].table_refs == ["#/tables/0"]
def test_a_table_nominates_its_page_only_when_the_verdicts_are_given():
doc = _document([], tables=[(16, (0.05, 0.1, 0.95, 0.8))])
assert nominate_pages(doc) == []
# --- reading a scan --------------------------------------------------------
def test_figure_becomes_a_chart_record_on_the_source_page():
response = _response([_chunk("figure", (0.08, 0.15, 0.48, 0.64), BAR_CHART_TEXT, "fig-1")])
scan = page_scan(response, page=7, model="dpt-2", picture_classes=["bar_chart"])
assert scan.status == "figures"
assert scan.page == 7
assert scan.version == "dpt-2-20260410"
assert scan.credits == 3.0
assert len(scan.figures) == 1
chart = scan.figures[0]
# The source page is carried in; the response numbers its own page from zero.
assert chart.page == 7
assert chart.text == BAR_CHART_TEXT
assert chart.box == {"left": 0.08, "top": 0.15, "right": 0.48, "bottom": 0.64}
assert chart.chunk_id == "fig-1"
assert chart.job_id == "cmrm81gqe0566ej01mtn47e4n"
def test_title_and_note_are_kept_beside_the_charts():
response = _response(
[
_chunk("text", (0.08, 0.07, 0.62, 0.10), "Adjusted Revenue(1) by Year", "t-1"),
_chunk("figure", (0.08, 0.15, 0.48, 0.64), BAR_CHART_TEXT, "fig-1"),
_chunk("marginalia", (0.04, 0.95, 0.96, 0.98), "Note: (1) See Appendix A.", "m-1"),
_chunk("table", (0.08, 0.70, 0.92, 0.90), "", "tb-1"),
]
)
scan = page_scan(response, page=7, model="dpt-2", picture_classes=["bar_chart"])
assert [c.kind for c in scan.context] == ["text", "marginalia"]
assert scan.context[0].text == "Adjusted Revenue(1) by Year"
assert scan.context[1].text == "Note: (1) See Appendix A."
# The title and the note are attached to no chart.
assert len(scan.figures) == 1
def test_two_charts_on_one_page_stay_separate():
response = _response(
[
_chunk("figure", (0.05, 0.15, 0.45, 0.64), BAR_CHART_TEXT, "fig-1"),
_chunk("figure", (0.55, 0.15, 0.95, 0.64), "<::pie chart\n: pie chart::>", "fig-2"),
]
)
scan = page_scan(response, page=9, model="dpt-2", picture_classes=["bar_chart", "pie_chart"])
assert [c.chunk_id for c in scan.figures] == ["fig-1", "fig-2"]
assert scan.figures[0].box != scan.figures[1].box
def test_a_page_returning_no_figure_is_recorded_as_scanned_and_empty():
response = _response([_chunk("text", (0.1, 0.1, 0.9, 0.2), "Segment results", "t-1")])
scan = page_scan(
response, page=12, model="dpt-2", picture_classes=["bar_chart"], response_artifact="a.json"
)
assert scan.status == "empty"
assert scan.figures == []
# The page was submitted, so it was billed, and it still names its response.
assert scan.credits == 3.0
assert scan.response_artifact == "a.json"
# --- which parse gets enriched ---------------------------------------------
def test_a_directory_resolves_to_the_fused_parse_over_the_raw_one(tmp_path):
# Both stages have written into the same directory. The fused document is
# the one every later stage reads, so it is the one enriched.
(tmp_path / "deck.docling.json").write_text("{}", encoding="utf-8")
(tmp_path / "deck.unified.json").write_text("{}", encoding="utf-8")
assert resolve_parse(str(tmp_path), "deck").name == "deck.unified.json"
def test_a_directory_falls_back_to_the_raw_parse():
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / "deck.docling.json").write_text("{}", encoding="utf-8")
assert resolve_parse(tmp, "deck").name == "deck.docling.json"
def test_a_file_is_taken_as_given(tmp_path):
path = tmp_path / "anything.json"
path.write_text("{}", encoding="utf-8")
assert resolve_parse(str(path), "deck") == path
def test_a_directory_holding_no_parse_says_what_it_wanted(tmp_path):
with pytest.raises(FileNotFoundError) as err:
resolve_parse(str(tmp_path), "deck")
assert "deck.unified.json" in str(err.value)
assert "deck.docling.json" in str(err.value)
# --- what a run reports ----------------------------------------------------
def test_a_dropped_page_still_appears_and_costs_nothing():
run = FigureRun(
document="deck",
scans=[
PageScan(page=3, status="dropped", reason="no chart on the page", picture_classes=["bar_chart"]),
PageScan(page=7, status="figures", credits=3.0, figures=[_chart(7, (0.1, 0.1, 0.5, 0.5))]),
],
)
assert run.nominated == 2
assert run.scanned == 1
assert run.submitted == 1
assert run.credits == 3.0
# The dropped page is in the output, so a nominated page never disappears.
assert [s.page for s in run.scans] == [3, 7]
def test_a_reused_scan_is_not_billed_to_the_run_that_read_it():
reused = FigureRun(
document="deck",
scans=[
PageScan(
page=7, status="figures", credits=3.0, reused=True, figures=[_chart(7, (0.1, 0.1, 0.5, 0.5))]
)
],
)
# The page has a scan and its records, but this run submitted nothing.
assert reused.scanned == 1
assert reused.submitted == 0
assert reused.credits == 0.0
assert len(reused.figures) == 1
# The scan still records what it cost when it was made.
assert reused.scans[0].credits == 3.0
# --- grafting --------------------------------------------------------------
def _scan(page: int, figures: Sequence[FigureRecord]) -> PageScan:
return PageScan(
page=page,
status="figures" if figures else "empty",
job_id="job-1",
model="dpt-2",
version="dpt-2-20260410",
figures=list(figures),
)
def _chart(page: int, box: NormBox, text: str = BAR_CHART_TEXT, chunk_id: str = "fig-1") -> FigureRecord:
x1, y1, x2, y2 = box
return FigureRecord(
page=page,
text=text,
box={"left": x1, "top": y1, "right": x2, "bottom": y2},
chunk_id=chunk_id,
job_id="job-1",
)
def test_chart_text_lands_on_the_picture_the_parse_found():
doc = _document([(7, (0.08, 0.15, 0.48, 0.64), "bar_chart")])
scans = [_scan(7, [_chart(7, (0.07, 0.14, 0.49, 0.65))])]
refined, errors = graft_figures(doc, scans, _page_dims([7]))
assert errors == []
picture = refined.pictures[0]
assert picture.meta is not None
assert picture.meta.description is not None
assert picture.meta.description.text == BAR_CHART_TEXT
# The class the parse predicted survives the graft.
assert picture.meta.classification is not None
assert picture.meta.classification.predictions[0].class_name == "bar_chart"
meta = picture.meta.model_dump()
assert meta["quber__figure_job_id"] == "job-1"
assert meta["quber__figure_model"] == "dpt-2"
assert meta["quber__figure_version"] == "dpt-2-20260410"
assert meta["quber__figure_chunk_id"] == "fig-1"
# The record names the picture it filled, both directions of the pairing.
assert scans[0].figures[0].picture_ref == picture.self_ref
def test_the_input_parse_is_not_mutated():
doc = _document([(7, (0.08, 0.15, 0.48, 0.64), "bar_chart")])
refined, _errors = graft_figures(doc, [_scan(7, [_chart(7, (0.07, 0.14, 0.49, 0.65))])], _page_dims([7]))
assert refined.pictures[0].meta is not None and refined.pictures[0].meta.description is not None
assert doc.pictures[0].meta is not None and doc.pictures[0].meta.description is None
def test_side_by_side_charts_land_on_their_own_pictures():
doc = _document(
[
(9, (0.05, 0.15, 0.45, 0.64), "pie_chart"),
(9, (0.55, 0.15, 0.95, 0.64), "pie_chart"),
]
)
scans = [
_scan(
9,
[
_chart(9, (0.05, 0.15, 0.45, 0.64), text="left donut", chunk_id="fig-1"),
_chart(9, (0.55, 0.15, 0.95, 0.64), text="right donut", chunk_id="fig-2"),
],
)
]
refined, errors = graft_figures(doc, scans, _page_dims([9]))
assert errors == []
texts = [
p.meta.description.text
for p in refined.pictures
if p.meta is not None and p.meta.description is not None
]
assert texts == ["left donut", "right donut"]
def test_a_figure_over_no_picture_is_surfaced_not_dropped():
# The parse recorded its picture at the top of the page; the scan found the
# chart at the bottom.
doc = _document([(7, (0.05, 0.05, 0.45, 0.20), "bar_chart")])
scans = [_scan(7, [_chart(7, (0.05, 0.70, 0.45, 0.95))])]
refined, errors = graft_figures(doc, scans, _page_dims([7]))
assert len(errors) == 1
assert "overlaps no picture" in errors[0]
assert refined.pictures[0].meta is not None
assert refined.pictures[0].meta.description is None
# The record survives with no picture to point at.
assert scans[0].figures[0].picture_ref is None
assert scans[0].figures[0].text == BAR_CHART_TEXT
def test_one_picture_over_two_figures_is_split_into_two():
# The parse drew a single wide region across two charts printed side by side;
# the scan returned them separately, each with its own box.
doc = _document([(9, (0.17, 0.61, 0.87, 0.92), "pie_chart")])
scans = [
_scan(
9,
[
_chart(9, (0.05, 0.51, 0.49, 0.92), text="left donut", chunk_id="fig-1"),
_chart(9, (0.50, 0.51, 0.95, 0.92), text="right donut", chunk_id="fig-2"),
],
)
]
refined, errors = graft_figures(doc, scans, _page_dims([9]))
assert errors == []
assert len(refined.pictures) == 2
left, right = refined.pictures
assert left.meta is not None and left.meta.description is not None
assert right.meta is not None and right.meta.description is not None
assert left.meta.description.text == "left donut"
assert right.meta.description.text == "right donut"
# Each part carries the box the scan drew, not the merged region's box.
assert prov_box(left, PAGE_W, PAGE_H) == pytest.approx((0.05, 0.51, 0.49, 0.92))
assert prov_box(right, PAGE_W, PAGE_H) == pytest.approx((0.50, 0.51, 0.95, 0.92))
# The parse's class prediction reaches both parts.
for part in (left, right):
assert part.meta is not None and part.meta.classification is not None
assert part.meta.classification.predictions[0].class_name == "pie_chart"
# Both records name their own part, and the parts are distinct elements.
refs = [c.picture_ref for c in scans[0].figures]
assert refs == [left.self_ref, right.self_ref]
assert left.self_ref != right.self_ref
def test_a_split_keeps_the_regions_reading_order_position():
doc = _document(
[
(9, (0.53, 0.19, 0.93, 0.50), "bar_chart"),
(9, (0.17, 0.61, 0.87, 0.92), "pie_chart"),
(9, (0.02, 0.94, 0.22, 0.97), "icon"),
]
)
scans = [
_scan(
9,
[
_chart(9, (0.05, 0.51, 0.49, 0.92), text="left donut", chunk_id="fig-1"),
_chart(9, (0.51, 0.12, 0.95, 0.51), text="waterfall", chunk_id="fig-2"),
_chart(9, (0.50, 0.51, 0.95, 0.92), text="right donut", chunk_id="fig-3"),
],
)
]
refined, errors = graft_figures(doc, scans, _page_dims([9]))
assert errors == []
# The split parts sit where the merged region sat: after the bar chart and
# before the icon that followed it. Reading order is the body's order, not
# the order the flat picture list happens to have been appended in.
by_ref = {p.self_ref: p for p in refined.pictures}
in_body = [by_ref[child.cref] for child in refined.body.children if child.cref in by_ref]
texts = [
(p.meta.description.text if p.meta is not None and p.meta.description is not None else None)
for p in in_body
]
assert texts == ["waterfall", "left donut", "right donut", None]
def _two_panel_pdf(dest: Path) -> Path:
"""A one-page PDF whose left half is solid red and right half solid blue.
A split part's bitmap can then be checked by colour: a part cut from the left
box must be red throughout, and one cut from the right box blue, while the
merged region's bitmap would contain both.
"""
import fitz
doc = fitz.open()
page = doc.new_page(width=PAGE_W, height=PAGE_H)
page.draw_rect(fitz.Rect(0, 0, PAGE_W / 2, PAGE_H), color=None, fill=(1, 0, 0))
page.draw_rect(fitz.Rect(PAGE_W / 2, 0, PAGE_W, PAGE_H), color=None, fill=(0, 0, 1))
doc.save(str(dest))
doc.close()
return dest
def _colours(ref: ImageRef) -> set[Tuple[int, int, int]]:
"""The distinct colours in a stored bitmap, sampled across its whole area."""
image = ref.pil_image
assert image is not None
rgb = image.convert("RGB")
width, height = rgb.size
return {
cast(Tuple[int, int, int], rgb.getpixel((x, y)))
for x in (1, width // 2, width - 2)
for y in (1, height // 2, height - 2)
}
def _merged_region_document() -> DoclingDocument:
"""One picture spanning both panels, carrying a bitmap of the merged region."""
doc = _document([(1, (0.10, 0.20, 0.90, 0.80), "pie_chart")])
picture = doc.pictures[0]
picture.image = ImageRef.from_pil(image=PILImage.new("RGB", (80, 60), (0, 255, 0)), dpi=72)
return doc
def test_a_split_part_gets_its_own_bitmap_cut_from_the_page(tmp_path):
pdf = _two_panel_pdf(tmp_path / "two-panel.pdf")
doc = _merged_region_document()
figures = [
_chart(1, (0.05, 0.20, 0.45, 0.80), text="left panel", chunk_id="fig-1"),
_chart(1, (0.55, 0.20, 0.95, 0.80), text="right panel", chunk_id="fig-2"),
]
refined, errors = graft_figures(doc, [_scan(1, figures)], _page_dims([1]), source=pdf)
assert errors == []
left, right = refined.pictures
assert left.image is not None and right.image is not None
# Each part's bitmap is a crop of its own box, not of the merged region: the
# left part is entirely red and the right entirely blue.
assert _colours(left.image) == {(255, 0, 0)}
assert _colours(right.image) == {(0, 0, 255)}
# Neither is the placeholder bitmap the merged picture carried.
assert (0, 255, 0) not in _colours(left.image)
# The bitmaps are cut at the resolution the merged picture carried, and their
# pixel size follows each part's own box.
assert left.image.dpi == 72 and right.image.dpi == 72
assert left.image.size.width == pytest.approx(0.40 * PAGE_W, abs=2)
assert left.image.size.height == pytest.approx(0.60 * PAGE_H, abs=2)
def test_a_split_without_the_source_keeps_boxes_and_drops_bitmaps(tmp_path):
_ = tmp_path
doc = _merged_region_document()
figures = [
_chart(1, (0.05, 0.20, 0.45, 0.80), text="left panel", chunk_id="fig-1"),
_chart(1, (0.55, 0.20, 0.95, 0.80), text="right panel", chunk_id="fig-2"),
]
refined, errors = graft_figures(doc, [_scan(1, figures)], _page_dims([1]))
# The split still happens and the text is still grafted; only the bitmaps are
# absent, because a bitmap of the merged region would depict the wrong thing.
assert errors == []
assert len(refined.pictures) == 2
assert all(p.image is None for p in refined.pictures)
assert [c.picture_ref for c in figures] == [p.self_ref for p in refined.pictures]
def test_an_unsplit_picture_keeps_the_bitmap_the_parse_cut():
doc = _merged_region_document()
refined, errors = graft_figures(
doc, [_scan(1, [_chart(1, (0.10, 0.20, 0.90, 0.80), text="one chart")])], _page_dims([1])
)
assert errors == []
assert len(refined.pictures) == 1
picture = refined.pictures[0]
assert picture.image is not None
assert _colours(picture.image) == {(0, 255, 0)}
def test_two_pictures_the_scan_read_as_one_figure_become_one():
# The parse split a chart pair the scan describes together. The scan looked at
# the page, so its count wins and the two pictures merge.
doc = _document(
[
(7, (0.06, 0.36, 0.50, 0.85), "bar_chart"),
(7, (0.54, 0.44, 0.95, 0.78), "bar_chart"),
]
)
figures = [_chart(7, (0.05, 0.15, 0.96, 0.86), text="both bar charts", chunk_id="fig-1")]
refined, errors = graft_figures(doc, [_scan(7, figures)], _page_dims([7]))
assert errors == []
assert len(refined.pictures) == 1
picture = refined.pictures[0]
assert picture.meta is not None and picture.meta.description is not None
assert picture.meta.description.text == "both bar charts"
# The surviving element carries the box the scan drew, spanning both figures.
assert prov_box(picture, PAGE_W, PAGE_H) == pytest.approx((0.05, 0.15, 0.96, 0.86))
assert picture.meta.classification is not None
assert picture.meta.classification.predictions[0].class_name == "bar_chart"
assert figures[0].picture_ref == picture.self_ref
def test_a_merge_keeps_the_regions_reading_order_position():
doc = _document(
[
(7, (0.10, 0.05, 0.90, 0.12), "line_chart"),
(7, (0.06, 0.36, 0.50, 0.85), "bar_chart"),
(7, (0.54, 0.44, 0.95, 0.78), "bar_chart"),
(7, (0.02, 0.95, 0.19, 0.98), "calendar"),
]
)
figures = [
_chart(7, (0.10, 0.05, 0.90, 0.12), text="banner chart", chunk_id="fig-0"),
_chart(7, (0.05, 0.15, 0.96, 0.86), text="both bar charts", chunk_id="fig-1"),
_chart(7, (0.02, 0.95, 0.19, 0.98), text="the calendar", chunk_id="fig-2"),
]
refined, errors = graft_figures(doc, [_scan(7, figures)], _page_dims([7]))
assert errors == []
by_ref = {p.self_ref: p for p in refined.pictures}
in_body = [by_ref[c.cref] for c in refined.body.children if c.cref in by_ref]
texts = [
(p.meta.description.text if p.meta is not None and p.meta.description is not None else None)
for p in in_body
]
assert texts == ["banner chart", "both bar charts", "the calendar"]
def test_a_region_the_two_sides_agree_on_keeps_the_box_the_parse_measured():
doc = _document([(7, (0.08, 0.15, 0.48, 0.64), "bar_chart")])
figures = [_chart(7, (0.07, 0.14, 0.49, 0.65))]
refined, errors = graft_figures(doc, [_scan(7, figures)], _page_dims([7]))
assert errors == []
# Nothing was reshaped, so the parse's own measurement stands.
assert prov_box(refined.pictures[0], PAGE_W, PAGE_H) == pytest.approx((0.08, 0.15, 0.48, 0.64))
def test_a_region_the_parse_called_page_furniture_is_dropped():
doc = _document(
[
(7, (0.08, 0.15, 0.48, 0.64), "bar_chart"),
(7, (0.02, 0.94, 0.19, 0.98), "logo"),
]
)
figures = [
_chart(7, (0.08, 0.15, 0.48, 0.64), text="a real chart", chunk_id="fig-1"),
_chart(7, (0.02, 0.94, 0.19, 0.98), text="a description of the wordmark", chunk_id="fig-2"),
]
scan = _scan(7, figures)
refined, errors = graft_figures(doc, [scan], _page_dims([7]))
assert errors == []
# The record is gone from the run, so a branded deck does not count its
# letterhead among the figures.
assert [c.chunk_id for c in scan.figures] == ["fig-1"]
# The logo picture survives in the document, carrying no description.
boxed = [(p, prov_box(p, PAGE_W, PAGE_H)) for p in refined.pictures]
logo = next(p for p, box in boxed if box is not None and box[1] > 0.9)
assert logo.meta is not None and logo.meta.description is None
def test_a_page_whose_only_figure_was_furniture_reads_as_empty():
doc = _document([(3, (0.02, 0.94, 0.19, 0.98), "logo")])
scan = _scan(3, [_chart(3, (0.02, 0.94, 0.19, 0.98), text="the wordmark", chunk_id="fig-1")])
_refined, errors = graft_figures(doc, [scan], _page_dims([3]))
assert errors == []
assert scan.figures == []
# The page was scanned and produced nothing worth keeping, which is what the
# run should report rather than claiming a figure.
assert scan.status == "empty"
def test_a_figure_takes_the_picture_it_fits_best_not_the_first_one_offered():
# Both figures overlap the wide region, the second one more than the first.
# Assigning greedily in arrival order would give the region to the first and
# orphan the better match.
doc = _document([(9, (0.17, 0.61, 0.87, 0.92), "pie_chart")])
figures = [
_chart(9, (0.05, 0.51, 0.49, 0.92), text="worse fit", chunk_id="fig-1"),
_chart(9, (0.50, 0.51, 0.95, 0.92), text="better fit", chunk_id="fig-2"),
]
refined, errors = graft_figures(doc, [_scan(9, figures)], _page_dims([9]))
# Neither is orphaned: they are both kept and the region is split.
assert errors == []
assert [c.picture_ref for c in figures] == [p.self_ref for p in refined.pictures]
def test_a_page_with_no_charts_leaves_its_pictures_alone():
doc = _document([(12, (0.08, 0.15, 0.48, 0.64), "bar_chart")])
refined, errors = graft_figures(doc, [_scan(12, [])], _page_dims([12]))
assert errors == []
assert refined.pictures[0].meta is not None
assert refined.pictures[0].meta.description is None
# --- tables read off the page image ----------------------------------------
def _table_response(cells: List[Dict[str, Any]], table_id: str = "t-1") -> Dict[str, Any]:
"""A response holding one table chunk, with the grounding map a scan returns.
`cells` are given as {"id", "row", "col", "text"} plus optional spans; the
HTML grid and the per-cell grounding entries are both built from them, the
way a live response carries the two.
"""
rows = "".join(
""
+ "".join(
f'| {c["text"]} | '
for c in cells
if c["row"] == r
)
+ "
"
for r in sorted({c["row"] for c in cells})
)
chunk = _chunk("table", (0.05, 0.10, 0.95, 0.80), f'', "chunk-t")
grounding: Dict[str, Any] = {
table_id: {
"box": {"left": 0.06, "top": 0.11, "right": 0.94, "bottom": 0.78},
"page": 0,
"type": "table",
}
}
for c in cells:
grounding[c["id"]] = {
"box": {"left": 0.1, "top": 0.2, "right": 0.4, "bottom": 0.25},
"page": 0,
"type": "tableCell",
"confidence": 0.99,
"position": {
"chunk_id": "chunk-t",
"row": c["row"],
"col": c["col"],
"rowspan": c.get("rowspan", 1),
"colspan": c.get("colspan", 1),
},
}
response = _response([chunk])
response["grounding"] = grounding
return response
def test_a_returned_table_becomes_a_dense_grid_with_a_box_on_every_cell():
response = _table_response(
[
{"id": "c1", "row": 0, "col": 0, "text": "Assets"},
{"id": "c2", "row": 0, "col": 1, "text": "2026", "colspan": 2},
{"id": "c3", "row": 1, "col": 0, "text": "Cash"},
{"id": "c4", "row": 1, "col": 1, "text": "$ (1,707,526)"},
{"id": "c5", "row": 1, "col": 2, "text": "12%"},
]
)
scan = page_scan(response, page=16, model="dpt-2", picture_classes=[])
assert scan.status == "tables"
assert len(scan.tables) == 1
table = scan.tables[0]
# The spanning header cell is written once; the position it covers is blank.
assert table.cells == [
["Assets", "2026", ""],
["Cash", "$ (1,707,526)", "12%"],
]
assert [[b is not None for b in row] for row in table.cell_boxes] == [
[True, True, False],
[True, True, True],
]
# The table's own box, drawn tighter than the chunk that carries it.
assert table.box == {"left": 0.06, "top": 0.11, "right": 0.94, "bottom": 0.78}
def test_markup_inside_a_cell_reads_as_the_text_the_page_prints():
response = _table_response([{"id": "c1", "row": 0, "col": 0, "text": "Net income
1Q26"}])
table = page_scan(response, page=16, model="dpt-2", picture_classes=[]).tables[0]
assert table.cells == [["Net income 1Q26"]]
def test_a_page_returning_a_figure_and_a_table_reports_the_figure():
response = _table_response([{"id": "c1", "row": 0, "col": 0, "text": "Cash"}])
response["chunks"].append(_chunk("figure", (0.1, 0.1, 0.4, 0.3), BAR_CHART_TEXT, "fig-1"))
scan = page_scan(response, page=16, model="dpt-2", picture_classes=["bar_chart"])
assert scan.status == "figures"
assert len(scan.tables) == 1
assert len(scan.figures) == 1
def test_a_nominated_table_the_scan_returned_nothing_over_is_reported():
doc = _document([], tables=[(16, (0.05, 0.1, 0.95, 0.8))])
scan = PageScan(page=16, status="empty", table_refs=["#/tables/0"])
tables, errors = asyncio.run(
capture_tables(doc, [scan], Path("doc.pdf"), _page_dims([16]), cast(Any, None))
)
assert tables == []
assert len(errors) == 1
assert "#/tables/0" in errors[0]
def test_the_scans_reading_replaces_the_parses_own_reading_of_the_image():
doc = _document([], tables=[(16, (0.05, 0.1, 0.95, 0.8))])
scanned = ScannedTable(
page=16,
cells=[["Cash"], ["1,707,526"]],
cell_boxes=[[None], [None]],
chunk_id="chunk-t",
table_ref="#/tables/0",
table_id="doc-p16-t1",
)
scan = PageScan(page=16, status="tables", table_refs=["#/tables/0"], tables=[scanned])
extracted = ExtractedTable(
table_id="doc-p16-t1",
title="Consolidated Balance Sheets",
markdown="| Cash |\n| --- |\n| 1,707,526 |",
page=16,
kind="image_table",
)
errors = graft_tables(doc, [scan], [extracted], _page_dims([16]))
assert errors == []
table = doc.tables[0]
assert [c.text for c in table.data.table_cells] == ["Cash", "1,707,526"]
assert [doc.texts[int(ref.cref.rsplit("/", 1)[1])].text for ref in table.captions] == [
"Consolidated Balance Sheets"
]
def test_a_scan_that_read_no_cells_leaves_the_parses_reading_standing():
doc = _document([], tables=[(16, (0.05, 0.1, 0.95, 0.8))])
doc.tables[0].data = TableData(
table_cells=[
TableCell(
text="S 1.707.526",
start_row_offset_idx=0,
end_row_offset_idx=1,
start_col_offset_idx=0,
end_col_offset_idx=1,
)
],
num_rows=1,
num_cols=1,
)
scanned = ScannedTable(page=16, cells=[], cell_boxes=[], table_ref="#/tables/0", table_id="doc-p16-t1")
scan = PageScan(page=16, status="tables", table_refs=["#/tables/0"], tables=[scanned])
extracted = ExtractedTable(table_id="doc-p16-t1", markdown="", page=16, kind="image_table")
errors = graft_tables(doc, [scan], [extracted], _page_dims([16]))
assert len(errors) == 1
assert [c.text for c in doc.tables[0].data.table_cells] == ["S 1.707.526"]
# --- checking a scanned cell against the parse ------------------------------
def _scanned(cells, boxes) -> ScannedTable:
return ScannedTable(page=16, cells=cells, cell_boxes=boxes, chunk_id="chunk-t")
def _box(x1, y1, x2, y2):
return {"left": x1, "top": y1, "right": x2, "bottom": y2}
def _parse_table(entries) -> TableItem:
"""A docling table whose cells are given as (row, col, text, normalized box)."""
doc = DoclingDocument(name="synthetic")
doc.add_page(page_no=16, size=Size(width=PAGE_W, height=PAGE_H))
cells = [
TableCell(
text=text,
start_row_offset_idx=r,
end_row_offset_idx=r + 1,
start_col_offset_idx=c,
end_col_offset_idx=c + 1,
bbox=BoundingBox(
l=b[0] * PAGE_W,
t=b[1] * PAGE_H,
r=b[2] * PAGE_W,
b=b[3] * PAGE_H,
coord_origin=CoordOrigin.TOPLEFT,
),
)
for r, c, text, b in entries
]
rows = max((c.end_row_offset_idx for c in cells), default=0)
cols = max((c.end_col_offset_idx for c in cells), default=0)
return doc.add_table(
data=TableData(table_cells=cells, num_rows=rows, num_cols=cols),
prov=ProvenanceItem(page_no=16, bbox=_bottomleft((0.0, 0.0, 1.0, 1.0)), charspan=(0, 0)),
)
def test_a_cell_both_readers_agree_on_is_not_asked_about():
scanned = _scanned([["21,249,107"]], [[_box(0.1, 0.1, 0.3, 0.12)]])
table = _parse_table([(0, 0, "21,249,107", (0.1, 0.1, 0.3, 0.12))])
reader = MockCellReader()
cells, read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", reader))
assert cells == [["21,249,107"]]
assert read == 0
assert reader.calls == []
def test_a_cell_the_two_readers_differ_on_is_read_off_the_page():
# The scan turned a thousands separator into a decimal point; the parse did not.
scanned = _scanned([["19.543.903"]], [[_box(0.1, 0.1, 0.3, 0.12)]])
table = _parse_table([(0, 0, "19,543,903", (0.1, 0.1, 0.3, 0.12))])
reader = MockCellReader({"A1": "19,543,903"})
cells, read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", reader))
assert cells == [["19,543,903"]]
assert read == 1
assert [(q.address, q.scan_read, q.parse_read) for q in reader.calls[0]] == [
("A1", "19.543.903", "19,543,903")
]
def test_the_agent_may_answer_with_neither_reading():
# Both readers corrupted the same cell in different places.
scanned = _scanned([["$ 3.000.00"]], [[_box(0.1, 0.1, 0.3, 0.12)]])
table = _parse_table([(0, 0, "S 3,000,00", (0.1, 0.1, 0.3, 0.12))])
reader = MockCellReader({"A1": "$ 3,000.00"})
cells, _read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", reader))
assert cells == [["$ 3,000.00"]]
def test_a_cell_the_parse_never_read_is_still_checked():
scanned = _scanned([["524.001"]], [[_box(0.1, 0.1, 0.3, 0.12)]])
table = _parse_table([(0, 0, "elsewhere", (0.8, 0.8, 0.9, 0.9))])
reader = MockCellReader({"A1": "524,001"})
cells, read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", reader))
assert cells == [["524,001"]]
assert read == 1
assert reader.calls[0][0].parse_read is None
def test_a_tight_box_inside_a_wide_one_is_the_same_cell():
"""The parse bounds a row heading to the word, the scan to the whole band.
Measured against their combined area they barely overlap; against the smaller
box the tight one is wholly inside the wide one."""
scanned = _scanned([["ASSETS"]], [[_box(0.244, 0.174, 0.672, 0.191)]])
table = _parse_table([(0, 0, "ASSETS", (0.244, 0.175, 0.279, 0.190))])
reader = MockCellReader()
cells, read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", reader))
assert cells == [["ASSETS"]]
assert read == 0
assert reader.calls == []
def test_a_failed_reading_leaves_every_cell_as_the_scan_read_it():
scanned = _scanned([["19.543.903"]], [[_box(0.1, 0.1, 0.3, 0.12)]])
table = _parse_table([(0, 0, "19,543,903", (0.1, 0.1, 0.3, 0.12))])
reader = MockCellReader() # answers nothing
cells, read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", reader))
assert cells == [["19.543.903"]]
assert read == 0
def test_no_reader_leaves_the_grid_alone():
scanned = _scanned([["19.543.903"]], [[_box(0.1, 0.1, 0.3, 0.12)]])
table = _parse_table([(0, 0, "19,543,903", (0.1, 0.1, 0.3, 0.12))])
cells, read = asyncio.run(crosscheck_cells(scanned, table, (PAGE_W, PAGE_H), b"png", None))
assert cells == [["19.543.903"]]
assert read == 0
# --- FigureCorrection: the text a read figure leaves behind on the page ---------------
def _blank_pdf(dest: Path, pages: int) -> Path:
"""A PDF of `pages` blank pages at the test page size, for the page render."""
import fitz
doc = fitz.open()
for _ in range(pages):
doc.new_page(width=PAGE_W, height=PAGE_H)
doc.save(str(dest))
doc.close()
return dest
def _with_texts(doc: DoclingDocument, texts: Sequence[Tuple[int, NormBox, str, DocItemLabel]]):
"""Add texts to `doc` as (page, normalized box, text, label), parented to the body."""
for page, box, body, label in texts:
doc.add_text(
label=label,
text=body,
prov=ProvenanceItem(page_no=page, bbox=_bottomleft(box), charspan=(0, len(body))),
)
return doc
def test_only_plain_text_overlapping_a_read_figure_is_a_candidate():
doc = _with_texts(
_document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")]),
[
(7, (0.11, 0.45, 0.16, 0.47), "90.0%", DocItemLabel.TEXT),
(7, (0.11, 0.50, 0.16, 0.52), "80.0%", DocItemLabel.TEXT),
# Inside the figure but labelled: never a candidate, whatever it overlaps.
(7, (0.20, 0.42, 0.45, 0.44), "Asset Performance", DocItemLabel.SECTION_HEADER),
(7, (0.20, 0.75, 0.45, 0.78), "(1) Based on carrying value.", DocItemLabel.FOOTNOTE),
# Plain text, but printed clear of the figure.
(7, (0.10, 0.05, 0.60, 0.08), "During Q1 2026 the Company amended", DocItemLabel.TEXT),
],
)
scan = PageScan(page=7, status="figures", figures=[_chart(7, (0.10, 0.40, 0.50, 0.80))])
found = candidates(doc, scan, _page_dims([7]))
assert [t.text for t in found] == ["90.0%", "80.0%"]
def test_a_page_the_scan_read_no_figure_on_has_no_candidates():
doc = _with_texts(
_document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")]),
[(7, (0.11, 0.45, 0.16, 0.47), "90.0%", DocItemLabel.TEXT)],
)
scan = PageScan(page=7, status="empty", figures=[])
assert candidates(doc, scan, _page_dims([7])) == []
def test_removed_text_is_recorded_with_its_page_box_and_reason():
doc = _with_texts(
_document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")]),
[
(7, (0.11, 0.45, 0.16, 0.47), "90.0%", DocItemLabel.TEXT),
(7, (0.11, 0.50, 0.16, 0.52), "80.0%", DocItemLabel.TEXT),
],
)
scan = PageScan(page=7, status="figures", figures=[_chart(7, (0.10, 0.40, 0.50, 0.80))])
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "doc.pdf"
_blank_pdf(source, pages=7)
removed = asyncio.run(
correct_figures(doc, [scan], _page_dims([7]), source, MockFigureCorrector(["90.0%"]))
)
assert [(r.page, r.text) for r in removed] == [(7, "90.0%")]
assert removed[0].box is not None
x1, y1, x2, y2 = removed[0].box
assert (round(x1, 2), round(y1, 2), round(x2, 2), round(y2, 2)) == (0.11, 0.45, 0.16, 0.47)
assert [t.text for t in doc.texts] == ["80.0%"]
def test_a_figures_markers_land_on_the_picture_it_was_grafted_onto():
"""The agent reads a marker off the chart's label and it becomes an attribute.
Recorded on the picture with what it points at, so resolving it later reads a
field rather than the description it was printed in.
"""
doc = _document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")])
chart = _chart(7, (0.10, 0.40, 0.50, 0.80))
chart.picture_ref = doc.pictures[0].self_ref
scan = PageScan(page=7, status="figures", figures=[chart])
finder = MockFigureCorrector(
markers=[
Marker(figure=0, marker="1", label="Book Equity Value", kind="footnote"),
Marker(figure=0, marker="2", label="Book Equity Value", kind="section"),
]
)
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "doc.pdf"
_blank_pdf(source, pages=7)
asyncio.run(correct_figures(doc, [scan], _page_dims([7]), source, finder))
recorded = getattr(doc.pictures[0].meta, FOOTNOTE_MARKS_FIELD, None)
assert recorded == [
{"marker": "1", "kind": "footnote", "label": "Book Equity Value"},
{"marker": "2", "kind": "section", "label": "Book Equity Value"},
]
def test_a_fragment_named_furniture_twice_is_removed_once():
"""Two verdicts on one fragment are one text item, and the document model
rejects the same element queued for deletion twice, failing the whole sweep."""
correction = FigureCorrection(
furniture=[
Furniture(index=3, reason="axis tick"),
Furniture(index=3, reason="gridline label"),
Furniture(index=5, reason="legend key"),
]
)
assert [(f.index, f.reason) for f in correction.furniture] == [(3, "axis tick"), (5, "legend key")]
def test_a_stacked_marker_group_returned_as_one_string_is_split():
"""The prompt asks for one entry per marker; compliance is not certain.
A label printed '(b)(c)(d)' that comes back as a single marker becomes three,
the same normalization a table's markers get. A parenthesized negative value
is not a group and is left alone.
"""
correction = FigureCorrection(
markers=[
Marker(figure=0, marker="(b)(c)(d)", label="Segment EBITDA"),
Marker(figure=0, marker="(84)", label="Net loss"),
]
)
assert [(m.marker, m.label) for m in correction.markers] == [
("(b)", "Segment EBITDA"),
("(c)", "Segment EBITDA"),
("(d)", "Segment EBITDA"),
("(84)", "Net loss"),
]
def test_the_notes_the_agent_read_off_the_page_land_on_the_picture():
"""The definitions themselves, not just the markers pointing at them.
A table's correction agent returns the notes printed below it as marker and
text pairs, and resolution starts from those rather than from a scan over the
page's lines. A figure states them the same way.
"""
doc = _document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")])
chart = _chart(7, (0.10, 0.40, 0.50, 0.80))
chart.picture_ref = doc.pictures[0].self_ref
scan = PageScan(page=7, status="figures", figures=[chart])
finder = MockFigureCorrector(
markers=[Marker(figure=0, marker="1", kind="footnote")],
notes=[
Note(figure=0, marker="1", text="Excludes noncontrolling interest"),
Note(figure=0, marker="", text="As of 03/31/2026"),
],
)
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "doc.pdf"
_blank_pdf(source, pages=7)
asyncio.run(correct_figures(doc, [scan], _page_dims([7]), source, finder))
assert getattr(doc.pictures[0].meta, FOOTNOTES_FIELD, None) == [
{"marker": "1", "text": "Excludes noncontrolling interest"},
{"marker": "", "text": "As of 03/31/2026"},
]
def test_a_figure_carrying_markers_is_read_on_a_page_with_nothing_to_sweep():
"""A chart printing a marker on its title leaves no fragments behind. The page
is still put to the agent, because the marker is on the figure, not beside it."""
doc = _document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")])
chart = _chart(7, (0.10, 0.40, 0.50, 0.80))
chart.picture_ref = doc.pictures[0].self_ref
scan = PageScan(page=7, status="figures", figures=[chart])
finder = MockFigureCorrector(markers=[Marker(figure=0, marker="*", kind="footnote")])
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "doc.pdf"
_blank_pdf(source, pages=7)
removed = asyncio.run(correct_figures(doc, [scan], _page_dims([7]), source, finder))
assert removed == []
assert finder.calls == [[]]
assert [m["marker"] for m in getattr(doc.pictures[0].meta, FOOTNOTE_MARKS_FIELD)] == ["*"]
def test_the_sweep_switched_off_removes_nothing():
doc = _with_texts(
_document([(7, (0.10, 0.40, 0.50, 0.80), "bar_chart")]),
[(7, (0.11, 0.45, 0.16, 0.47), "90.0%", DocItemLabel.TEXT)],
)
scan = PageScan(page=7, status="figures", figures=[_chart(7, (0.10, 0.40, 0.50, 0.80))])
removed = asyncio.run(correct_figures(doc, [scan], _page_dims([7]), Path("unused.pdf"), None))
assert removed == []
assert [t.text for t in doc.texts] == ["90.0%"]
def test_a_table_the_parse_filed_as_a_picture_is_added_beside_it():
"""The parse detected the region and called it a picture. The scan read a
grid over it, so the page prints a table and the document gains one. The
picture stays: it is a real element and the values were read off it."""
doc = _document([(14, (0.48, 0.23, 0.96, 0.92), "table")])
scanned = ScannedTable(
page=14,
cells=[["Year", "Sales"], ["2025", "$13"]],
cell_boxes=[[None, None], [None, None]],
box={"left": 0.49, "top": 0.24, "right": 0.95, "bottom": 0.91},
)
scan = PageScan(page=14, status="tables", tables=[scanned])
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "doc.pdf"
_blank_pdf(source, pages=14)
assembly = TableAssembly(source, cast(Any, None), asyncio.Semaphore(1), 200)
tables, errors = asyncio.run(
capture_tables(doc, [scan], source, _page_dims([14]), assembly, MockCellReader())
)
assert errors == []
assert scanned.picture_ref == "#/pictures/0"
assert scanned.table_ref is None
assert len(tables) == 1
graft_errors = graft_tables(doc, [scan], tables, _page_dims([14]))
assert graft_errors == []
assert len(doc.tables) == 1
assert len(doc.pictures) == 1
def test_a_picture_a_parse_table_already_covers_is_not_a_second_home():
"""Docling detects the same region twice, once as a table and once as a
picture. The table engine read the text layer there, so the scan's reading is
ignored the way it is for any table the page nominated nothing for. Adding it
beside the parse table would state the same figures twice."""
doc = _document(
[(7, (0.05, 0.14, 0.49, 0.48), "table")],
tables=[(7, (0.05, 0.14, 0.49, 0.48))],
)
scanned = ScannedTable(
page=7,
cells=[["CRE Loan Summary", "$2,202.8"]],
cell_boxes=[[None, None]],
box={"left": 0.05, "top": 0.14, "right": 0.49, "bottom": 0.48},
)
scan = PageScan(page=7, status="tables", tables=[scanned])
with tempfile.TemporaryDirectory() as tmp:
source = Path(tmp) / "doc.pdf"
_blank_pdf(source, pages=7)
assembly = TableAssembly(source, cast(Any, None), asyncio.Semaphore(1), 200)
tables, errors = asyncio.run(
capture_tables(doc, [scan], source, _page_dims([7]), assembly, MockCellReader())
)
assert tables == []
assert errors == []
assert scanned.picture_ref is None
assert len(doc.tables) == 1
def test_a_picture_nothing_read_is_reported():
"""A figure covering no picture is already an error. The reverse was silent,
which is how a table the parse filed as a picture went missing on a page the
run reported nothing for."""
doc = _document(
[
(3, (0.1, 0.1, 0.4, 0.4), "bar_chart"),
(3, (0.5, 0.1, 0.9, 0.4), "bar_code"),
(3, (0.02, 0.94, 0.2, 0.98), "logo"),
]
)
doc.pictures[0].meta = PictureMeta(
description=DescriptionMetaField(text="a bar chart", created_by="test")
)
errors = unread_pictures(doc, _page_dims([3]))
assert len(errors) == 1
assert "#/pictures/1" in errors[0] and "bar_code" in errors[0]
def test_a_picture_a_table_covers_is_not_reported():
"""The parse detected one region twice, as a table and as a picture. The
table holds the content; the picture is a duplicate outline of it."""
doc = _document(
[(7, (0.05, 0.14, 0.49, 0.48), "table")],
tables=[(7, (0.05, 0.14, 0.49, 0.48))],
)
assert unread_pictures(doc, _page_dims([7])) == []
COMPOSITE_TEXT = (
"<::chart and table::>\n"
"<::chart: Cumulative Bonds Issued::>\n"
"Sep. 2012: $325\nJul. 2014: $625\n"
"<::table::>\n"
"| Category | Sep. 2012 | Jul. 2014 |\n"
"|:---|:---|:---|\n"
"| New Issuance Size ($M) | $325 | $300 |\n"
"| Bond Issue Credit Ratings | Ba3 / B+ / BB | Ba3 / B+ / BB |\n"
"<::/table::>\n"
"<::/chart and table::>"
)
def test_a_table_printed_inside_a_figure_is_read_as_a_table():
"""A page printing a chart and a table as one composite comes back as a
single figure with the table written into its text between markers the scan
emits itself. The markers are read; nothing infers where the table is."""
response = _response([_chunk("figure", (0.02, 0.24, 0.99, 0.84), COMPOSITE_TEXT, "fig-1")])
tables = scanned_tables(response, page=18, job_id="job-1")
assert len(tables) == 1
table = tables[0]
assert table.cells == [
["Category", "Sep. 2012", "Jul. 2014"],
["New Issuance Size ($M)", "$325", "$300"],
["Bond Issue Credit Ratings", "Ba3 / B+ / BB", "Ba3 / B+ / BB"],
]
# The composite is grounded as one box and nothing per cell, so the cells
# arrive without geometry and the correction reads the page image instead.
assert table.cell_boxes == [[None] * 3 for _ in range(3)]
assert table.box == {"left": 0.02, "top": 0.24, "right": 0.99, "bottom": 0.84}
def test_a_figure_with_no_table_block_yields_no_table():
response = _response([_chunk("figure", (0.1, 0.1, 0.5, 0.5), BAR_CHART_TEXT, "fig-1")])
assert scanned_tables(response, page=7, job_id="job-1") == []