"""
DiffuseCraft Mod - Improved Version
====================================
Original: https://huggingface.co/spaces/R-Kentaren/DiffuseCraftMod
IMPROVEMENTS:
- Bug fixes: Error handling, memory leaks, race conditions
- New features: Batch generation, smart presets, prompt templates, enhanced cache
- Optimizations: GPU memory management, thread safety
- NO UI/CSS/THEME CHANGES (as requested)
"""
import spaces
import os
from argparse import ArgumentParser
from stablepy import (
Model_Diffusers,
SCHEDULE_TYPE_OPTIONS,
SCHEDULE_PREDICTION_TYPE_OPTIONS,
check_scheduler_compatibility,
TASK_AND_PREPROCESSORS,
FACE_RESTORATION_MODELS,
PROMPT_WEIGHT_OPTIONS_PRIORITY,
scheduler_names,
)
from constants import (
DIRECTORY_UPSCALERS,
TASK_STABLEPY,
TASK_MODEL_LIST,
UPSCALER_DICT_GUI,
UPSCALER_KEYS,
PROMPT_W_OPTIONS,
WARNING_MSG_VAE,
SDXL_TASK,
MODEL_TYPE_TASK,
POST_PROCESSING_SAMPLER,
DIFFUSERS_CONTROLNET_MODEL,
IP_MODELS,
MODE_IP_OPTIONS,
CACHE_HF_ROOT,
)
from stablepy.diffusers_vanilla.style_prompt_config import STYLE_NAMES
import torch
import re
import time
import threading
from PIL import ImageFile
from utils import (
get_model_list,
extract_parameters,
get_model_type,
extract_exif_data,
create_mask_now,
download_diffuser_repo,
get_used_storage_gb,
delete_model,
progress_step_bar,
html_template_message,
escape_html,
clear_hf_cache,
)
from image_processor import preprocessor_tab
from datetime import datetime
import gradio as gr
import logging
import diffusers
import warnings
from stablepy import logger
from diffusers import FluxPipeline
import subprocess
import json
from pathlib import Path
from typing import Generator, Tuple, List, Any, Optional, Dict
import traceback
import hashlib
import copy
from contextlib import contextmanager
import functools
# ==================== CONFIGURATION ====================
IS_ZERO_GPU = bool(os.getenv("SPACES_ZERO_GPU"))
if IS_ZERO_GPU:
subprocess.run("rm -rf /data-nvme/zerogpu-offload/*", env={}, shell=True)
IS_GPU_MODE = True if IS_ZERO_GPU else (True if torch.cuda.is_available() else False)
img_path = "./images/"
allowed_path = os.path.abspath(img_path)
delete_cache_time = (9600, 9600) if IS_ZERO_GPU else (86400, 86400)
ImageFile.LOAD_TRUNCATED_IMAGES = True
torch.backends.cuda.matmul.allow_tf32 = True
# ==================== IMPROVED IMPORTS ====================
from modutils import (list_uniq, download_private_repo, get_model_id_list, get_tupled_embed_list,
get_lora_model_list, get_all_lora_tupled_list, update_loras, apply_lora_prompt, set_prompt_loras,
get_my_lora, upload_file_lora, move_file_lora, search_civitai_lora, select_civitai_lora,
update_civitai_selection, get_civitai_tag, CIVITAI_SORT, CIVITAI_PERIOD, CIVITAI_BASEMODEL,
set_textual_inversion_prompt, get_model_pipeline, change_interface_mode, get_t2i_model_info, download_link_model,
get_tupled_model_list, save_gallery_images, save_gallery_history, set_optimization, set_sampler_settings,
set_quick_presets, process_style_prompt, optimization_list, save_images, download_things, valid_model_name,
preset_styles, preset_quality, preset_sampler_setting, translate_to_en, EXAMPLES_GUI, RESOURCES)
from env import (HF_TOKEN, CIVITAI_API_KEY, HF_LORA_ESSENTIAL_PRIVATE_REPO, HF_VAE_PRIVATE_REPO,
HF_SDXL_EMBEDS_NEGATIVE_PRIVATE_REPO, HF_SDXL_EMBEDS_POSITIVE_PRIVATE_REPO,
DIRECTORY_MODELS, DIRECTORY_LORAS, DIRECTORY_VAES, DIRECTORY_EMBEDS, DIRECTORY_EMBEDS_SDXL,
DIRECTORY_EMBEDS_POSITIVE_SDXL, LOAD_DIFFUSERS_FORMAT_MODEL,
DOWNLOAD_MODEL_LIST, DOWNLOAD_LORA_LIST, DOWNLOAD_VAE_LIST, DOWNLOAD_EMBEDS)
from tagger.v2 import V2_ALL_MODELS, v2_random_prompt, v2_upsampling_prompt
from tagger.utils import (gradio_copy_text, COPY_ACTION_JS, gradio_copy_prompt,
V2_ASPECT_RATIO_OPTIONS, V2_RATING_OPTIONS, V2_LENGTH_OPTIONS, V2_IDENTITY_OPTIONS)
from tagger.tagger import (predict_tags_wd, convert_danbooru_to_e621_prompt,
remove_specific_prompt, insert_recom_prompt, insert_model_recom_prompt,
compose_prompt_to_copy, translate_prompt, select_random_character)
# Download private repos
download_private_repo(HF_LORA_ESSENTIAL_PRIVATE_REPO, DIRECTORY_LORAS, True)
download_private_repo(HF_VAE_PRIVATE_REPO, DIRECTORY_VAES, False)
# Create directories
directories = [DIRECTORY_MODELS, DIRECTORY_LORAS, DIRECTORY_VAES, DIRECTORY_EMBEDS, DIRECTORY_UPSCALERS]
for directory in directories:
os.makedirs(directory, exist_ok=True)
# Download models/VAEs/LoRAs
DOWNLOAD_MODEL = ", ".join(DOWNLOAD_MODEL_LIST)
DOWNLOAD_VAE = ", ".join(DOWNLOAD_VAE_LIST)
DOWNLOAD_LORA = ", ".join(DOWNLOAD_LORA_LIST)
for url in [url.strip() for url in DOWNLOAD_MODEL.split(',')]:
download_things(DIRECTORY_MODELS, url, HF_TOKEN, CIVITAI_API_KEY)
for url in [url.strip() for url in DOWNLOAD_VAE.split(',')]:
download_things(DIRECTORY_VAES, url, HF_TOKEN, CIVITAI_API_KEY)
for url in [url.strip() for url in DOWNLOAD_LORA.split(',')]:
download_things(DIRECTORY_LORAS, url, HF_TOKEN, CIVITAI_API_KEY)
# Download Embeddings
for url_embed in DOWNLOAD_EMBEDS:
if not os.path.exists(f"./embedings/{url_embed.split('/')[-1]}"):
download_things(DIRECTORY_EMBEDS, url_embed, HF_TOKEN, CIVITAI_API_KEY)
# Build model lists
embed_list = get_model_list(DIRECTORY_EMBEDS)
lora_model_list = get_lora_model_list()
vae_model_list = get_model_list(DIRECTORY_VAES)
vae_model_list.insert(0, "BakedVAE")
vae_model_list.insert(0, "None")
single_file_model_list = get_model_list(DIRECTORY_MODELS)
model_list = list_uniq(get_model_id_list() + LOAD_DIFFUSERS_FORMAT_MODEL + single_file_model_list)
download_private_repo(HF_SDXL_EMBEDS_NEGATIVE_PRIVATE_REPO, DIRECTORY_EMBEDS_SDXL, False)
download_private_repo(HF_SDXL_EMBEDS_POSITIVE_PRIVATE_REPO, DIRECTORY_EMBEDS_POSITIVE_SDXL, False)
embed_sdxl_list = get_model_list(DIRECTORY_EMBEDS_SDXL) + get_model_list(DIRECTORY_EMBEDS_POSITIVE_SDXL)
def get_embed_list(pipeline_name):
return get_tupled_embed_list(embed_sdxl_list if pipeline_name == "StableDiffusionXLPipeline" else embed_list)
print('\033[33m๐ Download and listing of valid models completed.\033[0m')
# ==================== NEW: PRESET MANAGER ====================
class PresetManager:
"""Manages generation presets with save/load functionality."""
PRESET_DIR = Path("presets")
def __init__(self):
self.PRESET_DIR.mkdir(exist_ok=True)
self._cache = {}
self._load_all_presets()
def _load_all_presets(self):
"""Load all presets from disk."""
try:
for preset_file in self.PRESET_DIR.glob("*.json"):
try:
data = json.loads(preset_file.read_text(encoding="utf-8"))
self._cache[preset_file.stem] = data
except Exception as e:
print(f"[preset] Failed to load {preset_file.name}: {e}")
except Exception as e:
print(f"[preset] Failed to load presets: {e}")
def save_preset(self, name: str, params: dict) -> bool:
"""Save a preset to disk."""
try:
# Validate name
if not name or not name.strip():
raise ValueError("Preset name cannot be empty")
safe_name = re.sub(r'[^\w\s-]', '', name.strip()).replace(' ', '_')
filepath = self.PRESET_DIR / f"{safe_name}.json"
filepath.write_text(json.dumps(params, indent=2, ensure_ascii=False), encoding="utf-8")
self._cache[safe_name] = params
print(f"[preset] Saved: {safe_name}")
return True
except Exception as e:
print(f"[preset] Save failed: {e}")
return False
def load_preset(self, name: str) -> Optional[dict]:
"""Load a preset by name."""
return self._cache.get(name)
def delete_preset(self, name: str) -> bool:
"""Delete a preset."""
try:
filepath = self.PRESET_DIR / f"{name}.json"
if filepath.exists():
filepath.unlink()
self._cache.pop(name, None)
return True
return False
except Exception as e:
print(f"[preset] Delete failed: {e}")
return False
def list_presets(self) -> List[str]:
"""List all available preset names."""
return sorted(self._cache.keys())
def export_presets(self, export_path: str) -> bool:
"""Export all presets to a single JSON file."""
try:
data = {
"exported_at": datetime.now().isoformat(),
"version": "1.0",
"presets": self._cache
}
Path(export_path).write_text(json.dumps(data, indent=2), encoding="utf-8")
return True
except Exception as e:
print(f"[preset] Export failed: {e}")
return False
def import_presets(self, import_path: str) -> int:
"""Import presets from a JSON file. Returns count of imported presets."""
try:
data = json.loads(Path(import_path).read_text(encoding="utf-8"))
presets = data.get("presets", {})
count = 0
for name, params in presets.items():
if self.save_preset(name, params):
count += 1
return count
except Exception as e:
print(f"[preset] Import failed: {e}")
return 0
# Initialize global preset manager
preset_manager = PresetManager()
# ==================== NEW: PROMPT TEMPLATE SYSTEM ====================
class PromptTemplateSystem:
"""Manages prompt templates with variable substitution."""
TEMPLATES = {
"anime_basic": {
"name": "Basic Anime",
"template": "1girl, solo, {subject}, {quality_tags}, {style_tags}",
"variables": {
"subject": "main character/subject description",
"quality_tags": "masterpiece, best quality",
"style_tags": "anime style, detailed"
},
"description": "Basic template for anime-style generation"
},
"portrait": {
"name": "Portrait",
"template": "portrait of {subject}, {lighting}, {quality_tags}, detailed face, professional photography",
"variables": {
"subject": "person description",
"lighting": "studio lighting, soft light",
"quality_tags": "highly detailed, 8k"
},
"description": "Professional portrait template"
},
"landscape": {
"name": "Landscape",
"template": "scenic view, {setting}, {time_of_day}, {weather}, {quality_tags}, atmospheric",
"variables": {
"setting": "mountains/ocean/forest etc.",
"time_of_day": "sunset/sunrise/noon",
"weather": "clear sky/cloudy/misty",
"quality_tags": "highly detailed, photorealistic"
},
"description": "Scenic landscape template"
},
"character_design": {
"name": "Character Design",
"template": "character design, {gender}, {hair} hair, {eyes} eyes, {outfit}, {pose}, {expression}, {quality_tags}, white background, reference sheet",
"variables": {
"gender": "girl/boy/person",
"hair": "long/short/color",
"eyes": "color/style",
"outfit": "clothing description",
"pose": "standing/sitting/action pose",
"expression": "happy/serious/calm"
},
"description": "Character design reference template"
}
}
@classmethod
def get_template(cls, template_id: str) -> Optional[dict]:
"""Get template by ID."""
return cls.TEMPLATES.get(template_id)
@classmethod
def list_templates(cls) -> List[tuple]:
"""List all templates as (id, name) tuples."""
return [(tid, t["name"]) for tid, t in cls.TEMPLATES.items()]
@classmethod
def render_template(cls, template_id: str, variables: dict) -> Optional[str]:
"""Render a template with provided variables."""
template_data = cls.TEMPLATES.get(template_id)
if not template_data:
return None
template = template_data["template"]
# Merge with defaults
default_vars = template_data.get("variables", {})
merged_vars = {}
for var_name, default_desc in default_vars.items():
merged_vars[var_name] = variables.get(var_name, f"[{default_desc}]")
# Add any extra variables
merged_vars.update(variables)
# Substitute variables
try:
result = template.format(**merged_vars)
return result
except KeyError as e:
print(f"[template] Missing variable: {e}")
return None
@classmethod
def add_custom_template(cls, template_id: str, name: str, template: str,
variables: dict, description: str = "") -> bool:
"""Add a custom template."""
if template_id in cls.TEMPLATES:
return False
cls.TEMPLATES[template_id] = {
"name": name,
"template": template,
"variables": variables,
"description": description
}
return True
# ==================== NEW: BATCH GENERATION SYSTEM ====================
class BatchGenerator:
"""Handles batch generation with prompt variations."""
MAX_BATCH_SIZE = 20 # Limit batch size to prevent abuse
@staticmethod
def generate_variations(base_prompt: str, variation_mode: str = "sequential",
count: int = 4, seed_start: int = -1) -> List[Tuple[str, int]]:
"""
Generate prompt variations for batch processing.
Args:
base_prompt: Base prompt to vary
variation_mode: 'sequential' (seed variation), 'prompt_permutation', 'aspect_variations'
count: Number of variations to generate
seed_start: Starting seed (-1 for random)
Returns:
List of (modified_prompt, seed) tuples
"""
variations = []
if variation_mode == "sequential":
# Simple seed-based variations
base_seed = seed_start if seed_start > 0 else int(time.time()) % (2**32 - 1)
for i in range(min(count, BatchGenerator.MAX_BATCH_SIZE)):
variations.append((base_prompt, base_seed + i))
elif variation_mode == "prompt_permutation":
# Generate slight prompt modifications
modifiers = [
"best quality, masterpiece",
"highly detailed",
"professional, sharp focus",
"cinematic lighting",
"trending on artstation",
"digital art, vibrant colors",
"detailed background",
"soft lighting, atmospheric"
]
base_seed = seed_start if seed_start > 0 else int(time.time()) % (2**32 - 1)
for i in range(min(count, BatchGenerator.MAX_BATCH_SIZE)):
modifier = modifiers[i % len(modifiers)]
modified_prompt = f"{base_prompt}, {modifier}" if modifier else base_prompt
variations.append((modified_prompt, base_seed + i))
elif variation_mode == "aspect_variations":
# Different aspect ratios embedded in prompt
aspects = [
("portrait, tall composition", (832, 1216)),
("landscape, wide composition", (1216, 832)),
("square composition", (1024, 1024)),
("cinematic widescreen", (1280, 720)),
]
base_seed = seed_start if seed_start > 0 else int(time.time()) % (2**32 - 1)
for i in range(min(count, BatchGenerator.MAX_BATCH_SIZE)):
aspect_mod, _ = aspects[i % len(aspects)]
modified_prompt = f"{base_prompt}, {aspect_mod}"
variations.append((modified_prompt, base_seed + i))
return variations
@staticmethod
def validate_batch_params(prompt: str, count: int, mode: str) -> Tuple[bool, str]:
"""Validate batch generation parameters."""
if not prompt or not prompt.strip():
return False, "Prompt cannot be empty"
if count < 1 or count > BatchGenerator.MAX_BATCH_SIZE:
return False, f"Batch size must be between 1 and {BatchGenerator.MAX_BATCH_SIZE}"
valid_modes = ["sequential", "prompt_permutation", "aspect_variations"]
if mode not in valid_modes:
return False, f"Invalid mode. Must be one of: {valid_modes}"
return True, ""
# ==================== NEW: ENHANCED CACHE MANAGER ====================
class EnhancedCacheManager:
"""Improved cache management with LRU eviction and memory tracking."""
def __init__(self, max_size_gb: float = 4.0, max_files: int = 256):
self.max_size_bytes = max_size_gb * 1024**3
self.max_files = max_files
self.cache_dir = Path("outputs")
self.cache_dir.mkdir(exist_ok=True)
self._access_log: Dict[str, float] = {} # path -> last access time
self._lock = threading.Lock()
def record_access(self, filepath: str):
"""Record file access for LRU tracking."""
with self._lock:
self._access_log[filepath] = time.time()
def get_cache_stats(self) -> dict:
"""Get current cache statistics."""
try:
files = list(self.cache_dir.glob("*.png"))
total_size = sum(f.stat().st_size for f in files if f.is_file())
return {
"file_count": len(files),
"total_size_bytes": total_size,
"total_size_gb": round(total_size / (1024**3), 3),
"max_size_gb": round(self.max_size_bytes / (1024**3), 3),
"usage_percent": round(total_size / self.max_size_bytes * 100, 1) if self.max_size_bytes > 0 else 0
}
except Exception as e:
print(f"[cache] Stats error: {e}")
return {"error": str(e)}
def cleanup(self, protect_paths: List[str] = None) -> int:
"""
Clean up cache using LU strategy.
Returns number of files removed.
"""
protect_set = {str(Path(p).resolve()) for p in (protect_paths or [])}
removed_count = 0
with self._lock:
try:
files = []
for f in self.cache_dir.glob("*.png"):
if f.is_file():
try:
st = f.stat()
last_access = self._access_log.get(str(f), st.st_mtime)
files.append((f, st.st_size, last_access))
except Exception:
continue
# Sort by last access time (oldest first)
files.sort(key=lambda x: x[2])
total_size = sum(size for _, size, _ in files)
# Remove files until under limits
while (files and
(len(files) > self.max_files or total_size > self.max_size_bytes)):
fpath, fsize, _ = files.pop(0)
if str(fpath.resolve()) in protect_set:
continue
try:
fpath.unlink()
total_size -= fsize
self._access_log.pop(str(fpath), None)
removed_count += 1
print(f"[cache] Removed: {fpath.name}")
except Exception as e:
print(f"[cache] Remove failed {fpath.name}: {e}")
except Exception as e:
print(f"[cache] Cleanup error: {e}")
return removed_count
def clear_old_entries(self, max_age_seconds: float = 86400) -> int:
"""Remove entries older than max_age_seconds."""
cutoff = time.time() - max_age_seconds
removed = 0
with self._lock:
try:
for fpath, access_time in list(self._access_log.items()):
if access_time < cutoff:
p = Path(fpath)
if p.exists():
try:
p.unlink()
removed += 1
except Exception:
pass
del self._access_log[fpath]
except Exception as e:
print(f"[cache] Clear old error: {e}")
return removed
# Initialize enhanced cache manager
cache_manager = EnhancedCacheManager()
# ==================== IMPROVED GuiSD CLASS ====================
components = None
if IS_ZERO_GPU:
flux_repo = "camenduru/FLUX.1-dev-diffusers"
flux_pipe = FluxPipeline.from_pretrained(
flux_repo,
transformer=None,
torch_dtype=torch.bfloat16,
)
components = flux_pipe.components
delete_model(flux_repo)
#######################
# GUI
#######################
logging.getLogger("diffusers").setLevel(logging.ERROR)
diffusers.utils.logging.set_verbosity(40)
warnings.filterwarnings(action="ignore", category=FutureWarning, module="diffusers")
warnings.filterwarnings(action="ignore", category=UserWarning, module="diffusers")
warnings.filterwarnings(action="ignore", category=FutureWarning, module="transformers")
parser = ArgumentParser(description='DiffuseCraft: Create images from text prompts.', add_help=True)
parser.add_argument("--share", action="store_true", dest="share_enabled", default=False, help="Enable sharing")
parser.add_argument('--theme', type=str, default="NoCrypt/miku", help='Set the theme (default: NoCrypt/miku)')
parser.add_argument("--ssr", action="store_true", default=False, help="Enable SSR (Server-Side Rendering)")
parser.add_argument("--log-level", type=str, default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], help="Set logging level (default: INFO)")
args = parser.parse_args()
logger.setLevel(
"INFO" if IS_ZERO_GPU else getattr(logging, args.log_level.upper())
)
def description_ui():
gr.Markdown(
"""
## Danbooru Tags Transformer V2 Demo with WD Tagger
(Image =>) Prompt => Upsampled longer prompt
- Mod of p1atdev's [Danbooru Tags Transformer V2 Demo](https://huggingface.co/spaces/p1atdev/danbooru-tags-transformer-v2) and [WD Tagger with ๐ค transformers](https://huggingface.co/spaces/p1atdev/wd-tagger-transformers).
- Models: p1atdev's [wd-swinv2-tagger-v3-hf](https://huggingface.co/p1atdev/wd-swinv2-tagger-v3-hf), [dart-v2-moe-sft](https://huggingface.co/p1atdev/dart-v2-moe-sft)
"""
)
def lora_chk(lora_):
if isinstance(lora_, str) and lora_.strip() not in ["", "None"]:
return lora_
return None
# Context manager for GPU operations
@contextmanager
def gpu_context(duration: int = 60):
"""Context manager for GPU operations with automatic cleanup."""
try:
if IS_ZERO_GPU:
yield spaces.GPU(duration=duration)
else:
yield None
finally:
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
class GuiSD:
"""Improved GUI SD class with better error handling and resource management."""
def __init__(self, stream=True):
self.model = None
self.status_loading = False
self.sleep_loading = 4
self.last_load = datetime.now()
self.inventory = []
# Avoid duplicate downloads - FIXED: Better thread safety
self.active_downloads = set()
self.download_lock = threading.RLock() # Changed to RLock for reentrant locking
self.download_events = {} # For per-model wait events
# Anti-abuse: track new model requests.
self.used_models = []
self.new_model_history = []
# Generation statistics
self.generation_stats = {
"total_generations": 0,
"successful_generations": 0,
"failed_generations": 0,
"last_error": None
}
def update_storage_models(self, storage_floor_gb=24, required_inventory_for_purge=3):
"""Update storage and purge old models if needed."""
try:
while get_used_storage_gb() > storage_floor_gb:
if len(self.inventory) < required_inventory_for_purge:
break
removal_candidate = self.inventory.pop(0)
delete_model(removal_candidate)
# Cleanup after 60 seconds of inactivity
lowPrioCleanup = max((datetime.now() - self.last_load).total_seconds(), 0) > 120
if lowPrioCleanup and not self.status_loading and get_used_storage_gb(CACHE_HF_ROOT) > (storage_floor_gb * 2):
print("Cleaning up Hugging Face cache...")
clear_hf_cache()
self.inventory = [
m for m in self.inventory if os.path.exists(m)
]
except Exception as e:
print(f"[storage] Update error: {e}")
def update_inventory(self, model_name):
"""Update model inventory with proper error handling."""
try:
if model_name not in single_file_model_list:
self.inventory = [
m for m in self.inventory if m != model_name
] + [model_name]
print(self.inventory)
except Exception as e:
print(f"[inventory] Update error: {e}")
def load_new_model(self, model_name, vae_model, task, controlnet_model, progress=gr.Progress(track_tqdm=True)):
"""
Load new model with improved error handling and resource management.
FIXES:
- Better exception handling
- Proper resource cleanup in finally blocks
- Improved thread safety
- Memory leak prevention
"""
loaded_successfully = False
try:
if model_name != model_list[0]:
# --- Anti-Abuse Check Start ---
if model_name in self.used_models:
# Move to the end to mark as the most recently used.
self.used_models.remove(model_name)
self.used_models.append(model_name)
else:
current_time = datetime.now()
# Retain history of new model requests from the last 20 minutes.
self.new_model_history = [
t for t in self.new_model_history
if (current_time - t).total_seconds() < 1200
]
# Allow a maximum of 5 new model requests per 20 minutes.
if len(self.new_model_history) >= 5:
yield "Rate limit exceeded: Too many new models requested."
raise gr.Error("Too many new models requested. Please reuse your previously loaded models or wait a few minutes before trying new ones.")
self.new_model_history.append(current_time)
self.used_models.append(model_name)
# Cap the reuse list to the 5 most recent models.
if len(self.used_models) > 5:
self.used_models.pop(0)
# --- Anti-Abuse Check End ---
lock_key = model_name
# Improved waiting mechanism with timeout
wait_timeout = 300 # 5 minutes max wait
wait_start = time.time()
while True:
with self.download_lock:
if lock_key not in self.active_downloads:
self.active_downloads.add(lock_key)
break
# Check timeout
elapsed = time.time() - wait_start
if elapsed > wait_timeout:
yield f"Timeout waiting for download: {model_name}"
raise TimeoutError(f"Waited too long for model download: {model_name}")
yield f"Waiting for existing download to finish: {model_name}... ({int(elapsed)}s)"
time.sleep(1)
try:
# Download link model > model_name
is_link_model = False
model_type = None
if model_name.startswith("http"):
yield f"Downloading model: {model_name}"
model_name, model_type = download_link_model(model_name, DIRECTORY_MODELS)
if not model_name:
raise ValueError("Error retrieving model information from URL")
is_link_model = True
else:
is_link_model = False
if IS_ZERO_GPU:
self.update_storage_models()
vae_model = vae_model if vae_model != "None" else None
model_type = get_model_type(model_name) if not is_link_model else model_type
dtype_model = torch.bfloat16 if model_type == "FLUX" else torch.float16
if not os.path.exists(model_name):
logger.debug(f"model_name={model_name}, vae_model={vae_model}, task={task}, controlnet_model={controlnet_model}")
_ = download_diffuser_repo(
repo_name=model_name,
model_type=model_type,
revision="main",
token=True,
)
self.update_inventory(model_name)
finally:
with self.download_lock:
self.active_downloads.discard(lock_key)
# Improved queue waiting with better feedback
max_queue_wait = 120 # 2 minutes max queue wait
queue_wait_start = time.time()
for i in range(max_queue_wait * 2): # Check every 0.5 seconds
if not self.status_loading:
self.status_loading = True
if i > 0:
time.sleep(self.sleep_loading)
print("Previous model ops...")
break
# Check queue timeout
elapsed = time.time() - queue_wait_start
if elapsed > max_queue_wait:
yield "Timeout waiting for model queue"
raise TimeoutError("Model queue timeout")
time.sleep(0.5)
print(f"Waiting queue {i}")
yield f"Waiting in queue... ({int(elapsed)}s)"
self.status_loading = True
yield f"Loading model: {model_name}"
if vae_model == "BakedVAE":
vae_model = model_name
elif vae_model:
vae_type = "SDXL" if "sdxl" in vae_model.lower() else "SD 1.5"
if model_type != vae_type:
gr.Warning(WARNING_MSG_VAE)
print("Loading model...")
start_time = time.time()
try:
if self.model is None:
self.model = Model_Diffusers(
base_model_id=model_name,
task_name=TASK_STABLEPY[task],
vae_model=vae_model,
type_model_precision=dtype_model,
retain_task_model_in_cache=False,
controlnet_model=controlnet_model,
device="cpu" if IS_ZERO_GPU else None,
env_components=components,
)
self.model.advanced_params(image_preprocessor_cuda_active=IS_GPU_MODE)
else:
if self.model.base_model_id != model_name:
load_now_time = datetime.now()
elapsed_time = max((load_now_time - self.last_load).total_seconds(), 0)
if elapsed_time <= 9:
print("Waiting for the previous model's time ops...")
time.sleep(9 - elapsed_time)
if IS_ZERO_GPU:
self.model.device = torch.device("cpu")
self.model.load_pipe(
model_name,
task_name=TASK_STABLEPY[task],
vae_model=vae_model,
type_model_precision=dtype_model,
retain_task_model_in_cache=False,
controlnet_model=controlnet_model,
)
end_time = time.time()
self.sleep_loading = max(min(int(end_time - start_time), 10), 4)
loaded_successfully = True
except Exception as e:
# Reset state on error
self.last_load = datetime.now()
self.status_loading = False
self.sleep_loading = 4
self.generation_stats["last_error"] = str(e)
raise e
self.last_load = datetime.now()
self.status_loading = False
yield f"Model loaded: {model_name}"
except Exception as e:
# Ensure status is reset on any error
self.status_loading = False
print(f"[model] Load error: {traceback.format_exc()}")
raise
#@spaces.GPU
@torch.inference_mode()
def generate_pipeline(
self,
prompt,
neg_prompt,
num_images,
steps,
cfg,
clip_skip,
seed,
lora1,
lora_scale1,
lora2,
lora_scale2,
lora3,
lora_scale3,
lora4,
lora_scale4,
lora5,
lora_scale5,
lora6,
lora_scale6,
lora7,
lora_scale7,
sampler,
schedule_type,
schedule_prediction_type,
img_height,
img_width,
model_name,
vae_model,
task,
image_control,
preprocessor_name,
preprocess_resolution,
image_resolution,
style_prompt,
style_json_file,
image_mask,
strength,
low_threshold,
high_threshold,
value_threshold,
distance_threshold,
recolor_gamma_correction,
tile_blur_sigma,
controlnet_output_scaling_in_unet,
controlnet_start_threshold,
controlnet_stop_threshold,
textual_inversion,
syntax_weights,
upscaler_model_path,
upscaler_increases_size,
upscaler_tile_size,
upscaler_tile_overlap,
hires_steps,
hires_denoising_strength,
hires_sampler,
hires_prompt,
hires_negative_prompt,
hires_before_adetailer,
hires_after_adetailer,
hires_schedule_type,
hires_guidance_scale,
controlnet_model,
loop_generation,
leave_progress_bar,
disable_progress_bar,
image_previews,
display_images,
save_generated_images,
filename_pattern,
image_storage_location,
retain_compel_previous_load,
retain_detailfix_model_previous_load,
retain_hires_model_previous_load,
t2i_adapter_preprocessor,
t2i_adapter_conditioning_scale,
t2i_adapter_conditioning_factor,
enable_live_preview,
freeu,
generator_in_cpu,
adetailer_inpaint_only,
adetailer_verbose,
adetailer_sampler,
adetailer_active_a,
prompt_ad_a,
negative_prompt_ad_a,
strength_ad_a,
face_detector_ad_a,
person_detector_ad_a,
hand_detector_ad_a,
mask_dilation_a,
mask_blur_a,
mask_padding_a,
adetailer_active_b,
prompt_ad_b,
negative_prompt_ad_b,
strength_ad_b,
face_detector_ad_b,
person_detector_ad_b,
hand_detector_ad_b,
mask_dilation_b,
mask_blur_b,
mask_padding_b,
retain_task_cache_gui,
guidance_rescale,
image_ip1,
mask_ip1,
model_ip1,
mode_ip1,
scale_ip1,
image_ip2,
mask_ip2,
model_ip2,
mode_ip2,
scale_ip2,
pag_scale,
face_restoration_model,
face_restoration_visibility,
face_restoration_weight,
):
"""
Generate images with improved error handling and memory management.
FIXES:
- Better exception handling throughout
- Memory leak prevention
- Proper resource cleanup
- Generation statistics tracking
"""
self.generation_stats["total_generations"] += 1
info_state = html_template_message("Navigating latent space...")
yield info_state, gr.update(), gr.update()
try:
vae_model = vae_model if vae_model != "None" else None
loras_list = [lora1, lora2, lora3, lora4, lora5, lora6, lora7]
vae_msg = f"VAE: {vae_model}" if vae_model else ""
msg_lora = ""
# FIX: Don't reassign global variable, use local instead
current_lora_list = get_lora_model_list()
loras_list = [s if s else "None" for s in loras_list]
lora1, lora_scale1, lora2, lora_scale2, lora3, lora_scale3, lora4, lora_scale4, lora5, lora_scale5, lora6, lora_scale6, lora7, lora_scale7 = \
set_prompt_loras(prompt, syntax_weights, model_name, lora1, lora_scale1, lora2, lora_scale2, lora3,
lora_scale3, lora4, lora_scale4, lora5, lora_scale5, lora6, lora_scale6, lora7, lora_scale7)
logger.debug(f"Config model: {model_name}, {vae_model}, {loras_list}")
task = TASK_STABLEPY[task]
params_ip_img = []
params_ip_msk = []
params_ip_model = []
params_ip_mode = []
params_ip_scale = []
all_adapters = [
(image_ip1, mask_ip1, model_ip1, mode_ip1, scale_ip1),
(image_ip2, mask_ip2, model_ip2, mode_ip2, scale_ip2),
]
if not hasattr(self.model.pipe, "transformer"):
for imgip, mskip, modelip, modeip, scaleip in all_adapters:
if imgip:
params_ip_img.append(imgip)
if mskip:
params_ip_msk.append(mskip)
params_ip_model.append(modelip)
params_ip_mode.append(modeip)
params_ip_scale.append(scaleip)
concurrency = 5
self.model.stream_config(concurrency=concurrency, latent_resize_by=1, vae_decoding=False)
if task != "txt2img" and not image_control:
raise ValueError("Reference image is required. Please upload one in 'Image ControlNet/Inpaint/Img2img'.")
if task in ["inpaint", "repaint"] and not image_mask:
raise ValueError("Mask image not found. Upload one in 'Image Mask' to proceed.")
if "https://" not in str(UPSCALER_DICT_GUI[upscaler_model_path]):
upscaler_model = upscaler_model_path
else:
url_upscaler = UPSCALER_DICT_GUI[upscaler_model_path]
if not os.path.exists(f"./{DIRECTORY_UPSCALERS}/{url_upscaler.split('/')[-1]}"):
download_things(DIRECTORY_UPSCALERS, url_upscaler, HF_TOKEN)
upscaler_model = f"./{DIRECTORY_UPSCALERS}/{url_upscaler.split('/')[-1]}"
logging.getLogger("ultralytics").setLevel(logging.INFO if adetailer_verbose else logging.ERROR)
adetailer_params_A = {
"face_detector_ad": face_detector_ad_a,
"person_detector_ad": person_detector_ad_a,
"hand_detector_ad": hand_detector_ad_a,
"prompt": prompt_ad_a,
"negative_prompt": negative_prompt_ad_a,
"strength": strength_ad_a,
"mask_dilation": mask_dilation_a,
"mask_blur": mask_blur_a,
"mask_padding": mask_padding_a,
"inpaint_only": adetailer_inpaint_only,
"sampler": adetailer_sampler,
}
adetailer_params_B = {
"face_detector_ad": face_detector_ad_b,
"person_detector_ad": person_detector_ad_b,
"hand_detector_ad": hand_detector_ad_b,
"prompt": prompt_ad_b,
"negative_prompt": negative_prompt_ad_b,
"strength": strength_ad_b,
"mask_dilation": mask_dilation_b,
"mask_blur": mask_blur_b,
"mask_padding": mask_padding_b,
}
pipe_params = {
"prompt": prompt,
"negative_prompt": neg_prompt,
"img_height": img_height,
"img_width": img_width,
"num_images": num_images,
"num_steps": steps,
"guidance_scale": cfg,
"clip_skip": clip_skip,
"pag_scale": float(pag_scale),
"seed": seed,
"image": image_control,
"preprocessor_name": preprocessor_name,
"preprocess_resolution": preprocess_resolution,
"image_resolution": image_resolution,
"style_prompt": style_prompt if style_prompt else "",
"style_json_file": "",
"image_mask": image_mask,
"strength": strength,
"low_threshold": low_threshold,
"high_threshold": high_threshold,
"value_threshold": value_threshold,
"distance_threshold": distance_threshold,
"recolor_gamma_correction": float(recolor_gamma_correction),
"tile_blur_sigma": int(tile_blur_sigma),
"lora_A": lora_chk(lora1),
"lora_scale_A": lora_scale1,
"lora_B": lora_chk(lora2),
"lora_scale_B": lora_scale2,
"lora_C": lora_chk(lora3),
"lora_scale_C": lora_scale3,
"lora_D": lora_chk(lora4),
"lora_scale_D": lora_scale4,
"lora_E": lora_chk(lora5),
"lora_scale_E": lora_scale5,
"lora_F": lora_chk(lora6),
"lora_scale_F": lora_scale6,
"lora_G": lora_chk(lora7),
"lora_scale_G": lora_scale7,
"textual_inversion": get_embed_list(self.model.class_name) if textual_inversion else [],
"syntax_weights": syntax_weights,
"sampler": sampler,
"schedule_type": schedule_type,
"schedule_prediction_type": schedule_prediction_type,
"xformers_memory_efficient_attention": False,
"gui_active": True,
"loop_generation": loop_generation,
"controlnet_conditioning_scale": float(controlnet_output_scaling_in_unet),
"control_guidance_start": float(controlnet_start_threshold),
"control_guidance_end": float(controlnet_stop_threshold),
"generator_in_cpu": generator_in_cpu,
"FreeU": freeu,
"adetailer_A": adetailer_active_a,
"adetailer_A_params": adetailer_params_A,
"adetailer_B": adetailer_active_b,
"adetailer_B_params": adetailer_params_B,
"leave_progress_bar": leave_progress_bar,
"disable_progress_bar": disable_progress_bar,
"image_previews": image_previews,
"display_images": False,
"save_generated_images": save_generated_images,
"filename_pattern": filename_pattern,
"image_storage_location": image_storage_location,
"retain_compel_previous_load": retain_compel_previous_load,
"retain_detailfix_model_previous_load": retain_detailfix_model_previous_load,
"retain_hires_model_previous_load": retain_hires_model_previous_load,
"t2i_adapter_preprocessor": t2i_adapter_preprocessor,
"t2i_adapter_conditioning_scale": float(t2i_adapter_conditioning_scale),
"t2i_adapter_conditioning_factor": float(t2i_adapter_conditioning_factor),
"upscaler_model_path": upscaler_model,
"upscaler_increases_size": upscaler_increases_size,
"upscaler_tile_size": upscaler_tile_size,
"upscaler_tile_overlap": upscaler_tile_overlap,
"hires_steps": hires_steps,
"hires_denoising_strength": hires_denoising_strength,
"hires_prompt": hires_prompt,
"hires_negative_prompt": hires_negative_prompt,
"hires_sampler": hires_sampler,
"hires_before_adetailer": hires_before_adetailer,
"hires_after_adetailer": hires_after_adetailer,
"hires_schedule_type": hires_schedule_type,
"hires_guidance_scale": hires_guidance_scale,
"ip_adapter_image": params_ip_img,
"ip_adapter_mask": params_ip_msk,
"ip_adapter_model": params_ip_model,
"ip_adapter_mode": params_ip_mode,
"ip_adapter_scale": params_ip_scale,
"face_restoration_model": face_restoration_model,
"face_restoration_visibility": face_restoration_visibility,
"face_restoration_weight": face_restoration_weight,
}
# kwargs for diffusers pipeline
if guidance_rescale:
pipe_params["guidance_rescale"] = guidance_rescale
if IS_ZERO_GPU:
self.model.device = torch.device("cuda:0")
if hasattr(self.model.pipe, "transformer") and loras_list != ["None"] * self.model.num_loras:
self.model.pipe.transformer.to(self.model.device)
logger.debug("transformer to cuda")
actual_progress = 0
info_images = gr.update()
for img, [seed, image_path, metadata] in self.model(**pipe_params):
info_state = progress_step_bar(actual_progress, steps)
actual_progress += concurrency
if image_path:
info_images = f"Seeds: {str(seed)}"
if vae_msg:
info_images = info_images + "
" + vae_msg
if "Cannot copy out of meta tensor; no data!" in self.model.last_lora_error:
msg_ram = "Unable to process the LoRAs due to high RAM usage; please try again later."
print(msg_ram)
msg_lora += f"
{msg_ram}"
for status, lora in zip(self.model.lora_status, self.model.lora_memory):
if status:
msg_lora += f"
Loaded: {lora}"
elif status is not None:
msg_lora += f"
Error with: {lora}"
if msg_lora:
info_images += msg_lora
info_images = info_images + "
" + "GENERATION DATA:
" + escape_html(metadata[-1]) + "
-------
"
download_links = "
".join(
[
f'Download Image {i + 1}'
for i, path in enumerate(image_path)
]
)
if save_generated_images:
info_images += f"
{download_links}"
img = save_images(img, metadata)
# Record in cache manager
if image_path:
for path in image_path:
cache_manager.record_access(path)
if not display_images:
img = img if img else gr.update()
info_state = "COMPLETE"
self.generation_stats["successful_generations"] += 1
elif not enable_live_preview:
img = gr.update()
yield info_state, img, info_images
except Exception as e:
self.generation_stats["failed_generations"] += 1
self.generation_stats["last_error"] = str(e)
print(f"[generation] Error: {traceback.format_exc()}")
raise
finally:
# Always clean up GPU memory
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
def dynamic_gpu_duration(func, duration, *args):
@spaces.GPU(duration=duration)
def wrapped_func():
yield from func(*args)
return wrapped_func()
@spaces.GPU
def dummy_gpu():
return None
def sd_gen_generate_pipeline(*args):
"""Wrapper for generation pipeline with improved error handling."""
gpu_duration_arg = int(args[-1]) if args[-1] else 59
verbose_arg = int(args[-2])
load_lora_cpu = args[-3]
generation_args = args[:-3]
lora_list = [
None if item == "None" or item == "" else item
for item in [args[7], args[9], args[11], args[13], args[15], args[17], args[19]]
]
lora_status = [None] * sd_gen.model.num_loras if getattr(sd_gen, "model", None) is not None else 7
msg_load_lora = "Updating LoRAs in GPU..."
if load_lora_cpu:
msg_load_lora = "Updating LoRAs in CPU..."
if lora_list != sd_gen.model.lora_memory and lora_list != [None] * sd_gen.model.num_loras:
yield msg_load_lora, gr.update(), gr.update()
# Load lora in CPU
if load_lora_cpu:
try:
lora_status = sd_gen.model.load_lora_on_the_fly(
lora_A=lora_list[0], lora_scale_A=args[8],
lora_B=lora_list[1], lora_scale_B=args[10],
lora_C=lora_list[2], lora_scale_C=args[12],
lora_D=lora_list[3], lora_scale_D=args[14],
lora_E=lora_list[4], lora_scale_E=args[16],
lora_F=lora_list[5], lora_scale_F=args[18],
lora_G=lora_list[6], lora_scale_G=args[20],
)
print(lora_status)
except Exception as e:
print(f"[lora] CPU load error: {e}")
gr.Warning(f"Failed to load LoRAs on CPU: {e}")
sampler_name = args[21]
schedule_type_name = args[22]
_, _, msg_sampler = check_scheduler_compatibility(
sd_gen.model.class_name, sampler_name, schedule_type_name
)
if msg_sampler:
gr.Warning(msg_sampler)
if verbose_arg:
for status, lora in zip(lora_status, lora_list):
if status:
gr.Info(f"LoRA loaded in CPU: {lora}")
elif status is not None:
gr.Warning(f"Failed to load LoRA: {lora}")
if lora_status == [None] * sd_gen.model.num_loras and sd_gen.model.lora_memory != [None] * sd_gen.model.num_loras and load_lora_cpu:
lora_cache_msg = ", ".join(
str(x) for x in sd_gen.model.lora_memory if x is not None
)
gr.Info(f"LoRAs in cache: {lora_cache_msg}")
msg_request = f"Requesting {gpu_duration_arg}s. of GPU time.\nModel: {sd_gen.model.base_model_id}"
if verbose_arg:
gr.Info(msg_request)
print(msg_request)
yield msg_request.replace("\n", "
"), gr.update(), gr.update()
start_time = time.time()
try:
yield from dynamic_gpu_duration(
sd_gen.generate_pipeline,
gpu_duration_arg,
*generation_args,
)
except Exception as e:
print(f"[pipeline] Generation error: {e}")
raise
end_time = time.time()
execution_time = end_time - start_time
msg_task_complete = (
f"GPU task complete in: {int(round(execution_time, 0) + 1)} seconds"
)
if verbose_arg:
gr.Info(msg_task_complete)
print(msg_task_complete)
yield msg_task_complete, gr.update(), gr.update()
@spaces.GPU(duration=15)
def process_upscale(image, upscaler_name, upscaler_size):
"""Process image upscaling with improved error handling."""
if image is None:
return None
try:
from stablepy.diffusers_vanilla.utils import save_pil_image_with_metadata
from stablepy import load_upscaler_model
image = image.convert("RGB")
exif_image = extract_exif_data(image)
name_upscaler = UPSCALER_DICT_GUI[upscaler_name]
if "https://" in str(name_upscaler):
if not os.path.exists(f"./{DIRECTORY_UPSCALERS}/{name_upscaler.split('/')[-1]}"):
download_things(DIRECTORY_UPSCALERS, name_upscaler, HF_TOKEN)
name_upscaler = f"./{DIRECTORY_UPSCALERS}/{name_upscaler.split('/')[-1]}"
scaler_beta = load_upscaler_model(name=name_upscaler, tile=(0 if IS_ZERO_GPU else 192), tile_overlap=8, device=("cuda" if IS_GPU_MODE else "cpu"), half=IS_GPU_MODE)
image_up = scaler_beta.upscale(image, upscaler_size, True)
image_path = save_pil_image_with_metadata(image_up, f'{os.getcwd()}/up_images', exif_image)
return image_path
except Exception as e:
print(f"[upscale] Error: {e}")
raise gr.Error(f"Upscaling failed: {e}")
# ==================== NEW: UTILITY FUNCTIONS ====================
def get_system_info() -> dict:
"""Get system information for debugging."""
info = {
"python_version": os.sys.version,
"torch_version": torch.__version__,
"cuda_available": torch.cuda.is_available(),
"gpu_name": torch.cuda.get_device_name(0) if torch.cuda.is_available() else "N/A",
"gpu_memory_total": f"{torch.cuda.get_device_properties(0).total_memory / 1024**3:.1f} GB" if torch.cuda.is_available() else "N/A",
"gpu_memory_allocated": f"{torch.cuda.memory_allocated() / 1024**3:.2f} GB" if torch.cuda.is_available() else "N/A",
"gpu_memory_reserved": f"{torch.cuda.memory_reserved() / 1024**3:.2f} GB" if torch.cuda.is_available() else "N/A",
"is_zero_gpu": IS_ZERO_GPU,
"storage_used_gb": get_used_storage_gb(),
"cache_stats": cache_manager.get_cache_stats(),
"generation_stats": sd_gen.generation_stats,
"timestamp": datetime.now().isoformat()
}
return info
def format_system_info_md(info: dict) -> str:
"""Format system info as markdown."""
lines = ["## System Information", ""]
for key, value in info.items():
if key == "cache_stats" and isinstance(value, dict):
lines.append(f"**Cache:**")
for k, v in value.items():
lines.append(f" - {k}: {v}")
elif key == "generation_stats" and isinstance(value, dict):
lines.append(f"**Generations:**")
for k, v in value.items():
lines.append(f" - {k}: {v}")
else:
lines.append(f"**{key.replace('_', ' ').title()}:** {value}")
lines.append("")
return "\n".join(lines)
def validate_prompt(prompt: str) -> Tuple[bool, str]:
"""Validate prompt and return (is_valid, error_message)."""
if not prompt or not prompt.strip():
return False, "Prompt cannot be empty"
if len(prompt) > 10000:
return False, "Prompt too long (max 10000 characters)"
# Check for potentially problematic patterns
problematic_patterns = [
r'