"""Terrarioum AI — interface de génération AlayaWorld (image + caméra + prompt).""" import os import shutil import subprocess import sys import tempfile import urllib.request import zipfile from pathlib import Path import gradio as gr import spaces from PIL import Image from huggingface_hub import hf_hub_download, snapshot_download ROOT = Path(__file__).resolve().parent CASE = ROOT / 'playground' / 'case1' / 'case1_camera.pt' WEIGHTS = ROOT / 'checkpoints' / 'merged_infer.safetensors' CONFIG = ROOT / 'configs' / 'infer.yaml' def check_ready(): missing = [] for name, path in [('poids AlayaWorld', WEIGHTS), ('configuration', CONFIG), ('trajectoire caméra case1', CASE), ('Gemma 3', ROOT / 'checkpoints' / 'gemma-3-12b-it-qat-q4_0-unquantized' / 'config.json'), ('code Depth Anything 3', ROOT / 'checkpoints' / 'Depth-Anything-3' / 'src' / 'depth_anything_3')]: if not path.exists(): missing.append(name) return missing def preparer_modeles(progress=gr.Progress()): """Télécharge automatiquement les fichiers manquants avant l'utilisation du GPU.""" checkpoints = ROOT / 'checkpoints' checkpoints.mkdir(parents=True, exist_ok=True) try: if not WEIGHTS.is_file() or WEIGHTS.stat().st_size < 1_000_000_000: progress(0, desc='AlayaWorld : téléchargement ~31,3 Gio') hf_hub_download(repo_id='AlayaLab/AlayaWorld', filename='merged_infer.safetensors', local_dir=str(checkpoints)) gemma = checkpoints / 'gemma-3-12b-it-qat-q4_0-unquantized' if not (gemma / 'config.json').is_file(): progress(0.4, desc='Téléchargement Gemma 3') snapshot_download(repo_id='google/gemma-3-12b-it-qat-q4_0-unquantized', local_dir=str(gemma), token=os.environ.get('HF_TOKEN') or None) da3 = checkpoints / 'Depth-Anything-3' if not (da3 / 'src' / 'depth_anything_3').is_dir(): progress(0.7, desc='Téléchargement Depth Anything 3') with tempfile.TemporaryDirectory() as tmp: archive = Path(tmp) / 'da3.zip' urllib.request.urlretrieve('https://github.com/ByteDance-Seed/Depth-Anything-3/archive/refs/heads/main.zip', archive) with zipfile.ZipFile(archive) as z: for entry in z.infolist(): parts = Path(entry.filename).parts[1:] if entry.is_dir() or not parts or '..' in parts: continue target = da3.joinpath(*parts) target.parent.mkdir(parents=True, exist_ok=True) with z.open(entry) as source, target.open('wb') as dest: shutil.copyfileobj(source, dest) progress(0.85, desc='Poids Depth Anything 3') snapshot_download(repo_id='depth-anything/DA3NESTED-GIANT-LARGE-1.1', cache_dir=str(checkpoints / 'hf_cache')) progress(1, desc='Fichiers prêts') return '✅ Modèles présents. Lancement de la génération…', True except Exception as exc: return f'❌ Téléchargement impossible : {type(exc).__name__}: {exc}', False @spaces.GPU(duration=120) def generate(image, prompt, rounds, ready): if not ready: return None, "Téléchargement incomplet : génération non lancée." if image is None: return None, 'Ajoute une image de départ.' if not prompt or not prompt.strip(): return None, 'Décris ton monde.' missing = check_ready() if missing: return None, ('Impossible de lancer AlayaWorld : fichiers absents : ' + ', '.join(missing) + '. Les fichiers de stockage temporaire peuvent disparaître après redémarrage.') # Une caméra d’exemple est utilisée : ce n’est pas encore un pilotage en temps réel. work = Path(tempfile.mkdtemp(prefix='terrarioum_')) prefix = work / 'scene' try: Image.open(image).convert('RGB').save(str(prefix) + '_image.png') shutil.copy2(CASE, str(prefix) + '_camera.pt') (work / 'scene_prompt.txt').write_text(prompt.strip(), encoding='utf-8') out = work / 'videos' cmd = [sys.executable, '-m', 'inference.run', '--input', str(prefix), '--cfg', str(CONFIG), '--output-dir', str(out), '--rounds', str(int(rounds)), '--compile', 'none', '--no-flex-attn', '--no-joystick', '--skill-sec', '0'] env = dict(os.environ) env['PYTORCH_ALLOC_CONF'] = 'expandable_segments:True' result = subprocess.run(cmd, cwd=str(ROOT), env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=105) videos = sorted(out.rglob('*.mp4')) if out.exists() else [] if result.returncode != 0 or not videos: return None, 'Génération échouée :\n' + result.stdout[-4500:] # Conserver le résultat en dehors du répertoire de travail temporaire. saved = ROOT / 'outputs' / (work.name + '.mp4') saved.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(videos[-1], saved) return str(saved), 'Vidéo générée avec AlayaWorld (trajectoire caméra de démonstration).' except subprocess.TimeoutExpired: return None, ('Temps GPU dépassé avant la fin de la génération. ' 'ZeroGPU peut être insuffisant pour charger et exécuter ce modèle volumineux.') except Exception as exc: return None, f'Erreur de génération : {type(exc).__name__}: {exc}' finally: shutil.rmtree(work, ignore_errors=True) with gr.Blocks(title='Terrarioum AI — Génération de monde') as demo: gr.Markdown('# 🌿 Terrarioum AI — AlayaWorld\nCrée une scène à partir d’une image et d’une description.') with gr.Row(): with gr.Column(): first_image = gr.Image(label='Image de départ du monde', type='filepath') prompt = gr.Textbox(label='Description du monde', lines=4, placeholder='Une forêt tropicale humide, des plantes luxuriantes, lumière naturelle...') rounds = gr.Slider(1, 3, value=1, step=1, label='Durée (1 à 3 séquences de ~1,33 s)') start = gr.Button('🌍 Générer mon monde', variant='primary') with gr.Column(): video = gr.Video(label='Monde généré') status = gr.Textbox(label='Résultat', lines=6) gr.Markdown('La caméra suit pour l’instant la trajectoire fournie dans `playground/case1`. ' 'Le déplacement libre en temps réel nécessitera une intégration supplémentaire. ' 'Sur ZeroGPU gratuit, le chargement du modèle peut échouer ou dépasser le temps disponible.') gr.Markdown('Les modèles manquants sont téléchargés automatiquement au clic. Le stockage ZeroGPU reste temporaire.') ready = gr.State(False) preparation = start.click(preparer_modeles, inputs=[], outputs=[status, ready], concurrency_limit=1) preparation.success(generate, inputs=[first_image, prompt, rounds, ready], outputs=[video, status], concurrency_limit=1) demo.queue() demo.launch()