Coverage for src / quber / core / extractors / camelot / correspondence / recovery.py: 31%

1""" 

2Guarded recovery for a detected table the full-page Camelot passes 

3missed: one targeted, region-constrained stream pass aimed at a box, 

4plus the pairing helper that picks a grid-locator region when the 

5detector's own box drifted. The recovery can only upgrade a miss or 

6no-op — never worse than reporting it. 

7""" 

8 

9from __future__ import annotations 

10 

11import warnings 

12from pathlib import Path 

13from typing import Any, List, Optional, Tuple 

14 

15from quber.agents.grid_locator import LocatedTable 

16from quber.core.extractors.camelot.acquire import ( 

17 CamelotCandidate, 

18 cells_to_boxes, 

19 df_to_cells, 

20 grid_to_markdown, 

21) 

22from quber.core.extractors.camelot.tighten import tighten_cell_boxes 

23 

24 

25def init_recovery_warning_filter() -> None: 

26 """Silence camelot's expected 'No tables found in table area' miss. 

27 

28 The region-constrained recovery pass deliberately aims a stream parse at a 

29 detector box that may hold no grid; that miss is reported separately as a 

30 detected-not-extracted table, so the camelot warning is pure noise. A 

31 persistent process-global filter is used (not a per-call context manager) 

32 because the recovery runs across worker threads, where a context manager's 

33 save/restore races and lets warnings leak through. 

34 """ 

35 warnings.filterwarnings("ignore", message="No tables found in table area", category=UserWarning) 

36 

37 

38init_recovery_warning_filter() 

39 

40 

41def camelot_targeted( 

42 source_str: str, 

43 page: int, 

44 area: str, 

45 ordinal: int, 

46 flavor: str = "stream", 

47 row_tol: Optional[int] = None, 

48 column_tol: Optional[int] = None, 

49) -> Optional[CamelotCandidate]: 

50 """Run a single region-constrained Camelot pass. Returns the 

51 first table found in the region, or None if Camelot found none. Imports 

52 camelot lazily (stream needs no OpenCV) and runs in-process; intended 

53 to be called via asyncio.to_thread. Raises on Camelot-internal errors 

54 (e.g. an empty region) — the caller guards and reports the table as 

55 detected_not_extracted. 

56 

57 `row_tol`/`column_tol` override stream's grouping tolerances for a 

58 capture-repair retry (e.g. a totals row typeset on a raised baseline 

59 splits into two bands at the default tolerance and its numbers are 

60 dropped at the band boundary). They apply to the stream flavor only. 

61 """ 

62 import camelot 

63 

64 from quber.agents.completeness import page_words 

65 

66 kwargs: dict[str, Any] = {} 

67 if flavor == "stream": 

68 if row_tol is not None: 

69 kwargs["row_tol"] = row_tol 

70 if column_tol is not None: 

71 kwargs["column_tol"] = column_tol 

72 tables = camelot.read_pdf( # pyright: ignore[reportPrivateImportUsage,reportArgumentType] 

73 source_str, pages=str(page), flavor=flavor, table_areas=[area], **kwargs 

74 ) 

75 if not tables: 

76 return None 

77 t = tables[0] 

78 # camelot exposes the table box only as the private `_bbox`; read it via 

79 # getattr so the access is not flagged as private use. 

80 raw_bbox = getattr(t, "_bbox", None) 

81 bbox = tuple(raw_bbox) if raw_bbox else None 

82 report = getattr(t, "parsing_report", {}) or {} 

83 cells = df_to_cells(t.df) 

84 cell_boxes = cells_to_boxes(getattr(t, "cells", None), cells) 

85 _, page_h, words = page_words(Path(source_str), page) 

86 cell_boxes = tighten_cell_boxes(cells, cell_boxes, words, page_h) 

87 return CamelotCandidate( 

88 candidate_id=f"recovery-p{page}-o{ordinal}", 

89 flavor="stream" if flavor == "stream" else "lattice", 

90 page=page, 

91 bbox=bbox, # type: ignore[arg-type] 

92 accuracy=float(report.get("accuracy", 0.0) or 0.0), 

93 cells=cells, 

94 cell_boxes=cell_boxes, 

95 markdown=grid_to_markdown(cells), 

96 ) 

97 

98 

99def nearest_grid_region( 

100 located: List[LocatedTable], 

101 consumed: set[int], 

102 detector_bbox: Optional[Tuple[float, float, float, float]], 

103) -> Optional[int]: 

104 """Pick the unconsumed grid-located region best matching a detected table. 

105 

106 The detector's box drifts, but its vertical neighbourhood is roughly right 

107 even when its extent is wrong, so we pair by nearest region center-y. When 

108 the detector gave no box, fall back to reading order (first unconsumed). 

109 Returns an index into `located`, or None if all regions are consumed. 

110 """ 

111 candidates = [i for i in range(len(located)) if i not in consumed] 

112 if not candidates: 

113 return None 

114 if detector_bbox is None: 

115 return candidates[0] 

116 center = (min(detector_bbox[1], detector_bbox[3]) + max(detector_bbox[1], detector_bbox[3])) / 2.0 

117 return min(candidates, key=lambda i: abs((located[i].region[1] + located[i].region[3]) / 2.0 - center))