import os
import math
import random
import logging
import requests
import numpy as np
import torch
import spaces
from fastapi import FastAPI, HTTPException
from diffusers import FlowMatchEulerDiscreteScheduler
from qwenimage.pipeline_qwenimage_edit_plus import QwenImageEditPlusPipeline
from qwenimage.transformer_qwenimage import QwenImageTransformer2DModel
from qwenimage.qwen_fa3_processor import QwenDoubleStreamAttnProcessorFA3
from optimization import optimize_pipeline_
from PIL import Image
import gradio as gr
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.")
dtype = torch.bfloat16
device = "cuda" if torch.cuda.is_available() else "cpu"
scheduler_cfg = {
"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_cfg)
pipe = QwenImageEditPlusPipeline.from_pretrained(
"Qwen/Qwen-Image-Edit-2509",
scheduler=scheduler,
torch_dtype=dtype,
).to(device)
pipe.load_lora_weights(
"lightx2v/Qwen-Image-Lightning",
weight_name="Qwen-Image-Lightning-4steps-V2.0.safetensors",
)
pipe.fuse_lora()
pipe.transformer.__class__ = QwenImageTransformer2DModel
pipe.transformer.set_attn_processor(QwenDoubleStreamAttnProcessorFA3())
optimize_pipeline_(
pipe,
image=[Image.new("RGB", (1024, 1024)), Image.new("RGB", (1024, 1024))],
prompt="prompt",
)
MAX_SEED = np.iinfo(np.int32).max
QUALITY_PROMPT = ", high quality, detailed, vibrant, professional lighting"
@spaces.GPU(duration=40)
def infer(image, prompt):
negative_prompt = ""
seed = random.randint(0, MAX_SEED)
generator = torch.Generator(device=device).manual_seed(seed)
prompt_en = translate_albanian_to_english(prompt.strip(), language="en") + QUALITY_PROMPT
pil_img = None
if image is not None:
if isinstance(image, Image.Image):
pil_img = image.convert("RGB")
elif isinstance(image, str):
pil_img = Image.open(image).convert("RGB")
elif hasattr(image, "name"):
pil_img = Image.open(image.name).convert("RGB")
output = pipe(
image=[pil_img] if pil_img is not None else None,
prompt=prompt_en,
height=None,
width=None,
negative_prompt=negative_prompt,
num_inference_steps=4,
generator=generator,
true_cfg_scale=1.0,
num_images_per_prompt=1,
).images
return output[0] if output else None
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("Edit your images with prompt descriptions", elem_id="subtitle")
with gr.Column(elem_id="input_column"):
input_image = gr.Image(
label="Input Image",
type="pil",
sources=["upload"],
show_download_button=False,
show_share_button=False,
interactive=True,
elem_classes=["gradio-component", "image-container"]
)
prompt = gr.Textbox(
label="Prompt",
lines=3,
elem_classes=["gradio-component"]
)
run_button = gr.Button(
"Edit!",
variant="primary",
elem_classes=["gradio-component", "gr-button-primary"]
)
result_image = gr.Image(
label="Result Image",
type="pil",
interactive=False,
show_download_button=True,
show_share_button=False,
elem_classes=["gradio-component", "image-container"]
)
run_button.click(fn=infer, inputs=[input_image, prompt], outputs=[result_image])
prompt.submit(fn=infer, inputs=[input_image, prompt], outputs=[result_image])
return demo
app = FastAPI()
demo = create_demo()
app.mount("/y7u8i9o0p1l2k3j4h5g6f7d8s9a0q1w2e3r4t5y6u7i8o9p0l1k2j3h4g5f6d7s8", demo.app)
@app.get("/{path:path}")
async def catch_all(path: str):
raise HTTPException(status_code=500, detail="Internal Server Error")
if __name__ == "__main__":
logger.info(f"Gradio version: {gr.__version__}")
demo.queue().launch(share=True)