import os os.system( 'pip install --upgrade --pre --extra-index-url https://download.pytorch.org/whl/nightly/cu126 ' '"torch==2.9.0" "torchvision==0.24.0" "torchaudio==2.9.0" ' '"transformers>=4.44" "huggingface-hub>=1.0.0rc6" spaces -q' ) import spaces import gradio as gr import torch import math from PIL import Image from diffusers import QwenImageEditPlusPipeline, FlowMatchEulerDiscreteScheduler import requests import logging import numpy as np import random from fastapi import FastAPI, HTTPException logging.basicConfig( level=logging.INFO, filename="qwen_image_editor.log", filemode="a", format="%(asctime)s - %(levelname)s - %(message)s", ) logger = logging.getLogger(__name__) @spaces.GPU def translate_albanian_to_english(text: str, language: str = "en"): if not text.strip(): raise gr.Error("Please enter a description.") for attempt in range(2): try: response = requests.post( "https://hal1993-mdftranslation1234567890abcdef1234567890-fc073a6.hf.space/v1/translate", json={"from_language": "sq", "to_language": "en", "input_text": text}, headers={"accept": "application/json", "Content-Type": "application/json"}, timeout=5, ) response.raise_for_status() translated = response.json().get("translate", "") logger.info(f"Translation response: {translated}") return translated except Exception as e: logger.error(f"Translation error (attempt {attempt + 1}): {e}") if attempt == 1: raise gr.Error("Translation failed. Please try again.") raise gr.Error("Translation failed. Please try again.") scheduler_config = { "base_image_seq_len": 256, "base_shift": math.log(3), "invert_sigmas": False, "max_image_seq_len": 8192, "max_shift": math.log(3), "num_train_timesteps": 1000, "shift": 1.0, "shift_terminal": None, "stochastic_sampling": False, "time_shift_type": "exponential", "use_beta_sigmas": False, "use_dynamic_shifting": True, "use_exponential_sigmas": False, "use_karras_sigmas": False, } scheduler = FlowMatchEulerDiscreteScheduler.from_config(scheduler_config) pipeline = QwenImageEditPlusPipeline.from_pretrained( "Qwen/Qwen-Image-Edit-2509", scheduler=scheduler, torch_dtype=torch.bfloat16, ) pipeline.to("cuda") pipeline.set_progress_bar_config(disable=None) pipeline.load_lora_weights( "lightx2v/Qwen-Image-Lightning", weight_name="Qwen-Image-Lightning-8steps-V2.0-bf16.safetensors", ) pipeline.fuse_lora() MAX_SEED = np.iinfo(np.int32).max QUALITY_PROMPT = ", high quality, detailed, vibrant, professional lighting" @spaces.GPU(duration=60) def edit_images(image1, image2, prompt): if image1 is None or image2 is None: raise gr.Error("Please upload both images") prompt_en = translate_albanian_to_english(prompt.strip(), language="en") prompt_final = prompt_en + QUALITY_PROMPT if not isinstance(image1, Image.Image): image1 = Image.fromarray(image1) if not isinstance(image2, Image.Image): image2 = Image.fromarray(image2) seed = random.randint(0, MAX_SEED) true_cfg_scale = 1.0 negative_prompt = "" num_steps = 8 guidance_scale = 1.0 inputs = { "image": [image1, image2], "prompt": prompt_final, "generator": torch.manual_seed(seed), "true_cfg_scale": true_cfg_scale, "negative_prompt": negative_prompt, "num_inference_steps": num_steps, "guidance_scale": guidance_scale, "num_images_per_prompt": 1, } logger.info(f"Calling pipeline – Prompt: {prompt_final}") logger.info(f"Seed: {seed} | Steps: {num_steps}") with torch.inference_mode(): output = pipeline(**inputs) return output.images[0] def create_demo(): with gr.Blocks(css="", title="Qwen Image Editor") as demo: gr.HTML( """ """ ) with gr.Row(elem_id="general_items"): gr.Markdown("# ") gr.Markdown("Blend images together guided by a prompt description.", elem_id="subtitle") with gr.Column(elem_id="input_column"): image1_input = gr.Image( label="First Image", type="pil", sources=["upload"], interactive=True, elem_classes=["gradio-component", "image-container"], ) image2_input = gr.Image( label="Second Image", type="pil", sources=["upload"], interactive=True, elem_classes=["gradio-component", "image-container"], ) prompt_input = gr.Textbox( label="Prompt", placeholder="Describe how you want the images combined or edited...", lines=3, elem_classes=["gradio-component"], ) run_button = gr.Button( "Edit!", variant="primary", elem_classes=["gradio-component", "gr-button-primary"], ) output_image = gr.Image( label="Result Image", type="pil", interactive=False, elem_classes=["gradio-component", "image-container"], ) gr.on( triggers=[run_button.click, prompt_input.submit], fn=edit_images, inputs=[image1_input, image2_input, prompt_input], outputs=[output_image], show_progress="full", ) return demo app = FastAPI() demo = create_demo() app.mount("/q3w4e5r6t7y8u9i0o1p2l3k4j5h6g7f8d9s0a1q2w3e4r5t6y7u8i9o0p1l2k3j4", demo.app) @app.get("/{path:path}") async def catch_all(path: str): if not path.startswith("spaceishere"): raise HTTPException(status_code=500, detail="Internal Server Error") return demo if __name__ == "__main__": logger.info(f"Gradio version: {gr.__version__}") demo.queue().launch(share=True)