Process image and perform OCR recognition. Args: image_path: Image file path, None if no image. Returns: tuple: (visualization image, Markdown content with base64 images, JSON content, ZIP file path, raw Markdown)
(
image_path: str | None
)
| 59 | |
| 60 | |
| 61 | def process_image( |
| 62 | image_path: str | None |
| 63 | ) -> tuple[Image.Image | None, str, str, str | None, str, str]: |
| 64 | """Process image and perform OCR recognition. |
| 65 | |
| 66 | Args: |
| 67 | image_path: Image file path, None if no image. |
| 68 | |
| 69 | Returns: |
| 70 | tuple: (visualization image, Markdown content with base64 images, JSON content, ZIP file path, raw Markdown) |
| 71 | """ |
| 72 | global current_pipeline |
| 73 | |
| 74 | if image_path is None: |
| 75 | return None, '', '', None, '', '' |
| 76 | |
| 77 | # Initialize pipeline on first use |
| 78 | if current_pipeline is None: |
| 79 | current_pipeline = get_pipeline() |
| 80 | |
| 81 | # Get original image name |
| 82 | base_name = os.path.splitext(os.path.basename(image_path))[0] |
| 83 | file_ext = os.path.splitext(image_path)[1] or '.jpg' |
| 84 | |
| 85 | # Create a directory with image name for this request |
| 86 | output_base_dir = 'gradio_outputs' |
| 87 | os.makedirs(output_base_dir, exist_ok=True) |
| 88 | |
| 89 | # Add timestamp to avoid conflicts if same filename is uploaded multiple times |
| 90 | timestamp = str(uuid.uuid4())[:8] |
| 91 | folder_name = f'{base_name}_{timestamp}' |
| 92 | tmp_dir = os.path.join(output_base_dir, folder_name) |
| 93 | os.makedirs(tmp_dir, exist_ok=True) |
| 94 | |
| 95 | try: |
| 96 | # Copy and rename the input image |
| 97 | tmp_img_path = os.path.join(tmp_dir, f'{base_name}{file_ext}') |
| 98 | image = Image.open(image_path) |
| 99 | image.save(tmp_img_path) |
| 100 | |
| 101 | # Predict |
| 102 | result = current_pipeline( |
| 103 | img_path=tmp_img_path, |
| 104 | merge_layout_blocks=True |
| 105 | ) |
| 106 | logger.info(f'Pipeline result type: {type(result)}, has content: {bool(result)}') |
| 107 | if result: |
| 108 | logger.info(f'Result keys: {result.keys()}') |
| 109 | if 'recognition_results' in result: |
| 110 | logger.info(f'Recognition results count: {len(result["recognition_results"])}') |
| 111 | |
| 112 | if not result: |
| 113 | logger.warning('Pipeline returned empty result') |
| 114 | return None, 'No results found.', '', None, '', '' |
| 115 | |
| 116 | # Save results |
| 117 | logger.info(f'Saving results to: {tmp_dir}') |
| 118 | current_pipeline.save_visualization(result, tmp_dir) |
nothing calls this directly
no test coverage detected