import os import time import random import logging from gradio.blocks import postprocess_update_dict import numpy as np from typing import Any, Dict, List, Optional, Union, Tuple, Iterator import spaces import gc import torch from PIL import Image import gradio as gr from tempfile import NamedTemporaryFile from diffusers import ( DiffusionPipeline, AutoencoderTiny, AutoencoderKL, AutoPipelineForImage2Image, FluxPipeline, Flux2Pipeline, AutoModel, FlowMatchEulerDiscreteScheduler, DPMSolverMultistepScheduler) from transformers import Mistral3ForConditionalGeneration from huggingface_hub import ( hf_hub_download, HfFileSystem, ModelCard, snapshot_download) from diffusers.utils import load_image from modules.version_info import ( versions_html, #initialize_cuda, #release_torch_resources, #get_torch_info ) from modules.image_utils import ( change_color, open_image, build_prerendered_images_by_quality, upscale_image, # lerp_imagemath, # shrink_and_paste_on_blank, show_lut, apply_lut_to_image_path, multiply_and_blend_images, alpha_composite_with_control, apply_alpha_mask, resize_and_crop_image, convert_to_rgba_png, get_image_from_dict ) from modules.constants import ( LORA_DETAILS, LORAS as loras, MODELS, LORA_TO_MODEL, default_lut_example_img, lut_files, MAX_SEED, IS_SHARED_SPACE, # lut_folder,cards, # cards_alternating, # card_colors, # card_colors_alternating, pre_rendered_maps_paths, PROMPTS, NEGATIVE_PROMPTS, TARGET_SIZE, temp_files, load_env_vars, dotenv_path ) from modules.color_utils import ( color_to_hex ) # from modules.excluded_colors import ( # add_color, # delete_color, # build_dataframe, # on_input, # excluded_color_list, # on_color_display_select # ) from modules.misc import ( get_filename, convert_ratio_to_dimensions, update_dimensions_on_ratio, wait5 ) from modules.lora_details import ( approximate_token_count, split_prompt_precisely, upd_prompt_notes_by_index, get_trigger_words_by_index ) from modules.mazlib.cli import run_maze as run_maze_maker input_image_palette = [] current_prerendered_image = gr.State("./images/Beeuty-1.png") user_info = { "username": "guest", "session_hash": None, "headers": None, "client": None, "query_params": None, "path_params": None, "level" : 0 } user_profile = None def get_profile() -> gr.OAuthProfile | None: global user_profile return user_profile # Define a function to handle the login button click and retrieve user information. def handle_login(profile: gr.OAuthProfile | None, request: gr.Request): # Extract user information from the request if profile is None: profile = gr.State(None) profile = get_profile() user_info = { "username": request.username, "session_hash": request.session_hash, "headers": dict(request.headers), "client": request.client, "query_params": dict(request.query_params), "path_params": dict(request.path_params), "level" : (0 if request.username == "guest" else 2) } return user_info, gr.update(logout_value=f"Logout {user_info['username']} ({user_info['level']})", value=f"Login {user_info['username']} ({user_info['level']})"), profile #---if workspace = local or colab--- # Authenticate with Hugging Face # from huggingface_hub import login # Log in to Hugging Face using the provided token # hf_token = 'hf-token-authentication' # login(hf_token) def calculate_shift( image_seq_len, base_seq_len: int = 256, max_seq_len: int = 4096, base_shift: float = 0.5, max_shift: float = 1.16, ): m = (max_shift - base_shift) / (max_seq_len - base_seq_len) b = base_shift - m * base_seq_len mu = image_seq_len * m + b return mu def retrieve_timesteps( scheduler, num_inference_steps: Optional[int] = None, device: Optional[Union[str, torch.device]] = None, timesteps: Optional[List[int]] = None, sigmas: Optional[List[float]] = None, **kwargs, ): if timesteps is not None and sigmas is not None: raise ValueError("Only one of `timesteps` or `sigmas` can be passed. Please choose one to set custom values") if timesteps is not None: scheduler.set_timesteps(timesteps=timesteps, device=device, **kwargs) timesteps = scheduler.timesteps num_inference_steps = len(timesteps) elif sigmas is not None: scheduler.set_timesteps(sigmas=sigmas, device=device, **kwargs) timesteps = scheduler.timesteps num_inference_steps = len(timesteps) else: scheduler.set_timesteps(num_inference_steps, device=device, **kwargs) timesteps = scheduler.timesteps return timesteps, num_inference_steps def is_flux2_model(model_name: str) -> bool: return "FLUX.2" in model_name.upper() or model_name.lower().startswith("fal/flux.2") def get_total_vram_gb(device_index: int = 0) -> float: if not torch.cuda.is_available(): return 0.0 try: total_bytes = torch.cuda.get_device_properties(device_index).total_memory return float(total_bytes) / (1024 ** 3) except Exception: return 0.0 def should_enable_flux2_cpu_offload(vram_threshold_gb: float = 80.0) -> bool: return get_total_vram_gb() < vram_threshold_gb def _coerce_lora_parameter_value(value): if isinstance(value, str): v = value.strip() try: if "." in v: return float(v) return int(v) except ValueError: return value return value def get_lora_runtime_parameters(lora_repo: str, selected_lora: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: params: Dict[str, Any] = {} detail_items = LORA_DETAILS.get(lora_repo, []) for item in detail_items: if isinstance(item, dict): detail_params = item.get("parameters") if isinstance(detail_params, dict): for key, value in detail_params.items(): params[key] = _coerce_lora_parameter_value(value) if selected_lora and isinstance(selected_lora.get("parameters"), dict): for key, value in selected_lora["parameters"].items(): params[key] = _coerce_lora_parameter_value(value) return params # Preload settings FLUX2_PRELOAD_REPO = "diffusers/FLUX.2-dev-bnb-4bit" FLUX2_PRELOAD_PATTERNS = [ "model_index.json", "transformer/*", "vae/*", "scheduler/*", "tokenizer/*", "tokenizer_2/*", "text_encoder/*", "text_encoder_2/*", ] FLUX2_TURBO_LORA_REPO = "fal/FLUX.2-dev-Turbo" FLUX2_TURBO_LORA_FILE = "flux.2-turbo-lora.safetensors" def preload_flux2_model_to_cache() -> None: """Warm HF cache at startup (no model instantiation).""" token = os.getenv("HF_TOKEN", None) # 1) Base FLUX.2 model cache warmup t0 = time.time() flux2_repo_id = FLUX2_PRELOAD_REPO.strip() print(f"[Startup] Preloading {flux2_repo_id} into Hugging Face cache...") try: local_path = snapshot_download( repo_id=flux2_repo_id, allow_patterns=FLUX2_PRELOAD_PATTERNS, resume_download=True, token=token, ) print(f"[Startup] FLUX.2 cache preload complete in {time.time() - t0:.2f}s") print(f"[Startup] FLUX.2 cached at: {local_path}") except Exception as ex: print(f"[Startup] FLUX.2 preload skipped/failed after {time.time() - t0:.2f}s: {ex}") # 2) FLUX.2 Turbo LoRA cache warmup t1 = time.time() print(f"[Startup] Preloading {FLUX2_TURBO_LORA_REPO}/{FLUX2_TURBO_LORA_FILE}...") try: lora_path = hf_hub_download( repo_id=FLUX2_TURBO_LORA_REPO, filename=FLUX2_TURBO_LORA_FILE, token=token, ) print(f"[Startup] FLUX.2 Turbo LoRA cached in {time.time() - t1:.2f}s") print(f"[Startup] FLUX.2 Turbo LoRA path: {lora_path}") except Exception as ex: print(f"[Startup] FLUX.2 Turbo LoRA preload skipped/failed after {time.time() - t1:.2f}s: {ex}") @torch.inference_mode() def flux_pipe_call_that_returns_an_iterable_of_images( self, prompt: Union[str, List[str]] = None, prompt_2: Optional[Union[str, List[str]]] = None, negative_prompt: Optional[Union[str, List[str]]] = None, height: Optional[int] = None, width: Optional[int] = None, num_inference_steps: int = 28, timesteps: List[int] = None, guidance_scale: float = 3.5, num_images_per_prompt: Optional[int] = 1, generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None, latents: Optional[torch.FloatTensor] = None, prompt_embeds: Optional[torch.FloatTensor] = None, pooled_prompt_embeds: Optional[torch.FloatTensor] = None, output_type: Optional[str] = "pil", return_dict: bool = True, joint_attention_kwargs: Optional[Dict[str, Any]] = None, max_sequence_length: int = 512, good_vae: Optional[Any] = None, ): # Set default height and width height = height or self.default_sample_size * self.vae_scale_factor width = width or self.default_sample_size * self.vae_scale_factor # Validate inputs self.check_inputs( prompt, prompt_2, height, width, prompt_embeds=prompt_embeds, pooled_prompt_embeds=pooled_prompt_embeds, max_sequence_length=max_sequence_length, ) self._guidance_scale = guidance_scale self._joint_attention_kwargs = joint_attention_kwargs self._interrupt = False batch_size = 1 if isinstance(prompt, str) else len(prompt) device = self._execution_device lora_scale = joint_attention_kwargs.get("scale", None) if joint_attention_kwargs is not None else None # Prepare prompt inputs. Flux2Pipeline does not accept prompt_2 or negative_prompt # as separate kwargs, so merge them into a single prompt string for Flux2. if isinstance(self, Flux2Pipeline): combined_prompt = None if prompt is None: combined_prompt = prompt_2 elif prompt_2 is None: combined_prompt = prompt else: combined_prompt = f"{prompt} {prompt_2}" prompt_to_encode = combined_prompt neg_prompt_to_encode = None else: prompt_to_encode = prompt neg_prompt_to_encode = None if negative_prompt is None else negative_prompt # Encode the positive prompt prompt_embeds_pos, pooled_prompt_embeds_pos, text_ids_pos = self.encode_prompt( prompt=prompt_to_encode, prompt_2=None, prompt_embeds=prompt_embeds, pooled_prompt_embeds=pooled_prompt_embeds, device=device, num_images_per_prompt=num_images_per_prompt, max_sequence_length=max_sequence_length, lora_scale=lora_scale, ) # Encode the negative prompt if provided # Encode negative prompt only for pipelines that support it if (not isinstance(self, Flux2Pipeline)) and (negative_prompt is not None): prompt_embeds_neg, pooled_prompt_embeds_neg, text_ids_neg = self.encode_prompt( prompt=negative_prompt, prompt_2=None, prompt_embeds=None, pooled_prompt_embeds=None, device=device, num_images_per_prompt=num_images_per_prompt, max_sequence_length=max_sequence_length, lora_scale=lora_scale, ) else: # Fallback to positive embeddings if no negative prompt or unsupported prompt_embeds_neg = prompt_embeds_pos pooled_prompt_embeds_neg = pooled_prompt_embeds_pos text_ids_neg = text_ids_pos # Prepare latents num_channels_latents = self.transformer.config.in_channels // 4 latents, latent_image_ids = self.prepare_latents( batch_size * num_images_per_prompt, num_channels_latents, height, width, prompt_embeds_pos.dtype, device, generator, latents, ) # Set up timesteps sigmas = np.linspace(1.0, 1 / num_inference_steps, num_inference_steps) image_seq_len = latents.shape[1] mu = calculate_shift( image_seq_len, self.scheduler.config.base_image_seq_len, self.scheduler.config.max_image_seq_len, self.scheduler.config.base_shift, self.scheduler.config.max_shift, ) timesteps, num_inference_steps = retrieve_timesteps( self.scheduler, num_inference_steps, device, timesteps, sigmas, mu=mu, ) self._num_timesteps = len(timesteps) guidance = ( torch.full([1], guidance_scale, device=device, dtype=torch.float32).expand(latents.shape[0]) if self.transformer.config.guidance_embeds else None ) # Denoising loop for i, t in enumerate(timesteps): if self._interrupt: continue timestep = t.expand(latents.shape[0]).to(latents.dtype) print(f"Step {i + 1}/{num_inference_steps} - Timestep: {timestep.item()}\n") # Compute noise prediction for positive prompt noise_pred_pos = self.transformer( hidden_states=latents, timestep=timestep / 1000, guidance=guidance, pooled_projections=pooled_prompt_embeds_pos, encoder_hidden_states=prompt_embeds_pos, txt_ids=text_ids_pos, img_ids=latent_image_ids, joint_attention_kwargs=self.joint_attention_kwargs, return_dict=False, )[0] # Compute noise prediction for negative prompt noise_pred_neg = self.transformer( hidden_states=latents, timestep=timestep / 1000, guidance=guidance, pooled_projections=pooled_prompt_embeds_neg, encoder_hidden_states=prompt_embeds_neg, txt_ids=text_ids_neg, img_ids=latent_image_ids, joint_attention_kwargs=self.joint_attention_kwargs, return_dict=False, )[0] # Combine noise predictions using guidance scale noise_pred = noise_pred_neg + guidance_scale * (noise_pred_pos - noise_pred_neg) # Generate intermediate image latents_for_image = self._unpack_latents(latents, height, width, self.vae_scale_factor) latents_for_image = (latents_for_image / self.vae.config.scaling_factor) + self.vae.config.shift_factor image = self.vae.decode(latents_for_image, return_dict=False)[0] yield self.image_processor.postprocess(image, output_type=output_type)[0] # Update latents with combined noise prediction latents = self.scheduler.step(noise_pred, t, latents, return_dict=False)[0] torch.cuda.empty_cache() # Final image generation latents = self._unpack_latents(latents, height, width, self.vae_scale_factor) latents = (latents / good_vae.config.scaling_factor) + good_vae.config.shift_factor image = good_vae.decode(latents, return_dict=False)[0] self.maybe_free_model_hooks() torch.cuda.empty_cache() yield self.image_processor.postprocess(image, output_type=output_type)[0] @torch.inference_mode() def flux_pipe_call_single_final_image( self, prompt: Union[str, List[str]] = None, prompt_2: Optional[Union[str, List[str]]] = None, negative_prompt: Optional[Union[str, List[str]]] = None, height: Optional[int] = None, width: Optional[int] = None, num_inference_steps: int = 28, guidance_scale: float = 3.5, generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None, joint_attention_kwargs: Optional[Dict[str, Any]] = None, output_type: Optional[str] = "pil", ): is_f2 = isinstance(self, Flux2Pipeline) final_steps = num_inference_steps if num_inference_steps is not None else (7 if is_f2 else 28) kwargs = { "prompt": prompt, "height": height, "width": width, "num_inference_steps": final_steps, "guidance_scale": guidance_scale, "generator": generator, "output_type": output_type, } if is_f2: # Flux2Pipeline does not accept prompt_2; merge into prompt. if prompt is None: kwargs["prompt"] = prompt_2 elif prompt_2 is not None: kwargs["prompt"] = f"{prompt} {prompt_2}" else: if prompt_2 is not None: kwargs["prompt_2"] = prompt_2 # Flux2Pipeline does not support negative_prompt if (negative_prompt is not None) and (not is_f2): kwargs["negative_prompt"] = negative_prompt if joint_attention_kwargs is not None: if is_f2: kwargs["attention_kwargs"] = joint_attention_kwargs else: kwargs["joint_attention_kwargs"] = joint_attention_kwargs return self(**kwargs).images[0] #--------------------------------------------------Model Initialization-----------------------------------------------------------------------------------------# # Pre-download FLUX.2 once at startup so later switching is cache-hit. preload_flux2_model_to_cache() dtype = torch.bfloat16 device = "cpu" # ZeroGPU-safe at startup; move to CUDA only inside @spaces.GPU base_model = "black-forest-labs/FLUX.1-dev" active_base_model = base_model flux2_cpu_offload_enabled = False taef1 = AutoencoderTiny.from_pretrained("madebyollin/taef1", torch_dtype=dtype) good_vae = AutoencoderKL.from_pretrained(base_model, subfolder="vae", torch_dtype=dtype) pipe = DiffusionPipeline.from_pretrained(base_model, torch_dtype=dtype, vae=taef1) pipe_i2i = None # pipe_i2i = AutoPipelineForImage2Image.from_pretrained(base_model, # vae=good_vae, # transformer=pipe.transformer, # text_encoder=pipe.text_encoder, # tokenizer=pipe.tokenizer, # text_encoder_2=pipe.text_encoder_2, # tokenizer_2=pipe.tokenizer_2, # torch_dtype=dtype # ) pipe.flux_pipe_call_that_returns_an_iterable_of_images = flux_pipe_call_that_returns_an_iterable_of_images.__get__(pipe) pipe.flux_pipe_call_single_final_image = flux_pipe_call_single_final_image.__get__(pipe) def _release_gpu_obj(obj): if obj is None: return None try: if hasattr(obj, "to"): obj.to("cpu") except Exception: pass del obj gc.collect() # IMPORTANT: no torch.cuda.* calls here (can run outside ZeroGPU allowance) return None def ensure_pipelines_for_model(model_name: str, use_safetensors: bool = False, image_input=None): global pipe, pipe_i2i, good_vae, active_base_model, flux2_cpu_offload_enabled need_i2i = image_input is not None already_correct_model = (model_name == active_base_model) # Fast path: correct model and requested variant already exists. if already_correct_model: if need_i2i and pipe_i2i is not None: pipe = _release_gpu_obj(pipe) # enforce exclusivity status_msg = f"Using base model: {active_base_model} (i2i ready)" print(status_msg) gr.Info(status_msg) return if (not need_i2i) and pipe is not None: pipe_i2i = _release_gpu_obj(pipe_i2i) # enforce exclusivity status_msg = f"Using base model: {active_base_model} (t2i ready)" print(status_msg) gr.Info(status_msg) return status_msg = f"Rebuilding {'i2i' if need_i2i else 't2i'} pipeline for {model_name}" print(status_msg) gr.Info(status_msg) else: status_msg = f"Switching base model from {active_base_model} to {model_name}" print(status_msg) gr.Info(status_msg) # release old model resources before rebuilding pipe = _release_gpu_obj(pipe) pipe_i2i = _release_gpu_obj(pipe_i2i) good_vae = _release_gpu_obj(good_vae) # Build t2i source pipe if needed (also needed as source when creating i2i). if pipe is None: if is_flux2_model(model_name): text_encoder = Mistral3ForConditionalGeneration.from_pretrained( model_name, subfolder="text_encoder", torch_dtype=torch.bfloat16, device_map="cpu", local_files_only=True, ) dit = AutoModel.from_pretrained( model_name, subfolder="transformer", torch_dtype=torch.bfloat16, device_map="cpu", local_files_only=True, ) pipe = Flux2Pipeline.from_pretrained( model_name, torch_dtype=dtype, text_encoder=text_encoder, transformer=dit, use_safetensors=use_safetensors, local_files_only=True, ) flux2_cpu_offload_enabled = should_enable_flux2_cpu_offload(60.0) if flux2_cpu_offload_enabled: pipe.enable_model_cpu_offload() print(f"FLUX.2 CPU offload enabled (VRAM: {get_total_vram_gb():.2f} GB)") else: print(f"FLUX.2 CPU offload disabled (VRAM: {get_total_vram_gb():.2f} GB)") good_vae = pipe.vae elif "FLUX" in model_name.upper(): flux2_cpu_offload_enabled = False good_vae = AutoencoderKL.from_pretrained( model_name, subfolder="vae", torch_dtype=dtype, use_safetensors=use_safetensors ) pipe = FluxPipeline.from_pretrained( model_name, torch_dtype=dtype, vae=taef1, use_safetensors=use_safetensors ) else: flux2_cpu_offload_enabled = False good_vae = AutoencoderKL.from_pretrained( model_name, subfolder="vae", torch_dtype=dtype, use_safetensors=use_safetensors ) pipe = DiffusionPipeline.from_pretrained( model_name, torch_dtype=dtype, vae=taef1, use_safetensors=use_safetensors ) pipe.flux_pipe_call_that_returns_an_iterable_of_images = flux_pipe_call_that_returns_an_iterable_of_images.__get__(pipe) pipe.flux_pipe_call_single_final_image = flux_pipe_call_single_final_image.__get__(pipe) if need_i2i: pipe_i2i = _release_gpu_obj(pipe_i2i) if is_flux2_model(model_name): # FLUX.2 i2i: reuse Flux2Pipeline (no AutoPipelineForImage2Image) pipe_i2i = pipe pipe = None else: pipe_i2i_kwargs = { "vae": good_vae, "transformer": pipe.transformer, "torch_dtype": dtype, "use_safetensors": use_safetensors, "local_files_only": True, } if hasattr(pipe, "text_encoder") and pipe.text_encoder is not None: pipe_i2i_kwargs["text_encoder"] = pipe.text_encoder if hasattr(pipe, "tokenizer") and pipe.tokenizer is not None: pipe_i2i_kwargs["tokenizer"] = pipe.tokenizer if hasattr(pipe, "text_encoder_2") and pipe.text_encoder_2 is not None: pipe_i2i_kwargs["text_encoder_2"] = pipe.text_encoder_2 if hasattr(pipe, "tokenizer_2") and pipe.tokenizer_2 is not None: pipe_i2i_kwargs["tokenizer_2"] = pipe.tokenizer_2 pipe_i2i = AutoPipelineForImage2Image.from_pretrained(model_name, **pipe_i2i_kwargs) pipe = _release_gpu_obj(pipe) else: pipe_i2i = _release_gpu_obj(pipe_i2i) active_base_model = model_name def _move_for_inference(main_pipe, vae_obj): if isinstance(main_pipe, Flux2Pipeline) and flux2_cpu_offload_enabled: return if torch.cuda.is_available(): try: main_pipe.to("cuda") vae_obj.to("cuda") except RuntimeError as ex: if "CUDACachingAllocator" in str(ex) or "NVML_SUCCESS" in str(ex): torch.cuda.empty_cache() try: torch.cuda.ipc_collect() except Exception: pass main_pipe.to("cuda") vae_obj.to("cuda") else: raise class calculateDuration: def __init__(self, activity_name=""): self.activity_name = activity_name def __enter__(self): self.start_time = time.time() return self def __exit__(self, exc_type, exc_value, traceback): self.end_time = time.time() self.elapsed_time = self.end_time - self.start_time if self.activity_name: print(f"Elapsed time for {self.activity_name}: {self.elapsed_time:.6f} seconds") else: print(f"Elapsed time: {self.elapsed_time:.6f} seconds") def update_selection(evt: gr.SelectData, width, height, aspect_ratio, current_steps): selected_lora = loras[evt.index] new_placeholder = f"Type a prompt for {selected_lora['title']}" new_aspect_ratio = aspect_ratio lora_repo = selected_lora["repo"] target_model = LORA_TO_MODEL.get(lora_repo, base_model) lora_runtime_parameters = get_lora_runtime_parameters(lora_repo, selected_lora) lora_default_steps = lora_runtime_parameters.get("num_inference_steps") if isinstance(lora_default_steps, float): lora_default_steps = int(lora_default_steps) if is_flux2_model(target_model): flux2_steps = lora_default_steps if isinstance(lora_default_steps, int) else 7 steps_update = gr.update( minimum=flux2_steps, maximum=flux2_steps, step=1, value=flux2_steps, interactive=False, label="Steps (from LoRA defaults for FLUX.2)", ) else: safe_steps = 7 try: if current_steps is not None: safe_steps = int(current_steps) except Exception: safe_steps = 7 steps_update = gr.update( minimum=1, maximum=50, step=1, value=max(1, min(50, safe_steps)), interactive=True, label="Steps", ) updated_text = f"### Selected: [{lora_repo}](https://huggingface.co/{lora_repo}) ✅" # aspect will now use ratios if implemented, like 16:9, 4:3, 1:1 if "aspect" in selected_lora: try: new_aspect_ratio = selected_lora["aspect"] width, height = update_dimensions_on_ratio(new_aspect_ratio, height) except Exception as e: print(f"\nError in update selection aspect ratios:{e}\nSkipping") new_aspect_ratio = aspect_ratio width = width height = height return ( gr.update(placeholder=new_placeholder), updated_text, evt.index, width, height, new_aspect_ratio, upd_prompt_notes_by_index(evt.index), steps_update, ) @spaces.GPU(duration=120,progress=gr.Progress(track_tqdm=True)) def generate_image(prompt_mash, negative_prompt, steps, seed, cfg_scale, width, height, lora_scale, progress): _move_for_inference(pipe, good_vae) generator = torch.Generator(device="cuda").manual_seed(seed) flash_attention_enabled = torch.backends.cuda.flash_sdp_enabled() if flash_attention_enabled: pipe.attn_implementation="flash_attention_2" if IS_SHARED_SPACE: pipe.vae.enable_tiling() # For larger resolutions if needed else: # Compile UNet #pipe.transformer = torch.compile(pipe.transformer, mode="reduce-overhead") #pipe.enable_model_cpu_offload() #for smaller GPUs pipe.vae.enable_slicing() # Disable unnecessary features pipe.safety_checker = None print(f"\nGenerating image with prompt: {prompt_mash}\n") approx_tokens= approximate_token_count(prompt_mash) if approx_tokens > 76: print(f"\nSplitting prompt due to length: {approx_tokens}\n") prompt, prompt2 = split_prompt_precisely(prompt_mash) else: prompt = prompt_mash prompt2 = None with calculateDuration("Generating image"): # Generate image if isinstance(pipe, Flux2Pipeline): final_image = pipe.flux_pipe_call_single_final_image( prompt=prompt, prompt_2=prompt2, num_inference_steps=steps, guidance_scale=cfg_scale, width=width, height=height, generator=generator, joint_attention_kwargs={"scale": lora_scale}, output_type="pil", ) yield final_image else: for img in pipe.flux_pipe_call_that_returns_an_iterable_of_images( prompt=prompt, prompt_2=prompt2, num_inference_steps=steps, guidance_scale=cfg_scale, width=width, height=height, generator=generator, joint_attention_kwargs={"scale": lora_scale}, output_type="pil", good_vae=good_vae, ): yield img @spaces.GPU(duration=120,progress=gr.Progress(track_tqdm=True)) def generate_image_to_image(prompt_mash, negative_prompt, image_input_path, image_strength, steps, cfg_scale, width, height, lora_scale, seed, progress): _move_for_inference(pipe_i2i, good_vae) generator = torch.Generator(device="cuda").manual_seed(seed) flash_attention_enabled = torch.backends.cuda.flash_sdp_enabled() if flash_attention_enabled: pipe_i2i.attn_implementation="flash_attention_2" if IS_SHARED_SPACE: pipe_i2i.vae.enable_tiling() # For larger resolutions if needed else: # Compile UNet # pipe.transformer = torch.compile(pipe.transformer, mode="reduce-overhead") # uses the other pipe's transformer # pipe_i2i.enable_model_cpu_offload() #for smaller GPUs pipe_i2i.vae.enable_slicing() # Disable unnecessary features pipe_i2i.safety_checker = None image_input = open_image(image_input_path) print(f"\nGenerating image with prompt: {prompt_mash} and {image_input_path}\n") approx_tokens= approximate_token_count(prompt_mash) if approx_tokens > 76: print(f"\nSplitting prompt due to length: {approx_tokens}\n") prompt, prompt2 = split_prompt_precisely(prompt_mash) else: prompt = prompt_mash prompt2 = None with calculateDuration("Generating image"): # Generate image if isinstance(pipe_i2i, Flux2Pipeline): # Flux2Pipeline does not accept prompt_2; merge prompt and prompt_2 if prompt is None: merged_prompt = prompt2 elif prompt2 is None: merged_prompt = prompt else: merged_prompt = f"{prompt} {prompt2}" final_image = pipe_i2i( prompt=merged_prompt, image=image_input, num_inference_steps=steps, guidance_scale=cfg_scale, width=width, height=height, generator=generator, attention_kwargs={"scale": lora_scale}, output_type="pil", ).images[0] else: final_image = pipe_i2i( prompt=prompt, prompt_2=prompt2, image=image_input, strength=image_strength, num_inference_steps=steps, guidance_scale=cfg_scale, width=width, height=height, generator=generator, joint_attention_kwargs={"scale": lora_scale}, output_type="pil", ).images[0] return final_image def run_lora(prompt, map_option, image_input, image_strength, cfg_scale, steps, selected_index, randomize_seed, seed, width, height, lora_scale, enlarge, use_conditioned_image=False, progress=gr.Progress(track_tqdm=True)) -> Iterator[Tuple[str, int, Any]]: if selected_index is None: raise gr.Error("You must select a LoRA before proceeding.🧨") print(f"input Image: {image_input}\n") # handle selecting a conditioned image from the gallery global current_prerendered_image conditioned_image=None formatted_map_option = map_option.lower().replace(' ', '_') negative_prompt = "" if use_conditioned_image: print(f"Conditioned path: {current_prerendered_image.value}.. converting to RGB\n") # ensure the conditioned image is an image and not a string, cannot use RGBA if isinstance(current_prerendered_image.value, str): conditioned_image = open_image(current_prerendered_image.value).convert("RGB") image_input = resize_and_crop_image(conditioned_image, width, height) print(f"Conditioned Image: {image_input.size}.. converted to RGB and resized\n") if map_option != "Prompt": prompt = PROMPTS[map_option] negative_prompt = NEGATIVE_PROMPTS.get(map_option, "") selected_lora = loras[selected_index] lora_path = selected_lora["repo"] target_model = LORA_TO_MODEL.get(lora_path, base_model) lora_runtime_parameters = get_lora_runtime_parameters(lora_path, selected_lora) # detect safetensor usage use_safetensors_flag = False # check explicit "weights" field on the lora entry weight_field = selected_lora.get("weights") or "" if weight_field and "safetensor" in weight_field.lower(): use_safetensors_flag = True else: # fallback: check LORA_DETAILS for weight_name entries details = LORA_DETAILS.get(lora_path) if details: for d in details: wn = d.get("weight_name") or d.get("weights") or "" if wn and "safetensor" in wn.lower(): use_safetensors_flag = True break ensure_pipelines_for_model(target_model, use_safetensors=use_safetensors_flag, image_input=image_input) runtime_steps = lora_runtime_parameters.get("num_inference_steps") runtime_guidance = lora_runtime_parameters.get("guidance_scale") if isinstance(runtime_steps, (int, float)): steps = int(runtime_steps) elif is_flux2_model(active_base_model): steps = 8 if isinstance(runtime_guidance, (int, float)): cfg_scale = float(runtime_guidance) trigger_word = selected_lora["trigger_word"] if(trigger_word): if "trigger_position" in selected_lora: if selected_lora["trigger_position"] == "prepend": prompt_mash = f"{trigger_word} {prompt}" else: prompt_mash = f"{prompt} {trigger_word}" else: prompt_mash = f"{trigger_word} {prompt}" else: prompt_mash = prompt with calculateDuration("Unloading LoRA"): if pipe is not None: pipe.unload_lora_weights() if pipe_i2i is not None: pipe_i2i.unload_lora_weights() #LoRA weights flow with calculateDuration(f"Loading LoRA weights for {selected_lora['title']}"): pipe_to_use = pipe_i2i if image_input is not None else pipe weight_name = selected_lora.get("weights", None) pipe_to_use.load_lora_weights( lora_path, weight_name=weight_name, low_cpu_mem_usage=True ) with calculateDuration("Randomizing seed"): if randomize_seed: seed = random.randint(0, MAX_SEED) if(image_input is not None): print(f"\nGenerating image to image with seed: {seed}\n") generated_image = generate_image_to_image(prompt_mash, negative_prompt, image_input, image_strength, steps, cfg_scale, width, height, lora_scale, seed, progress) if enlarge: upscale_factor = max(1.0, min((TARGET_SIZE[0] / width), (TARGET_SIZE[1] / height))) progress(0.75, desc=f"upscaling to {upscale_factor}") upscaled_image = upscale_image(generated_image, upscale_factor) progress(1.0, desc=f"upscaled to {upscale_factor}") else: upscaled_image = generated_image # Save the upscaled image to a temporary file with NamedTemporaryFile(delete=False, suffix=".png", prefix=f"{formatted_map_option}_") as tmp_upscaled: upscaled_image.save(tmp_upscaled.name, format="PNG") temp_files.append(tmp_upscaled.name) print(f"Upscaled image saved to {tmp_upscaled.name}") final_image = tmp_upscaled.name yield final_image, seed, gr.update(visible=False) else: image_generator = generate_image(prompt_mash, negative_prompt, steps, seed, cfg_scale, width, height, lora_scale, progress) final_image = None step_counter = 0 # create a single temporary file for intermediate previews with NamedTemporaryFile(delete=False, suffix=".png", prefix=f"{formatted_map_option}_step_") as tmp_step: tmp_preview_path = tmp_step.name for image in image_generator: step_counter += 1 generated_image = image # PIL.Image.Image # overwrite the same preview file to update UI without creating many files try: generated_image.save(tmp_preview_path, format="PNG") except Exception: # fallback: save to a new temp file if overwrite fails with NamedTemporaryFile(delete=False, suffix=".png", prefix=f"{formatted_map_option}_step_") as tf: generated_image.save(tf.name, format="PNG") tmp_preview_path = tf.name progress_bar = f'
' # yield preview file path, seed, progress yield tmp_preview_path, seed, gr.update(value=progress_bar, visible=True) if enlarge: upscale_factor = max(1.0, min((TARGET_SIZE[0] / width), (TARGET_SIZE[1] / height))) progress(0.98, desc=f"upscaling to {upscale_factor}") upscaled_image = upscale_image(generated_image, upscale_factor) else: upscaled_image = generated_image # Save the upscaled image to a temporary file with NamedTemporaryFile(delete=False, suffix=".png", prefix=f"{formatted_map_option}_") as tmp_upscaled: upscaled_image.save(tmp_upscaled.name, format="PNG") temp_files.append(tmp_upscaled.name) print(f"Upscaled image saved to {tmp_upscaled.name}") final_image = tmp_upscaled.name yield final_image, seed, gr.update(value=progress_bar, visible=False) def get_huggingface_safetensors(link): split_link = link.split("/") if len(split_link) == 2: model_card = ModelCard.load(link) base_model = model_card.data.get("base_model") print(base_model) # Allows Both if base_model not in MODELS: raise Exception("Flux LoRA Not Found!") image_path = model_card.data.get("widget", [{}])[0].get("output", {}).get("url", None) trigger_word = model_card.data.get("instance_prompt", "") image_url = f"https://huggingface.co/{link}/resolve/main/{image_path}" if image_path else None fs = HfFileSystem() safetensors_name = None try: list_of_files = fs.ls(link, detail=False) for file in list_of_files: if file.endswith(".safetensors"): safetensors_name = file.split("/")[-1] if (not image_url) and file.lower().endswith((".jpg", ".jpeg", ".png", ".webp")): image_elements = file.split("/") image_url = f"https://huggingface.co/{link}/resolve/main/{image_elements[-1]}" except Exception as e: print(e) gr.Warning("You didn't include a link neither a valid Hugging Face repository with a *.safetensors LoRA") raise Exception("You didn't include a link neither a valid Hugging Face repository with a *.safetensors LoRA") return split_link[1], link, safetensors_name, trigger_word, image_url, base_model def check_custom_model(link): if(link.startswith("https://")): if(link.startswith("https://huggingface.co") or link.startswith("https://www.huggingface.co")): link_split = link.split("huggingface.co/") return get_huggingface_safetensors(link_split[1]) else: return get_huggingface_safetensors(link) def add_custom_lora(custom_lora): global loras if(custom_lora): try: title, repo, path, trigger_word, image, detected_base_model = check_custom_model(custom_lora) print(f"Loaded custom LoRA: {repo}") card = f'''
Loaded custom LoRA:

{title}

{"Using: "+trigger_word+" as the trigger word" if trigger_word else "No trigger word found. If there's a trigger word, include it in your prompt"}
''' existing_item_index = next((index for (index, item) in enumerate(loras) if item['repo'] == repo), None) if(not existing_item_index): new_item = { "image": image, "title": title, "repo": repo, "weights": path, "trigger_word": trigger_word, "base_model": detected_base_model } print(new_item) existing_item_index = len(loras) loras.append(new_item) return gr.update(visible=True, value=card), gr.update(visible=True), gr.Gallery(selected_index=None), f"Custom: {path}", existing_item_index, trigger_word except Exception as e: gr.Warning(f"Invalid LoRA: either you entered an invalid link, or a non-FLUX LoRA") return gr.update(visible=True, value=f"Invalid LoRA: either you entered an invalid link, a non-FLUX LoRA"), gr.update(visible=False), gr.update(), "", None, "" else: return gr.update(visible=False), gr.update(visible=False), gr.update(), "", None, "" def remove_custom_lora(): return gr.update(visible=False), gr.update(visible=False), gr.update(), "", None, "" def on_prerendered_gallery_selection(event_data: gr.SelectData): global current_prerendered_image selected_index = event_data.index selected_image = pre_rendered_maps_paths[selected_index] print(f"Gallery Image Selected: {selected_image}\n") current_prerendered_image.value = selected_image return current_prerendered_image def update_prompt_visibility(map_option): is_visible = (map_option == "Prompt") return ( gr.update(visible=is_visible), gr.update(visible=is_visible), gr.update(visible=is_visible) ) def copy_map_prompts_to_manual(map_option: str): """Return the manual prompt and negative prompt values for the given map option.""" # Safely fetch values prompt_val = PROMPTS.get(map_option) if map_option is not None else None negative_val = NEGATIVE_PROMPTS.get(map_option) if map_option is not None else None # If a PROMPTS/NEGATIVE_PROMPTS entry is None, do not overwrite that textbox # Return gr.update() (no-op) for components that should remain unchanged. out_prompt = gr.update(value=prompt_val) if prompt_val is not None else gr.update() out_negative = gr.update(value=negative_val) if negative_val is not None else gr.update() return out_prompt, out_negative def composite_with_control_sync(input_image, sketch_image, slider_value): if input_image is None or sketch_image is None: return input_image # Load the images using open_image() if they are provided as file paths. in_img = open_image(input_image) if isinstance(input_image, str) else input_image if in_img is None: return input_image sk_img_path, _ = get_image_from_dict(sketch_image) if not sk_img_path: return input_image sk_img = open_image(sk_img_path) if sk_img is None: return input_image # Resize sketch image if dimensions don't match input image. if in_img.size != sk_img.size: sk_img = sk_img.resize(in_img.size, Image.LANCZOS) # Now composite using the original alpha_composite_with_control function. result_img = alpha_composite_with_control(in_img, sk_img, slider_value) return result_img def replace_input_with_sketch_image(sketch_image): print(f"Sketch Image: {sketch_image}\n") sketch, is_dict = get_image_from_dict(sketch_image) if sketch is None: gr.Warning("Sketch image not found.") return None try: sketch_img = open_image(sketch) if isinstance(sketch, str) else sketch with NamedTemporaryFile(delete=False, suffix=".png", prefix="sketch_input_") as tmp_sketch: sketch_img.convert("RGBA").save(tmp_sketch.name, format="PNG") temp_files.append(tmp_sketch.name) return tmp_sketch.name except Exception as e: print(f"Error preparing sketch image for input: {e}") return sketch def on_input_image_change(image_path): if image_path is None: gr.Warning("Please upload an Input Image to get started.") return None, gr.update() img, img_path = convert_to_rgba_png(image_path) with Image.open(img_path) as pil_img: width, height = pil_img.size return [img_path, gr.update(width=width, height=height)] def update_sketch_dimensions(input_image, sketch_image): if input_image is None or sketch_image is None: return [sketch_image, gr.update()] # Load the images using open_image() if they are provided as file paths. in_img = open_image(input_image) if isinstance(input_image, str) else input_image if in_img is None: return [sketch_image, gr.update()] sk_img_path, _ = get_image_from_dict(sketch_image) if not sk_img_path: return [sketch_image, gr.update()] sk_img = open_image(sk_img_path) if sk_img is None: return [sketch_image, gr.update()] # Resize sketch image if dimensions don't match input image. if (in_img) and (in_img.size != sk_img.size): sk_img = sk_img.resize(in_img.size, Image.LANCZOS) return [sk_img, gr.update(width=in_img.width, height=in_img.height)] else: return [sk_img, gr.update()] def update_maze_maker_shape_visibility(shape): shape_key = (shape or "rect").lower() is_hex = shape_key in ("hex", "hexagon", "hexagonal") return ( gr.update(visible=not is_hex), gr.update(visible=not is_hex), gr.update(visible=is_hex), ) def generate_and_apply_maze_to_input( shape: str, width: int, height: int, radius: int, cell_size: int, use_iterative: bool, show_solution: bool, show_doors: bool, input_image: Optional[str], opacity_percent: float, maze_wall_color: str = "#000000", maze_floor_color: str = "#ffffff", maze_door_color: str = "#B0B0B0", maze_path_color: str = "#ff0000", seed: int = 0, ) -> Tuple[str, str]: shape_key = (shape or "rect").lower() if shape_key in ("hex", "hexagon", "hexagonal"): normalized_size = max(1, int(radius)) else: w = max(5, int(width)) h = max(5, int(height)) if w % 2 == 0: w += 1 if h % 2 == 0: h += 1 normalized_size = (w, h) with NamedTemporaryFile(delete=False, suffix=".png", prefix="maze_maker_") as tmp_maze: maze_output_path = tmp_maze.name if seed != 0: name, ext = os.path.splitext(maze_output_path) maze_output_path = f"{name}_s{seed}{ext}" try: # build colors for drawing and pass through to the maze generator/drawer color_dict = { 'wall': color_to_hex(maze_wall_color) or '#000000', 'floor': color_to_hex(maze_floor_color) or '#ffffff', 'door': color_to_hex(maze_door_color) or '#B0B0B0', 'path': color_to_hex(maze_path_color) or '#ff0000', # keep sensible defaults for colors not exposed in the UI 'start': 'green', } _, solution_path = run_maze_maker( shape=shape_key, size=normalized_size, cell=max(4, int(cell_size)), use_iterative=bool(use_iterative), draw=True, output=maze_output_path, show_solution=bool(show_solution), show_doors=bool(show_doors), seed=int(seed) if seed is not None else 0, color_dict=color_dict, ) temp_files.append(maze_output_path) except Exception as ex: raise gr.Error(f"Failed to generate maze: {ex}") path_info = f"path nodes: {len(solution_path)}" if solution_path else "no solution path found" status = f"Maze created ({shape_key}, size={normalized_size}, {path_info}) and applied to Input Image." # If there's no input image provided, return the generated maze image path # (returning a file path avoids Gradio creating temporary PIL files which # previously led to intermittent FileNotFoundError on chained calls). if not input_image: return maze_output_path, status try: base_img = open_image(input_image) if isinstance(input_image, str) else input_image overlay = open_image(maze_output_path) # ensure RGBA and resize overlay to match base image if overlay.mode != 'RGBA': overlay = overlay.convert('RGBA') if overlay.size != base_img.size: overlay = overlay.resize(base_img.size, Image.LANCZOS) # Resize/convert done above; composite using utility (same pattern as composite_with_control_sync) result = alpha_composite_with_control(base_img, overlay, opacity_percent) # save to temp file and return path with NamedTemporaryFile(delete=False, suffix=".png", prefix="maze_composite_") as outf: result.convert('RGBA').save(outf.name, format='PNG') temp_files.append(outf.name) return outf.name, status except Exception as ex: # fallback: return generated maze image if compositing fails print(f"Maze compositing failed: {ex}") return maze_output_path, status @spaces.GPU() def getVersions(): return versions_html() gr.set_static_paths(paths=["images/","images/images","images/prerendered","LUT/","fonts/", "assets/"]) title = "MapLore Magic (Development Features for HexaGrid)" with gr.Blocks(css_paths="style_20260316.css", title=title, theme='Surn/beeuty@==0.5.12', delete_cache=(43200, 43200), head_paths="head.htm") as app: with gr.Row(): gr.Markdown(""" # MapLore Magic (Development Features for HexaGrid) ## This project includes features that did not make it into the main project! ⬢""", elem_classes="intro") with gr.Row(): with gr.Accordion("Welcome to MapLore Magic, the ultimate tool for transforming your images into mesmerizing artworks. Whether you're a tabletop game enthusiast, a digital artist, or someone who loves unique patterns, LandLore Magic has something for you.", open=False, elem_classes="intro"): gr.Markdown (""" ## Drop an image into the Input Image and get started! ## What is MapLore Magic? MapLore Magic is a web-based application that allows you to transform your images into stunning artworks. You can customize various aspects of your images, apply filters, and use advanced AI models to generate unique visuals. ### What Can You Do? - **AI-Powered Image Generation:** Use advanced AI models to generate images based on your prompts. - **Sketch, Upload, or Select:** Guide your AI image generation with sketches, uploaded images, or pre-rendered templates. - **Image Filter [Look-Up Table (LUT)] Application:** Apply filters (LUTs) to your images for color grading and enhancement. - **Pre-rendered Maps:** Access a library of pre-rendered maps for quick and easy customization. - **Composite Images:** Composite images with control over opacity and color. - **Sketch Pad:** Draw or upload sketches to guide image generation. ### Why You'll Love It - **Fun and Easy to Use:** With an intuitive interface and real-time previews, creating unique images has never been this fun! - **Endless Creativity:** Unleash your creativity with endless customization options and see your images transform in unique ways. - **Advanced AI Models:** Leverage advanced AI models and LoRA weights for high-quality image generation and customization. ### Get Started 1. **Upload or Generate an Image:** Start by uploading your own image or generate one using our AI-powered tool. 2. **Customize Your Image:** Play around with the settings to create the perfect image. 3. **Download and Share:** Once you're happy with your creation, download it and share it with the world! ### Advanced Features - **Generative AI Integration:** Utilize models like `black-forest-labs/FLUX.1-dev` and various LoRA weights for generating unique images. - **Pre-rendered Maps:** Access a library of pre-rendered maps for quick and easy customization. - **Image Filter [Look-Up Table (LUT)] Application:** Apply filters (LUTs) to your images for color grading and enhancement. - **Composite Images:** Composite images with control over opacity and color. - **Sketch Pad:** Draw or upload sketches to guide image generation. Join the creative journey with MapLore Magic today! """, elem_classes="intro") selected_index = gr.State(None) # create user_profile state inside the Blocks so it is registered with the app user_profile = gr.State(None) with gr.Row(): with gr.Column(scale=2): progress_bar = gr.Markdown(elem_id="progress",visible=False) input_image = gr.Image( label="Input Image", type="filepath", interactive=True, elem_classes="centered solid imgcontainer", key="imgInput", image_mode="RGB", format="PNG" ) with gr.Column(scale=1): with gr.Accordion("Image Filters", open = False): with gr.Row(): with gr.Column(): lut_filename = gr.Textbox( value="", label="Look Up Table (LUT) File Name", elem_id="lutFileName") with gr.Column(): lut_file = gr.File( value=None, file_count="single", file_types=[".cube"], type="filepath", label="LUT cube File", height=120) with gr.Row(): lut_intensity = gr.Slider(label="Filter Intensity", minimum=-200, maximum=200, value=100, info="0 none, negative inverts the filter", interactive=True) apply_lut_button = gr.Button("Apply Filter (LUT)", elem_classes="solid", elem_id="apply_lut_button") with gr.Row(): lut_example_image = gr.Image(type="pil", label="Filter (LUT) Example Image", value=default_lut_example_img, format="png") with gr.Row(): with gr.Column(): gr.Markdown(""" ### Included Filters (LUTs) Try on Example Image then APPLY FILTER to Input Image. *-none.cube files are placebo controls """, elem_id="lut_markdown") with gr.Column(): gr.Examples(elem_id="lut_examples", examples=[[f] for f in lut_files], inputs=[lut_filename], outputs=[lut_filename], label="Select a Filter (LUT) file to populate the LUT File Name field", examples_per_page = 25, ) lut_file.change(get_filename, inputs=[lut_file], outputs=[lut_filename]) lut_filename.change(show_lut, inputs=[lut_filename, input_image, lut_intensity], outputs=[lut_example_image], scroll_to_output=True) lut_intensity.change(show_lut, inputs=[lut_filename, input_image, lut_intensity], outputs=[lut_example_image]) apply_lut_button.click( lambda lut_filename, input_image, lut_intensity: gr.Warning("Please upload an Input Image to get started.") if input_image is None else apply_lut_to_image_path(lut_filename, input_image, lut_intensity)[1], inputs=[lut_filename, input_image, lut_intensity], outputs=[input_image], scroll_to_output=True ) with gr.Accordion("Color Composite", open = False): with gr.Row(): composite_color = gr.ColorPicker(label="Color", value="#ede9ac44") composite_opacity = gr.Slider(label="Opacity %", minimum=0, maximum=100, value=50, interactive=True) with gr.Row(): composite_button = gr.Button("Composite", elem_classes="solid") with gr.Accordion("Maze Maker", open=False, elem_id="maze_maker"): maze_shape = gr.Dropdown( label="Maze Shape", choices=["rect", "tri", "hex"], value="rect", elem_classes="solid" ) with gr.Row(): maze_width = gr.Slider(label="Maze Width (rect/tri)", minimum=5, maximum=199, step=2, value=37, interactive=True) maze_height = gr.Slider(label="Maze Height (rect/tri)", minimum=5, maximum=199, step=2, value=21, interactive=True) maze_radius = gr.Slider(label="Hex Radius", minimum=2, maximum=40, step=1, value=8, interactive=True, visible=False) maze_cell_size = gr.Slider(label="Cell Size", minimum=8, maximum=80, step=1, value=20, interactive=True) # opacity for maze overlay when compositing onto the Input Image maze_opacity = gr.Slider(label="Maze Opacity %", minimum=0, maximum=100, value=100, interactive=True) with gr.Row(): maze_use_iterative = gr.Checkbox(label="Use Iterative Generator", value=True) maze_show_solution = gr.Checkbox(label="Show Solution Path (red)", value=True) maze_show_doors = gr.Checkbox(label="Show Cell Outlines", value=True) maze_seed = gr.Slider(label="Maze Seed", minimum=0, maximum=MAX_SEED, step=1, value=0, randomize=True, elem_id="maze_rnd_seed", info="0 = randomize") maze_wall_color = gr.ColorPicker(label="Wall Color", value="#000000") maze_floor_color = gr.ColorPicker(label="Floor Color", value="#ffffff") maze_door_color = gr.ColorPicker(label="Door Color", value="#B0B0B0") maze_path_color = gr.ColorPicker(label="Path Color", value="#ff0000") maze_apply_button = gr.Button("Create Maze and Apply to Input Image", elem_classes="solid") maze_status = gr.Markdown(value="") with gr.Accordion("Sketch Pad", open = False, elem_id="sketchpd"): with gr.Row(min_height=480): sketch_image = gr.Sketchpad( label="Sketch Image", type="filepath", #invert_colors=True, sources=['upload','clipboard'], #tool=['editor','select','color-sketch'], placeholder="Draw a sketch or upload an image. Currently broken in gradio 5.17.1", interactive=True, elem_id="sketchpad", elem_classes="centered solid imgcontainer", key="imgSketch", image_mode="RGBA", format="PNG", brush=gr.Brush(colors=['#000000', '#3270cc', '#cc3232', '#32cc70', '#ffffff']), canvas_size=(640,480) ) with gr.Row(): with gr.Column(scale=1): sketch_replace_input_image_button = gr.Button( "Replace Input Image with sketch", elem_id="sketch_replace_input_image_button", elem_classes="solid" ) with gr.Column(scale=2): alpha_composite_slider = gr.Slider(0,100,50,0.5, label="Alpha Composite Sketch to Input Image", elem_id="alpha_composite_slider") with gr.Row(): with gr.Accordion("Generative AI", open = True ): with gr.Column(): with gr.Row(): with gr.Column(scale=5): map_options = gr.Dropdown( label="Map Options*", choices=list(PROMPTS.keys()), value="Alien Landscape", elem_classes="solid", scale=0, ) with gr.Column(scale=1): # Small button placed beside the dropdown copy_to_manual_prompt = gr.Button( "Copy to Manual Prompt", elem_classes="solid", elem_id="copy_to_manual_prompt", size="sm", ) prompt = gr.Textbox( label="Prompt", visible=False, elem_classes="solid", value="Planetary overhead view, directly from above, centered on the planet’s surface, (rectangular tabletop_map) alien planet map, Battletech_boardgame scifi world with forests, lakes, oceans, continents and snow at the top and bottom, (middle is dark, no_reflections, no_shadows), looking straight down.", lines=4 ) negative_prompt_textbox = gr.Textbox( label="Negative Prompt", visible=False, elem_classes="solid", value="Earth, low quality, bad anatomy, blurry, cropped, worst quality, shadows, people, humans, reflections, shadows, realistic map of the Earth, isometric, text" ) prompt_notes_label = gr.Label( "Choose a LoRa style or add an image. YOU MUST CLEAR THE IMAGE TO START OVER ", elem_classes="solid centered small", show_label=False, visible=False ) # Keep the change event to maintain functionality map_options.change( fn=update_prompt_visibility, inputs=[map_options], outputs=[prompt, negative_prompt_textbox, prompt_notes_label] ) # copy Map Options prompts into the manual prompt boxes copy_to_manual_prompt.click( fn=copy_map_prompts_to_manual, inputs=[map_options], outputs=[prompt, negative_prompt_textbox], ) with gr.Row(): with gr.Column(scale=1): generate_button = gr.Button("Generate From Map Options, Input Image and LoRa Style", variant="primary", elem_id="gen_btn") with gr.Accordion("Image Styles*", open=False): selected_info = gr.Markdown("") lora_gallery = gr.Gallery( [(item["image"], item["title"]) for item in loras], label="LoRA Styles", allow_preview=False, columns=3, elem_id="lora_gallery", show_share_button=False ) with gr.Accordion("Custom LoRA", open=False): with gr.Group(): custom_lora = gr.Textbox(label="Enter Custom LoRA. **NOT TESTED**", placeholder="prithivMLmods/Canopus-LoRA-Flux-Anime") gr.Markdown("[Check the list of FLUX LoRA's](https://huggingface.co/models?other=base_model:adapter:black-forest-labs/FLUX.1-dev)", elem_id="lora_list") custom_lora_info = gr.HTML(visible=False) custom_lora_button = gr.Button("Remove custom LoRA", visible=False) with gr.Column(scale=2): generate_input_image_from_gallery = gr.Button( "Generate AI Image from Template Image", elem_id="generate_input_image_from_gallery", elem_classes="solid", variant="primary" ) with gr.Accordion("Template Images", open = False): with gr.Row(): with gr.Column(scale=1): # Gallery from PRE_RENDERED_IMAGES GOES HERE prerendered_image_gallery = gr.Gallery(label="Template Gallery", show_label=True, value=build_prerendered_images_by_quality(3,'thumbnail'), elem_id="gallery", elem_classes="solid", type="filepath", columns=[3], rows=[3], preview=False ,object_fit="contain", height="auto", format="png",allow_preview=False) with gr.Column(scale=1): # def handle_login(request: gr.Request): # # Extract user information from the request # user_info = { # "username": request.username, # "session_hash": request.session_hash, # "headers": dict(request.headers), # "client": request.client, # "query_params": dict(request.query_params), # "path_params": dict(request.path_params) # } # print(f"\n{user_info}\n") # return user_info replace_input_image_button = gr.Button( "Replace Input Image", elem_id="prerendered_replace_input_image_button", elem_classes="solid" ) # login_button = gr.LoginButton() # user_info_output = gr.JSON(label="User Information") # login_button.click(fn=handle_login, inputs=[], outputs=user_info_output) with gr.Row(): with gr.Accordion("Advanced Settings", open=False): with gr.Row(): image_strength = gr.Slider(label="Image Guidance Strength (prompt percentage)", info="Lower means more image influence", minimum=0.1, maximum=1.0, step=0.01, value=0.85) with gr.Column(): with gr.Row(): cfg_scale = gr.Slider(label="CFG Scale", minimum=1, maximum=20, step=0.5, value=5.0) steps = gr.Slider(label="Steps", minimum=1, maximum=50, step=1, value=7) with gr.Row(): negative_prompt_textbox = gr.Textbox( label="Negative Prompt", visible=False, elem_classes="solid", value="Earth, low quality, bad anatomy, blurry, cropped, worst quality, shadows, people, humans, reflections, shadows, realistic map of the Earth, isometric, text" ) # Add Dropdown for sizing of Images, height and width based on selection. Options are 16x9, 16x10, 4x5, 1x1 # The values of height and width are based on common resolutions for each aspect ratio # Default to 16x9, 1024x576 image_size_ratio = gr.Dropdown(label="Image Aspect Ratio", choices=["16:9", "16:10", "4:5", "4:3", "2:1","3:2","1:1", "9:16", "10:16", "5:4", "3:4","1:2", "2:3"], value="16:9", elem_classes="solid", type="value", scale=0, interactive=True) width = gr.Slider(label="Width", minimum=256, maximum=2560, step=16, value=1024, interactive=False) height = gr.Slider(label="Height", minimum=256, maximum=1536, step=64, value=512) enlarge_to_default = gr.Checkbox(label="Auto Enlarge to Default Size", value=False) image_size_ratio.change( fn=update_dimensions_on_ratio, inputs=[image_size_ratio, height], outputs=[width, height] ) height.change( fn=lambda *args: update_dimensions_on_ratio(*args)[0], inputs=[image_size_ratio, height], outputs=[width] ) with gr.Row(): randomize_seed = gr.Checkbox(False, label="Randomize seed",elem_id="rnd_seed_chk") seed = gr.Slider(label="Seed", minimum=0, maximum=MAX_SEED, step=1, value=0, randomize=True, elem_id="rnd_seed") lora_scale = gr.Slider(label="LoRA Scale", minimum=0, maximum=3, step=0.01, value=1.01) with gr.Row(): login_button = gr.LoginButton(logout_value=f"Logout({user_info['username']} ({user_info['level']}))", size="md", elem_classes="solid centered", elem_id="hf_login_btn", icon="./assets/favicon.ico") # Create a JSON component to display the user information user_info_output = gr.JSON(label="User Information:") # Set up the event listener for the login button click login_button.click(fn=wait5, inputs=[gr.State("Login/logout"), gr.State(10)], outputs=[] ).then( fn=handle_login, inputs=[], outputs=[user_info_output, login_button, user_profile] ) with gr.Row(): gr.HTML(value=getVersions(), visible=True, elem_id="versions") # Event Handlers composite_button.click( fn=lambda input_image, composite_color, composite_opacity: gr.Warning("Please upload an Input Image to get started.") if input_image is None else change_color(input_image, composite_color, composite_opacity), inputs=[input_image, composite_color, composite_opacity], outputs=[input_image] ) maze_shape.change( fn=update_maze_maker_shape_visibility, inputs=[maze_shape], outputs=[maze_width, maze_height, maze_radius] ) maze_apply_button.click( fn=generate_and_apply_maze_to_input, inputs=[maze_shape, maze_width, maze_height, maze_radius, maze_cell_size, maze_use_iterative, maze_show_solution, maze_show_doors, input_image, maze_opacity, maze_wall_color, maze_floor_color, maze_door_color, maze_path_color, maze_seed], outputs=[input_image, maze_status], scroll_to_output=True, api_name="generate_and_apply_maze" ).then( fn=update_sketch_dimensions, inputs=[input_image, sketch_image], outputs=[sketch_image, sketch_image] ) input_image.input( fn=on_input_image_change, inputs=[input_image], outputs=[input_image,sketch_image], scroll_to_output=True, ) #use conditioned_image as the input_image for generate_input_image_click generate_input_image_from_gallery.click( fn=run_lora, inputs=[prompt, map_options, input_image, image_strength, cfg_scale, steps, selected_index, randomize_seed, seed, width, height, lora_scale, enlarge_to_default, gr.State(True)], outputs=[input_image, seed, progress_bar], scroll_to_output=True, api_name="run_lora" ).then( fn=update_sketch_dimensions, inputs=[input_image, sketch_image], outputs=[sketch_image, sketch_image] ) prerendered_image_gallery.select( fn=on_prerendered_gallery_selection, inputs=None, outputs=gr.State(current_prerendered_image), # Update the state with the selected image show_api=False, scroll_to_output=True ) alpha_composite_slider.change( fn=composite_with_control_sync, inputs=[input_image, sketch_image, alpha_composite_slider], outputs=[input_image], scroll_to_output=True ) sketch_replace_input_image_button.click( lambda sketch_image: replace_input_with_sketch_image(sketch_image), inputs=[sketch_image], outputs=[input_image], scroll_to_output=True ) # replace input image with selected prerendered image gallery selection replace_input_image_button.click( lambda: current_prerendered_image.value, inputs=None, outputs=[input_image], scroll_to_output=True ).then( fn=update_sketch_dimensions, inputs=[input_image, sketch_image], outputs=[sketch_image, sketch_image] ) lora_gallery.select( update_selection, inputs=[width, height, image_size_ratio, steps], outputs=[prompt, selected_info, selected_index, width, height, image_size_ratio, prompt_notes_label, steps] ) custom_lora.input( add_custom_lora, inputs=[custom_lora], outputs=[custom_lora_info, custom_lora_button, lora_gallery, selected_info, selected_index, prompt] ) custom_lora_button.click( remove_custom_lora, outputs=[custom_lora_info, custom_lora_button, lora_gallery, selected_info, selected_index, custom_lora] ) gr.on( triggers=[generate_button.click, prompt.submit], fn=run_lora, inputs=[prompt, map_options, input_image, image_strength, cfg_scale, steps, selected_index, randomize_seed, seed, width, height, lora_scale, enlarge_to_default, gr.State(False)], outputs=[input_image, seed, progress_bar] ).then( fn=update_sketch_dimensions, inputs=[input_image, sketch_image], outputs=[sketch_image, sketch_image] ) # gr.api(ping, api_name="ping") gr.api(generate_and_apply_maze_to_input, api_name="generate_and_apply_maze") load_env_vars(dotenv_path) logging.basicConfig( format="[%(levelname)s] %(asctime)s %(message)s", level=logging.INFO ) logging.info("Environment Variables: %s" % os.environ) app.queue(max_size=10, api_open=True) app.launch(allowed_paths=["assets","/","./assets","images","./images", "./images/prerendered"], favicon_path="./assets/favicon.ico", max_file_size="10mb", show_api=True, server_name='0.0.0.0', server_port=7860)