""" Docling & SOTA Layout Analysis Engine for Math OCR Pipeline V2.0. Implements Universal Recursive XY-Cut++ with Column-Aware Whitespace Analysis, Adaptive Row-Banding, and Block-Container Integrity. Guarantees clean reading order across any multi-column, table, or mixed-layout PDF. """ from typing import List, Dict, Any, Tuple import numpy as np def normalize_item_coords(item: Any, page_width: float, page_height: float) -> Dict[str, Any]: """ Normalizes coordinates from OCRBlock, dict, tuple or list into absolute PDF points. """ text = "" block_type = "text" is_gemini_norm = False coords = None if hasattr(item, 'box_2d'): coords = item.box_2d text = getattr(item, 'text_content', getattr(item, 'text', '')) block_type = getattr(item, 'block_type', 'text') is_gemini_norm = True elif isinstance(item, dict): if 'box_2d' in item: coords = item['box_2d'] is_gemini_norm = True else: coords = item.get('coords') or item.get('bbox') is_gemini_norm = False text = item.get('text_content') or item.get('text', '') block_type = item.get('block_type', 'text') elif isinstance(item, (list, tuple)) and len(item) >= 2: raw_coords = item[0] data = item[1] if isinstance(data, dict): text = data.get('text_content') or data.get('text', '') block_type = data.get('block_type', 'text') if 'box_2d' in data: coords = data['box_2d'] is_gemini_norm = True else: coords = raw_coords else: text = getattr(data, 'text_content', getattr(data, 'text', (data[1] if isinstance(data, tuple) else str(data)))) if hasattr(data, 'block_type'): block_type = data.block_type if hasattr(data, 'box_2d'): coords = data.box_2d is_gemini_norm = True else: coords = raw_coords if not is_gemini_norm: # If all 4 values are integers in [0, 1000], it is normalized [ymin, xmin, ymax, xmax] is_gemini_norm = (isinstance(coords, (list, tuple)) and len(coords) == 4 and all(isinstance(v, int) for v in coords) and max(coords) <= 1000 and min(coords) >= 0) else: return None if not text or not str(text).strip(): return None if not isinstance(coords, (list, tuple)) or len(coords) != 4: return None if is_gemini_norm: ymin, xmin, ymax, xmax = coords x0 = (xmin / 1000.0) * page_width y0 = (ymin / 1000.0) * page_height x1 = (xmax / 1000.0) * page_width y1 = (ymax / 1000.0) * page_height else: # Standard PDF points in [x0, y0, x1, y1] x0, y0, x1, y1 = coords w = max(x1 - x0, 2.0) h = max(y1 - y0, 2.0) return { 'x0': x0, 'y0': y0, 'x1': x1, 'y1': y1, 'w': w, 'h': h, 'x_mid': (x0 + x1) / 2.0, 'y_mid': (y0 + y1) / 2.0, 'text': str(text).strip(), 'block_type': block_type, 'raw': item, 'coords': (x0, y0, x1, y1) } def decompose_atomic_lines(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Decomposes multi-line boxes into single-line atomic spans with proportional heights. Preserves mathematical equations and table blocks intact to prevent interleaving. """ atomic = [] for it in items: raw_text = it['text'] # Never break formulas or tables into separate disjoint spans if it.get('block_type') in ['math', 'table'] or any(sym in raw_text for sym in ['∫', '∑', '∏', '|---', '\noder\n', '= ']): atomic.append(it) continue lines = [l.strip() for l in raw_text.split("\n") if l.strip()] if len(lines) <= 1: atomic.append(it) else: line_h = it['h'] / len(lines) for idx, line in enumerate(lines): ly0 = it['y0'] + idx * line_h ly1 = ly0 + line_h sub_item = dict(it) sub_item['y0'] = ly0 sub_item['y1'] = ly1 sub_item['h'] = line_h sub_item['y_mid'] = (ly0 + ly1) / 2.0 sub_item['text'] = line sub_item['coords'] = (it['x0'], ly0, it['x1'], ly1) atomic.append(sub_item) return atomic def merge_equation_lhs_and_rhs(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Pre-merges equation left-hand side conditions with the equation itself. If an item starts with '=' (or '= ∫', '= ∑', etc.) and one or more items exist immediately to its left sharing vertical overlap (e.g. '1,0 kN · δ_(ik)', 'oder', '1,0 kNm · φ_(ik)'), they are merged into ONE unified equation item in left-to-right order. """ if not items: return [] consumed = set() merged = [] for i, it in enumerate(items): if i in consumed: continue txt = it['text'].strip() if txt.startswith('=') or txt.startswith('=\n'): lhs_candidates = [] for j, other in enumerate(items): if j == i or j in consumed: continue # Is other to the left and within 250 pt? if other['x1'] <= it['x0'] + 20 and other['x0'] >= it['x0'] - 250: v_overlap = min(it['y1'], other['y1']) - max(it['y0'], other['y0']) if v_overlap > 0: lhs_candidates.append((j, other)) if lhs_candidates: lhs_candidates.sort(key=lambda c: (c[1]['y0'], c[1]['x0'])) combined_lhs_text = " ".join(c[1]['text'].strip() for c in lhs_candidates) new_text = f"({combined_lhs_text}) {txt}" all_boxes = [c[1] for c in lhs_candidates] + [it] new_x0 = min(b['x0'] for b in all_boxes) new_y0 = min(b['y0'] for b in all_boxes) new_x1 = max(b['x1'] for b in all_boxes) new_y1 = max(b['y1'] for b in all_boxes) merged_item = dict(it) merged_item['text'] = new_text merged_item['x0'] = new_x0 merged_item['y0'] = new_y0 merged_item['x1'] = new_x1 merged_item['y1'] = new_y1 merged_item['w'] = new_x1 - new_x0 merged_item['h'] = new_y1 - new_y0 merged_item['x_mid'] = (new_x0 + new_x1) / 2.0 merged_item['y_mid'] = (new_y0 + new_y1) / 2.0 merged_item['coords'] = (new_x0, new_y0, new_x1, new_y1) merged_item['block_type'] = 'math' for j, _ in lhs_candidates: consumed.add(j) consumed.add(i) merged.append(merged_item) continue merged.append(it) consumed.add(i) return merged def sort_row_banded(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Dynamic Baseline Sorting nach SOTA XY-Cut++ Optimierung. Verhindert Fußnoten- und Subskript-Kollaps durch adaptive Diskriminierungsschwellen basierend auf der typografischen Basislinie (y1 / ymax) und der individuellen Eigenhöhe. """ if not items: return [] # Sort primarily by typographic baseline (y1) sorted_by_baseline = sorted(items, key=lambda it: it['y1']) rows = [] curr_row = [] curr_row_baseline = None curr_row_min_h = None for it in sorted_by_baseline: it_y1 = it['y1'] it_h = it['h'] if curr_row_baseline is None: curr_row = [it] curr_row_baseline = it_y1 curr_row_min_h = it_h else: # Dynamische Toleranz: 35% der kleineren Boxhöhe (mindestens 4 pt) row_tol = max(4.0, min(curr_row_min_h, it_h) * 0.35) # Gehört zur aktuellen Zeile, wenn die Basislinie innerhalb der dynamischen Toleranz liegt if abs(it_y1 - curr_row_baseline) <= row_tol: curr_row.append(it) curr_row_baseline = sum(x['y1'] for x in curr_row) / len(curr_row) curr_row_min_h = min(curr_row_min_h, it_h) else: curr_row.sort(key=lambda x: x['x0']) rows.append(curr_row) curr_row = [it] curr_row_baseline = it_y1 curr_row_min_h = it_h if curr_row: curr_row.sort(key=lambda x: x['x0']) rows.append(curr_row) flat = [] for r in rows: flat.extend(r) return flat def find_vertical_gutter(items: List[Dict[str, Any]], page_width: float) -> float: """ Finds a clear vertical white-space corridor (gutter) that separates two text columns. Returns the x-coordinate of the gutter, or None if no clear column division exists. Distinguishes 2-column text/lists from multi-column tables via column width and row-sharing. """ if len(items) < 4: return None # If the majority of items are explicitly tagged as table, sort row-by-row table_count = sum(1 for it in items if it.get('block_type') == 'table') if table_count >= len(items) * 0.5: return None min_x = min(it['x0'] for it in items) max_x = max(it['x1'] for it in items) span_w = max_x - min_x if span_w < 120.0: return None # Search candidates in the central zone (35% to 65% of region) c_start = min_x + span_w * 0.35 c_end = min_x + span_w * 0.65 # Check for empty vertical slices (gutters) step = 4.0 best_gutter = None max_gap_width = 0.0 curr_gap_start = None x_scan = c_start while x_scan <= c_end: # Count overlapping boxes overlap_count = sum(1 for it in items if it['x0'] < x_scan < it['x1']) if overlap_count == 0: if curr_gap_start is None: curr_gap_start = x_scan else: if curr_gap_start is not None: gap_w = x_scan - curr_gap_start if gap_w > max_gap_width: max_gap_width = gap_w best_gutter = curr_gap_start + gap_w / 2.0 curr_gap_start = None x_scan += step if curr_gap_start is not None: gap_w = c_end - curr_gap_start if gap_w > max_gap_width: max_gap_width = gap_w best_gutter = curr_gap_start + gap_w / 2.0 # Require a minimum gutter width of at least 8 points if max_gap_width >= 8.0 and best_gutter is not None: col_left = [it for it in items if it['x1'] <= best_gutter or it['x_mid'] < best_gutter] col_right = [it for it in items if it not in col_left] if not col_left or not col_right: return None left_w = max(it['x1'] for it in col_left) - min(it['x0'] for it in col_left) right_w = max(it['x1'] for it in col_right) - min(it['x0'] for it in col_right) # In a genuine 2-column text layout, at least some items (headings/lines) span across the column width (max_w >= 60%) # In a multi-column table grid (>= 3 cols), every single cell is narrow (max_w < 60% of half-width) max_w_left = max((it['w'] for it in col_left), default=0.0) max_w_right = max((it['w'] for it in col_right), default=0.0) # Only reject as table grid if BOTH sides consist of narrow cells and the gap is narrow (< 15 pt). # If at least one side has wide blocks/headings or the gap is a wide gutter, it is a genuine column split! if (max_w_left < left_w * 0.45 and max_w_right < right_w * 0.45) and max_gap_width < 15.0: return None return best_gutter return None def find_horizontal_gap(items: List[Dict[str, Any]]) -> float: """ Finds a horizontal whitespace gap (X-Cut) that completely separates items above from items below. """ if len(items) < 4: return None min_y = min(it['y0'] for it in items) max_y = max(it['y1'] for it in items) span_h = max_y - min_y if span_h < 40.0: return None # Sort items by y0 sorted_items = sorted(items, key=lambda it: it['y0']) # Track running maximum of y1 running_max_y1 = sorted_items[0]['y1'] best_gap_y = None max_gap_h = 0.0 for i in range(1, len(sorted_items)): curr_y0 = sorted_items[i]['y0'] if curr_y0 > running_max_y1: gap_h = curr_y0 - running_max_y1 above_count = i below_count = len(sorted_items) - i # Require at least 1 item above and below; for single-item boundaries (e.g. section headings), require at least 6.0 pt gap min_gap_req = 4.0 if (above_count >= 2 and below_count >= 2) else 6.0 if gap_h > max_gap_h and gap_h >= min_gap_req: max_gap_h = gap_h best_gap_y = running_max_y1 + gap_h / 2.0 running_max_y1 = max(running_max_y1, sorted_items[i]['y1']) if max_gap_h >= 5.0 and best_gap_y is not None: return best_gap_y return None def recursive_xy_cut(items: List[Dict[str, Any]], page_width: float, page_height: float, depth: int = 0) -> List[Dict[str, Any]]: """ Recursive XY-Cut++ Layout Sorter. 1. Horizontal Banners partition the region vertically into independent horizontal bands. 2. Within each band, vertical gutters split parallel columns (Left -> Right). 3. Within each column, horizontal whitespace gaps split sections/containers. 4. Within each container, items are grouped into horizontal row bands (Top -> Bottom, Left -> Right). """ if not items: return [] if len(items) <= 2 or depth > 8: return sort_row_banded(items) min_x = min(it['x0'] for it in items) max_x = max(it['x1'] for it in items) span_w = max_x - min_x # 1. Identify horizontal banners (width >= 60% of region or >= 300 pt) banners = [] non_banners = [] for it in items: is_banner = (it['w'] >= span_w * 0.60 or it['w'] >= 300.0) and it['w'] >= 60.0 # Treat major horizontal math equations (w >= 220 pt) at top of page as banners if not is_banner and it.get('block_type') == 'math' and it['w'] >= 220.0 and it['y0'] < 160.0: if any(term in it['text'] for term in ['(M_k', 'EI', 'Arbeitsgleichung', 'GA']): is_banner = True if is_banner: banners.append(it) else: non_banners.append(it) if banners: banners.sort(key=lambda b: b['y0']) result = [] last_y = -1e9 for b in banners: b_y0 = b['y0'] b_y1 = b['y1'] band_items = [it for it in non_banners if last_y <= it['y_mid'] < b_y0] if band_items: result.extend(recursive_xy_cut(band_items, page_width, page_height, depth + 1)) # Preserve items sharing vertical span with the banner beside_left = [it for it in non_banners if b_y0 <= it['y_mid'] < b_y1 and it['x1'] <= b['x0'] + 10] if beside_left: result.extend(sort_row_banded(beside_left)) result.append(b) beside_right = [it for it in non_banners if b_y0 <= it['y_mid'] < b_y1 and it['x0'] >= b['x1'] - 10] if beside_right: result.extend(sort_row_banded(beside_right)) last_y = b_y1 bottom_items = [it for it in non_banners if it['y_mid'] >= last_y] if bottom_items: result.extend(recursive_xy_cut(bottom_items, page_width, page_height, depth + 1)) return result # 2. Inside a single column or sub-region (span_w < 70% of page), check horizontal gap FIRST! # This guarantees that stacked sections/containers (e.g. Greek alphabet table above Section 4) # are completely read in sequence before slicing them into internal sub-columns. if span_w < page_width * 0.70: gap_y = find_horizontal_gap(items) if gap_y is not None: items_above = [it for it in items if it['y1'] <= gap_y or it['y_mid'] < gap_y] items_below = [it for it in items if it not in items_above] if items_above and items_below: above_sorted = recursive_xy_cut(items_above, page_width, page_height, depth + 1) below_sorted = recursive_xy_cut(items_below, page_width, page_height, depth + 1) return above_sorted + below_sorted # 3. Check for vertical column split (Gutter / Y-Cut) gutter_x = find_vertical_gutter(items, page_width) if gutter_x is not None: col_left = [it for it in items if it['x1'] <= gutter_x or it['x_mid'] < gutter_x] col_right = [it for it in items if it not in col_left] if col_left and col_right: left_sorted = recursive_xy_cut(col_left, page_width, page_height, depth + 1) right_sorted = recursive_xy_cut(col_right, page_width, page_height, depth + 1) return left_sorted + right_sorted # 4. Check for horizontal whitespace gap (X-Cut fallback for full-width blocks) gap_y = find_horizontal_gap(items) if gap_y is not None: items_above = [it for it in items if it['y1'] <= gap_y or it['y_mid'] < gap_y] items_below = [it for it in items if it not in items_above] if items_above and items_below: above_sorted = recursive_xy_cut(items_above, page_width, page_height, depth + 1) below_sorted = recursive_xy_cut(items_below, page_width, page_height, depth + 1) return above_sorted + below_sorted # 5. Default: Dense single-column / table / paragraph -> sort via Row-Banding return sort_row_banded(items) def decompose_and_sort_spans_docling(boxes_with_data, page_width=595.0, page_height=842.0): """ Public entry point for backward compatibility with gradio_api.py. Accepts list of OCR boxes, normalizes, sorts in column/row-aware order, and returns tuples of ((x0, y0, x1, y1), text, raw_box). """ if not boxes_with_data: return [] normalized = [] for item in boxes_with_data: norm = normalize_item_coords(item, page_width, page_height) if norm is not None: normalized.append(norm) if not normalized: return [] # 1. Pre-merge stacked equation LHS conditions with equations merged = merge_equation_lhs_and_rhs(normalized) # 2. Decompose non-math multi-line blocks into atomic line spans atomic = decompose_atomic_lines(merged) # 3. Apply Universal Recursive XY-Cut++ sorted_items = recursive_xy_cut(atomic, page_width, page_height) # Return standard tuple format: (coords, text, box) return [(it['coords'], it['text'], it['raw']) for it in sorted_items] def sort_boxes_docling_style(boxes_with_data, page_width=595.0, page_height=842.0): """ Alias wrapper. """ return decompose_and_sort_spans_docling(boxes_with_data, page_width, page_height)