Spaces:
Running on Zero
Running on Zero
| try: | |
| import spaces | |
| ZERO_GPU = True | |
| print("🖥️ ZeroGPU Runtime Detected") | |
| except ImportError: | |
| spaces = None | |
| ZERO_GPU = False | |
| print("🖥️ Dedicated GPU Runtime") | |
| import os | |
| import re | |
| import gc | |
| import math | |
| import json | |
| import time | |
| import random | |
| import traceback | |
| import base64 | |
| import sys | |
| import uuid | |
| import threading | |
| import urllib.request | |
| from io import BytesIO | |
| from typing import Dict, Any, Optional, Tuple, List | |
| from urllib.parse import unquote | |
| from PIL import Image | |
| import torch | |
| import gradio as gr | |
| from huggingface_hub import hf_hub_download | |
| from diffusers import ( | |
| StableDiffusionXLPipeline, | |
| EulerAncestralDiscreteScheduler, | |
| EulerDiscreteScheduler, | |
| DPMSolverMultistepScheduler, | |
| DPMSolverSDEScheduler, | |
| KDPM2DiscreteScheduler, | |
| KDPM2AncestralDiscreteScheduler, | |
| HeunDiscreteScheduler, | |
| LMSDiscreteScheduler, | |
| DDIMScheduler, | |
| UniPCMultistepScheduler, | |
| TCDScheduler, | |
| DEISMultistepScheduler, | |
| SASolverScheduler, | |
| PNDMScheduler, | |
| DDPMScheduler, | |
| LCMScheduler, | |
| EDMDPMSolverMultistepScheduler, | |
| ) | |
| from compel import Compel, ReturnedEmbeddingsType | |
| from fastapi import Request, Response | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from starlette.responses import JSONResponse | |
| from starlette.exceptions import HTTPException as StarletteHTTPException | |
| try: | |
| from diffusers import StableDiffusionXLPAGPipeline | |
| PAG_AVAILABLE = True | |
| print("🔮 PAG Pipeline tersedia!") | |
| except ImportError: | |
| PAG_AVAILABLE = False | |
| print("⚠️ PAG Pipeline tidak tersedia") | |
| torch.backends.cuda.matmul.allow_tf32 = True | |
| torch.backends.cudnn.allow_tf32 = True | |
| torch.backends.cudnn.benchmark = True | |
| torch.set_float32_matmul_precision("high") | |
| os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True" | |
| try: | |
| torch.backends.cuda.enable_flash_sdp(True) | |
| print("⚡ Flash SDP enabled") | |
| except AttributeError: | |
| pass | |
| print("⚡ TF32=ON | bf16=ON | SDPA=FlashAttention") | |
| MODEL_REPO = "Bl4ckSpaces/FFD-XL-3.0" | |
| MODEL_FILE = "FFD_XL_3_0_v4_bf16.safetensors" | |
| DISPLAY_NAME = "FFD-XL-3.0" | |
| PREDICTION_TYPE = "epsilon" | |
| CLIP_SKIP = 2 | |
| DEFAULT_BATCH = 1 | |
| MAX_STEPS = 100 | |
| MAX_BATCH_SIZE = 8 | |
| MAX_WIDTH = 1536 | |
| MAX_HEIGHT = 1536 | |
| MIN_WIDTH = 512 | |
| MIN_HEIGHT = 512 | |
| def validate_params(steps, batch_size, width, height): | |
| errors = [] | |
| if steps < 1 or steps > MAX_STEPS: | |
| errors.append(f"Steps must be 1-{MAX_STEPS}, got {steps}") | |
| if batch_size < 1 or batch_size > MAX_BATCH_SIZE: | |
| errors.append(f"Batch size must be 1-{MAX_BATCH_SIZE}, got {batch_size}") | |
| if width < MIN_WIDTH or width > MAX_WIDTH: | |
| errors.append(f"Width must be {MIN_WIDTH}-{MAX_WIDTH}, got {width}") | |
| if height < MIN_HEIGHT or height > MAX_HEIGHT: | |
| errors.append(f"Height must be {MIN_HEIGHT}-{MAX_HEIGHT}, got {height}") | |
| return errors | |
| def is_quota_error(error_msg: str) -> bool: | |
| msg = error_msg.lower() | |
| return any(kw in msg for kw in ["exceeded", "quota", "rate limit", "too many"]) | |
| def is_gpu_cold_start_error(error_msg: str) -> bool: | |
| msg = error_msg.lower() | |
| return any( | |
| kw in msg | |
| for kw in [ | |
| "no gpu", | |
| "gpu was available", | |
| "available after", | |
| "no gpu was", | |
| "gpu is not available", | |
| "gpu not available", | |
| "not ready", | |
| ] | |
| ) | |
| KEEP_ALIVE_INTERVAL = 180 | |
| def _send_keep_alive_ping(): | |
| for endpoint in ["http://localhost:7860/", "http://localhost:7860/health"]: | |
| try: | |
| req = urllib.request.Request( | |
| endpoint, | |
| headers={"User-Agent": "FFD-XL-KeepAlive/1.0"}, | |
| method="GET", | |
| ) | |
| urllib.request.urlopen(req, timeout=10) | |
| except Exception: | |
| pass | |
| def keep_alive_worker(): | |
| time.sleep(30) | |
| print(f"🫀 Keep-alive started (interval: {KEEP_ALIVE_INTERVAL}s)") | |
| while True: | |
| time.sleep(KEEP_ALIVE_INTERVAL) | |
| _send_keep_alive_ping() | |
| print(f"🫀 Keep-alive ping @ {time.strftime('%H:%M:%S')}") | |
| def start_keep_alive(): | |
| thread = threading.Thread(target=keep_alive_worker, daemon=True, name="keep-alive") | |
| thread.start() | |
| return thread | |
| print(f"📥 Downloading model: {MODEL_REPO}/{MODEL_FILE} ...") | |
| model_path = None | |
| try: | |
| model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE) | |
| print(f"✅ Model downloaded (anonymous): {model_path}") | |
| except Exception as e: | |
| print(f"⚠️ Anonymous download failed: {str(e)[:100]}") | |
| if model_path is None: | |
| hf_env = os.environ.get("HF_TOKEN") | |
| if hf_env: | |
| try: | |
| model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE) | |
| print(f"✅ Model downloaded with HF_TOKEN env: {model_path}") | |
| except Exception as e2: | |
| print(f"⚠️ HF_TOKEN download also failed: {str(e2)[:100]}") | |
| if model_path is None: | |
| raise RuntimeError("Failed to download model. Set HF_TOKEN in Space secrets as fallback.") | |
| print(f"⏳ Loading {DISPLAY_NAME} Pipeline (bfloat16)...") | |
| pipe = StableDiffusionXLPipeline.from_single_file( | |
| model_path, | |
| torch_dtype=torch.bfloat16, | |
| use_safetensors=True, | |
| add_watermarker=False, | |
| config="stabilityai/stable-diffusion-xl-base-1.0", | |
| ) | |
| pipe.to("cuda") | |
| print("✅ Pipeline loaded!") | |
| try: | |
| from diffusers.models.attention_processor import AttnProcessor2_0 | |
| pipe.unet.set_attn_processor(AttnProcessor2_0()) | |
| print("✅ AttnProcessor2_0 (SDPA) aktif!") | |
| except ImportError: | |
| pipe.unet.set_default_attn_processor() | |
| pipe.vae.to(dtype=torch.bfloat16) | |
| try: | |
| pipe.vae.config.force_upcast = False | |
| except AttributeError: | |
| pass | |
| pipe.upcast_vae = lambda: None | |
| print("🔒 VAE STRICT BF16 aktif") | |
| pipe.unet.to(memory_format=torch.channels_last) | |
| pipe.text_encoder.to(memory_format=torch.channels_last) | |
| pipe.text_encoder_2.to(memory_format=torch.channels_last) | |
| pipe.vae.to(memory_format=torch.channels_last) | |
| pipe.vae.enable_tiling() | |
| print("📐 Channels Last + VAE Tiling aktif") | |
| _orig_vae_decode = pipe.vae.decode | |
| _orig_vae_encode = pipe.vae.encode | |
| def _patched_vae_decode(z, return_dict=True): | |
| torch.backends.cudnn.benchmark = False | |
| try: | |
| result = _orig_vae_decode(z, return_dict=return_dict) | |
| finally: | |
| torch.backends.cudnn.benchmark = True | |
| return result | |
| def _patched_vae_encode(x, return_dict=True): | |
| torch.backends.cudnn.benchmark = False | |
| try: | |
| result = _orig_vae_encode(x, return_dict=return_dict) | |
| finally: | |
| torch.backends.cudnn.benchmark = True | |
| return result | |
| pipe.vae.decode = _patched_vae_decode | |
| pipe.vae.encode = _patched_vae_encode | |
| print("🛡️ cuDNN Benchmark Isolation aktif") | |
| _emb_type = ( | |
| ReturnedEmbeddingsType.PENULTIMATE_HIDDEN_STATES_NON_NORMALIZED | |
| if CLIP_SKIP == 2 | |
| else ReturnedEmbeddingsType.LAST_HIDDEN_STATES_NON_NORMALIZED | |
| ) | |
| compel = Compel( | |
| tokenizer=[pipe.tokenizer, pipe.tokenizer_2], | |
| text_encoder=[pipe.text_encoder, pipe.text_encoder_2], | |
| returned_embeddings_type=_emb_type, | |
| requires_pooled=[False, True], | |
| ) | |
| print("✅ Compel prompt encoder ready") | |
| try: | |
| from compel.embeddings_provider import EmbeddingsProviderMulti | |
| if not hasattr(EmbeddingsProviderMulti, "empty_z"): | |
| @property | |
| def _multi_empty_z(self): | |
| return self.embedding_providers[0].empty_z | |
| EmbeddingsProviderMulti.empty_z = _multi_empty_z | |
| print("🔧 HOTFIX #1 APPLIED: EmbeddingsProviderMulti.empty_z patched") | |
| else: | |
| print("✅ EmbeddingsProviderMulti already has empty_z") | |
| except (ImportError, AttributeError) as e: | |
| print(f"⚠️ Could not apply empty_z hotfix: {e}") | |
| CONDITIONING_CACHE = {} | |
| def _manual_pad_conditioning(tensor_a, tensor_b): | |
| if tensor_a.shape[1] == tensor_b.shape[1]: | |
| return tensor_a, tensor_b | |
| max_len = max(tensor_a.shape[1], tensor_b.shape[1]) | |
| if tensor_a.shape[1] < max_len: | |
| pad = torch.zeros( | |
| tensor_a.shape[0], | |
| max_len - tensor_a.shape[1], | |
| tensor_a.shape[2], | |
| dtype=tensor_a.dtype, | |
| device=tensor_a.device, | |
| ) | |
| tensor_a = torch.cat([tensor_a, pad], dim=1) | |
| if tensor_b.shape[1] < max_len: | |
| pad = torch.zeros( | |
| tensor_b.shape[0], | |
| max_len - tensor_b.shape[1], | |
| tensor_b.shape[2], | |
| dtype=tensor_b.dtype, | |
| device=tensor_b.device, | |
| ) | |
| tensor_b = torch.cat([tensor_b, pad], dim=1) | |
| return tensor_a, tensor_b | |
| def get_conditioning(pos, neg): | |
| cache_key = (pos, neg) | |
| if cache_key in CONDITIONING_CACHE: | |
| return CONDITIONING_CACHE[cache_key] | |
| def encode_with_break(prompt_text): | |
| parts = prompt_text.split("BREAK") | |
| embeds_list, pooled_list = [], [] | |
| for part in parts: | |
| part = part.strip() | |
| if not part: | |
| continue | |
| e, p = compel(part) | |
| embeds_list.append(e) | |
| pooled_list.append(p) | |
| if not embeds_list: | |
| return compel("") | |
| return torch.cat(embeds_list, dim=1), pooled_list[0] | |
| with torch.no_grad(): | |
| pc, pp = encode_with_break(pos) | |
| nc, np_pool = encode_with_break(neg) | |
| try: | |
| padded = compel.pad_conditioning_tensors_to_same_length([pc, nc]) | |
| pc, nc = padded[0], padded[1] | |
| except (AttributeError, RuntimeError, ValueError) as pad_err: | |
| print(f"⚠️ compel padding failed ({pad_err}), using manual fallback") | |
| pc, nc = _manual_pad_conditioning(pc, nc) | |
| print(f"✅ Manual padding applied: pc={list(pc.shape)}, nc={list(nc.shape)}") | |
| res = (pc, pp, nc, np_pool) | |
| CONDITIONING_CACHE[cache_key] = res | |
| return res | |
| base_config = dict(pipe.scheduler.config) | |
| SAMPLER_REGISTRY = { | |
| "Euler": { | |
| "class": EulerDiscreteScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "Euler a": { | |
| "class": EulerAncestralDiscreteScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "Heun": { | |
| "class": HeunDiscreteScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "LMS": { | |
| "class": LMSDiscreteScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta"], | |
| "tag": "general", | |
| }, | |
| "DPM2": { | |
| "class": KDPM2DiscreteScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta"], | |
| "tag": "general", | |
| }, | |
| "DPM2 a": { | |
| "class": KDPM2AncestralDiscreteScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta"], | |
| "tag": "general", | |
| }, | |
| "DPM++ 2M": { | |
| "class": DPMSolverMultistepScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "DPM++ 2M SDE": { | |
| "class": DPMSolverMultistepScheduler, | |
| "params": {"algorithm_type": "sde-dpmsolver++"}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "DPM++ 3M": { | |
| "class": DPMSolverMultistepScheduler, | |
| "params": {"solver_order": 3}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "DPM++ 3M SDE": { | |
| "class": DPMSolverMultistepScheduler, | |
| "params": {"solver_order": 3, "algorithm_type": "sde-dpmsolver++"}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "DPM++ SDE": { | |
| "class": DPMSolverSDEScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta"], | |
| "tag": "general", | |
| }, | |
| "EDM DPM++ 2M": { | |
| "class": EDMDPMSolverMultistepScheduler, | |
| "params": {}, | |
| "schedules": ["Karras"], | |
| "tag": "general", | |
| }, | |
| "DDIM": { | |
| "class": DDIMScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Beta"], | |
| "tag": "general", | |
| }, | |
| "PNDM": { | |
| "class": PNDMScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Beta"], | |
| "tag": "general", | |
| }, | |
| "DDPM": { | |
| "class": DDPMScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Beta"], | |
| "tag": "general", | |
| }, | |
| "UniPC": { | |
| "class": UniPCMultistepScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta", "SGM Uniform"], | |
| "tag": "general", | |
| }, | |
| "DEIS": { | |
| "class": DEISMultistepScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta"], | |
| "tag": "general", | |
| }, | |
| "SA Solver": { | |
| "class": SASolverScheduler, | |
| "params": {}, | |
| "schedules": ["Normal", "Karras", "Exponential", "Polyexponential", "Beta"], | |
| "tag": "general", | |
| }, | |
| "LCM": { | |
| "class": LCMScheduler, | |
| "params": {}, | |
| "schedules": ["Fixed"], | |
| "tag": "special", | |
| }, | |
| "TCD": { | |
| "class": TCDScheduler, | |
| "params": {}, | |
| "schedules": ["Fixed"], | |
| "tag": "special", | |
| }, | |
| } | |
| SCHEDULE_REGISTRY = { | |
| "Normal": { | |
| "params": {}, | |
| "description": "Default linear sigma schedule", | |
| }, | |
| "Karras": { | |
| "params": {"use_karras_sigmas": True}, | |
| "description": "Karras noise schedule", | |
| }, | |
| "Exponential": { | |
| "params": {"use_exponential_sigmas": True}, | |
| "description": "Exponential decay", | |
| }, | |
| "Polyexponential": { | |
| "params": {"use_polyexponential_sigmas": True}, | |
| "description": "Poly-exponential hybrid", | |
| }, | |
| "Beta": { | |
| "params": {"use_beta_sigmas": True}, | |
| "description": "Beta distribution", | |
| }, | |
| "SGM Uniform": { | |
| "params": {"timestep_spacing": "leading"}, | |
| "description": "SGM uniform spacing", | |
| }, | |
| "Fixed": { | |
| "params": {}, | |
| "description": "Fixed schedule (LCM/TCD)", | |
| }, | |
| } | |
| def build_scheduler(sampler_name, schedule_name): | |
| if sampler_name not in SAMPLER_REGISTRY: | |
| return None, f"Sampler '{sampler_name}' tidak ditemukan!" | |
| if schedule_name not in SCHEDULE_REGISTRY: | |
| return None, f"Schedule '{schedule_name}' tidak ditemukan!" | |
| sampler_info = SAMPLER_REGISTRY[sampler_name] | |
| schedule_info = SCHEDULE_REGISTRY[schedule_name] | |
| if schedule_name not in sampler_info["schedules"]: | |
| compatible = ", ".join(sampler_info["schedules"]) | |
| return None, f"'{schedule_name}' tidak compatible dengan '{sampler_name}'. Compatible: {compatible}" | |
| merged_config = dict(base_config) | |
| merged_config.update(sampler_info["params"]) | |
| merged_config.update(schedule_info["params"]) | |
| try: | |
| return sampler_info["class"].from_config(merged_config), None | |
| except Exception as e: | |
| return None, f"Gagal membangun scheduler: {str(e)}" | |
| default_scheduler, default_scheduler_error = build_scheduler("Euler a", "Normal") | |
| if default_scheduler is not None: | |
| pipe.scheduler = default_scheduler | |
| TOTAL_COMBOS = sum(len(info["schedules"]) for info in SAMPLER_REGISTRY.values()) | |
| print( | |
| f"✅ SAMPLER SYSTEM: {len(SAMPLER_REGISTRY)} samplers × {len(SCHEDULE_REGISTRY)} schedules = {TOTAL_COMBOS} combos" | |
| ) | |
| PAG_PIPE = None | |
| def get_pag_pipeline(): | |
| global PAG_PIPE | |
| if not PAG_AVAILABLE: | |
| return None | |
| if PAG_PIPE is None: | |
| try: | |
| PAG_PIPE = StableDiffusionXLPAGPipeline( | |
| vae=pipe.vae, | |
| text_encoder=pipe.text_encoder, | |
| text_encoder_2=pipe.text_encoder_2, | |
| tokenizer=pipe.tokenizer, | |
| tokenizer_2=pipe.tokenizer_2, | |
| unet=pipe.unet, | |
| scheduler=pipe.scheduler, | |
| ) | |
| print("✅ PAG Pipeline berhasil dibuat!") | |
| except Exception as e: | |
| print(f"⚠️ Gagal membuat PAG Pipeline: {e}") | |
| return None | |
| PAG_PIPE.scheduler = pipe.scheduler | |
| return PAG_PIPE | |
| CFG_DECAY_CURVES = ["linear", "cosine", "exponential", "step"] | |
| def _compute_cfg(progress, cfg_start, cfg_end, curve_type): | |
| if curve_type == "linear": | |
| return cfg_start + (cfg_end - cfg_start) * progress | |
| if curve_type == "cosine": | |
| return cfg_end + (cfg_start - cfg_end) * (1 + math.cos(math.pi * progress)) / 2 | |
| if curve_type == "exponential": | |
| if cfg_start > 0 and cfg_end > 0: | |
| return cfg_start * (cfg_end / cfg_start) ** progress | |
| return cfg_start + (cfg_end - cfg_start) * progress | |
| if curve_type == "step": | |
| return cfg_start if progress < 0.5 else cfg_end | |
| return cfg_start + (cfg_end - cfg_start) * progress | |
| def build_cfg_decay_callback(cfg_start, cfg_end, curve_type, total_steps): | |
| def callback(pipeline, step_index, timestep, callback_kwargs): | |
| progress = step_index / max(total_steps - 1, 1) | |
| pipeline._guidance_scale = _compute_cfg(progress, cfg_start, cfg_end, curve_type) | |
| return callback_kwargs | |
| return callback | |
| print(f"✅ GUIDANCE: CFG Rescale ✅ | CFG Decay ✅ | PAG {'✅' if PAG_AVAILABLE else '❌'}") | |
| LORA_DB = {} | |
| ACTIVE_LORA_ADAPTER = None | |
| ACTIVE_LORA_SCALE = 0.0 | |
| GPU_DURATION_TIERS = [10, 15, 30, 60, 120] | |
| GPU_WRAPPED_FNS = {} | |
| def estimate_gpu_duration(steps, batch_size, width, height, enable_pag, enable_cfg_decay): | |
| megapixels = (width * height) / (1024 * 1024) | |
| base_time = megapixels * steps * 0.12 * batch_size | |
| if enable_pag: | |
| base_time *= 1.4 | |
| if enable_cfg_decay: | |
| base_time *= 1.05 | |
| estimated = base_time * 1.3 | |
| for dur in GPU_DURATION_TIERS: | |
| if estimated <= dur: | |
| return dur, f"~{base_time:.1f}s est → {dur}s tier" | |
| return GPU_DURATION_TIERS[-1], f"~{base_time:.1f}s est → {GPU_DURATION_TIERS[-1]}s (MAX)" | |
| def encode_images_to_base64(images: list, seed: int, quality: int = 92) -> list: | |
| results = [] | |
| for i, img in enumerate(images): | |
| buf = BytesIO() | |
| img.save(buf, format="JPEG", quality=quality) | |
| img_bytes = buf.getvalue() | |
| b64_str = base64.b64encode(img_bytes).decode("utf-8") | |
| results.append( | |
| { | |
| "data": f"data:image/jpeg;base64,{b64_str}", | |
| "name": f"ffd_{seed}_{i}.jpg", | |
| "seed": seed, | |
| "index": i, | |
| "width": img.width, | |
| "height": img.height, | |
| "bytes": len(img_bytes), | |
| } | |
| ) | |
| return results | |
| RAM_GALLERIES = {} | |
| RAM_GALLERY_LOCK = threading.RLock() | |
| RAM_GALLERY_START_TIME = time.time() | |
| RAM_GALLERY_MAX_GALLERIES = 100 | |
| RAM_GALLERY_MAX_IMAGES_PER_GALLERY = 1000 | |
| RAM_GALLERY_MAX_TOTAL_IMAGES = 10000 | |
| RAM_GALLERY_MAX_NAME_LENGTH = 80 | |
| RAM_GALLERY_MAX_DESC_LENGTH = 500 | |
| def _sanitize_gallery_name(name: str) -> str: | |
| name = "" if name is None else str(name) | |
| name = name.strip() | |
| if not name: | |
| return "" | |
| name = name.replace("/", "-").replace("\\", "-") | |
| name = re.sub(r'[<>:"|?#&%*\x00-\x1f]', "", name).strip() | |
| name = re.sub(r"\s+", " ", name) | |
| return name | |
| def _estimate_base64_bytes(data_url: str) -> int: | |
| if not data_url or "," not in data_url: | |
| return 0 | |
| b64_part = data_url.split(",", 1)[-1] | |
| return int(len(b64_part) * 3 / 4) | |
| def _get_total_image_count(): | |
| total = 0 | |
| for g in RAM_GALLERIES.values(): | |
| total += len(g["images"]) | |
| return total | |
| def _get_total_ram_bytes(): | |
| total = 0 | |
| for g in RAM_GALLERIES.values(): | |
| for img in g["images"]: | |
| total += img.get("bytes", 0) | |
| return total | |
| def _gallery_summary(gallery, include_images=False): | |
| total_bytes = sum(img.get("bytes", 0) for img in gallery["images"]) | |
| summary = { | |
| "id": gallery["id"], | |
| "name": gallery["name"], | |
| "description": gallery["description"], | |
| "image_count": len(gallery["images"]), | |
| "total_bytes": total_bytes, | |
| "total_mb": round(total_bytes / (1024 * 1024), 2), | |
| "created_at": gallery["created_at"], | |
| "last_modified": gallery["last_modified"], | |
| } | |
| if include_images: | |
| summary["images"] = gallery["images"] | |
| return summary | |
| def ram_gallery_create(name: str, description: str = ""): | |
| name = _sanitize_gallery_name(name) | |
| description = "" if description is None else str(description) | |
| description = description.strip() | |
| if not name: | |
| return False, "Gallery name cannot be empty", None | |
| if len(name) > RAM_GALLERY_MAX_NAME_LENGTH: | |
| return False, f"Name too long (max {RAM_GALLERY_MAX_NAME_LENGTH} chars)", None | |
| if len(description) > RAM_GALLERY_MAX_DESC_LENGTH: | |
| description = description[:RAM_GALLERY_MAX_DESC_LENGTH] | |
| with RAM_GALLERY_LOCK: | |
| if name in RAM_GALLERIES: | |
| return False, f"Gallery '{name}' already exists", None | |
| if len(RAM_GALLERIES) >= RAM_GALLERY_MAX_GALLERIES: | |
| return False, f"Max galleries limit reached ({RAM_GALLERY_MAX_GALLERIES})", None | |
| gallery_id = str(uuid.uuid4())[:8] | |
| now = time.strftime("%Y-%m-%d %H:%M:%S") | |
| RAM_GALLERIES[name] = { | |
| "id": gallery_id, | |
| "name": name, | |
| "description": description, | |
| "images": [], | |
| "created_at": now, | |
| "last_modified": now, | |
| } | |
| print(f"🏞️ Gallery created: '{name}' (id: {gallery_id})") | |
| return True, f"Gallery '{name}' created", gallery_id | |
| def ram_gallery_add_images(gallery_name: str, images: list): | |
| gallery_name = _sanitize_gallery_name(gallery_name) | |
| with RAM_GALLERY_LOCK: | |
| if gallery_name not in RAM_GALLERIES: | |
| return False, f"Gallery '{gallery_name}' not found", 0 | |
| gallery = RAM_GALLERIES[gallery_name] | |
| current_count = len(gallery["images"]) | |
| total_count = _get_total_image_count() | |
| new_count = len(images) | |
| if current_count + new_count > RAM_GALLERY_MAX_IMAGES_PER_GALLERY: | |
| allowed = RAM_GALLERY_MAX_IMAGES_PER_GALLERY - current_count | |
| if allowed <= 0: | |
| return False, f"Gallery '{gallery_name}' is full ({RAM_GALLERY_MAX_IMAGES_PER_GALLERY} images max)", 0 | |
| images = images[:allowed] | |
| new_count = allowed | |
| if total_count + new_count > RAM_GALLERY_MAX_TOTAL_IMAGES: | |
| allowed = RAM_GALLERY_MAX_TOTAL_IMAGES - total_count | |
| if allowed <= 0: | |
| return False, f"Total gallery limit reached ({RAM_GALLERY_MAX_TOTAL_IMAGES} images)", 0 | |
| images = images[:allowed] | |
| new_count = allowed | |
| added = 0 | |
| for img_data in images: | |
| img_id = str(uuid.uuid4())[:12] | |
| img_entry = { | |
| "id": img_id, | |
| "data": img_data.get("data", ""), | |
| "seed": img_data.get("seed", "N/A"), | |
| "sampler": img_data.get("sampler", "N/A"), | |
| "schedule": img_data.get("schedule", "N/A"), | |
| "steps": img_data.get("steps", 0), | |
| "cfg_scale": img_data.get("cfg_scale", 0), | |
| "width": img_data.get("width", 0), | |
| "height": img_data.get("height", 0), | |
| "prompt": img_data.get("prompt", ""), | |
| "negative_prompt": img_data.get("negative_prompt", ""), | |
| "duration": img_data.get("duration", 0), | |
| "pipeline": img_data.get("pipeline", "N/A"), | |
| "guidance": img_data.get("guidance", []), | |
| "lora": img_data.get("lora", "None"), | |
| "lora_scale": img_data.get("lora_scale", 0), | |
| "bytes": _estimate_base64_bytes(img_data.get("data", "")), | |
| "added_at": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| } | |
| gallery["images"].append(img_entry) | |
| added += 1 | |
| gallery["last_modified"] = time.strftime("%Y-%m-%d %H:%M:%S") | |
| total_bytes = sum(img.get("bytes", 0) for img in gallery["images"]) | |
| print( | |
| f"🏞️ +{added} images → '{gallery_name}' " | |
| f"(total: {len(gallery['images'])}, {total_bytes // (1024 * 1024)}MB)" | |
| ) | |
| return True, f"Added {added} images to '{gallery_name}'", added | |
| def ram_gallery_list(): | |
| with RAM_GALLERY_LOCK: | |
| galleries = [] | |
| for name, gallery in RAM_GALLERIES.items(): | |
| galleries.append(_gallery_summary(gallery, include_images=False)) | |
| return galleries | |
| def ram_gallery_get_images(gallery_name: str): | |
| gallery_name = _sanitize_gallery_name(gallery_name) | |
| with RAM_GALLERY_LOCK: | |
| if gallery_name not in RAM_GALLERIES: | |
| return None, f"Gallery '{gallery_name}' not found" | |
| gallery = RAM_GALLERIES[gallery_name] | |
| return _gallery_summary(gallery, include_images=True), None | |
| def ram_gallery_delete(gallery_name: str): | |
| gallery_name = _sanitize_gallery_name(gallery_name) | |
| with RAM_GALLERY_LOCK: | |
| if gallery_name not in RAM_GALLERIES: | |
| return False, f"Gallery '{gallery_name}' not found" | |
| count = len(RAM_GALLERIES[gallery_name]["images"]) | |
| del RAM_GALLERIES[gallery_name] | |
| print(f"🏞️ Gallery deleted: '{gallery_name}' ({count} images removed)") | |
| return True, f"Gallery '{gallery_name}' deleted ({count} images removed)" | |
| def ram_gallery_delete_image(gallery_name: str, image_id: str): | |
| gallery_name = _sanitize_gallery_name(gallery_name) | |
| image_id = str(image_id).strip() | |
| with RAM_GALLERY_LOCK: | |
| if gallery_name not in RAM_GALLERIES: | |
| return False, f"Gallery '{gallery_name}' not found" | |
| gallery = RAM_GALLERIES[gallery_name] | |
| original_count = len(gallery["images"]) | |
| gallery["images"] = [img for img in gallery["images"] if img["id"] != image_id] | |
| if len(gallery["images"]) == original_count: | |
| return False, f"Image '{image_id}' not found in '{gallery_name}'" | |
| gallery["last_modified"] = time.strftime("%Y-%m-%d %H:%M:%S") | |
| print(f"🏞️ Image removed: '{image_id}' from '{gallery_name}'") | |
| return True, f"Image '{image_id}' removed from '{gallery_name}'" | |
| def ram_gallery_rename(old_name: str, new_name: str): | |
| old_name = _sanitize_gallery_name(unquote(str(old_name))) | |
| new_name = _sanitize_gallery_name(new_name) | |
| if not new_name: | |
| return False, "New name cannot be empty" | |
| if len(new_name) > RAM_GALLERY_MAX_NAME_LENGTH: | |
| return False, f"Name too long (max {RAM_GALLERY_MAX_NAME_LENGTH} chars)" | |
| with RAM_GALLERY_LOCK: | |
| if old_name not in RAM_GALLERIES: | |
| return False, f"Gallery '{old_name}' not found" | |
| if new_name in RAM_GALLERIES: | |
| return False, f"Gallery '{new_name}' already exists" | |
| gallery = RAM_GALLERIES.pop(old_name) | |
| gallery["name"] = new_name | |
| gallery["last_modified"] = time.strftime("%Y-%m-%d %H:%M:%S") | |
| RAM_GALLERIES[new_name] = gallery | |
| print(f"🏞️ Gallery renamed: '{old_name}' → '{new_name}'") | |
| return True, f"Gallery renamed to '{new_name}'" | |
| def ram_gallery_stats(): | |
| with RAM_GALLERY_LOCK: | |
| total_galleries = len(RAM_GALLERIES) | |
| total_images = _get_total_image_count() | |
| total_bytes = _get_total_ram_bytes() | |
| uptime = time.time() - RAM_GALLERY_START_TIME | |
| uptime_hours = uptime / 3600 | |
| sys_ram_used_mb = 0 | |
| try: | |
| import resource | |
| sys_ram_used_mb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 | |
| except Exception: | |
| pass | |
| gpu_allocated = "N/A" | |
| gpu_reserved = "N/A" | |
| try: | |
| gpu_allocated = f"{torch.cuda.memory_allocated() / (1024 ** 3):.2f} GB" | |
| gpu_reserved = f"{torch.cuda.memory_reserved() / (1024 ** 3):.2f} GB" | |
| except Exception: | |
| pass | |
| gallery_details = [] | |
| for name, gallery in RAM_GALLERIES.items(): | |
| gallery_details.append( | |
| { | |
| "name": name, | |
| "id": gallery["id"], | |
| "image_count": len(gallery["images"]), | |
| "total_mb": round( | |
| sum(img.get("bytes", 0) for img in gallery["images"]) / (1024 * 1024), | |
| 2, | |
| ), | |
| "created_at": gallery["created_at"], | |
| "last_modified": gallery["last_modified"], | |
| } | |
| ) | |
| return { | |
| "total_galleries": total_galleries, | |
| "max_galleries": RAM_GALLERY_MAX_GALLERIES, | |
| "total_images": total_images, | |
| "max_images_per_gallery": RAM_GALLERY_MAX_IMAGES_PER_GALLERY, | |
| "max_total_images": RAM_GALLERY_MAX_TOTAL_IMAGES, | |
| "total_ram_used_mb": round(total_bytes / (1024 * 1024), 2), | |
| "total_ram_used_gb": round(total_bytes / (1024 ** 3), 4), | |
| "process_rss_mb": round(sys_ram_used_mb, 1), | |
| "gpu_allocated": gpu_allocated, | |
| "gpu_reserved": gpu_reserved, | |
| "uptime_hours": round(uptime_hours, 2), | |
| "galleries": gallery_details, | |
| } | |
| print("🏞️ RAM Gallery System ready") | |
| print(f" Max galleries: {RAM_GALLERY_MAX_GALLERIES}") | |
| print(f" Max images/gallery: {RAM_GALLERY_MAX_IMAGES_PER_GALLERY}") | |
| print(f" Max total images: {RAM_GALLERY_MAX_TOTAL_IMAGES}") | |
| _retry_depth = {"count": 0} | |
| def generate_image( | |
| prompt, | |
| negative_prompt, | |
| sampler, | |
| schedule, | |
| steps, | |
| cfg_scale, | |
| seed, | |
| batch_size, | |
| width, | |
| height, | |
| lora_name, | |
| lora_scale, | |
| enable_cfg_rescale, | |
| cfg_rescale_value, | |
| enable_cfg_decay, | |
| cfg_decay_start, | |
| cfg_decay_end, | |
| cfg_decay_curve, | |
| enable_pag, | |
| pag_scale, | |
| token_budget, | |
| hf_token, | |
| ): | |
| global ACTIVE_LORA_ADAPTER, ACTIVE_LORA_SCALE | |
| _scheduler_hooked = False | |
| _original_scheduler_step = None | |
| active_pipe = pipe | |
| try: | |
| if hf_token and str(hf_token).strip(): | |
| os.environ["HF_TOKEN"] = str(hf_token).strip() | |
| safe_token = str(hf_token)[:8] + "..." | |
| print(f"🔑 Token from frontend: {safe_token}") | |
| else: | |
| safe_token = "none" | |
| print("⚠️ No token received from frontend") | |
| param_errors = validate_params(int(steps), int(batch_size), int(width), int(height)) | |
| if param_errors: | |
| err_text = f"Parameter Error: {' | '.join(param_errors)}" | |
| print(f"❌ {err_text}") | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": err_text, | |
| "images": [], | |
| "retryable": False, | |
| } | |
| ) | |
| steps = min(int(steps), MAX_STEPS) | |
| batch_size = min(int(batch_size), MAX_BATCH_SIZE) | |
| width = max(MIN_WIDTH, min(int(width), MAX_WIDTH)) | |
| height = max(MIN_HEIGHT, min(int(height), MAX_HEIGHT)) | |
| req_sampler, req_schedule = sampler, schedule | |
| if req_schedule == "Normal" and " " in req_sampler: | |
| parts = req_sampler.rsplit(" ", 1) | |
| if parts[-1] in SCHEDULE_REGISTRY: | |
| req_sampler, req_schedule = parts[0], parts[-1] | |
| scheduler, err = build_scheduler(req_sampler, req_schedule) | |
| if scheduler is not None: | |
| pipe.scheduler = scheduler | |
| sampler_used, schedule_used = req_sampler, req_schedule | |
| else: | |
| fallback_scheduler, _ = build_scheduler("Euler a", "Normal") | |
| if fallback_scheduler is not None: | |
| pipe.scheduler = fallback_scheduler | |
| sampler_used, schedule_used = "Euler a (Fallback)", "Normal" | |
| final_seed = random.randint(0, 2**32 - 1) if seed == -1 else int(seed) | |
| while "{" in prompt: | |
| prompt = re.sub( | |
| r"\{([^{}]*)\}", | |
| lambda m: random.choice(m.group(1).split("|")).strip(), | |
| prompt, | |
| ) | |
| base_cfg = float(cfg_scale) | |
| call_guidance_scale = float(cfg_decay_start) if enable_cfg_decay else base_cfg | |
| pag_active = enable_pag and PAG_AVAILABLE | |
| if pag_active: | |
| active_pipe = get_pag_pipeline() | |
| if active_pipe is None: | |
| active_pipe = pipe | |
| pag_active = False | |
| pipeline_name = "PAG" | |
| else: | |
| active_pipe = pipe | |
| pipeline_name = "Standard" | |
| guidance_summary = [] | |
| if enable_cfg_rescale: | |
| guidance_summary.append(f"Rescale={cfg_rescale_value}") | |
| if enable_cfg_decay: | |
| guidance_summary.append(f"Decay={cfg_decay_start}→{cfg_decay_end}({cfg_decay_curve})") | |
| else: | |
| guidance_summary.append(f"CFG={call_guidance_scale}") | |
| if pag_active: | |
| guidance_summary.append(f"PAG={pag_scale}") | |
| print(f"🎨 {sampler_used}+{schedule_used} | {' | '.join(guidance_summary)} | {pipeline_name}") | |
| lora_id = None | |
| req_lora_name = "None" | |
| match_n = re.search(r"LCSN:([a-zA-Z0-9_]+)", prompt, re.IGNORECASE) | |
| if match_n: | |
| lora_id = match_n.group(1).upper() | |
| prompt = re.sub(r"LCSN:[a-zA-Z0-9_]+", "", prompt, flags=re.IGNORECASE) | |
| match_w = re.search(r"LCS:([0-9]*\.?[0-9]+)", prompt, re.IGNORECASE) | |
| if match_w: | |
| lora_scale = float(match_w.group(1)) | |
| prompt = re.sub(r"LCS:[0-9]*\.?[0-9]+", "", prompt, flags=re.IGNORECASE) | |
| prompt = re.sub(r",\s*,", ",", prompt) | |
| prompt = re.sub(r"\s+", " ", prompt).strip(", ") | |
| if lora_id and lora_id in LORA_DB: | |
| target = LORA_DB[lora_id] | |
| req_lora_name = target["name"] | |
| if ACTIVE_LORA_ADAPTER != target["adapter"] or ACTIVE_LORA_SCALE != lora_scale: | |
| if ACTIVE_LORA_ADAPTER is not None: | |
| pipe.unfuse_lora() | |
| if ACTIVE_LORA_ADAPTER != target["adapter"]: | |
| pipe.unload_lora_weights() | |
| gc.collect() | |
| torch.cuda.empty_cache() | |
| if ACTIVE_LORA_ADAPTER != target["adapter"]: | |
| if os.path.exists(target["path"]): | |
| pipe.load_lora_weights(target["path"], adapter_name=target["adapter"]) | |
| pipe.fuse_lora(lora_scale=lora_scale) | |
| ACTIVE_LORA_ADAPTER = target["adapter"] | |
| ACTIVE_LORA_SCALE = lora_scale | |
| else: | |
| if ACTIVE_LORA_ADAPTER is not None: | |
| pipe.unfuse_lora() | |
| pipe.unload_lora_weights() | |
| gc.collect() | |
| torch.cuda.empty_cache() | |
| ACTIVE_LORA_ADAPTER = None | |
| ACTIVE_LORA_SCALE = 0.0 | |
| st = time.time() | |
| generator = torch.Generator("cuda").manual_seed(final_seed) | |
| with torch.inference_mode(): | |
| pc, pp, nc, np_pool = get_conditioning(prompt, negative_prompt) | |
| gen_kwargs = dict( | |
| prompt_embeds=pc, | |
| pooled_prompt_embeds=pp, | |
| negative_prompt_embeds=nc, | |
| negative_pooled_prompt_embeds=np_pool, | |
| num_inference_steps=int(steps), | |
| guidance_scale=call_guidance_scale, | |
| width=int(width), | |
| height=int(height), | |
| num_images_per_prompt=int(batch_size), | |
| generator=generator, | |
| ) | |
| if enable_cfg_rescale: | |
| gen_kwargs["guidance_rescale"] = float(cfg_rescale_value) | |
| if pag_active: | |
| gen_kwargs["pag_scale"] = float(pag_scale) | |
| if enable_cfg_decay: | |
| if pag_active: | |
| _original_scheduler_step = active_pipe.scheduler.step | |
| _scheduler_hooked = True | |
| _step_counter = [0] | |
| def hooked_step(*args, **kwargs): | |
| result = _original_scheduler_step(*args, **kwargs) | |
| _step_counter[0] += 1 | |
| progress = _step_counter[0] / max(int(steps) - 1, 1) | |
| active_pipe._guidance_scale = _compute_cfg( | |
| progress, | |
| float(cfg_decay_start), | |
| float(cfg_decay_end), | |
| cfg_decay_curve, | |
| ) | |
| return result | |
| active_pipe.scheduler.step = hooked_step | |
| else: | |
| gen_kwargs["callback_on_step_end"] = build_cfg_decay_callback( | |
| float(cfg_decay_start), | |
| float(cfg_decay_end), | |
| cfg_decay_curve, | |
| int(steps), | |
| ) | |
| gen_kwargs["callback_on_step_end_tensor_kwargs"] = ["latents"] | |
| try: | |
| result = active_pipe(**gen_kwargs) | |
| images = result.images | |
| except TypeError as te: | |
| err_str = str(te) | |
| removed = [] | |
| for param in [ | |
| "pag_scale", | |
| "guidance_rescale", | |
| "callback_on_step_end", | |
| "callback_on_step_end_tensor_kwargs", | |
| ]: | |
| if param in err_str and param in gen_kwargs: | |
| gen_kwargs.pop(param) | |
| removed.append(param) | |
| if param == "callback_on_step_end": | |
| gen_kwargs.pop("callback_on_step_end_tensor_kwargs", None) | |
| if removed: | |
| print(f"⚠️ Fallback: removed {removed}") | |
| if "pag_scale" in removed: | |
| pag_active = False | |
| active_pipe = pipe | |
| result = active_pipe(**gen_kwargs) | |
| images = result.images | |
| else: | |
| raise | |
| finally: | |
| if _scheduler_hooked and _original_scheduler_step is not None: | |
| try: | |
| active_pipe.scheduler.step = _original_scheduler_step | |
| except Exception: | |
| pass | |
| duration = time.time() - st | |
| print(f"✅ Generated in {duration:.2f}s | Seed: {final_seed} | Token: {safe_token}") | |
| _retry_depth["count"] = 0 | |
| base64_data = encode_images_to_base64(images, final_seed) | |
| gpu_mem = "N/A" | |
| try: | |
| allocated = torch.cuda.memory_allocated() / (1024 ** 3) | |
| gpu_mem = f"{allocated:.1f}GB" | |
| except Exception: | |
| pass | |
| response_data = { | |
| "success": True, | |
| "images": base64_data, | |
| "metadata": { | |
| "seed": final_seed, | |
| "sampler": sampler_used, | |
| "schedule": schedule_used, | |
| "steps": steps, | |
| "cfg_scale": call_guidance_scale, | |
| "width": width, | |
| "height": height, | |
| "batch_size": batch_size, | |
| "duration": round(duration, 2), | |
| "pipeline": pipeline_name, | |
| "guidance": guidance_summary, | |
| "lora": req_lora_name, | |
| "lora_scale": ACTIVE_LORA_SCALE if ACTIVE_LORA_ADAPTER else 0.0, | |
| "token": safe_token, | |
| "gpu_memory": gpu_mem, | |
| "prompt": prompt, | |
| "negative_prompt": negative_prompt, | |
| }, | |
| } | |
| result_json = json.dumps(response_data) | |
| print(f"✅ Encoded {len(base64_data)} images ({sum(d['bytes'] for d in base64_data) // 1024}KB)") | |
| return result_json | |
| except Exception as e: | |
| print(f"❌ ERROR: {str(e)[:200]}") | |
| traceback.print_exc() | |
| err_msg = str(e) | |
| if is_gpu_cold_start_error(err_msg): | |
| print("❄️ GPU Cold Start - waiting 10s...") | |
| time.sleep(10) | |
| if _retry_depth["count"] < 2: | |
| _retry_depth["count"] += 1 | |
| print(f"🔄 Cold start retry #{_retry_depth['count']}/2...") | |
| return generate_image( | |
| prompt, | |
| negative_prompt, | |
| sampler, | |
| schedule, | |
| steps, | |
| cfg_scale, | |
| seed, | |
| batch_size, | |
| width, | |
| height, | |
| lora_name, | |
| lora_scale, | |
| enable_cfg_rescale, | |
| cfg_rescale_value, | |
| enable_cfg_decay, | |
| cfg_decay_start, | |
| cfg_decay_end, | |
| cfg_decay_curve, | |
| enable_pag, | |
| pag_scale, | |
| token_budget, | |
| hf_token, | |
| ) | |
| _retry_depth["count"] = 0 | |
| lower = err_msg.lower() | |
| retryable = any( | |
| kw in lower | |
| for kw in [ | |
| "gpu", | |
| "cuda", | |
| "oom", | |
| "memory", | |
| "timeout", | |
| "cold start", | |
| "not ready", | |
| "unavailable", | |
| "quota", | |
| "rate limit", | |
| ] | |
| ) | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": err_msg[:500], | |
| "images": [], | |
| "retryable": retryable, | |
| } | |
| ) | |
| finally: | |
| if "HF_TOKEN" in os.environ: | |
| del os.environ["HF_TOKEN"] | |
| def smart_generate( | |
| prompt, | |
| negative_prompt, | |
| sampler, | |
| schedule, | |
| steps, | |
| cfg_scale, | |
| seed, | |
| batch_size, | |
| width, | |
| height, | |
| lora_name, | |
| lora_scale, | |
| enable_cfg_rescale, | |
| cfg_rescale_value, | |
| enable_cfg_decay, | |
| cfg_decay_start, | |
| cfg_decay_end, | |
| cfg_decay_curve, | |
| enable_pag, | |
| pag_scale, | |
| token_budget, | |
| hf_token, | |
| ): | |
| try: | |
| param_errors = validate_params(int(steps), int(batch_size), int(width), int(height)) | |
| if param_errors: | |
| err_text = f"Parameter Error: {' | '.join(param_errors)}" | |
| print(f"❌ {err_text}") | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": err_text, | |
| "images": [], | |
| "retryable": False, | |
| } | |
| ) | |
| if token_budget == "Auto": | |
| actual_duration, est_text = estimate_gpu_duration( | |
| int(steps), | |
| int(batch_size), | |
| int(width), | |
| int(height), | |
| enable_pag, | |
| enable_cfg_decay, | |
| ) | |
| print(f"⏱️ Auto Estimate: {est_text}") | |
| else: | |
| actual_duration = int(str(token_budget).replace("s", "")) | |
| if actual_duration not in GPU_DURATION_TIERS: | |
| actual_duration = min(GPU_DURATION_TIERS, key=lambda x: abs(x - actual_duration)) | |
| if ZERO_GPU and actual_duration in GPU_WRAPPED_FNS: | |
| print(f"🚀 Using GPU wrapper: {actual_duration}s") | |
| try: | |
| result = GPU_WRAPPED_FNS[actual_duration]( | |
| prompt, | |
| negative_prompt, | |
| sampler, | |
| schedule, | |
| steps, | |
| cfg_scale, | |
| seed, | |
| batch_size, | |
| width, | |
| height, | |
| lora_name, | |
| lora_scale, | |
| enable_cfg_rescale, | |
| cfg_rescale_value, | |
| enable_cfg_decay, | |
| cfg_decay_start, | |
| cfg_decay_end, | |
| cfg_decay_curve, | |
| enable_pag, | |
| pag_scale, | |
| actual_duration, | |
| hf_token, | |
| ) | |
| if result is None: | |
| print("❌ GPU wrapper returned None!") | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": "GPU wrapper returned None — GPU may be unavailable or quota exceeded", | |
| "images": [], | |
| "retryable": True, | |
| } | |
| ) | |
| if not isinstance(result, str): | |
| print(f"❌ GPU wrapper returned non-string: {type(result)}") | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": f"GPU wrapper returned {type(result).__name__} instead of JSON string", | |
| "images": [], | |
| "retryable": True, | |
| } | |
| ) | |
| try: | |
| parsed = json.loads(result) | |
| if not isinstance(parsed, dict): | |
| print(f"❌ Result is not dict: {type(parsed)}") | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": f"Invalid result format: {type(parsed).__name__}", | |
| "images": [], | |
| "retryable": True, | |
| } | |
| ) | |
| except json.JSONDecodeError as je: | |
| print(f"❌ Result is not valid JSON: {je}") | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": f"Backend returned invalid JSON: {str(je)}", | |
| "images": [], | |
| "retryable": True, | |
| } | |
| ) | |
| return result | |
| except Exception as gpu_err: | |
| err_msg = str(gpu_err) | |
| print(f"❌ GPU WRAPPER EXCEPTION: {err_msg}") | |
| traceback.print_exc() | |
| lower = err_msg.lower() | |
| is_cold_start = any( | |
| kw in lower | |
| for kw in [ | |
| "no gpu", | |
| "gpu was available", | |
| "available after", | |
| "no gpu was", | |
| "gpu is not available", | |
| "not ready", | |
| "cold start", | |
| "gpu timeout", | |
| "quota", | |
| "rate limit", | |
| "exceeded", | |
| "too many", | |
| "sleeping", | |
| "waking up", | |
| ] | |
| ) | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": f"GPU {'Cold Start' if is_cold_start else 'Error'}: {err_msg[:300]}", | |
| "images": [], | |
| "retryable": True, | |
| } | |
| ) | |
| else: | |
| return generate_image( | |
| prompt, | |
| negative_prompt, | |
| sampler, | |
| schedule, | |
| steps, | |
| cfg_scale, | |
| seed, | |
| batch_size, | |
| width, | |
| height, | |
| lora_name, | |
| lora_scale, | |
| enable_cfg_rescale, | |
| cfg_rescale_value, | |
| enable_cfg_decay, | |
| cfg_decay_start, | |
| cfg_decay_end, | |
| cfg_decay_curve, | |
| enable_pag, | |
| pag_scale, | |
| actual_duration, | |
| hf_token, | |
| ) | |
| except Exception as outer_err: | |
| err_msg = str(outer_err) | |
| print(f"❌ OUTER EXCEPTION in smart_generate: {err_msg}") | |
| traceback.print_exc() | |
| return json.dumps( | |
| { | |
| "success": False, | |
| "error": f"Unexpected error: {err_msg[:300]}", | |
| "images": [], | |
| "retryable": True, | |
| } | |
| ) | |
| if ZERO_GPU: | |
| for dur in GPU_DURATION_TIERS: | |
| GPU_WRAPPED_FNS[dur] = spaces.GPU(duration=dur)(generate_image) | |
| print(f"✅ GPU wrappers created: {list(GPU_WRAPPED_FNS.keys())}") | |
| with gr.Blocks( | |
| title="FFD-XL-3.0 Backend API", | |
| theme=gr.themes.Soft(), | |
| css="footer { display: none !important; }", | |
| ) as demo: | |
| gr.Markdown("# 🚀 FFD-XL-3.0 Backend API (Simplified)\nReturns single JSON string.") | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| prompt_input = gr.Textbox( | |
| label="Prompt", | |
| lines=4, | |
| value="masterpiece, best quality, 1girl", | |
| ) | |
| negative_prompt_input = gr.Textbox( | |
| label="Negative Prompt", | |
| lines=2, | |
| value="lowres, bad anatomy, worst quality", | |
| ) | |
| with gr.Accordion("Settings", open=True): | |
| token_budget_dropdown = gr.Dropdown( | |
| label="Token Budget", | |
| choices=["Auto", "10s", "15s", "30s", "60s", "120s"], | |
| value="Auto", | |
| ) | |
| with gr.Row(): | |
| sampler_dropdown = gr.Dropdown( | |
| label="Sampler", | |
| choices=list(SAMPLER_REGISTRY.keys()), | |
| value="Euler a", | |
| scale=2, | |
| ) | |
| schedule_dropdown = gr.Dropdown( | |
| label="Schedule", | |
| choices=[ | |
| "Normal", | |
| "Karras", | |
| "Exponential", | |
| "Polyexponential", | |
| "Beta", | |
| "SGM Uniform", | |
| "Fixed", | |
| ], | |
| value="Normal", | |
| scale=1, | |
| ) | |
| with gr.Row(): | |
| steps_slider = gr.Slider( | |
| label="Steps", | |
| minimum=1, | |
| maximum=MAX_STEPS, | |
| value=28, | |
| step=1, | |
| ) | |
| cfg_slider = gr.Slider( | |
| label="CFG Scale", | |
| minimum=1.0, | |
| maximum=30.0, | |
| value=7.0, | |
| step=0.1, | |
| ) | |
| with gr.Row(): | |
| seed_input = gr.Number(label="Seed (-1=random)", value=-1, precision=0) | |
| batch_slider = gr.Slider( | |
| label="Batch", | |
| minimum=1, | |
| maximum=MAX_BATCH_SIZE, | |
| value=1, | |
| step=1, | |
| ) | |
| with gr.Row(): | |
| width_dropdown = gr.Dropdown( | |
| label="Width", | |
| choices=[ | |
| 512, | |
| 576, | |
| 640, | |
| 704, | |
| 768, | |
| 832, | |
| 896, | |
| 960, | |
| 1024, | |
| 1088, | |
| 1152, | |
| 1216, | |
| 1280, | |
| 1344, | |
| 1408, | |
| 1472, | |
| 1536, | |
| ], | |
| value=832, | |
| ) | |
| height_dropdown = gr.Dropdown( | |
| label="Height", | |
| choices=[ | |
| 512, | |
| 576, | |
| 640, | |
| 704, | |
| 768, | |
| 832, | |
| 896, | |
| 1024, | |
| 1088, | |
| 1152, | |
| 1216, | |
| 1280, | |
| 1344, | |
| 1408, | |
| 1472, | |
| 1536, | |
| ], | |
| value=1216, | |
| ) | |
| with gr.Accordion("Advanced", open=False): | |
| lora_dropdown = gr.Dropdown( | |
| label="LoRA", | |
| choices=["None"] + list(LORA_DB.keys()), | |
| value="None", | |
| ) | |
| lora_scale_slider = gr.Slider( | |
| label="LoRA Scale", | |
| minimum=0.0, | |
| maximum=2.0, | |
| value=1.0, | |
| step=0.05, | |
| ) | |
| with gr.Row(): | |
| enable_cfg_rescale = gr.Checkbox(label="CFG Rescale", value=False) | |
| cfg_rescale_slider = gr.Slider( | |
| label="Rescale", | |
| minimum=0.0, | |
| maximum=1.0, | |
| value=0.7, | |
| step=0.05, | |
| ) | |
| enable_cfg_decay = gr.Checkbox(label="CFG Decay", value=False) | |
| with gr.Row(): | |
| cfg_decay_start = gr.Slider( | |
| label="Start CFG", | |
| minimum=1.0, | |
| maximum=30.0, | |
| value=7.0, | |
| step=0.5, | |
| ) | |
| cfg_decay_end = gr.Slider( | |
| label="End CFG", | |
| minimum=1.0, | |
| maximum=30.0, | |
| value=2.0, | |
| step=0.5, | |
| ) | |
| cfg_decay_curve_dropdown = gr.Dropdown( | |
| label="Decay Curve", | |
| choices=CFG_DECAY_CURVES, | |
| value="cosine", | |
| ) | |
| with gr.Row(): | |
| enable_pag = gr.Checkbox( | |
| label="PAG", | |
| value=False, | |
| interactive=PAG_AVAILABLE, | |
| ) | |
| pag_scale_slider = gr.Slider( | |
| label="PAG Scale", | |
| minimum=0.0, | |
| maximum=15.0, | |
| value=3.0, | |
| step=0.5, | |
| ) | |
| hf_token_input = gr.Textbox(label="HF Token", visible=False, value="") | |
| generate_btn = gr.Button("🎨 Generate", variant="primary") | |
| with gr.Column(scale=3): | |
| json_output = gr.Textbox(label="Result JSON", lines=20, value="") | |
| def update_schedule_ui(sampler_name): | |
| if sampler_name in SAMPLER_REGISTRY: | |
| compatible = SAMPLER_REGISTRY[sampler_name]["schedules"] | |
| return gr.update(choices=compatible, value=compatible[0]) | |
| return gr.update() | |
| sampler_dropdown.change( | |
| fn=update_schedule_ui, | |
| inputs=[sampler_dropdown], | |
| outputs=[schedule_dropdown], | |
| ) | |
| generate_btn.click( | |
| fn=smart_generate, | |
| inputs=[ | |
| prompt_input, | |
| negative_prompt_input, | |
| sampler_dropdown, | |
| schedule_dropdown, | |
| steps_slider, | |
| cfg_slider, | |
| seed_input, | |
| batch_slider, | |
| width_dropdown, | |
| height_dropdown, | |
| lora_dropdown, | |
| lora_scale_slider, | |
| enable_cfg_rescale, | |
| cfg_rescale_slider, | |
| enable_cfg_decay, | |
| cfg_decay_start, | |
| cfg_decay_end, | |
| cfg_decay_curve_dropdown, | |
| enable_pag, | |
| pag_scale_slider, | |
| token_budget_dropdown, | |
| hf_token_input, | |
| ], | |
| outputs=[json_output], | |
| api_name="generate", | |
| ) | |
| demo.queue(max_size=20, default_concurrency_limit=1) | |
| print("✅ Gradio queue enabled (max_size=20, concurrency=1)") | |
| demo.app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=False, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| expose_headers=["*"], | |
| max_age=86400, | |
| ) | |
| print("✅ CORS: ALLOW ALL ORIGINS") | |
| @demo.app.exception_handler(StarletteHTTPException) | |
| async def api_http_exception_handler(request: Request, exc: StarletteHTTPException): | |
| status_code = getattr(exc, "status_code", 500) | |
| detail = getattr(exc, "detail", "HTTP error") | |
| return JSONResponse( | |
| content={ | |
| "success": False, | |
| "error": str(detail), | |
| "message": str(detail), | |
| "path": str(request.url.path), | |
| }, | |
| status_code=status_code, | |
| ) | |
| @demo.app.get("/health") | |
| def health_check(): | |
| gpu_mem = "N/A" | |
| try: | |
| allocated = torch.cuda.memory_allocated() / (1024 ** 3) | |
| reserved = torch.cuda.memory_reserved() / (1024 ** 3) | |
| gpu_mem = f"{allocated:.1f}GB / {reserved:.1f}GB" | |
| except Exception: | |
| pass | |
| with RAM_GALLERY_LOCK: | |
| gallery_count = len(RAM_GALLERIES) | |
| gallery_images = sum(len(g["images"]) for g in RAM_GALLERIES.values()) | |
| gallery_mb = round( | |
| sum(sum(img.get("bytes", 0) for img in g["images"]) for g in RAM_GALLERIES.values()) | |
| / (1024 * 1024), | |
| 2, | |
| ) | |
| return { | |
| "status": "online", | |
| "model": DISPLAY_NAME, | |
| "gpu": "RTX 6000 Blackwell (ZeroGPU)", | |
| "gpu_memory": gpu_mem, | |
| "zero_gpu": ZERO_GPU, | |
| "pag_available": PAG_AVAILABLE, | |
| "output_format": "Single JSON string {success, images, metadata}", | |
| "ram_gallery": { | |
| "enabled": True, | |
| "galleries": gallery_count, | |
| "total_images": gallery_images, | |
| "used_mb": gallery_mb, | |
| }, | |
| "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| } | |
| @demo.app.get("/api/config") | |
| def get_config_api(): | |
| return { | |
| "model": DISPLAY_NAME, | |
| "prediction_type": PREDICTION_TYPE, | |
| "clip_skip": CLIP_SKIP, | |
| "default_batch": DEFAULT_BATCH, | |
| "samplers": { | |
| n: { | |
| "schedules": i["schedules"], | |
| "tag": i["tag"], | |
| } | |
| for n, i in SAMPLER_REGISTRY.items() | |
| }, | |
| "schedules": { | |
| n: { | |
| "description": i["description"], | |
| } | |
| for n, i in SCHEDULE_REGISTRY.items() | |
| }, | |
| "cfg_decay_curves": CFG_DECAY_CURVES, | |
| "gpu_duration_tiers": GPU_DURATION_TIERS, | |
| "pag_available": PAG_AVAILABLE, | |
| "loras": [ | |
| { | |
| "id": k, | |
| "name": v["name"], | |
| } | |
| for k, v in LORA_DB.items() | |
| ] | |
| if LORA_DB | |
| else [], | |
| "limits": { | |
| "max_steps": MAX_STEPS, | |
| "max_batch_size": MAX_BATCH_SIZE, | |
| "max_width": MAX_WIDTH, | |
| "max_height": MAX_HEIGHT, | |
| "min_width": MIN_WIDTH, | |
| "min_height": MIN_HEIGHT, | |
| }, | |
| "ram_gallery": { | |
| "enabled": True, | |
| "max_galleries": RAM_GALLERY_MAX_GALLERIES, | |
| "max_images_per_gallery": RAM_GALLERY_MAX_IMAGES_PER_GALLERY, | |
| "max_total_images": RAM_GALLERY_MAX_TOTAL_IMAGES, | |
| }, | |
| "output_format": "Single JSON string", | |
| } | |
| async def _read_json(request: Request): | |
| try: | |
| return await request.json() | |
| except Exception: | |
| return {} | |
| def _ok(payload: dict, status_code: int = 200): | |
| base = { | |
| "success": True, | |
| "error": None, | |
| "message": "OK", | |
| } | |
| base.update(payload) | |
| return JSONResponse(content=base, status_code=status_code) | |
| def _err(message: str, status_code: int = 400, extra: dict = None): | |
| payload = { | |
| "success": False, | |
| "error": message, | |
| "message": message, | |
| } | |
| if extra: | |
| payload.update(extra) | |
| return JSONResponse(content=payload, status_code=status_code) | |
| @demo.app.get("/api/gallery/list") | |
| async def api_gallery_list(): | |
| try: | |
| galleries = ram_gallery_list() | |
| return _ok( | |
| { | |
| "galleries": galleries, | |
| "total": len(galleries), | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.get("/api/gallery/stats") | |
| async def api_gallery_stats(): | |
| try: | |
| stats = ram_gallery_stats() | |
| return _ok(stats) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.post("/api/gallery/create") | |
| async def api_gallery_create(request: Request): | |
| try: | |
| body = await _read_json(request) | |
| name = _sanitize_gallery_name(body.get("name", "")) | |
| description = str(body.get("description", "")).strip() | |
| if not name: | |
| return _err("Gallery name is required", 400) | |
| success, message, gallery_id = ram_gallery_create(name, description) | |
| if not success: | |
| return _err(message, 400, {"name": name}) | |
| return _ok( | |
| { | |
| "message": message, | |
| "gallery_id": gallery_id, | |
| "name": name, | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.post("/api/gallery/clear-all") | |
| async def api_gallery_clear_all(): | |
| try: | |
| with RAM_GALLERY_LOCK: | |
| count = len(RAM_GALLERIES) | |
| total_images = sum(len(g["images"]) for g in RAM_GALLERIES.values()) | |
| RAM_GALLERIES.clear() | |
| print(f"🏞️ ALL galleries cleared: {count} galleries, {total_images} images removed") | |
| return _ok( | |
| { | |
| "message": f"All galleries cleared ({count} galleries, {total_images} images)", | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.post("/api/gallery/{gallery_name}/add") | |
| async def api_gallery_add(request: Request, gallery_name: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| body = await _read_json(request) | |
| images = body.get("images", []) | |
| if not images or not isinstance(images, list): | |
| return _err("No images provided", 400) | |
| success, message, added = ram_gallery_add_images(gallery_name, images) | |
| if not success: | |
| return _err(message, 400, {"gallery_name": gallery_name}) | |
| summary, get_err = ram_gallery_get_images(gallery_name) | |
| if get_err: | |
| return _err(get_err, 404, {"gallery_name": gallery_name}) | |
| return _ok( | |
| { | |
| "message": message, | |
| "added": added, | |
| "gallery_name": gallery_name, | |
| "gallery_image_count": summary.get("image_count", 0) if summary else 0, | |
| "gallery_total_mb": summary.get("total_mb", 0) if summary else 0, | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.get("/api/gallery/{gallery_name}/images") | |
| async def api_gallery_images(gallery_name: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| result, error = ram_gallery_get_images(gallery_name) | |
| if error: | |
| return _err(error, 404, {"gallery_name": gallery_name}) | |
| images_meta = [] | |
| for img in result.get("images", []): | |
| meta = {k: v for k, v in img.items() if k != "data"} | |
| meta["has_data"] = True | |
| images_meta.append(meta) | |
| return _ok( | |
| { | |
| "name": result.get("name"), | |
| "description": result.get("description"), | |
| "image_count": result.get("image_count"), | |
| "total_mb": result.get("total_mb"), | |
| "created_at": result.get("created_at"), | |
| "last_modified": result.get("last_modified"), | |
| "images": images_meta, | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.get("/api/gallery/{gallery_name}/image/{image_id}") | |
| async def api_gallery_get_single_image(gallery_name: str, image_id: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| image_id = unquote(image_id).strip() | |
| with RAM_GALLERY_LOCK: | |
| if gallery_name not in RAM_GALLERIES: | |
| return _err(f"Gallery '{gallery_name}' not found", 404) | |
| gallery = RAM_GALLERIES[gallery_name] | |
| for img in gallery["images"]: | |
| if img["id"] == image_id: | |
| return _ok({"image": img}) | |
| return _err(f"Image '{image_id}' not found", 404) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.get("/api/gallery/{gallery_name}/download") | |
| async def api_gallery_download_all(gallery_name: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| result, error = ram_gallery_get_images(gallery_name) | |
| if error: | |
| return _err(error, 404, {"gallery_name": gallery_name}) | |
| return _ok( | |
| { | |
| "name": result.get("name"), | |
| "image_count": result.get("image_count"), | |
| "images": result.get("images", []), | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.delete("/api/gallery/{gallery_name}") | |
| async def api_gallery_delete(gallery_name: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| success, message = ram_gallery_delete(gallery_name) | |
| if not success: | |
| return _err(message, 404, {"gallery_name": gallery_name}) | |
| return _ok( | |
| { | |
| "message": message, | |
| "gallery_name": gallery_name, | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.delete("/api/gallery/{gallery_name}/image/{image_id}") | |
| async def api_gallery_delete_image(gallery_name: str, image_id: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| image_id = unquote(image_id).strip() | |
| success, message = ram_gallery_delete_image(gallery_name, image_id) | |
| if not success: | |
| return _err( | |
| message, | |
| 404, | |
| { | |
| "gallery_name": gallery_name, | |
| "image_id": image_id, | |
| }, | |
| ) | |
| return _ok( | |
| { | |
| "message": message, | |
| "gallery_name": gallery_name, | |
| "image_id": image_id, | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| @demo.app.post("/api/gallery/{gallery_name}/rename") | |
| async def api_gallery_rename(request: Request, gallery_name: str): | |
| try: | |
| gallery_name = _sanitize_gallery_name(unquote(gallery_name)) | |
| body = await _read_json(request) | |
| new_name = _sanitize_gallery_name(body.get("new_name", "")) | |
| if not new_name: | |
| return _err("New name is required", 400) | |
| success, message = ram_gallery_rename(gallery_name, new_name) | |
| if not success: | |
| return _err(message, 400, {"gallery_name": gallery_name}) | |
| return _ok( | |
| { | |
| "message": message, | |
| "old_name": gallery_name, | |
| "new_name": new_name, | |
| } | |
| ) | |
| except Exception as e: | |
| return _err(str(e), 500) | |
| print("") | |
| print("=" * 70) | |
| print("🚀 FFD-XL-3.0 BACKEND v2.0.1 — Simplified Mode + RAM Gallery") | |
| print(f" 📦 {DISPLAY_NAME} | bfloat16 | SDPA") | |
| print(f" 📋 {len(SAMPLER_REGISTRY)} samplers × {len(SCHEDULE_REGISTRY)} schedules = {TOTAL_COMBOS}") | |
| print(f" 🔮 PAG: {'✅' if PAG_AVAILABLE else '❌'}") | |
| print(f" 🎯 ZeroGPU: {'✅' if ZERO_GPU else '❌'}") | |
| print(" 📤 Output: Single JSON string") | |
| print(" 🛡️ Safe wrapper: Catch ALL exceptions") | |
| print( | |
| f" 🏞️ RAM Gallery: ✅ ({RAM_GALLERY_MAX_GALLERIES} max galleries, " | |
| f"{RAM_GALLERY_MAX_IMAGES_PER_GALLERY} img/gallery)" | |
| ) | |
| print(" 📡 Gallery API: /api/gallery/*") | |
| print("=" * 70) | |
| start_keep_alive() | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=False, | |
| show_error=True, | |
| ssr_mode=False, | |
| ) |