Coverage for src / quber / core / figures / crosscheck.py: 94%

1"""Check every cell the scan read against the parse's reading of the same cell. 

2 

3The scan is the reader for a table printed as an image, and it has one failure 

4mode: it turns thousands separators into decimal points. The parse reads the same 

5image and is already in the artifacts. It is not a second table and contributes 

6no structure — it is a second opinion on what a cell says, available for free, 

7and it catches that failure. 

8 

9Every cell in the scan's grid is checked. A cell leaves here either confirmed by 

10a second reader or read off the page by an agent. None stands on the scan's word 

11alone: 

12 

13- The parse read the same text. Two independent readers agree, and the cell 

14 stands. 

15- The parse read something different. The agent reads the cell. 

16- The parse has no cell there. The agent reads the cell. Having no second 

17 opinion is a reason to look harder, not a reason to wave it through. 

18 

19Cells pair by how much one box covers the other, taken against the smaller of 

20the two. The readers agree on where a cell is and disagree on how much room it 

21takes: a row heading is bound by the parse to the word itself and by the scan to 

22the whole row band, one box sitting entirely inside the other. Measured against 

23their combined area those overlap almost not at all. 

24 

25Texts are compared as they stand. A currency symbol read as a letter, a comma 

26read as a period, a parenthesis dropped from a negative — each changes what the 

27document states, and a rule deciding which of those were too cosmetic to ask 

28about would be deciding the thing the agent is there to decide. 

29 

30This checks values, not coverage. If the page prints a line the scan drew no box 

31around, the table is short a row and nothing here notices. 

32""" 

33 

34from __future__ import annotations 

35 

36from typing import Dict, List, Optional, Tuple 

37 

38from docling_core.types.doc.base import CoordOrigin 

39from docling_core.types.doc.document import TableItem 

40from loguru import logger 

41 

42from quber.agents.cell_reader import CellQuestion, CellReader 

43from quber.core.extractors.camelot.acquire import column_letter, grid_to_addressed_markdown 

44from quber.core.extractors.camelot.correspondence.geometry import coverage_fraction 

45from quber.core.figures.geometry import NormBox, norm_box 

46from quber.core.figures.models import ScannedTable 

47 

48#: Minimum coverage for a scan cell and a parse cell to be the same cell, 

49#: measured against the smaller box so a tight box inside a wide one registers. 

50MATCH_FRACTION = 0.10 

51 

52 

53async def crosscheck_cells( 

54 scanned: ScannedTable, 

55 table_item: Optional[TableItem], 

56 page_dims: Tuple[float, float], 

57 table_image: bytes, 

58 reader: Optional[CellReader], 

59 match_fraction: float = MATCH_FRACTION, 

60) -> Tuple[List[List[str]], int]: 

61 """The scan's grid with every unconfirmed cell read off the page. 

62 

63 Returns the grid and how many cells the agent read. The grid keeps its shape: 

64 only cell text changes, never the rows, the columns or the boxes. 

65 

66 Without a reader, or without a parse table to check against, the grid is 

67 returned unchanged and the reason is logged. A run that cannot check is not 

68 the same as a run that checked and found nothing. 

69 """ 

70 if reader is None: 

71 logger.info("Cell check: page {} no reader configured; the scan's grid stands", scanned.page) 

72 return scanned.cells, 0 

73 

74 questions, addresses = _questions(scanned, table_item, page_dims, match_fraction) 

75 if not questions: 

76 logger.info("Cell check: page {} every cell confirmed by the parse", scanned.page) 

77 return scanned.cells, 0 

78 

79 logger.info( 

80 "Cell check: page {} asking about {} of {} cell(s)", 

81 scanned.page, 

82 len(questions), 

83 sum(1 for row in scanned.cells for c in row if c.strip()), 

84 ) 

85 readings = await reader.read_cells(table_image, grid_to_addressed_markdown(scanned.cells), questions) 

86 

87 grid = [list(row) for row in scanned.cells] 

88 read = 0 

89 asked = {q.address for q in questions} 

90 for cell in readings.cells: 

91 if cell.address not in asked: 

92 logger.warning( 

93 "Cell check: page {} the reader returned {} which it was not asked about; ignored", 

94 scanned.page, 

95 cell.address, 

96 ) 

97 continue 

98 row, col = addresses[cell.address] 

99 if grid[row][col] != cell.value: 

100 logger.info( 

101 "Cell check: page {} {} {!r} -> {!r}", 

102 scanned.page, 

103 cell.address, 

104 grid[row][col], 

105 cell.value, 

106 ) 

107 grid[row][col] = cell.value 

108 read += 1 

109 

110 missing = asked - {c.address for c in readings.cells} 

111 if missing: 

112 logger.warning( 

113 "Cell check: page {} the reader returned nothing for {}; those cells keep the scan's reading", 

114 scanned.page, 

115 sorted(missing), 

116 ) 

117 return grid, read 

118 

119 

120def _questions( 

121 scanned: ScannedTable, 

122 table_item: Optional[TableItem], 

123 page_dims: Tuple[float, float], 

124 match_fraction: float, 

125) -> Tuple[List[CellQuestion], Dict[str, Tuple[int, int]]]: 

126 """The cells the parse did not confirm, and where each address sits in the grid.""" 

127 parse_cells = _parse_cells(table_item, page_dims) 

128 

129 questions: List[CellQuestion] = [] 

130 addresses: Dict[str, Tuple[int, int]] = {} 

131 for row, texts in enumerate(scanned.cells): 

132 for col, text in enumerate(texts): 

133 if not text.strip(): 

134 continue 

135 box = norm_box(scanned.cell_boxes[row][col] if col < len(scanned.cell_boxes[row]) else None) 

136 other = _best_match(box, parse_cells, match_fraction) 

137 if other is not None and other == text: 

138 continue 

139 address = f"{column_letter(col)}{row + 1}" 

140 addresses[address] = (row, col) 

141 questions.append(CellQuestion(address=address, scan_read=text, parse_read=other)) 

142 return questions, addresses 

143 

144 

145def _parse_cells( 

146 table_item: Optional[TableItem], page_dims: Tuple[float, float] 

147) -> List[Tuple[str, NormBox]]: 

148 """The parse's cells for this table, boxed in the normalized top-left frame.""" 

149 if table_item is None: 

150 return [] 

151 width, height = page_dims 

152 out: List[Tuple[str, NormBox]] = [] 

153 for cell in table_item.data.table_cells: 

154 bbox = cell.bbox 

155 if bbox is None or not (cell.text or "").strip(): 

156 continue 

157 if bbox.coord_origin == CoordOrigin.TOPLEFT: 

158 box = (bbox.l / width, min(bbox.t, bbox.b) / height, bbox.r / width, max(bbox.t, bbox.b) / height) 

159 else: 

160 box = ( 

161 bbox.l / width, 

162 (height - max(bbox.t, bbox.b)) / height, 

163 bbox.r / width, 

164 (height - min(bbox.t, bbox.b)) / height, 

165 ) 

166 out.append((cell.text, box)) 

167 return out 

168 

169 

170def _best_match( 

171 box: Optional[NormBox], parse_cells: List[Tuple[str, NormBox]], match_fraction: float 

172) -> Optional[str]: 

173 """What the parse read in the cell covering `box` best, or none past the floor.""" 

174 if box is None: 

175 return None 

176 best_text: Optional[str] = None 

177 best = match_fraction 

178 for text, other in parse_cells: 

179 cov = max(coverage_fraction(box, other), coverage_fraction(other, box)) 

180 if cov >= best: 

181 best, best_text = cov, text 

182 return best_text