Table Semantic Parser

Bases: BaseJob

End-to-end table semantic parsing pipeline for a document image.

This class detects tables/paragraphs, detects and recognizes text (OCR), detects table cells, and then builds a semantic representation of each table including: - grid tables (row/column structure) - key-value items inferred from adjacency heuristics - all cells with aggregated OCR content It also returns sorted paragraphs and global OCR words.

Attributes:

Name	Type	Description
layout_parser	LayoutParser	Detects tables and paragraph regions.
cell_detector	CellDetector	Extracts cell candidates inside tables.
text_detector	TextDetector	Detects word-level text regions.
text_recognizer	TextRecognizer	Recognizes text for detected regions.
visualize	bool	Whether to produce visualization images.
grid_only	bool	If True, skips clustering and attempts to parse full table as a grid.
merge_same_column_values	bool	Passed to grid parser to optionally merge values.

Notes

• The device argument is passed to submodules that support GPU/accelerator execution.
• The output is a TableSemanticParserSchema containing tables/paragraphs/words.

Source code in src/yomitoku/table_semantic_parser.py

class TableSemanticParser(BaseJob):
    """
    End-to-end table semantic parsing pipeline for a document image.

    This class detects tables/paragraphs, detects and recognizes text (OCR), detects
    table cells, and then builds a semantic representation of each table including:
    - grid tables (row/column structure)
    - key-value items inferred from adjacency heuristics
    - all cells with aggregated OCR content
    It also returns sorted paragraphs and global OCR words.

    Attributes:
        layout_parser (LayoutParser): Detects tables and paragraph regions.
        cell_detector (CellDetector): Extracts cell candidates inside tables.
        text_detector (TextDetector): Detects word-level text regions.
        text_recognizer (TextRecognizer): Recognizes text for detected regions.
        visualize (bool): Whether to produce visualization images.
        grid_only (bool): If True, skips clustering and attempts to parse full table as a grid.
        merge_same_column_values (bool): Passed to grid parser to optionally merge values.

    Notes:
        - The `device` argument is passed to submodules that support GPU/accelerator execution.
        - The output is a `TableSemanticParserSchema` containing tables/paragraphs/words.
    """

    def __init__(
        self,
        configs={},
        device="cuda:0",
        visualize=True,
        enable_preprocess=False,
        text_detector=None,
        text_recognizer=None,
        sync_count=False,
    ):
        """
        Initialize the table semantic parser and its submodules.

        Args:
            configs (dict): Configuration overrides for submodules. Expected keys (optional):
                - "table_detector": kwargs for LayoutParser
                - "table_parser": kwargs for CellDetector
                - "text_detector": kwargs for TextDetector
                - "text_recognizer": kwargs for TextRecognizer
            device (str): Device identifier passed to submodules (e.g., "cuda:0", "cpu").
            visualize (bool): If True, the parser will generate debug visualization images.
            sync_count (bool): Wait for usage-count transmission before returning inference results.

        Raises:
            ValueError: If `configs` is not a dict.
        """
        super().__init__(sync_count=sync_count)

        table_detector_kwargs = {
            "device": device,
            "visualize": visualize,
            # LayoutParser 全体の既定は rtdetrv2v3 だが、セマンティック
            # パーサーは public 版と同じ rtdetrv2v2 で揃える
            "model_name": "rtdetrv2v2",
        }
        table_cell_parser_kwargs = {
            "device": device,
            "visualize": visualize,
        }

        text_detector_kwargs = {
            "device": device,
        }

        text_recognizer_kwargs = {
            "device": device,
        }

        # table_structure_recognizer_kwargs = {
        #    "device": device,
        # }

        if isinstance(configs, dict):
            if "table_detector" in configs:
                table_detector_kwargs.update(configs["table_detector"])

            if "table_cell_parser" in configs:
                table_cell_parser_kwargs.update(configs["table_cell_parser"])

            if "text_detector" in configs:
                text_detector_kwargs.update(configs["text_detector"])

            if "text_recognizer" in configs:
                text_recognizer_kwargs.update(configs["text_recognizer"])

            # if "table_structure_recognizer" in configs:
            #    table_structure_recognizer_kwargs.update(
            #        configs["table_structure_recognizer"]
            #    )
        else:
            raise ValueError(
                "configs must be a dict. See the https://kotaro-kinoshita.github.io/yomitoku-dev/usage/"
            )

        self.layout_parser = LayoutParser(
            **table_detector_kwargs,
        )
        self.cell_detector = CellDetector(
            **table_cell_parser_kwargs,
        )

        self.text_detector = (
            text_detector
            if text_detector is not None
            else TextDetector(**text_detector_kwargs)
        )
        self.text_recognizer = (
            text_recognizer
            if text_recognizer is not None
            else TextRecognizer(**text_recognizer_kwargs)
        )

        # self.table_structure_recognizer = TableStructureRecognizer(
        #    **table_structure_recognizer_kwargs,
        # )

        self.enable_preprocess = enable_preprocess
        if self.enable_preprocess:
            preprocess_configs = {
                "rotate_detector": {
                    "device": device,
                },
            }
            if isinstance(configs, dict) and "preprocess" in configs:
                preprocess_configs.update(configs["preprocess"])
            self.preprocessor = Preprocessor(
                configs=preprocess_configs,
                sync_count=sync_count,
            )

        self.visualize = visualize

        self.merge_same_column_values = False

    def aggregate(self, ocr_res, cells, overlap_th=0.2):
        from collections import defaultdict

        cell_words = defaultdict(list)

        for word in ocr_res.words:
            word_box = quad_to_xyxy(word.points)
            best_cell = None
            best_ratio = 0

            for cell in cells:
                if cell.role == "group":
                    continue
                ratio, _ = calc_overlap_ratio(cell.box, word_box)
                if ratio > best_ratio:
                    best_ratio = ratio
                    best_cell = cell

            if best_cell is None or best_ratio < overlap_th:
                continue

            word_element = ParagraphSchema(
                box=word_box,
                contents=word.content,
                direction=word.direction,
                order=0,
                role=None,
            )
            # 段落要素は id が未採番(None)のまま渡されるため、cell.id を
            # キーにすると全要素が同一キーに集約され、全単語が全段落に
            # 割り当たってしまう。オブジェクト同一性をキーにする。
            cell_words[id(best_cell)].append(word_element)

        for cell in cells:
            contained = cell_words.get(id(cell), [])
            if not contained:
                cell.contents = ""
                continue

            dirs = [w.direction for w in contained]
            direction = (
                "horizontal"
                if dirs.count("horizontal") >= dirs.count("vertical")
                else "vertical"
            )
            order = "left2right" if direction == "horizontal" else "right2left"
            prediction_reading_order(contained, order)
            contained = sorted(contained, key=lambda x: x.order)
            text = "\n".join([w.contents for w in contained])
            cell.contents = text.replace("\n", "").strip()

    def replace_table_to_paragraphs(self, tables, paragraphs):
        new_table_list = []
        for table in tables:
            cnt_cell = 0
            for cell in table.cells:
                if cell.role in ["cell", "header"]:
                    cnt_cell += 1

            if cnt_cell < 2:
                paragraphs.append(
                    Element(
                        id=None,
                        box=table.box,
                        contents="",
                        score=1.0,
                        role=None,
                    )
                )
            else:
                new_table_list.append(table)

        return new_table_list

    def parse_detected_table(self, table, results_ocr, *, kv_only=False):
        """Build semantic structure for one already-detected table.

        Used by Studio's region analysis, where the user supplies the table
        box and the current (possibly edited) OCR words.
        """
        self.aggregate(results_ocr, table.cells)
        cells = {}
        for cell in table.cells:
            if not isinstance(cell, CellSchema):
                cell = CellSchema(
                    meta={},
                    id=cell.id,
                    box=cell.box,
                    role=cell.role,
                    row=cell.row,
                    col=cell.col,
                    row_span=cell.row_span,
                    col_span=cell.col_span,
                    contents=cell.contents,
                )
            cells[cell.id] = cell

        info = {
            "id": "t0",
            "box": table.box,
            "cells": {},
            "style": "border",
            "kv_items": [],
            "grids": [],
        }
        value_cells = [c for c in table.cells if c.role in ("cell", "header", "empty")]
        grid_regions = [] if kv_only else list(table.grid_regions)
        kv_regions = list(table.kv_regions)
        grid_regions, kv_regions = _resolve_overlapping_regions(
            grid_regions, kv_regions, value_cells
        )

        grid_claimed_ids = set()
        for region in grid_regions:
            region_cells = [
                c for c in value_cells if is_contained(region.box, c.box, threshold=0.5)
            ]
            if not region_cells:
                continue
            result = parse_grid_from_bottom_up(
                cells,
                _split_nodes_with_role(region_cells),
                self.merge_same_column_values,
            )
            if result is None:
                continue
            grid, grid_cells, _ = result
            info["grids"].append(grid)
            info["cells"].update(grid_cells)
            grid_claimed_ids.update(c.id for c in region_cells)

        remaining = [c for c in value_cells if c.id not in grid_claimed_ids]
        if remaining:
            for index, region in enumerate(kv_regions):
                region.id = f"kvr{index}"
            kv_items, _, kv_cells = parse_kv_items(
                _split_nodes_with_role(remaining), cells, kv_regions
            )
            info["kv_items"].extend(kv_items)
            info["cells"].update(kv_cells)

        for cell in cells.values():
            info["cells"].setdefault(cell.id, cell)
        info["kv_items"] = sorted(
            info["kv_items"], key=lambda item: info["cells"][item.value].box[1]
        )
        info["grids"] = sorted(info["grids"], key=lambda grid: grid.box[1])
        _assign_ids(info)
        return TableSemanticContentsSchema(**info)

    async def run_models(self, img):
        with ThreadPoolExecutor(max_workers=2) as executor:
            loop = asyncio.get_running_loop()
            tasks = [
                loop.run_in_executor(executor, self.text_detector, img),
                loop.run_in_executor(executor, self.layout_parser, img),
            ]

            results = await asyncio.gather(*tasks)

        results_det, _ = results[0]
        results_layout, _ = results[1]

        # borderless_table = [
        #    t.box for t in results_layout.tables if t.role == "borderless_table"
        # ]

        # bordered_table = [
        #    t for t in results_layout.tables if t.role != "borderless_table"
        # ]

        bordered_table = [t for t in results_layout.tables]

        results_table = self.cell_detector(img, bordered_table)
        # results_borderless_table, _ = self.table_structure_recognizer(
        #    img, borderless_table
        # )

        results_table = self.replace_table_to_paragraphs(
            results_table, results_layout.paragraphs
        )

        # word_dicts = [
        #    {"poly": quad_to_poly(q), "score": s}
        #    for q, s in zip(results_det.points, results_det.scores)
        # ]

        # cell_dicts = []
        # for table in results_table:
        #    for c in table.cells:
        #        if c.role in ["group", "empty"]:
        #            continue
        #        cell_dicts.append({"id": str(c.id), "poly": box_to_poly(c.box)})

        # セルにまたがるテキスト領域の分割
        # split_words = replace_spanning_words_with_clipped_polys_poly(
        #    words=word_dicts,
        #    cells=cell_dicts,
        #    min_area_ratio=0.05,
        #    keep_unsplit=True,
        # )

        # schema_dict = build_text_detector_schema_from_split_words_rotated_quad(
        #    split_words, cell_dicts, use_cell=False
        # )

        # results_det = TextDetectorSchema(**word_dicts)

        results_rec = self.text_recognizer(img, results_det.points)
        outputs = {"words": ocr_aggregate(results_rec)}
        results_ocr = OCRSchema(**outputs)

        return (results_ocr, results_table, results_layout.paragraphs)

    def visualizer_ocr(self, img, semantic_info):
        vis_ocr = _ocr_visualizer(
            img,
            semantic_info,
            font_size=self.text_recognizer._cfg.visualize.font_size,
            font_color=tuple(self.text_recognizer._cfg.visualize.color[::-1]),
            font_path=self.text_recognizer._cfg.visualize.font,
        )

        return vis_ocr

    def visualizer_layout(self, img, semantic_info):
        vis_layout = img.copy()

        vis_layout = _layout_visualizer(
            semantic_info.tables,
            vis_layout,
            prefix="Table",
        )

        vis_layout = _layout_visualizer(
            semantic_info.paragraphs,
            vis_layout,
            prefix="Paragraph",
        )

        for results_table in semantic_info.tables:
            vis_layout, _ = cell_detector_visualizer(
                vis_layout,
                vis_layout,
                results_table.cells.values(),
            )

            # kv_items のキー連鎖を緑の矢印で描画する (救済リンク含む)
            vis_layout = kv_items_visualizer(results_table, vis_layout)

            for grid in results_table.grids:
                box = grid.box
                cv2.rectangle(
                    vis_layout,
                    (box[0], box[1]),
                    (box[2], box[3]),
                    (255, 0, 0),
                    3,
                )

        return vis_layout

    def __call__(self, img, template=None, id=None, grid_only=False, kv_only=False):
        """
        Parse an input document image and return table semantics + visualizations.

        Steps:
        1) Run layout detection, text detection, cell detection, OCR recognition.
        2) Aggregate OCR results into each cell and paragraph (`contents` field).
        3) For each table:
            - resolve overlaps between model-predicted grid / kv_item regions
            - parse the row/column structure inside each grid region
            - parse key-value items from the remaining cells using kv regions
              (with fallback rescues for cells/headers outside the regions)
            - ensure all cells are included in the output
            - sort kv/grids and normalize IDs
        4) Sort tables and paragraphs and wrap into `TableSemanticParserSchema`.
        5) Optionally load a template JSON to align output with a predefined structure.
        6) Optionally generate visualization images.

        Args:
            img (np.ndarray): Input image (OpenCV ndarray).
            template (Optional[dict|str]): Template definition loaded into the schema.
                If provided, `semantic_info.load_template_json(template)` is called.
            id (Optional[str]): Reserved for future use (currently not used).
            grid_only (bool): If True, ignore kv_item regions and parse grids only.
            kv_only (bool): If True, ignore grid regions and parse key-values only.

        Returns:
            Tuple[TableSemanticParserSchema, np.ndarray, np.ndarray]:
                - semantic_info: tables/paragraphs/words semantic structure
                - vis_layout: visualization image for layout/cells/grids/kv links
                - vis_ocr: visualization image for OCR polygons/text

        Notes:
            - If `self.visualize` is False, `vis_layout` and `vis_ocr` are still returned
            as copies of the input image but without overlays.
            - This method uses `asyncio.run(...)` internally; calling it from an already
            running event loop (e.g., inside async frameworks) may require refactoring
            (e.g., exposing an async entrypoint).
        """

        if self.enable_preprocess:
            _, img = self.preprocessor(img)

        try:
            results_ocr, results_table, paragraphs = asyncio.run(self.run_models(img))
        except torch.cuda.OutOfMemoryError as e:
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
            logger.error("GPU out of memory in TableSemanticParser: %s", e)
            raise make_error(ErrorCode.GPU_OUT_OF_MEMORY) from e

        semantic_info = []
        for table in results_table:
            self.aggregate(results_ocr, table.cells)

        self.aggregate(results_ocr, paragraphs)

        vis_layout = img.copy()
        vis_ocr = img.copy()

        # テーブル構造のグラフ (DAG) を最後にまとめて上描きするため保持する
        dags = []

        for i, table in enumerate(results_table):
            cells = {}
            for cell in table.cells:
                if isinstance(cell, TableCellSchema):
                    cell = CellSchema(
                        meta={},
                        id=cell.id,
                        box=cell.box,
                        role=cell.role,
                        row=cell.row,
                        col=cell.col,
                        row_span=cell.row_span,
                        col_span=cell.col_span,
                        contents=cell.contents,
                    )

                cells[cell.id] = cell

            table_information = {
                "id": f"t{i}",
                "box": table.box,
                "cells": {},
                "style": "border",
                "kv_items": [],
                "grids": [],
            }
            if template is None:
                # モデルが直接予測した grid / kv_item 領域を利用する
                value_cells = [
                    c for c in table.cells if c.role in ("cell", "header", "empty")
                ]

                grid_regions = list(table.grid_regions)
                kv_regions = list(table.kv_regions)

                if grid_only:
                    kv_regions = []
                if kv_only:
                    grid_regions = []

                # kv_item と grid が重複検知された場合の優先度をルールベースで解決
                grid_regions, kv_regions = _resolve_overlapping_regions(
                    grid_regions,
                    kv_regions,
                    value_cells,
                )

                # --- grid 領域ごとに内部構造（行列）を推定 ---
                grid_claimed_ids = set()
                for region in grid_regions:
                    region_cells = [
                        c
                        for c in value_cells
                        if is_contained(region.box, c.box, threshold=0.5)
                    ]
                    if len(region_cells) == 0:
                        continue

                    clustered_nodes = _split_nodes_with_role(region_cells)
                    result = parse_grid_from_bottom_up(
                        cells,
                        clustered_nodes,
                        self.merge_same_column_values,
                    )

                    if result is None:
                        continue

                    grid, grid_cells, dag = result

                    table_information["grids"].append(grid)
                    table_information["cells"].update(grid_cells)
                    grid_claimed_ids.update(c.id for c in region_cells)

                    dags.append(dag)

                # --- grid に属さないセルを kv として推定 ---
                remaining_cells = [
                    c for c in value_cells if c.id not in grid_claimed_ids
                ]
                if remaining_cells:
                    nodes = _split_nodes_with_role(remaining_cells)

                    # モデル予測の kv_item 領域内のセルで隣接グラフを構築する
                    for ki, region in enumerate(kv_regions):
                        region.id = f"kvr{ki}"

                    kv_items, dag, kv_cells = parse_kv_items(
                        nodes,
                        cells,
                        kv_regions,
                    )

                    table_information["kv_items"].extend(kv_items)
                    table_information["cells"].update(kv_cells)

                    # NOTE: kv側はDAGではなく確定した kv_items のキー連鎖を
                    # kv_items_visualizer (緑矢印) で描画するため、DAGの
                    # 上描き対象には追加しない (dagsはgrid構造の可視化用)

            for cell in cells.values():
                if cell.id not in table_information["cells"]:
                    table_information["cells"][cell.id] = cell

            table_information["kv_items"] = sorted(
                table_information["kv_items"],
                key=lambda kv: table_information["cells"][kv.value].box[1],
            )

            table_information["grids"] = sorted(
                table_information["grids"],
                key=lambda g: g.box[1],
            )

            for i, grid in enumerate(table_information["grids"]):
                grid.id = f"g{i}"

            for i, kv in enumerate(table_information["kv_items"]):
                kv.id = f"kv{i}"

            _assign_ids(table_information)

            semantic_info.append(TableSemanticContentsSchema(**table_information))

        semantic_info = _sort_elements(semantic_info, prefix="t")
        paragraphs = _sort_elements(paragraphs, prefix="p")

        semantic_info = TableSemanticParserSchema(
            tables=semantic_info,
            paragraphs=paragraphs,
            words=results_ocr.words,
        )

        if template is not None:
            semantic_info.load_template_json(template)

        if self.visualize:
            vis_layout = self.visualizer_layout(vis_layout, semantic_info)
            vis_ocr = self.visualizer_ocr(vis_ocr, semantic_info)

            # テーブル構造のグラフ (DAG) をセル塗り・枠の上に描画して見やすくする
            for dag in dags:
                vis_layout = dag_visualizer(dag, vis_layout)

        return semantic_info, vis_layout, vis_ocr

