try: import spaces ZERO_GPU = True print("🖥️ ZeroGPU Runtime Detected") except ImportError: spaces = None ZERO_GPU = False print("🖥️ Dedicated GPU Runtime") # ===================================================================== # 📦 IMPORTS # ===================================================================== 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 # ===================================================================== # 🔮 PAG # ===================================================================== try: from diffusers import StableDiffusionXLPAGPipeline PAG_AVAILABLE = True print("🔮 PAG Pipeline tersedia!") except ImportError: PAG_AVAILABLE = False print("⚠️ PAG Pipeline tidak tersedia") # ===================================================================== # ⚡ GPU OPTIMIZATION # ===================================================================== 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 CONFIG # ===================================================================== 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 # ===================================================================== # 🔒 VALIDATION LIMITS # ===================================================================== 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 # ===================================================================== # 📊 ERROR DETECTION # ===================================================================== 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 ENGINE # ===================================================================== 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 # ===================================================================== # 📥 MODEL LOADING # ===================================================================== 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.") # ===================================================================== # 📥 PIPELINE LOADING # ===================================================================== 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") # ===================================================================== # 🔧 HOTFIX #1: compel bug — EmbeddingsProviderMulti missing 'empty_z' # ===================================================================== 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 # ===================================================================== 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 # ===================================================================== # 🎯 SAMPLER × SCHEDULE SYSTEM # ===================================================================== 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 PIPELINE MANAGER # ===================================================================== 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 # ===================================================================== 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 # ===================================================================== LORA_DB = {} ACTIVE_LORA_ADAPTER = None ACTIVE_LORA_SCALE = 0.0 # ===================================================================== # ⏱️ GPU TOKEN BUDGET SYSTEM # ===================================================================== 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)" # ===================================================================== # 🖼️ IMAGE ENCODING UTILITY # ===================================================================== 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 GALLERY SYSTEM v1.1 # ===================================================================== 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}") # ===================================================================== # 🎨 CORE GENERATION — RETURN ONLY JSON STRING # ===================================================================== _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"] # ===================================================================== # ⏱️ SMART GENERATE WRAPPER # ===================================================================== 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, } ) # ===================================================================== # 🔧 CREATE GPU WRAPPERS # ===================================================================== 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())}") # ===================================================================== # 🎨 GRADIO UI # ===================================================================== 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", ) # ===================================================================== # 🔧 QUEUE + CORS # ===================================================================== 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") # ===================================================================== # 🧯 GLOBAL API EXCEPTION HANDLER # ===================================================================== @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, ) # ===================================================================== # 📡 CUSTOM REST ENDPOINTS — HEALTH & CONFIG # ===================================================================== @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", } # ===================================================================== # 🏞️ RAM GALLERY REST API ENDPOINTS — FINAL FIXED # ===================================================================== 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) # ===================================================================== # 🚀 LAUNCH # ===================================================================== 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, )