| import os |
| import gradio as gr |
| from typing import Dict, Any |
|
|
| from core.settings import LORA_DIR |
| from utils.app_utils import sanitize_filename, get_lora_path |
|
|
| def process_pipeline_inputs(ui_inputs: Dict[str, Any], progress: gr.Progress, workflow_model_type: str = "h3") -> Dict[str, Any]: |
| active_loras_for_gpu, active_loras_for_meta = [], [] |
| lora_data = ui_inputs.get('lora_data', []) |
| if 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 == "File": |
| lora_filename = sanitize_filename(lora_id) |
| local_path = os.path.join(LORA_DIR, lora_filename) |
| if not os.path.exists(local_path): |
| raise gr.Error(f"Uploaded LoRA file '{lora_id}' no longer exists on server. Please re-upload it.") |
| 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.basename(local_path) |
| 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}") |
|
|
| return { |
| "active_loras_for_gpu": active_loras_for_gpu, |
| "active_loras_for_meta": active_loras_for_meta, |
| "temp_files_to_clean": [] |
| } |