__call__(img, template=None, id=None, grid_only=False, kv_only=False)

Parse an input document image and return table semantics + visualizations.

Steps: 1) Run layout detection, text detection, cell detection, OCR recognition. 2) Aggregate OCR results into each cell and paragraph (contents field). 3) For each table: - resolve overlaps between model-predicted grid / kv_item regions - parse the row/column structure inside each grid region - parse key-value items from the remaining cells using kv regions (with fallback rescues for cells/headers outside the regions) - ensure all cells are included in the output - sort kv/grids and normalize IDs 4) Sort tables and paragraphs and wrap into TableSemanticParserSchema. 5) Optionally load a template JSON to align output with a predefined structure. 6) Optionally generate visualization images.

Parameters:

Name	Type	Description	Default
img	ndarray	Input image (OpenCV ndarray).	required
template	Optional[dict | str]	Template definition loaded into the schema. If provided, semantic_info.load_template_json(template) is called.	None
id	Optional[str]	Reserved for future use (currently not used).	None
grid_only	bool	If True, ignore kv_item regions and parse grids only.	False
kv_only	bool	If True, ignore grid regions and parse key-values only.	False

Returns:

Type	Description
	Tuple[TableSemanticParserSchema, np.ndarray, np.ndarray]: - semantic_info: tables/paragraphs/words semantic structure - vis_layout: visualization image for layout/cells/grids/kv links - vis_ocr: visualization image for OCR polygons/text

