| """Bounded metadata cache for display previews, never for inference inputs.""" |
| from collections import OrderedDict |
| from functools import lru_cache |
| from pathlib import Path |
| from threading import RLock |
| from urllib.parse import quote |
| import html |
| import json |
|
|
| from PIL import Image |
|
|
| from documents import load_inputs |
| from presentation import input_summary |
| from region_selection import cached_image, region_image, source_key |
|
|
| _CACHE = OrderedDict() |
| _LOCK = RLock() |
| _LIMIT = 64 |
| PREVIEW_EDGE = 1280 |
|
|
|
|
| def display_image(image): |
| preview = image.copy() |
| preview.thumbnail((PREVIEW_EDGE, PREVIEW_EDGE), Image.Resampling.LANCZOS) |
| return preview |
|
|
|
|
| @lru_cache(maxsize=64) |
| def example_thumbnail(path): |
| image = load_inputs([path], 'document_markdown', 1)[0] |
| image.thumbnail((384, 256)) |
| return cached_image(image, image_format='webp', quality=75) |
|
|
|
|
| def source_preview(paths, task, page, drawing=False, temp_files=None): |
| identity = source_key(paths, page) |
| key = (identity, tuple(Path(path).suffix.lower() for path in paths), task == 'crosspage_table', drawing) |
| with _LOCK: |
| entry = _CACHE.get(key) |
| if entry is not None and all(Path(path).is_file() for path in (*entry['files'], *entry['originals'])): |
| _CACHE.move_to_end(key) |
| else: |
| entry = None |
| if entry is None: |
| images = load_inputs(paths, task, page) |
| pdf = Path(paths[0]).suffix.lower() == '.pdf' |
| originals = [cached_image(image) for image in images] if pdf else list(paths) |
| pages = [{'url': './gradio_api/file=' + quote(str(path), safe='/'), |
| 'name': f'Page {int(page)+index}' if pdf else Path(path).name, |
| 'width': image.width, 'height': image.height} |
| for index, (path, image) in enumerate(zip(originals, images))] |
| if drawing: |
| files = set() |
| region = region_image(images[0], paths, page, temp_files=files, |
| preview=display_image(images[0]), highres_url=pages[0]['url']) |
| values = [] |
| else: |
| files = [cached_image(display_image(image), image_format='webp', quality=75) for image in images] |
| region = None |
| values = [(path, f'Page {int(page)+index}' if Path(paths[0]).suffix.lower()=='.pdf' else f'Image {index+1}') |
| for index, path in enumerate(files)] |
| if pdf: |
| files = [*files, *originals] |
| metadata = html.escape(json.dumps(pages), quote=True) |
| entry = {'files': tuple(files), 'originals': tuple(originals), 'region': region, 'values': values, |
| 'summary': input_summary(images, pdf=pdf, page=page) |
| + f'<span hidden data-source-pages="{metadata}"></span>'} |
| with _LOCK: |
| _CACHE[key] = entry |
| _CACHE.move_to_end(key) |
| while len(_CACHE) > _LIMIT: |
| _CACHE.popitem(last=False) |
| if temp_files is not None: |
| temp_files.update(entry['files']) |
| return entry |
|
|
