Spaces:
Sleeping
Sleeping
Implement strict QC suite, restore German umlauts, isolate ds labels, and fix formula completeness
2e57a86 verified | """ | |
| 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) | |