Notes

• If self.visualize is False, vis_layout and vis_ocr are still returned as copies of the input image but without overlays.
• This method uses asyncio.run(...) internally; calling it from an already running event loop (e.g., inside async frameworks) may require refactoring (e.g., exposing an async entrypoint).

Source code in src/yomitoku/table_semantic_parser.py

def __call__(self, img, template=None, id=None, grid_only=False, kv_only=False):
    """
    Parse an input document image and return table semantics + visualizations.

    Steps:
    1) Run layout detection, text detection, cell detection, OCR recognition.
    2) Aggregate OCR results into each cell and paragraph (`contents` field).
    3) For each table:
        - resolve overlaps between model-predicted grid / kv_item regions
        - parse the row/column structure inside each grid region
        - parse key-value items from the remaining cells using kv regions
          (with fallback rescues for cells/headers outside the regions)
        - ensure all cells are included in the output
        - sort kv/grids and normalize IDs
    4) Sort tables and paragraphs and wrap into `TableSemanticParserSchema`.
    5) Optionally load a template JSON to align output with a predefined structure.
    6) Optionally generate visualization images.

    Args:
        img (np.ndarray): Input image (OpenCV ndarray).
        template (Optional[dict|str]): Template definition loaded into the schema.
            If provided, `semantic_info.load_template_json(template)` is called.
        id (Optional[str]): Reserved for future use (currently not used).
        grid_only (bool): If True, ignore kv_item regions and parse grids only.
        kv_only (bool): If True, ignore grid regions and parse key-values only.

    Returns:
        Tuple[TableSemanticParserSchema, np.ndarray, np.ndarray]:
            - semantic_info: tables/paragraphs/words semantic structure
            - vis_layout: visualization image for layout/cells/grids/kv links
            - vis_ocr: visualization image for OCR polygons/text

    Notes:
        - If `self.visualize` is False, `vis_layout` and `vis_ocr` are still returned
        as copies of the input image but without overlays.
        - This method uses `asyncio.run(...)` internally; calling it from an already
        running event loop (e.g., inside async frameworks) may require refactoring
        (e.g., exposing an async entrypoint).
    """

    if self.enable_preprocess:
        _, img = self.preprocessor(img)

    try:
        results_ocr, results_table, paragraphs = asyncio.run(self.run_models(img))
    except torch.cuda.OutOfMemoryError as e:
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
        logger.error("GPU out of memory in TableSemanticParser: %s", e)
        raise make_error(ErrorCode.GPU_OUT_OF_MEMORY) from e

    semantic_info = []
    for table in results_table:
        self.aggregate(results_ocr, table.cells)

    self.aggregate(results_ocr, paragraphs)

    vis_layout = img.copy()
    vis_ocr = img.copy()

    # テーブル構造のグラフ (DAG) を最後にまとめて上描きするため保持する
    dags = []

    for i, table in enumerate(results_table):
        cells = {}
        for cell in table.cells:
            if isinstance(cell, TableCellSchema):
                cell = CellSchema(
                    meta={},
                    id=cell.id,
                    box=cell.box,
                    role=cell.role,
                    row=cell.row,
                    col=cell.col,
                    row_span=cell.row_span,
                    col_span=cell.col_span,
                    contents=cell.contents,
                )

            cells[cell.id] = cell

        table_information = {
            "id": f"t{i}",
            "box": table.box,
            "cells": {},
            "style": "border",
            "kv_items": [],
            "grids": [],
        }
        if template is None:
            # モデルが直接予測した grid / kv_item 領域を利用する
            value_cells = [
                c for c in table.cells if c.role in ("cell", "header", "empty")
            ]

            grid_regions = list(table.grid_regions)
            kv_regions = list(table.kv_regions)

            if grid_only:
                kv_regions = []
            if kv_only:
                grid_regions = []

            # kv_item と grid が重複検知された場合の優先度をルールベースで解決
            grid_regions, kv_regions = _resolve_overlapping_regions(
                grid_regions,
                kv_regions,
                value_cells,
            )

            # --- grid 領域ごとに内部構造（行列）を推定 ---
            grid_claimed_ids = set()
            for region in grid_regions:
                region_cells = [
                    c
                    for c in value_cells
                    if is_contained(region.box, c.box, threshold=0.5)
                ]
                if len(region_cells) == 0:
                    continue

                clustered_nodes = _split_nodes_with_role(region_cells)
                result = parse_grid_from_bottom_up(
                    cells,
                    clustered_nodes,
                    self.merge_same_column_values,
                )

                if result is None:
                    continue

                grid, grid_cells, dag = result

                table_information["grids"].append(grid)
                table_information["cells"].update(grid_cells)
                grid_claimed_ids.update(c.id for c in region_cells)

                dags.append(dag)

            # --- grid に属さないセルを kv として推定 ---
            remaining_cells = [
                c for c in value_cells if c.id not in grid_claimed_ids
            ]
            if remaining_cells:
                nodes = _split_nodes_with_role(remaining_cells)

                # モデル予測の kv_item 領域内のセルで隣接グラフを構築する
                for ki, region in enumerate(kv_regions):
                    region.id = f"kvr{ki}"

                kv_items, dag, kv_cells = parse_kv_items(
                    nodes,
                    cells,
                    kv_regions,
                )

                table_information["kv_items"].extend(kv_items)
                table_information["cells"].update(kv_cells)

                # NOTE: kv側はDAGではなく確定した kv_items のキー連鎖を
                # kv_items_visualizer (緑矢印) で描画するため、DAGの
                # 上描き対象には追加しない (dagsはgrid構造の可視化用)

        for cell in cells.values():
            if cell.id not in table_information["cells"]:
                table_information["cells"][cell.id] = cell

        table_information["kv_items"] = sorted(
            table_information["kv_items"],
            key=lambda kv: table_information["cells"][kv.value].box[1],
        )

        table_information["grids"] = sorted(
            table_information["grids"],
            key=lambda g: g.box[1],
        )

        for i, grid in enumerate(table_information["grids"]):
            grid.id = f"g{i}"

        for i, kv in enumerate(table_information["kv_items"]):
            kv.id = f"kv{i}"

        _assign_ids(table_information)

        semantic_info.append(TableSemanticContentsSchema(**table_information))

    semantic_info = _sort_elements(semantic_info, prefix="t")
    paragraphs = _sort_elements(paragraphs, prefix="p")

    semantic_info = TableSemanticParserSchema(
        tables=semantic_info,
        paragraphs=paragraphs,
        words=results_ocr.words,
    )

    if template is not None:
        semantic_info.load_template_json(template)

    if self.visualize:
        vis_layout = self.visualizer_layout(vis_layout, semantic_info)
        vis_ocr = self.visualizer_ocr(vis_ocr, semantic_info)

        # テーブル構造のグラフ (DAG) をセル塗り・枠の上に描画して見やすくする
        for dag in dags:
            vis_layout = dag_visualizer(dag, vis_layout)

    return semantic_info, vis_layout, vis_ocr

