Coverage for src / quber / core / extractors / dual / orchestrator.py: 62%

1"""Run the visual and Camelot flows side by side for one document. 

2 

3Both flows run concurrently and each returns its own result; the extractor 

4hands back a `DualResult` carrying both, unmatched. There is no detector, 

5no box repair, no targeted recovery, and no grid escalation — the two 

6flows are independent first-class inputs, to be cross-referenced later. 

7""" 

8 

9from __future__ import annotations 

10 

11import asyncio 

12from pathlib import Path 

13from typing import Optional 

14 

15from quber.agents.classifier import TableClassifier, get_classifier 

16from quber.agents.grid_locator import GridLocator, get_grid_locator 

17from quber.core.extractors.dual.camelot import run_camelot_flow 

18from quber.core.extractors.dual.models import DualResult 

19from quber.core.extractors.dual.vision import run_vision_flow 

20 

21 

22class DualFlowExtractor: 

23 """Produce a document's visual tables and Camelot grids as parallel flows. 

24 

25 `locator` and `classifier` default to the configured API backends but 

26 can be injected (e.g. mocks in tests). `classify_camelot` controls only 

27 whether the Camelot grids get an is-table verdict attached; it never 

28 changes which grids are returned. 

29 """ 

30 

31 def __init__( 

32 self, 

33 locator: Optional[GridLocator] = None, 

34 classifier: Optional[TableClassifier] = None, 

35 dpi: int = 200, 

36 classify_camelot: bool = True, 

37 ) -> None: 

38 self.locator = locator or get_grid_locator() 

39 self.classifier = classifier or get_classifier() 

40 self.dpi = dpi 

41 self.classify_camelot = classify_camelot 

42 

43 async def run(self, source: Path) -> DualResult: 

44 classifier = self.classifier if self.classify_camelot else None 

45 visual, camelot = await asyncio.gather( 

46 run_vision_flow(source, self.locator, dpi=self.dpi), 

47 run_camelot_flow(source, classifier), 

48 ) 

49 return DualResult(source=str(source), visual=visual, camelot=camelot) 

50 

51 def run_sync(self, source: Path) -> DualResult: 

52 return asyncio.run(self.run(source))