| import os
|
| import random
|
| import numpy as np
|
| import gradio as gr
|
| from PIL import Image, ImageChops
|
| from typing import Dict, Any, List
|
|
|
| from core.settings import (
|
| INPUT_DIR, MULTIPLIERS_MAP, LORA_DIR, EMBEDDING_DIR, VAE_DIR,
|
| FEATURES_CONFIG, TASK_FEATURES_CONFIG
|
| )
|
| from utils.app_utils import (
|
| sanitize_filename,
|
| get_lora_path,
|
| get_embedding_path,
|
| ensure_controlnet_model_downloaded,
|
| ensure_ipadapter_models_downloaded,
|
| _ensure_model_downloaded,
|
| ensure_sd3_ipadapter_models_downloaded,
|
| get_vae_path,
|
| )
|
|
|
| def process_pipeline_inputs(ui_inputs: Dict[str, Any], progress: gr.Progress, workflow_model_type: str) -> Dict[str, Any]:
|
| task_type = ui_inputs['task_type']
|
| temp_files_to_clean = []
|
|
|
| arch_enabled_chains = FEATURES_CONFIG.get(workflow_model_type, {}).get('enabled_chains', [])
|
| task_enabled_chains = TASK_FEATURES_CONFIG.get(task_type, {}).get('enabled_chains', [])
|
| enabled_chains = set([c for c in arch_enabled_chains if c in task_enabled_chains])
|
|
|
| multiplier = MULTIPLIERS_MAP.get(workflow_model_type, 8)
|
| img_w, img_h = 0, 0
|
| if task_type == 'txt2img':
|
| img_w = int(ui_inputs.get('width', 0))
|
| img_h = int(ui_inputs.get('height', 0))
|
| elif task_type == 'img2img':
|
| input_image_pil = ui_inputs.get('img2img_image')
|
| if input_image_pil:
|
| img_w, img_h = input_image_pil.width, input_image_pil.height
|
| elif task_type == 'inpaint':
|
| inpaint_img = ui_inputs.get('inpaint_image')
|
| inpaint_dict = ui_inputs.get('inpaint_image_dict')
|
| if inpaint_img:
|
| img_w, img_h = inpaint_img.width, inpaint_img.height
|
| elif inpaint_dict and inpaint_dict.get('background'):
|
| img_w, img_h = inpaint_dict['background'].width, inpaint_dict['background'].height
|
| elif task_type == 'outpaint':
|
| input_image_pil = ui_inputs.get('outpaint_image')
|
| if input_image_pil:
|
| img_w, img_h = input_image_pil.width, input_image_pil.height
|
| elif task_type == 'hires_fix':
|
| input_image_pil = ui_inputs.get('hires_image')
|
| if input_image_pil:
|
| img_w, img_h = input_image_pil.width, input_image_pil.height
|
|
|
| if img_w > 0 and img_h > 0:
|
| if (img_w % multiplier != 0) or (img_h % multiplier != 0):
|
| warning_msg = f"Width and height must be multiples of {multiplier} for this model."
|
| raise gr.Error(warning_msg)
|
|
|
| lora_data = ui_inputs.get('lora_data', [])
|
| active_loras_for_gpu, active_loras_for_meta = [], []
|
| if 'lora' in enabled_chains and lora_data:
|
| sources, ids, scales, files = lora_data[0::4], lora_data[1::4], lora_data[2::4], lora_data[3::4]
|
| for i, (source, lora_id, scale, _) in enumerate(zip(sources, ids, scales, files)):
|
| if scale > 0 and lora_id and lora_id.strip():
|
| lora_filename = None
|
| if source in ("Upload File", "File"):
|
| raw_id = lora_id.strip().replace('\\', '/')
|
| rel_name = raw_id if raw_id.startswith("upload_file/") else f"upload_file/{raw_id}"
|
| local_path = os.path.join(LORA_DIR, rel_name)
|
| if not os.path.exists(local_path):
|
| if os.path.exists(os.path.join(LORA_DIR, raw_id)):
|
| local_path = os.path.join(LORA_DIR, raw_id)
|
| else:
|
| raise gr.Error(f"Uploaded LoRA file '{lora_id}' no longer exists on server. Please re-upload it.")
|
| lora_filename = os.path.relpath(local_path, LORA_DIR).replace('\\', '/')
|
| elif source in ("Civitai", "Hugging Face"):
|
| local_path, status = get_lora_path(source, lora_id, os.environ.get("CIVITAI_API_KEY", ""), progress)
|
| if local_path: lora_filename = os.path.relpath(local_path, LORA_DIR).replace('\\', '/')
|
| else: raise gr.Error(f"Failed to prepare LoRA {lora_id}: {status}")
|
|
|
| if lora_filename:
|
| active_loras_for_gpu.append({"lora_name": lora_filename, "strength_model": scale, "strength_clip": scale})
|
| active_loras_for_meta.append(f"{source} {lora_id}:{scale}")
|
|
|
| ui_inputs['denoise'] = 1.0
|
| if task_type == 'img2img': ui_inputs['denoise'] = ui_inputs.get('img2img_denoise', 0.7)
|
| elif task_type == 'hires_fix': ui_inputs['denoise'] = ui_inputs.get('hires_denoise', 0.55)
|
| elif task_type == 'inpaint': ui_inputs['denoise'] = ui_inputs.get('inpaint_denoise', 1.0)
|
|
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
|
|
| if task_type == 'img2img':
|
| input_image_pil = ui_inputs.get('img2img_image')
|
| if not input_image_pil:
|
| raise gr.Error("Please upload an image for Image-to-Image.")
|
| temp_file_path = os.path.join(INPUT_DIR, f"temp_input_{random.randint(1000, 9999)}.png")
|
| input_image_pil.save(temp_file_path, "PNG")
|
| ui_inputs['input_image'] = os.path.basename(temp_file_path)
|
| temp_files_to_clean.append(temp_file_path)
|
| ui_inputs['width'] = input_image_pil.width
|
| ui_inputs['height'] = input_image_pil.height
|
|
|
| elif task_type == 'inpaint':
|
| inpaint_img = ui_inputs.get('inpaint_image')
|
| inpaint_dict = ui_inputs.get('inpaint_image_dict')
|
|
|
| if inpaint_img:
|
| temp_file_path = os.path.join(INPUT_DIR, f"temp_inpaint_{random.randint(1000, 9999)}.png")
|
| inpaint_img.save(temp_file_path, "PNG")
|
| ui_inputs['input_image'] = os.path.basename(temp_file_path)
|
| temp_files_to_clean.append(temp_file_path)
|
| ui_inputs['width'] = inpaint_img.width
|
| ui_inputs['height'] = inpaint_img.height
|
| elif inpaint_dict and inpaint_dict.get('background') and inpaint_dict.get('layers'):
|
| background_img = inpaint_dict['background'].convert("RGBA")
|
| composite_mask_pil = Image.new('L', background_img.size, 0)
|
| for layer in inpaint_dict['layers']:
|
| if layer:
|
| layer_alpha = layer.split()[-1]
|
| composite_mask_pil = ImageChops.lighter(composite_mask_pil, layer_alpha)
|
|
|
| inverted_mask_alpha = Image.fromarray(255 - np.array(composite_mask_pil), mode='L')
|
| r, g, b, _ = background_img.split()
|
| composite_image_with_mask = Image.merge('RGBA', [r, g, b, inverted_mask_alpha])
|
|
|
| temp_file_path = os.path.join(INPUT_DIR, f"temp_inpaint_composite_{random.randint(1000, 9999)}.png")
|
| composite_image_with_mask.save(temp_file_path, "PNG")
|
|
|
| ui_inputs['input_image'] = os.path.basename(temp_file_path)
|
| temp_files_to_clean.append(temp_file_path)
|
| ui_inputs.pop('inpaint_mask', None)
|
| else:
|
| raise gr.Error("Inpainting requires an input image with a mask.")
|
|
|
| elif task_type == 'outpaint':
|
| input_image_pil = ui_inputs.get('outpaint_image')
|
| if not input_image_pil:
|
| raise gr.Error("Please upload an image for Outpainting.")
|
| temp_file_path = os.path.join(INPUT_DIR, f"temp_input_{random.randint(1000, 9999)}.png")
|
| input_image_pil.save(temp_file_path, "PNG")
|
| ui_inputs['input_image'] = os.path.basename(temp_file_path)
|
| temp_files_to_clean.append(temp_file_path)
|
|
|
| ui_inputs['megapixels'] = 0.25
|
| ui_inputs['grow_mask_by'] = ui_inputs.get('feathering', 10)
|
|
|
| elif task_type == 'hires_fix':
|
| input_image_pil = ui_inputs.get('hires_image')
|
| if not input_image_pil:
|
| raise gr.Error("Please upload an image for Hires Fix.")
|
| temp_file_path = os.path.join(INPUT_DIR, f"temp_input_{random.randint(1000, 9999)}.png")
|
| input_image_pil.save(temp_file_path, "PNG")
|
| ui_inputs['input_image'] = os.path.basename(temp_file_path)
|
| temp_files_to_clean.append(temp_file_path)
|
|
|
| embedding_data = ui_inputs.get('embedding_data', [])
|
| if 'embedding' in enabled_chains and embedding_data:
|
| emb_sources, emb_ids, emb_files = embedding_data[0::3], embedding_data[1::3], embedding_data[2::3]
|
| for i, (source, emb_id, _) in enumerate(zip(emb_sources, emb_ids, emb_files)):
|
| if emb_id and emb_id.strip():
|
| if source in ("Civitai", "Hugging Face"):
|
| local_path, status = get_embedding_path(source, emb_id, os.environ.get("CIVITAI_API_KEY", ""), progress)
|
| if not local_path:
|
| raise gr.Error(f"Failed to prepare Embedding {emb_id}: {status}")
|
| elif source in ("Upload File", "File"):
|
|
|
|
|
| pass
|
|
|
| controlnet_data = ui_inputs.get('controlnet_data', [])
|
| active_controlnets = []
|
| if 'controlnet' in enabled_chains and controlnet_data:
|
| (cn_images, _, _, cn_strengths, cn_filepaths) = [controlnet_data[i::5] for i in range(5)]
|
| for i in range(len(cn_images)):
|
| if cn_images[i] and cn_strengths[i] > 0 and cn_filepaths[i] and cn_filepaths[i] != "None":
|
| ensure_controlnet_model_downloaded(cn_filepaths[i], progress)
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| cn_temp_path = os.path.join(INPUT_DIR, f"temp_cn_{i}_{random.randint(1000, 9999)}.png")
|
| cn_images[i].save(cn_temp_path, "PNG")
|
| temp_files_to_clean.append(cn_temp_path)
|
| active_controlnets.append({
|
| "image": os.path.basename(cn_temp_path), "strength": cn_strengths[i],
|
| "start_percent": 0.0, "end_percent": 1.0, "control_net_name": cn_filepaths[i]
|
| })
|
|
|
| anima_controlnet_lllite_data = ui_inputs.get('anima_controlnet_lllite_data', [])
|
| active_anima_controlnets = []
|
| if 'anima_controlnet_lllite' in enabled_chains and anima_controlnet_lllite_data:
|
| (cn_images, _, _, cn_strengths, cn_filepaths, cn_starts, cn_ends) = [anima_controlnet_lllite_data[i::7] for i in range(7)]
|
| for i in range(len(cn_images)):
|
| if cn_images[i] and cn_strengths[i] > 0 and cn_filepaths[i] and cn_filepaths[i] != "None":
|
| _ensure_model_downloaded(cn_filepaths[i], progress)
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| cn_temp_path = os.path.join(INPUT_DIR, f"temp_anima_cn_{i}_{random.randint(1000, 9999)}.png")
|
| cn_images[i].save(cn_temp_path, "PNG")
|
| temp_files_to_clean.append(cn_temp_path)
|
| active_anima_controlnets.append({
|
| "image": os.path.basename(cn_temp_path), "strength": cn_strengths[i],
|
| "start_percent": cn_starts[i], "end_percent": cn_ends[i], "control_net_name": cn_filepaths[i]
|
| })
|
|
|
| diffsynth_controlnet_data = ui_inputs.get('diffsynth_controlnet_data', [])
|
| active_diffsynth_controlnets = []
|
| if 'diffsynth_controlnet' in enabled_chains and diffsynth_controlnet_data:
|
| (cn_images, _, _, cn_strengths, cn_filepaths) = [diffsynth_controlnet_data[i::5] for i in range(5)]
|
| for i in range(len(cn_images)):
|
| if cn_images[i] and cn_strengths[i] > 0 and cn_filepaths[i] and cn_filepaths[i] != "None":
|
| ensure_controlnet_model_downloaded(cn_filepaths[i], progress)
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| cn_temp_path = os.path.join(INPUT_DIR, f"temp_diffsynth_cn_{i}_{random.randint(1000, 9999)}.png")
|
| cn_images[i].save(cn_temp_path, "PNG")
|
| temp_files_to_clean.append(cn_temp_path)
|
| active_diffsynth_controlnets.append({
|
| "image": os.path.basename(cn_temp_path), "strength": cn_strengths[i],
|
| "control_net_name": cn_filepaths[i]
|
| })
|
|
|
| krea2_controlnet_data = ui_inputs.get('krea2_controlnet_data', [])
|
| active_krea2_controlnets = []
|
| if 'krea2_controlnet' in enabled_chains and krea2_controlnet_data:
|
| (cn_images, _, _, cn_strengths, cn_filepaths) = [krea2_controlnet_data[i::5] for i in range(5)]
|
| for i in range(len(cn_images)):
|
| if cn_images[i] and cn_strengths[i] > 0 and cn_filepaths[i] and cn_filepaths[i] != "None":
|
| ensure_controlnet_model_downloaded(cn_filepaths[i], progress)
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| cn_temp_path = os.path.join(INPUT_DIR, f"temp_krea2_cn_{i}_{random.randint(1000, 9999)}.png")
|
| cn_images[i].save(cn_temp_path, "PNG")
|
| temp_files_to_clean.append(cn_temp_path)
|
| active_krea2_controlnets.append({
|
| "image": os.path.basename(cn_temp_path), "strength": cn_strengths[i],
|
| "control_net_name": cn_filepaths[i]
|
| })
|
|
|
| ipadapter_data = ui_inputs.get('ipadapter_data', [])
|
| active_ipadapters = []
|
| if 'ipadapter' in enabled_chains and ipadapter_data:
|
| num_ipa_units = (len(ipadapter_data) - 5) // 3
|
| final_preset, final_weight, final_lora_strength, final_embeds_scaling, final_combine_method = ipadapter_data[-5:]
|
| ipa_images, ipa_weights, ipa_lora_strengths = [ipadapter_data[i*num_ipa_units:(i+1)*num_ipa_units] for i in range(3)]
|
| all_presets_to_download = set()
|
| for i in range(num_ipa_units):
|
| if ipa_images[i] and ipa_weights[i] > 0 and final_preset:
|
| all_presets_to_download.add(final_preset)
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| ipa_temp_path = os.path.join(INPUT_DIR, f"temp_ipa_{i}_{random.randint(1000, 9999)}.png")
|
| ipa_images[i].save(ipa_temp_path, "PNG")
|
| temp_files_to_clean.append(ipa_temp_path)
|
| active_ipadapters.append({
|
| "image": os.path.basename(ipa_temp_path), "preset": final_preset,
|
| "weight": ipa_weights[i], "lora_strength": ipa_lora_strengths[i]
|
| })
|
| if active_ipadapters and final_preset:
|
| all_presets_to_download.add(final_preset)
|
| for preset in all_presets_to_download:
|
| ensure_ipadapter_models_downloaded(preset, progress)
|
|
|
| model_type_key = 'sd15' if workflow_model_type == 'sd15' else 'sdxl'
|
| if active_ipadapters:
|
| active_ipadapters.append({
|
| 'is_final_settings': True, 'model_type': model_type_key, 'final_preset': final_preset,
|
| 'final_weight': final_weight, 'final_lora_strength': final_lora_strength,
|
| 'final_embeds_scaling': final_embeds_scaling, 'final_combine_method': final_combine_method
|
| })
|
|
|
| flux1_ipadapter_data = ui_inputs.get('flux1_ipadapter_data', [])
|
| active_flux1_ipadapters = []
|
| if 'flux1_ipadapter' in enabled_chains and flux1_ipadapter_data:
|
| num_units = len(flux1_ipadapter_data) // 4
|
| f_images = flux1_ipadapter_data[0*num_units : 1*num_units]
|
| f_weights = flux1_ipadapter_data[1*num_units : 2*num_units]
|
| f_starts = flux1_ipadapter_data[2*num_units : 3*num_units]
|
| f_ends = flux1_ipadapter_data[3*num_units : 4*num_units]
|
| for i in range(len(f_images)):
|
| if f_images[i] and f_weights[i] > 0:
|
| for filename in ["ip-adapter.bin"]:
|
| _ensure_model_downloaded(filename, progress)
|
|
|
| from huggingface_hub import snapshot_download
|
| progress(0.5, desc="Caching HF SigLIP model...")
|
| snapshot_download(
|
| repo_id="google/siglip-so400m-patch14-384",
|
| allow_patterns=["*.json", "*.safetensors", "*.txt"],
|
| ignore_patterns=["*.msgpack", "*.h5", "*.bin"]
|
| )
|
|
|
| temp_path = os.path.join(INPUT_DIR, f"temp_fipa_{i}_{random.randint(1000, 9999)}.png")
|
| f_images[i].save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_flux1_ipadapters.append({
|
| "image": os.path.basename(temp_path),
|
| "weight": f_weights[i], "start_percent": f_starts[i], "end_percent": f_ends[i]
|
| })
|
|
|
| sd3_ipadapter_data = ui_inputs.get('sd3_ipadapter_chain', [])
|
| active_sd3_ipadapters = []
|
| if 'sd3_ipadapter' in enabled_chains and sd3_ipadapter_data:
|
| num_units = len(sd3_ipadapter_data) // 4
|
| s_images = sd3_ipadapter_data[0*num_units : 1*num_units]
|
| s_weights = sd3_ipadapter_data[1*num_units : 2*num_units]
|
| s_starts = sd3_ipadapter_data[2*num_units : 3*num_units]
|
| s_ends = sd3_ipadapter_data[3*num_units : 4*num_units]
|
| sd3_ipa_downloaded = False
|
| for i in range(len(s_images)):
|
| if s_images[i] and s_weights[i] > 0:
|
| if not sd3_ipa_downloaded:
|
| ensure_sd3_ipadapter_models_downloaded(progress)
|
| sd3_ipa_downloaded = True
|
| temp_path = os.path.join(INPUT_DIR, f"temp_s3ipa_{i}_{random.randint(1000, 9999)}.png")
|
| s_images[i].save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_sd3_ipadapters.append({
|
| "image": os.path.basename(temp_path),
|
| "weight": s_weights[i], "start_percent": s_starts[i], "end_percent": s_ends[i]
|
| })
|
|
|
| style_data = ui_inputs.get('style_data', [])
|
| active_styles = []
|
| if 'style' in enabled_chains and style_data:
|
| num_units = len(style_data) // 2
|
| st_images = style_data[0*num_units : 1*num_units]
|
| st_strengths = style_data[1*num_units : 2*num_units]
|
| style_models_downloaded = False
|
| for i in range(len(st_images)):
|
| if st_images[i] and st_strengths[i] > 0:
|
| if not style_models_downloaded:
|
| _ensure_model_downloaded("sigclip_vision_patch14_384.safetensors", progress)
|
| _ensure_model_downloaded("flux1-redux-dev.safetensors", progress)
|
| style_models_downloaded = True
|
| temp_path = os.path.join(INPUT_DIR, f"temp_style_{i}_{random.randint(1000, 9999)}.png")
|
| st_images[i].save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_styles.append({
|
| "image": os.path.basename(temp_path), "strength": st_strengths[i]
|
| })
|
|
|
| reference_latent_data = ui_inputs.get('reference_latent_data', [])
|
| active_reference_latents = []
|
| if 'reference_latent' in enabled_chains and reference_latent_data:
|
| for img in reference_latent_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_reference_latents.append(os.path.basename(temp_path))
|
|
|
| hidream_o1_reference_data = ui_inputs.get('hidream_o1_reference_data', [])
|
| active_hidream_o1_reference = []
|
| if 'hidream_o1_reference' in enabled_chains and hidream_o1_reference_data:
|
| for img in hidream_o1_reference_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_ho1_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_hidream_o1_reference.append(os.path.basename(temp_path))
|
|
|
| sensenova_reference_data = ui_inputs.get('sensenova_reference_data', [])
|
| active_sensenova_reference = []
|
| if 'sensenova_reference' in enabled_chains and sensenova_reference_data:
|
| for img in sensenova_reference_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_sn_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_sensenova_reference.append(os.path.basename(temp_path))
|
|
|
| joyai_reference_data = ui_inputs.get('joyai_reference_data', [])
|
| active_joyai_reference = []
|
| if 'joyai_image' in enabled_chains and joyai_reference_data:
|
| for img in joyai_reference_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_joyai_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_joyai_reference.append(os.path.basename(temp_path))
|
|
|
| krea2_identity_edit_data = ui_inputs.get('krea2_identity_edit_data', [])
|
| active_krea2_identity_edit = []
|
| if 'krea2_identity_edit' in enabled_chains and krea2_identity_edit_data:
|
| for img in krea2_identity_edit_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_krea2_id_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_krea2_identity_edit.append(os.path.basename(temp_path))
|
|
|
| krea2_reference_edit_data = ui_inputs.get('krea2_reference_edit_data', [])
|
| active_krea2_reference_edit = []
|
| if 'krea2_style_reference' in enabled_chains and krea2_reference_edit_data:
|
| for img in krea2_reference_edit_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_krea2_reference_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_krea2_reference_edit.append(os.path.basename(temp_path))
|
|
|
| qwen_image_edit_data = ui_inputs.get('qwen_image_edit_data', [])
|
| active_qwen_image_edit = []
|
| if 'qwen_image_edit' in enabled_chains and qwen_image_edit_data:
|
| for img in qwen_image_edit_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_qwen_edit_ref_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_qwen_image_edit.append(os.path.basename(temp_path))
|
|
|
| reference_image_data = ui_inputs.get('reference_image_data', [])
|
| active_reference_images = []
|
| if 'reference_image' in enabled_chains and reference_image_data:
|
| for img in reference_image_data:
|
| if img:
|
| if not os.path.exists(INPUT_DIR): os.makedirs(INPUT_DIR)
|
| temp_path = os.path.join(INPUT_DIR, f"temp_ref_img_{random.randint(1000, 9999)}.png")
|
| img.save(temp_path, "PNG")
|
| temp_files_to_clean.append(temp_path)
|
| active_reference_images.append(os.path.basename(temp_path))
|
|
|
| vae_source = ui_inputs.get('vae_source')
|
| vae_id = ui_inputs.get('vae_id')
|
| vae_name_override = None
|
| if 'vae' in enabled_chains and vae_source and vae_id and str(vae_id).strip():
|
| vae_id_clean = str(vae_id).strip()
|
| if vae_source in ("Upload File", "File"):
|
| raw_id = vae_id_clean.replace('\\', '/')
|
| rel_name = raw_id if raw_id.startswith("upload_file/") else f"upload_file/{raw_id}"
|
| local_path = os.path.join(VAE_DIR, rel_name)
|
| if not os.path.exists(local_path):
|
| if os.path.exists(os.path.join(VAE_DIR, raw_id)):
|
| local_path = os.path.join(VAE_DIR, raw_id)
|
| else:
|
| raise gr.Error(f"Uploaded VAE file '{vae_id_clean}' no longer exists on server. Please re-upload it.")
|
| vae_name_override = os.path.relpath(local_path, VAE_DIR).replace('\\', '/')
|
| elif vae_source in ("Civitai", "Hugging Face"):
|
| local_path, status = get_vae_path(vae_source, vae_id_clean, os.environ.get("CIVITAI_API_KEY", ""), progress)
|
| if local_path: vae_name_override = os.path.relpath(local_path, VAE_DIR).replace('\\', '/')
|
| else: raise gr.Error(f"Failed to prepare VAE {vae_id_clean}: {status}")
|
| if vae_name_override:
|
| ui_inputs['vae_name'] = vae_name_override
|
|
|
| conditioning_data = ui_inputs.get('conditioning_data', [])
|
| active_conditioning = []
|
| if 'conditioning' in enabled_chains and conditioning_data:
|
| num_units = len(conditioning_data) // 6
|
| prompts, widths, heights, xs, ys, strengths = [conditioning_data[i*num_units : (i+1)*num_units] for i in range(6)]
|
| for i in range(num_units):
|
| if prompts[i] and prompts[i].strip():
|
| active_conditioning.append({
|
| "prompt": prompts[i], "width": int(widths[i]), "height": int(heights[i]),
|
| "x": int(xs[i]), "y": int(ys[i]), "strength": float(strengths[i])
|
| })
|
|
|
| pe_enable = ui_inputs.get('qwen_image_2_1_prompt_enhancer_enable', False)
|
| active_qwen_image_2_1_prompt_enhancer = []
|
| if 'qwen_image_2_1_prompt_enhancer' in enabled_chains and pe_enable:
|
| active_qwen_image_2_1_prompt_enhancer.append({
|
| "enable": True,
|
| "thinking": bool(ui_inputs.get('qwen_image_2_1_prompt_enhancer_thinking', False)),
|
| "max_length": int(ui_inputs.get('qwen_image_2_1_prompt_enhancer_max_length', 4096))
|
| })
|
|
|
| ming_pe_enable = ui_inputs.get('ming_image_prompt_enhancer_enable', False)
|
| active_ming_image_prompt_enhancer = []
|
| if 'ming_image_prompt_enhancer' in enabled_chains and ming_pe_enable:
|
| item = {
|
| "enable": True,
|
| "thinking": bool(ui_inputs.get('ming_image_prompt_enhancer_thinking', False)),
|
| "max_length": int(ui_inputs.get('ming_image_prompt_enhancer_max_length', 4096))
|
| }
|
| sys_p = ui_inputs.get('ming_image_prompt_enhancer_system_prompt')
|
| if sys_p:
|
| item["system_prompt"] = str(sys_p)
|
| active_ming_image_prompt_enhancer.append(item)
|
|
|
| return {
|
| "active_loras_for_gpu": active_loras_for_gpu,
|
| "active_loras_for_meta": active_loras_for_meta,
|
| "active_controlnets": active_controlnets,
|
| "active_anima_controlnets": active_anima_controlnets,
|
| "active_diffsynth_controlnets": active_diffsynth_controlnets,
|
| "active_krea2_controlnets": active_krea2_controlnets,
|
| "active_ipadapters": active_ipadapters,
|
| "active_flux1_ipadapters": active_flux1_ipadapters,
|
| "active_sd3_ipadapters": active_sd3_ipadapters,
|
| "active_styles": active_styles,
|
| "active_reference_latents": active_reference_latents,
|
| "active_hidream_o1_reference": active_hidream_o1_reference,
|
| "active_sensenova_reference": active_sensenova_reference,
|
| "active_joyai_reference": active_joyai_reference,
|
| "active_krea2_identity_edit": active_krea2_identity_edit,
|
| "active_krea2_reference_edit": active_krea2_reference_edit,
|
| "active_qwen_image_edit": active_qwen_image_edit,
|
| "active_reference_images": active_reference_images,
|
| "active_conditioning": active_conditioning,
|
| "active_qwen_image_2_1_prompt_enhancer": active_qwen_image_2_1_prompt_enhancer,
|
| "active_ming_image_prompt_enhancer": active_ming_image_prompt_enhancer,
|
| "temp_files_to_clean": temp_files_to_clean
|
| } |