__init__(configs={}, device='cuda:0', visualize=True, enable_preprocess=False, text_detector=None, text_recognizer=None, sync_count=False)

Initialize the table semantic parser and its submodules.

Parameters:

Name	Type	Description	Default
configs	dict	Configuration overrides for submodules. Expected keys (optional): - "table_detector": kwargs for LayoutParser - "table_parser": kwargs for CellDetector - "text_detector": kwargs for TextDetector - "text_recognizer": kwargs for TextRecognizer	{}
device	str	Device identifier passed to submodules (e.g., "cuda:0", "cpu").	'cuda:0'
visualize	bool	If True, the parser will generate debug visualization images.	True
sync_count	bool	Wait for usage-count transmission before returning inference results.	False

Raises:

Type	Description
ValueError	If configs is not a dict.

Source code in src/yomitoku/table_semantic_parser.py

def __init__(
    self,
    configs={},
    device="cuda:0",
    visualize=True,
    enable_preprocess=False,
    text_detector=None,
    text_recognizer=None,
    sync_count=False,
):
    """
    Initialize the table semantic parser and its submodules.

    Args:
        configs (dict): Configuration overrides for submodules. Expected keys (optional):
            - "table_detector": kwargs for LayoutParser
            - "table_parser": kwargs for CellDetector
            - "text_detector": kwargs for TextDetector
            - "text_recognizer": kwargs for TextRecognizer
        device (str): Device identifier passed to submodules (e.g., "cuda:0", "cpu").
        visualize (bool): If True, the parser will generate debug visualization images.
        sync_count (bool): Wait for usage-count transmission before returning inference results.

    Raises:
        ValueError: If `configs` is not a dict.
    """
    super().__init__(sync_count=sync_count)

    table_detector_kwargs = {
        "device": device,
        "visualize": visualize,
        # LayoutParser 全体の既定は rtdetrv2v3 だが、セマンティック
        # パーサーは public 版と同じ rtdetrv2v2 で揃える
        "model_name": "rtdetrv2v2",
    }
    table_cell_parser_kwargs = {
        "device": device,
        "visualize": visualize,
    }

    text_detector_kwargs = {
        "device": device,
    }

    text_recognizer_kwargs = {
        "device": device,
    }

    # table_structure_recognizer_kwargs = {
    #    "device": device,
    # }

    if isinstance(configs, dict):
        if "table_detector" in configs:
            table_detector_kwargs.update(configs["table_detector"])

        if "table_cell_parser" in configs:
            table_cell_parser_kwargs.update(configs["table_cell_parser"])

        if "text_detector" in configs:
            text_detector_kwargs.update(configs["text_detector"])

        if "text_recognizer" in configs:
            text_recognizer_kwargs.update(configs["text_recognizer"])

        # if "table_structure_recognizer" in configs:
        #    table_structure_recognizer_kwargs.update(
        #        configs["table_structure_recognizer"]
        #    )
    else:
        raise ValueError(
            "configs must be a dict. See the https://kotaro-kinoshita.github.io/yomitoku-dev/usage/"
        )

    self.layout_parser = LayoutParser(
        **table_detector_kwargs,
    )
    self.cell_detector = CellDetector(
        **table_cell_parser_kwargs,
    )

    self.text_detector = (
        text_detector
        if text_detector is not None
        else TextDetector(**text_detector_kwargs)
    )
    self.text_recognizer = (
        text_recognizer
        if text_recognizer is not None
        else TextRecognizer(**text_recognizer_kwargs)
    )

    # self.table_structure_recognizer = TableStructureRecognizer(
    #    **table_structure_recognizer_kwargs,
    # )

    self.enable_preprocess = enable_preprocess
    if self.enable_preprocess:
        preprocess_configs = {
            "rotate_detector": {
                "device": device,
            },
        }
        if isinstance(configs, dict) and "preprocess" in configs:
            preprocess_configs.update(configs["preprocess"])
        self.preprocessor = Preprocessor(
            configs=preprocess_configs,
            sync_count=sync_count,
        )

    self.visualize = visualize

    self.merge_same_column_values = False

