Spaces:
Running on Zero
Running on Zero
| """Local Gradio demo and Hugging Face Docker Space entry point.""" | |
| import argparse | |
| import json | |
| import os | |
| if os.environ.get('INFINISPLAT_ZEROGPU') == '1': | |
| import spaces # Must initialize CUDA emulation before importing torch. | |
| from pathlib import Path | |
| import re | |
| import tempfile | |
| import threading | |
| import uuid | |
| import time | |
| import warnings | |
| import gradio as gr | |
| import numpy as np | |
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.responses import FileResponse, HTMLResponse | |
| from fastapi.staticfiles import StaticFiles | |
| import uvicorn | |
| from privacy import FileAccess | |
| from infinisplat import Pipeline, export_ply, export_html | |
| from infinisplat.panorama import create_template, slice_panorama, load_template | |
| from infinisplat.cameras import load_colmap | |
| from infinisplat.viewer import require_html_converter | |
| MODES = [('Provided cameras / GT or COLMAP', 'provided'), | |
| ('DA3 estimated cameras → Pose conditioning', 'da3-pose'), | |
| ('Pose-free', 'pose-free')] | |
| def create_app(model, device='cuda:0', output_root=None, examples_manifest=None, inference_fn=None): | |
| output_root = Path(output_root or tempfile.mkdtemp(prefix='infinisplat-demo-')).resolve() | |
| output_root.mkdir(parents=True, exist_ok=True) | |
| examples = {} | |
| example_files = [] | |
| if examples_manifest: | |
| manifest = Path(examples_manifest).resolve() | |
| entries = json.loads(manifest.read_text())['examples'] | |
| panorama_manifest = os.environ.get('INFINISPLAT_PANORAMA_EXAMPLES') | |
| if panorama_manifest: | |
| extra = Path(panorama_manifest).resolve() | |
| # Each manifest has its own separately verified public asset root. | |
| entries = [(entry, manifest.parent) for entry in entries] + [ | |
| (entry, extra.parent) for entry in json.loads(extra.read_text())['examples']] | |
| else: | |
| entries = [(entry, manifest.parent) for entry in entries] | |
| for entry, asset_root in entries: | |
| def asset(relative): | |
| path = (asset_root / relative).resolve() | |
| if asset_root not in path.parents or not path.is_file(): | |
| raise ValueError('Invalid example asset: ' + relative) | |
| example_files.append(str(path)) | |
| return str(path) | |
| key = entry['id'] | |
| # Withdraw the Japanese Room case from both the UI and example APIs. | |
| if key == 'ai_006_001': | |
| continue | |
| if not re.fullmatch(r'[a-z0-9_]+', key) or key in examples: | |
| raise ValueError('Invalid or duplicate example ID') | |
| images = [asset(p) for p in entry['images']] | |
| if not 2 <= len(images) <= 50: | |
| raise ValueError('Examples require 2–50 images') | |
| examples[key] = dict(title=entry['title'], images=images, | |
| panorama=asset(entry['panorama']) if entry.get('panorama') else None, | |
| camera=asset(entry['camera']) if entry.get('camera') else None, | |
| pose_mode=entry.get('pose_mode', 'da3-pose'), | |
| long_side=int(entry.get('long_side', 1036)), | |
| processed_wh=entry.get('processed_wh'), | |
| video=asset(entry['video']) if entry.get('video') else None, | |
| credit=entry.get('credit', ''), | |
| html=asset(entry['html']), | |
| ply=asset(entry['ply']) if entry.get('ply') else None, | |
| thumbnail=asset(entry.get('preview', entry['images'][0]))) | |
| if examples[key]['pose_mode'] == 'provided' and not examples[key]['camera']: | |
| raise ValueError('Provided-camera examples require a camera file') | |
| access = FileAccess(example_files) | |
| # Gradio copies selected files into its cache. Match ordered image contents | |
| # and names so the project-page crop survives those changed file paths. | |
| example_preprocessing = { | |
| tuple(access.cache_key(path) for path in e['images']): e['processed_wh'] | |
| for e in examples.values() if e['processed_wh'] | |
| } | |
| pipeline = None | |
| lock = threading.Lock() | |
| def preview(files, request: gr.Request): | |
| access.require(request, files or []) | |
| files = files or [] | |
| return files, files, '\n'.join(f'{i+1}. {Path(p).name}' for i, p in enumerate(files)), f'Input: **{len(files)} views**' | |
| def extract_video(video, count, request: gr.Request): | |
| access.require(request, [video]) | |
| import cv2 | |
| if not video: | |
| raise gr.Error('Upload a video first.') | |
| capture = cv2.VideoCapture(str(video)) | |
| try: | |
| total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT)) | |
| if total < 2: | |
| raise gr.Error('Cannot read the video, or it contains fewer than two frames.') | |
| folder = output_root / ('frames-' + uuid.uuid4().hex) | |
| folder.mkdir() | |
| paths = [] | |
| for index in np.linspace(0, total - 1, min(int(count), total), dtype=int): | |
| capture.set(cv2.CAP_PROP_POS_FRAMES, int(index)) | |
| ok, frame = capture.read() | |
| if not ok: | |
| continue | |
| path = folder / f'{int(index):08d}.jpg' | |
| if cv2.imwrite(str(path), frame): | |
| paths.append(str(path)) | |
| if len(paths) < 2: | |
| raise gr.Error('Fewer than two frames could be decoded. Try another video.') | |
| access.grant(request, paths) | |
| return preview(paths, request) | |
| finally: | |
| capture.release() | |
| def reconstruct_impl(files, mode, camera_npz, colmap_cameras, colmap_images, size, | |
| request, progress, panorama=False, source_metadata=None): | |
| nonlocal pipeline | |
| started = time.monotonic() | |
| files = list(files or []) | |
| if not 2 <= len(files) <= 50: | |
| raise gr.Error('Upload 2–50 overlapping images of the same static scene.') | |
| if mode not in {value for _, value in MODES}: | |
| raise gr.Error('Select a valid reconstruction mode.') | |
| cameras = None | |
| try: | |
| access.require(request, [*files, camera_npz, colmap_cameras, colmap_images]) | |
| if mode == 'provided': | |
| if camera_npz and (colmap_cameras or colmap_images): | |
| raise ValueError('Choose either NPZ or COLMAP cameras, not both.') | |
| if camera_npz: | |
| with np.load(camera_npz, allow_pickle=False) as data: | |
| cameras = {key: data[key] for key in ('c2w', 'intrinsics')} | |
| elif colmap_cameras and colmap_images: | |
| with tempfile.TemporaryDirectory() as temp: | |
| for source, name in [(colmap_cameras, 'cameras.txt'), | |
| (colmap_images, 'images.txt')]: | |
| (Path(temp) / name).write_bytes(Path(source).read_bytes()) | |
| cameras = load_colmap(temp, files) | |
| else: | |
| raise ValueError('Upload cameras.npz, or both COLMAP cameras.txt and images.txt.') | |
| target_wh = None | |
| example_wh = example_preprocessing.get(tuple(access.cache_key(path) for path in files)) | |
| if example_wh: | |
| w, h = example_wh | |
| target_wh = (round(int(size) / 14) * 14, | |
| round(int(size) * h / w / 14) * 14) | |
| require_html_converter() | |
| if not model: | |
| raise ValueError('Set INFINISPLAT_MODEL or pass --model with an inference weights directory before starting.') | |
| with lock: | |
| if pipeline is None and inference_fn is None: | |
| progress(0.05, desc='Loading model (first run)') | |
| pipeline = Pipeline.from_pretrained(model, device=device) | |
| progress(0.2, desc='Reconstructing Gaussian scene') | |
| if inference_fn is not None: | |
| kwargs = {'panorama_shared_center': True} if panorama else {} | |
| scene = inference_fn(files, cameras, mode, int(size), target_wh=target_wh, **kwargs) | |
| else: | |
| scene = pipeline.reconstruct(files, cameras=cameras, pose_mode=mode, | |
| long_side=int(size), max_views=50, target_wh=target_wh, | |
| panorama_shared_center=panorama) | |
| run_id = uuid.uuid4().hex | |
| progress(0.8, desc='Exporting PLY and HTML') | |
| if panorama: | |
| scene['metadata']['panorama']['source'] = source_metadata | |
| ply = export_ply(scene, output_root / run_id, html=False) | |
| else: | |
| ply = export_ply(scene, output_root / run_id, html=True) | |
| del scene | |
| if panorama: | |
| export_html(ply, ply.with_suffix('.html'), display_transform=dict( | |
| quaternion=[1,0,0,0], translation=[0,0,0], | |
| camera=dict(position=[0,0,0],target=[0,0,-1],fov=90))) | |
| access.grant(request, [ply, ply.with_suffix('.html')]) | |
| access.session(request)[1]['runs'].add(run_id) | |
| viewer_url = access.viewer_link(request, run_id) | |
| viewer = f'<iframe src=".{viewer_url}" title="InfiniSplat 3D viewer" style="width:100%;height:520px;border:0;background:#666" referrerpolicy="no-referrer" allow="fullscreen" allowfullscreen></iframe>' | |
| return viewer, str(ply), str(ply.with_suffix('.html')), ( | |
| f"Complete · {'360° panorama / 24 calibrated views' if panorama else str(len(files))+' images / '+mode} · {int(size)} px · " | |
| f'{time.monotonic() - started:.1f} s (including loading and export)\n\n' | |
| f'[Open result in a new tab](.{viewer_url})') | |
| except gr.Error: | |
| # Preserve actionable ZeroGPU quota and queue messages. | |
| raise | |
| except Exception as error: | |
| import torch | |
| if isinstance(error, torch.cuda.OutOfMemoryError): | |
| torch.cuda.empty_cache() | |
| raise gr.Error('GPU memory is full. Use fewer images or a lower resolution.') from error | |
| raise gr.Error('Reconstruction failed. Check the input files and camera settings, or contact the operator.') from None | |
| def reconstruct(files, mode, camera_npz, colmap_cameras, colmap_images, size, | |
| request: gr.Request, progress=gr.Progress()): | |
| return reconstruct_impl(files, mode, camera_npz, colmap_cameras, colmap_images, | |
| size, request, progress) | |
| def inspect_panorama(path, request): | |
| from PIL import Image | |
| access.require(request, [path]) | |
| if not path: | |
| raise gr.Error('Upload one full 360° panorama first.') | |
| if Path(path).stat().st_size > 100 * 1024**2: | |
| raise gr.Error('Use a panorama file smaller than 100 MB.') | |
| try: | |
| with warnings.catch_warnings(): | |
| warnings.simplefilter('ignore', Image.DecompressionBombWarning) | |
| with Image.open(path) as image: | |
| if image.format not in ('JPEG','PNG','WEBP'): | |
| raise ValueError('Use a JPG, PNG or WebP panorama.') | |
| if image.width * image.height > 128_000_000 or max(image.size) >= 32767: | |
| raise ValueError('Use at most 128 megapixels and less than 32767 pixels per edge.') | |
| if abs(image.width / image.height - 2) > .005: | |
| raise ValueError('Use a full 360° × 180° equirectangular panorama with a 2:1 aspect ratio.') | |
| if image.getexif().get(274,1) != 1: | |
| raise ValueError('Remove EXIF rotation before uploading the panorama.') | |
| return image.size | |
| except (OSError, ValueError, Image.DecompressionBombError) as error: | |
| raise gr.Error(str(error)) from None | |
| def preview_panorama(path, request: gr.Request): | |
| width,height = inspect_panorama(path, request) | |
| return path, f'360° panorama · {width} × {height} · ready to reconstruct' | |
| def reconstruct_panorama(path, size, request: gr.Request, progress=gr.Progress()): | |
| inspect_panorama(path, request) | |
| if int(size) not in (224,336,518,756,768,1036,1078): | |
| raise gr.Error('Select a supported reconstruction resolution.') | |
| progress(.02, desc='Preparing panorama views') | |
| with tempfile.TemporaryDirectory(prefix='panorama-',dir=output_root) as temp: | |
| root=Path(temp);create_template(root/'template',round(int(size)/14)*14) | |
| try: | |
| metadata=slice_panorama(path,root/'template',root/'prepared') | |
| except (ValueError,OSError) as error: | |
| raise gr.Error(str(error)) from None | |
| spec,_=load_template(root/'template') | |
| paths=[str(root/'prepared/inputs'/v['file']) for v in spec['views']] | |
| camera=str(root/'prepared/cameras.npz') | |
| access.grant(request,[*paths,camera]) | |
| return reconstruct_impl(paths,'provided',camera,None,None,size,request,progress, | |
| panorama=True,source_metadata=metadata) | |
| def load_example(key, request: gr.Request): | |
| if key not in examples: | |
| raise gr.Error('Select an example.') | |
| example = examples[key] | |
| url = f'./example-viewer/{key}?noanim' | |
| iframe = f'<iframe src="{url}" title="InfiniSplat example" style="width:100%;height:520px;border:0;background:#666" allow="fullscreen" allowfullscreen></iframe>' | |
| camera_label = 'GT camera pose' if example['pose_mode'] == 'provided' else 'Estimated cameras' | |
| return (*preview(example['images'], request), example['pose_mode'], example['long_side'], iframe, | |
| example['ply'], example['html'], | |
| f"Example loaded · {len(example['images'])} views · {camera_label} · {example['long_side']} px\n\n[Open 3D viewer]({url}) · " + | |
| ('Click Reconstruct to run again.' if example['ply'] else | |
| 'HTML preview available. Click Reconstruct to generate a Gaussian PLY download.') + | |
| (f"\n\n{example['credit']}" if example['credit'] else ''), | |
| example['camera'], None, None, gr.Group(visible=example['pose_mode'] == 'provided'), example['video'], | |
| gr.Tabs(selected='panorama' if example['panorama'] else ('video' if example['video'] else 'images')), 16, example['panorama']) | |
| example_keys = list(examples) | |
| empty_viewer = ('<div class="empty-scene"><span>3D reconstruction</span>' | |
| '<p>Upload images or select an example to begin.</p></div>') | |
| with gr.Blocks(title='InfiniSplat v2', analytics_enabled=False, delete_cache=(3600, 86400)) as demo: | |
| selected = gr.File(file_count='multiple', type='filepath', visible=False) | |
| gr.HTML('<header class="masthead"><h1>InfiniSplat v2</h1>' | |
| '<p>Feed-forward 3D Gaussian reconstruction from images or video.</p></header>', elem_id='masthead') | |
| with gr.Row(elem_id='workspace'): | |
| with gr.Column(scale=1, min_width=300, elem_id='input-panel'): | |
| gr.HTML('<div class="section-heading"><h2>Input</h2></div>') | |
| with gr.Tabs(elem_id='input-tabs') as input_tabs: | |
| with gr.Tab('Images', id='images'): | |
| files = gr.File(label='Drop images here or click to upload', file_count='multiple', | |
| file_types=['.jpg', '.jpeg', '.png', '.webp'], type='filepath', | |
| height=155, show_label=False, elem_id='image-upload') | |
| gr.Markdown('2–50 overlapping images · JPG / PNG / WebP', elem_id='upload-hint') | |
| with gr.Tab('360° Panorama', id='panorama'): | |
| panorama_file = gr.File(label='360° panorama', file_types=['.jpg','.jpeg','.png','.webp'], type='filepath') | |
| panorama_status = gr.Markdown('One full 2:1 equirectangular panorama.') | |
| panorama_run = gr.Button('Reconstruct panorama') | |
| with gr.Tab('Video', id='video'): | |
| video = gr.Video(label='Upload a video of your scene', sources=['upload'], | |
| height=180, elem_id='video-upload', include_audio=True) | |
| frame_count = gr.Slider(2, 50, value=16, step=1, label='Number of frames') | |
| extract = gr.Button('Extract frames', size='sm') | |
| gallery = gr.Gallery(label='Input preview', columns=4, height=148, | |
| object_fit='cover', buttons=[], elem_id='input-gallery') | |
| view_count = gr.Markdown('Input: **0 views**', elem_id='input-view-count') | |
| with gr.Accordion('Reconstruction settings', open=False, elem_id='settings'): | |
| mode = gr.Radio(MODES, value='da3-pose', label='Camera Pose (optional)') | |
| gr.Markdown('Reliable pose conditioning generally improves quality. DA3 estimates cameras by default.') | |
| with gr.Group(visible=False) as camera_group: | |
| gr.Markdown('Upload OpenCV c2w / intrinsics in NPZ format matching the input order, or a COLMAP text model.') | |
| camera_npz = gr.File(label='cameras.npz', file_types=['.npz'], type='filepath') | |
| colmap_cameras = gr.File(label='COLMAP cameras.txt', file_types=['.txt'], type='filepath') | |
| colmap_images = gr.File(label='COLMAP images.txt', file_types=['.txt'], type='filepath') | |
| size = gr.Dropdown([224, 336, 518, 756, 768, 1036, 1078], value=518, label='Resolution (long edge)') | |
| order = gr.Textbox(label='Input order', interactive=False, lines=3) | |
| run = gr.Button('Reconstruct', variant='primary', elem_id='generate') | |
| gr.Markdown('Default uploads: estimated cameras · 518 px · 16 video frames. Examples include GT cameras.', elem_id='default-hint') | |
| with gr.Column(scale=2, min_width=360, elem_id='result-panel'): | |
| gr.HTML('<div class="section-heading"><h2>Result</h2></div>', elem_id='result-heading') | |
| viewer = gr.HTML(empty_viewer, elem_id='scene-viewer') | |
| gr.Markdown('Drag to orbit · Scroll to zoom · Fullscreen available', elem_id='viewer-help') | |
| # Retain the file outputs for API clients; visible buttons make | |
| # download availability explicit in the browser. | |
| ply_file = gr.File(label='Gaussian PLY', interactive=False, visible=False) | |
| html_file = gr.File(label='HTML', interactive=False, visible=False) | |
| with gr.Row(elem_id='downloads'): | |
| ply_download = gr.DownloadButton(label='Download Gaussian PLY', interactive=False, | |
| elem_id='download-ply', scale=1, min_width=0) | |
| html_download = gr.DownloadButton(label='Download HTML', interactive=False, | |
| elem_id='download-html', scale=1, min_width=0) | |
| gr.Markdown('PLY downloads become available after reconstruction.', elem_id='download-hint') | |
| status = gr.Markdown('Ready. Upload your scene or choose an example below.', elem_id='result-status') | |
| if examples: | |
| with gr.Column(elem_id='examples-section'): | |
| gr.HTML(f'<div class="section-heading"><h2>Examples</h2><span>{len(examples)} scenes · GT camera pose</span></div>' | |
| , elem_id='examples-heading') | |
| example_picker = gr.Dropdown(choices=example_keys, visible=False, value=None) | |
| example_gallery = gr.Gallery( | |
| [(v['thumbnail'], f"{v['title']} · {len(v['images'])} views") for v in examples.values()], | |
| columns=min(4, len(examples)), | |
| rows=(len(examples) + 3) // 4, | |
| height=310 if len(examples) > 4 else 174, object_fit='cover', allow_preview=False, | |
| interactive=False, show_label=False, container=False, buttons=[], elem_id='example-gallery') | |
| files.change(preview, files, [selected, gallery, order, view_count], queue=False, api_name="preview_inputs") | |
| extract.click(extract_video, [video, frame_count], [selected, gallery, order, view_count], api_name='extract_video') | |
| if examples: | |
| example_picker.change(load_example, example_picker, | |
| [selected, gallery, order, view_count, mode, size, viewer, ply_file, html_file, | |
| status, camera_npz, colmap_cameras, colmap_images, camera_group, video, input_tabs, frame_count, panorama_file], | |
| api_name='select_example', queue=False) | |
| def choose_thumbnail(evt: gr.SelectData, request: gr.Request): | |
| return load_example(example_keys[evt.index], request) | |
| example_gallery.select(choose_thumbnail, None, | |
| [selected, gallery, order, view_count, mode, size, viewer, ply_file, html_file, | |
| status, camera_npz, colmap_cameras, colmap_images, camera_group, video, input_tabs, frame_count, panorama_file], | |
| api_name=False, queue=False) | |
| mode.change(lambda value: gr.Group(visible=value == 'provided'), mode, camera_group, queue=False) | |
| run.click(reconstruct, [selected, mode, camera_npz, colmap_cameras, colmap_images, size], | |
| [viewer, ply_file, html_file, status], concurrency_limit=1, concurrency_id='reconstruction', api_name='reconstruct') | |
| panorama_file.change(preview_panorama, panorama_file, [panorama_file, panorama_status], | |
| queue=False, api_name='preview_panorama') | |
| panorama_run.click(reconstruct_panorama, [panorama_file, size], | |
| [viewer, ply_file, html_file, status], concurrency_limit=1, | |
| concurrency_id='reconstruction', api_name='reconstruct_panorama') | |
| def ready_download(path, request: gr.Request): | |
| access.require(request, [path]) | |
| return gr.update(value=path, interactive=bool(path)) | |
| ply_file.change(ready_download, ply_file, ply_download, queue=False, api_name=False) | |
| html_file.change(ready_download, html_file, html_download, queue=False, api_name=False) | |
| demo.queue(max_size=8) | |
| app = FastAPI() | |
| if os.environ.get('INFINISPLAT_ZEROGPU') == '1': | |
| from owner_quota import exchange_identity | |
| def owner_identity(request: Request): | |
| return exchange_identity(request) | |
| frontend = Path(__file__).parent / 'frontend' | |
| thumbnail_root = Path(os.environ.get('INFINISPLAT_THUMBNAILS', frontend / 'thumbs-v1')) | |
| if os.environ.get('INFINISPLAT_PANORAMA_THUMBNAILS'): | |
| app.mount('/studio-assets/panorama-thumbs', StaticFiles(directory=os.environ['INFINISPLAT_PANORAMA_THUMBNAILS']), name='panorama-thumbnails') | |
| app.mount('/studio-assets/thumbs-v1', StaticFiles(directory=thumbnail_root), name='example-thumbnails') | |
| app.mount('/studio-assets', StaticFiles(directory=frontend), name='studio-assets') | |
| def studio_index(): | |
| return FileResponse(frontend / 'index.html', media_type='text/html', | |
| headers={'Cache-Control': 'no-store'}) | |
| def studio_examples(): | |
| return [dict(id=key, title=e['title'], count=len(e['images']), | |
| pose_mode=e['pose_mode'], long_side=e['long_side'], processed_wh=e['processed_wh'], | |
| camera=f'/example-cameras/{key}' if e['camera'] else None, | |
| video=bool(e['video']), panorama=bool(e['panorama']), credit=e['credit']) | |
| for key, e in examples.items()] | |
| def studio_cameras(example_id: str): | |
| if example_id not in examples or not examples[example_id]['camera']: | |
| raise HTTPException(status_code=404) | |
| return FileResponse(examples[example_id]['camera'], media_type='application/octet-stream', | |
| filename=f'{example_id}_cameras.npz') | |
| def studio_video(example_id: str): | |
| if example_id not in examples or not examples[example_id]['video']: | |
| raise HTTPException(status_code=404) | |
| return FileResponse(examples[example_id]['video'], media_type='video/mp4') | |
| def studio_image(example_id: str, index: int): | |
| if example_id not in examples or not 0 <= index < len(examples[example_id]['images']): | |
| raise HTTPException(status_code=404) | |
| return FileResponse(examples[example_id]['images'][index]) | |
| def serve_viewer(run_id: str, request: Request): | |
| if not access.can_view(request, run_id): | |
| raise HTTPException(status_code=404) | |
| if not re.fullmatch(r'[0-9a-f]{32}', run_id): | |
| raise HTTPException(status_code=404) | |
| path = output_root / run_id / 'scene.html' | |
| if not path.is_file(): | |
| raise HTTPException(status_code=404) | |
| return FileResponse(path, media_type='text/html', headers={ | |
| 'Cache-Control': 'private, no-store', 'Referrer-Policy': 'no-referrer'}) | |
| def serve_download(run_id: str, filename: str, request: Request): | |
| if not re.fullmatch(r'[0-9a-f]{32}', run_id) or filename not in ('scene.ply', 'scene.html'): | |
| raise HTTPException(status_code=404) | |
| # A download may leave the iframe's partitioned cookie context. Reuse | |
| # the expiring result capability, scoped to this run's two outputs. | |
| if not access.can_view(request, run_id): | |
| raise HTTPException(status_code=404) | |
| path = output_root / run_id / filename | |
| if not path.is_file(): | |
| raise HTTPException(status_code=404) | |
| return FileResponse(path, media_type='application/octet-stream', filename=filename, | |
| headers={'Cache-Control': 'private, no-store', | |
| 'Referrer-Policy': 'no-referrer', | |
| 'X-Content-Type-Options': 'nosniff'}) | |
| def serve_example(example_id: str): | |
| if example_id not in examples: | |
| raise HTTPException(status_code=404) | |
| return FileResponse(examples[example_id]['html'], media_type='text/html') | |
| def serve_example_preview(example_id: str): | |
| if example_id not in examples: | |
| raise HTTPException(status_code=404) | |
| return FileResponse(examples[example_id]['thumbnail']) | |
| theme = gr.themes.Base(primary_hue='blue', neutral_hue='slate', | |
| font=['Inter', 'system-ui', 'sans-serif']) | |
| # Gradio 6.9 has no app-wide locale option. Set the locale exposed to this | |
| # document before its frontend initializes; browser preferences stay unchanged. | |
| english_head = """<script> | |
| Object.defineProperty(window.navigator, 'language', {get: () => 'en-US', configurable: true}); | |
| Object.defineProperty(window.navigator, 'languages', {get: () => ['en-US', 'en'], configurable: true}); | |
| document.documentElement.lang = 'en'; | |
| </script>""" | |
| # The mount's head hook runs after locale initialization. Insert before the | |
| # frontend module scripts in the initial document instead. | |
| async def english_document(request, call_next): | |
| response = await call_next(request) | |
| if request.url.path == '/' and 'text/html' in response.headers.get('content-type', ''): | |
| body = b''.join([chunk async for chunk in response.body_iterator]) | |
| body = body.replace(b'<head>', b'<head>' + english_head.encode(), 1) | |
| headers = dict(response.headers) | |
| headers.pop('content-length', None) | |
| headers['cache-control'] = 'no-store' | |
| return HTMLResponse(body, status_code=response.status_code, headers=headers) | |
| if request.url.path.startswith(('/studio-assets/thumbs-v1/', '/studio-assets/previews-v1/')): | |
| response.headers['Cache-Control'] = 'public, max-age=86400' | |
| if request.url.path == '/config': | |
| response.headers['Cache-Control'] = 'no-store' | |
| return response | |
| access.middleware(app) | |
| app.state.demo = demo | |
| return gr.mount_gradio_app(app, demo, path='/', allowed_paths=example_files, show_error=False, | |
| theme=theme, css=(Path(__file__).parent/'style.css').read_text(), footer_links=[], ssr_mode=False) | |
| if __name__ == '__main__': | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| parser.add_argument('--model', default=os.environ.get('INFINISPLAT_MODEL')) | |
| parser.add_argument('--device', default=os.environ.get('INFINISPLAT_DEVICE', 'cuda:0')) | |
| parser.add_argument('--host', default=os.environ.get('HOST', '127.0.0.1')) | |
| parser.add_argument('--port', type=int, default=int(os.environ.get('PORT', '7860'))) | |
| parser.add_argument('--output-root', default=os.environ.get('INFINISPLAT_OUTPUT')) | |
| parser.add_argument('--examples', default=os.environ.get('INFINISPLAT_EXAMPLES')) | |
| args = parser.parse_args() | |
| inference_fn = None | |
| if os.environ.get('INFINISPLAT_ZEROGPU') == '1': | |
| from zero_gpu import initialize, reconstruct_on_gpu | |
| initialize(args.model) | |
| inference_fn = reconstruct_on_gpu | |
| app = create_app(args.model, args.device, args.output_root, args.examples, | |
| inference_fn=inference_fn) | |
| # Generated viewer URLs contain per-result access tokens. | |
| server = uvicorn.Server(uvicorn.Config(app, host=args.host, port=args.port, access_log=False)) | |
| app.state.demo.server = server | |
| if inference_fn is not None: | |
| # Mounted Gradio uses Uvicorn rather than Blocks.launch. Execute the | |
| # same startup hook from pinned spaces 0.51.3 to pack emulated CUDA. | |
| from spaces.config import Config | |
| if Config.zero_gpu: | |
| from spaces.zero import startup | |
| startup() | |
| server.run() | |