parse_detected_table(table, results_ocr, *, kv_only=False)

Build semantic structure for one already-detected table.

Used by Studio's region analysis, where the user supplies the table box and the current (possibly edited) OCR words.

Source code in src/yomitoku/table_semantic_parser.py

def parse_detected_table(self, table, results_ocr, *, kv_only=False):
    """Build semantic structure for one already-detected table.

    Used by Studio's region analysis, where the user supplies the table
    box and the current (possibly edited) OCR words.
    """
    self.aggregate(results_ocr, table.cells)
    cells = {}
    for cell in table.cells:
        if not isinstance(cell, CellSchema):
            cell = CellSchema(
                meta={},
                id=cell.id,
                box=cell.box,
                role=cell.role,
                row=cell.row,
                col=cell.col,
                row_span=cell.row_span,
                col_span=cell.col_span,
                contents=cell.contents,
            )
        cells[cell.id] = cell

    info = {
        "id": "t0",
        "box": table.box,
        "cells": {},
        "style": "border",
        "kv_items": [],
        "grids": [],
    }
    value_cells = [c for c in table.cells if c.role in ("cell", "header", "empty")]
    grid_regions = [] if kv_only else list(table.grid_regions)
    kv_regions = list(table.kv_regions)
    grid_regions, kv_regions = _resolve_overlapping_regions(
        grid_regions, kv_regions, value_cells
    )

    grid_claimed_ids = set()
    for region in grid_regions:
        region_cells = [
            c for c in value_cells if is_contained(region.box, c.box, threshold=0.5)
        ]
        if not region_cells:
            continue
        result = parse_grid_from_bottom_up(
            cells,
            _split_nodes_with_role(region_cells),
            self.merge_same_column_values,
        )
        if result is None:
            continue
        grid, grid_cells, _ = result
        info["grids"].append(grid)
        info["cells"].update(grid_cells)
        grid_claimed_ids.update(c.id for c in region_cells)

    remaining = [c for c in value_cells if c.id not in grid_claimed_ids]
    if remaining:
        for index, region in enumerate(kv_regions):
            region.id = f"kvr{index}"
        kv_items, _, kv_cells = parse_kv_items(
            _split_nodes_with_role(remaining), cells, kv_regions
        )
        info["kv_items"].extend(kv_items)
        info["cells"].update(kv_cells)

    for cell in cells.values():
        info["cells"].setdefault(cell.id, cell)
    info["kv_items"] = sorted(
        info["kv_items"], key=lambda item: info["cells"][item.value].box[1]
    )
    info["grids"] = sorted(info["grids"], key=lambda grid: grid.box[1])
    _assign_ids(info)
    return TableSemanticContentsSchema(**info)