Update app.py

app.py

@@ -1,222 +1,176 @@
-import os
-import random
-import uuid
-import json
-import gradio as gr
-import numpy as np
-from PIL import Image
 import spaces
+import gradio as gr
 import torch
-from diffusers import StableDiffusionXLPipeline, EulerAncestralDiscreteScheduler
-
-DESCRIPTIONx = """## STABLE HAMSTER 🐹
-
-"""
-
-css = '''
-.gradio-container{max-width: 560px !important}
-h1{text-align:center}
-footer {
-    visibility: hidden
-}
-'''
-
-examples = [
-    "3d image, cute girl, in the style of Pixar --ar 1:2 --stylize 750, 4K resolution highlights, Sharp focus, octane render, ray tracing, Ultra-High-Definition, 8k, UHD, HDR, (Masterpiece:1.5), (best quality:1.5)",
-    "Cold coffee in a cup bokeh --ar 85:128 --v 6.0 --style raw5, 4K",
-    "Vector illustration of a horse, vector graphic design with flat colors on an brown background in the style of vector art, using simple shapes and graphics with simple details, professionally designed as a tshirt logo ready for print on a white background. --ar 89:82 --v 6.0 --style raw",
-    "Man in brown leather jacket posing for camera, in the style of sleek and stylized, clockpunk, subtle shades, exacting precision, ferrania p30  --ar 67:101 --v 5",
-    "Commercial photography, giant burger, white lighting, studio light, 8k octane rendering, high resolution photography, insanely detailed, fine details, on white isolated plain, 8k, commercial photography, stock photo, professional color grading, --v 4 --ar 9:16 "
-    
-]
+from PIL import Image
+from transformers import AutoProcessor, AutoModelForCausalLM, pipeline, Qwen2VLForConditionalGeneration
+from diffusers import DiffusionPipeline
+import random
+import numpy as np
+import os
+import subprocess
+from qwen_vl_utils import process_vision_info
+from threading import Thread
+import uuid
+import io
 
+# Initialize models
+device = "cuda" if torch.cuda.is_available() else "cpu"
+dtype = torch.bfloat16
 
-MODEL_ID = os.getenv("MODEL_VAL_PATH") #Use SDXL Model as "MODEL_REPO" --------->>> ”VALUE”. 
-MAX_IMAGE_SIZE = int(os.getenv("MAX_IMAGE_SIZE", "4096"))
-USE_TORCH_COMPILE = os.getenv("USE_TORCH_COMPILE", "0") == "1"
-ENABLE_CPU_OFFLOAD = os.getenv("ENABLE_CPU_OFFLOAD", "0") == "1"
-BATCH_SIZE = int(os.getenv("BATCH_SIZE", "1"))  # Allow generating multiple images at once
+huggingface_token = os.getenv("HUGGINGFACE_TOKEN")
 
-#Load model outside of function
-device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
-pipe = StableDiffusionXLPipeline.from_pretrained(
-    MODEL_ID,
-    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
-    use_safetensors=True,
-    add_watermarker=False,
-).to(device)
-pipe.scheduler = EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)
+# FLUX.1-dev model
+pipe = DiffusionPipeline.from_pretrained("black-forest-labs/FLUX.1-dev", torch_dtype=dtype, token=huggingface_token).to(device)
 
-# <compile speedup >
-if USE_TORCH_COMPILE:
-    pipe.compile()
+# Initialize Qwen2VL model
+qwen_model = Qwen2VLForConditionalGeneration.from_pretrained(
+    "prithivMLmods/JSONify-Flux", trust_remote_code=True, torch_dtype=torch.float16
+).to(device).eval()
+qwen_processor = AutoProcessor.from_pretrained("prithivMLmods/JSONify-Flux", trust_remote_code=True)
 
-# Offloading capacity (RAM)
-if ENABLE_CPU_OFFLOAD:
-    pipe.enable_model_cpu_offload()
+# Prompt Enhancer
+enhancer_long = pipeline("summarization", model="gokaygokay/Lamini-Prompt-Enchance-Long", device=device)
 
 MAX_SEED = np.iinfo(np.int32).max
-
-def save_image(img):
-    unique_name = str(uuid.uuid4()) + ".png"
-    img.save(unique_name)
-    return unique_name
-
-def randomize_seed_fn(seed: int, randomize_seed: bool) -> int:
+MAX_IMAGE_SIZE = 2048
+
+# Qwen2VL caption function
+@spaces.GPU
+def qwen_caption(image):
+    # Convert image to PIL if it's not already
+    if not isinstance(image, Image.Image):
+        image = Image.fromarray(image)
+    
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "image", "image": image},
+                {"type": "text", "text": "Caption the image"},
+            ],
+        }
+    ]
+
+    text = qwen_processor.apply_chat_template(
+        messages, tokenize=False, add_generation_prompt=True
+    )
+    image_inputs, video_inputs = process_vision_info(messages)
+    inputs = qwen_processor(
+        text=[text],
+        images=image_inputs,
+        videos=video_inputs,
+        padding=True,
+        return_tensors="pt",
+    ).to(device)
+
+    generated_ids = qwen_model.generate(**inputs, max_new_tokens=1024)
+    generated_ids_trimmed = [
+        out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
+    ]
+    output_text = qwen_processor.batch_decode(
+        generated_ids_trimmed,
+        skip_special_tokens=True,
+        clean_up_tokenization_spaces=False,
+    )[0]
+    
+    return output_text
+
+# Prompt Enhancer function
+def enhance_prompt(input_prompt):
+    result = enhancer_long("Enhance the description: " + input_prompt)
+    enhanced_text = result[0]['summary_text']
+    return enhanced_text
+
+@spaces.GPU(duration=190)
+def process_workflow(image, text_prompt, use_enhancer, seed, randomize_seed, width, height, guidance_scale, num_inference_steps, progress=gr.Progress(track_tqdm=True)):
+    if image is not None:
+        # Convert image to PIL if it's not already
+        if not isinstance(image, Image.Image):
+            image = Image.fromarray(image)
+        
+        prompt = qwen_caption(image)
+        print(prompt)
+    else:
+        prompt = text_prompt
+    
+    if use_enhancer:
+        prompt = enhance_prompt(prompt)
+    
     if randomize_seed:
         seed = random.randint(0, MAX_SEED)
-    return seed
-
-@spaces.GPU(duration=60, enable_queue=True)
-def generate(
-    prompt: str,
-    negative_prompt: str = "",
-    use_negative_prompt: bool = False,
-    seed: int = 1,
-    width: int = 1024,
-    height: int = 1024,
-    guidance_scale: float = 3,
-    num_inference_steps: int = 25,
-    randomize_seed: bool = False,
-    use_resolution_binning: bool = True, 
-    num_images: int = 1,  # Number of images to generate
-    progress=gr.Progress(track_tqdm=True),
-):
-    seed = int(randomize_seed_fn(seed, randomize_seed))
+    
     generator = torch.Generator(device=device).manual_seed(seed)
+    
+    image = pipe(
+        prompt=prompt,
+        generator=generator,
+        num_inference_steps=num_inference_steps,
+        width=width,
+        height=height,
+        guidance_scale=guidance_scale
+    ).images[0]
+    
+    return image, prompt, seed
+
+custom_css = """
+.input-group, .output-group {
+    border: 1px solid #e0e0e0;
+    border-radius: 10px;
+    padding: 20px;
+    margin-bottom: 20px;
+    background-color: #f9f9f9;
+}
+.submit-btn {
+    background-color: #2980b9 !important;
+    color: white !important;
+}
+.submit-btn:hover {
+    background-color: #3498db !important;
+}
+"""
 
-    #Options 
-    options = {
-        "prompt": [prompt] * num_images,
-        "negative_prompt": [negative_prompt] * num_images if use_negative_prompt else None,
-        "width": width,
-        "height": height,
-        "guidance_scale": guidance_scale,
-        "num_inference_steps": num_inference_steps,
-        "generator": generator,
-        "output_type": "pil",
-    }
-
-    if use_resolution_binning:
-        options["use_resolution_binning"] = True
-
-    #Images potential batches
-    images = []
-    for i in range(0, num_images, BATCH_SIZE):
-        batch_options = options.copy()
-        batch_options["prompt"] = options["prompt"][i:i+BATCH_SIZE]
-        if "negative_prompt" in batch_options:
-            batch_options["negative_prompt"] = options["negative_prompt"][i:i+BATCH_SIZE]
-        images.extend(pipe(**batch_options).images)
-
-    image_paths = [save_image(img) for img in images]
-    return image_paths, seed
-
-with gr.Blocks(css=css, theme="bethecloud/storj_theme") as demo:
-    gr.Markdown(DESCRIPTIONx)  
-    with gr.Group():
-        with gr.Row():
-            prompt = gr.Text(
-                label="Prompt",
-                show_label=False,
-                max_lines=1,
-                placeholder="Enter your prompt",
-                container=False,
-            )
-            run_button = gr.Button("Run", scale=0)
-        result = gr.Gallery(label="Result", columns=1, show_label=False) 
-    with gr.Accordion("Advanced options", open=False, visible=False):
-        num_images = gr.Slider(
-            label="Number of Images",
-            minimum=1,
-            maximum=4,
-            step=1,
-            value=1,
-        )
-        with gr.Row():
-            use_negative_prompt = gr.Checkbox(label="Use negative prompt", value=True)
-            negative_prompt = gr.Text(
-                label="Negative prompt",
-                max_lines=5,
-                lines=4,
-                placeholder="Enter a negative prompt",
-                value="(deformed, distorted, disfigured:1.3), poorly drawn, bad anatomy, wrong anatomy, extra limb, missing limb, floating limbs, (mutated hands and fingers:1.4), disconnected limbs, mutation, mutated, ugly, disgusting, blurry, amputation",
-                visible=True,
-            )
-        seed = gr.Slider(
-            label="Seed",
-            minimum=0,
-            maximum=MAX_SEED,
-            step=1,
-            value=0,
-        )
-        randomize_seed = gr.Checkbox(label="Randomize seed", value=True)
-        with gr.Row(visible=True):
-            width = gr.Slider(
-                label="Width",
-                minimum=512,
-                maximum=MAX_IMAGE_SIZE,
-                step=64,
-                value=1024,
-            )
-            height = gr.Slider(
-                label="Height",
-                minimum=512,
-                maximum=MAX_IMAGE_SIZE,
-                step=64,
-                value=1024,
-            )
-        with gr.Row():
-            guidance_scale = gr.Slider(
-                label="Guidance Scale",
-                minimum=0.1,
-                maximum=6,
-                step=0.1,
-                value=3.0,
-            )
-            num_inference_steps = gr.Slider(
-                label="Number of inference steps",
-                minimum=1,
-                maximum=25,
-                step=1,
-                value=23,
-            )
-
-    gr.Examples(
-        examples=examples,
-        inputs=prompt,
-        cache_examples=False
-    )
-
-    use_negative_prompt.change(
-        fn=lambda x: gr.update(visible=x),
-        inputs=use_negative_prompt,
-        outputs=negative_prompt,
-        api_name=False,
-    )
+title = """<h1 align="center">FLUX.1-dev with Qwen2VL Captioner and Prompt Enhancer</h1>
+<p><center>
+<a href="https://huggingface.co/black-forest-labs/FLUX.1-dev" target="_blank">[FLUX.1-dev Model]</a>
+<a href="https://huggingface.co/prithivMLmods/JSONify-Flux" target="_blank">[Qwen2VL Model]</a>
+<a href="https://huggingface.co/gokaygokay/Lamini-Prompt-Enchance-Long" target="_blank">[Prompt Enhancer Long]</a>
+<p align="center">Create long prompts from images or enhance your short prompts with prompt enhancer</p>
+</center></p>
+"""
 
-    gr.on(
-        triggers=[
-            prompt.submit,
-            negative_prompt.submit,
-            run_button.click,
-        ],
-        fn=generate,
+with gr.Blocks(css=custom_css, theme=gr.themes.Soft(primary_hue="blue", secondary_hue="gray")) as demo:
+    gr.HTML(title)
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            with gr.Group(elem_classes="input-group"):
+                input_image = gr.Image(label="Input Image (Qwen2VL Captioner)")
+            
+            with gr.Accordion("Advanced Settings", open=False):
+                text_prompt = gr.Textbox(label="Text Prompt (optional, used if no image is uploaded)")
+                use_enhancer = gr.Checkbox(label="Use Prompt Enhancer", value=False)
+                seed = gr.Slider(label="Seed", minimum=0, maximum=MAX_SEED, step=1, value=0)
+                randomize_seed = gr.Checkbox(label="Randomize Seed", value=True)
+                width = gr.Slider(label="Width", minimum=256, maximum=MAX_IMAGE_SIZE, step=32, value=1024)
+                height = gr.Slider(label="Height", minimum=256, maximum=MAX_IMAGE_SIZE, step=32, value=1024)
+                guidance_scale = gr.Slider(label="Guidance Scale", minimum=1, maximum=15, step=0.1, value=3.5)
+                num_inference_steps = gr.Slider(label="Inference Steps", minimum=1, maximum=50, step=1, value=28)
+            
+            generate_btn = gr.Button("Generate Image", elem_classes="submit-btn")
+        
+        with gr.Column(scale=1):
+            with gr.Group(elem_classes="output-group"):
+                output_image = gr.Image(label="Result", elem_id="gallery", show_label=False)
+                final_prompt = gr.Textbox(label="Final Prompt Used")
+                used_seed = gr.Number(label="Seed Used")
+    
+    generate_btn.click(
+        fn=process_workflow,
         inputs=[
-            prompt,
-            negative_prompt,
-            use_negative_prompt,
-            seed,
-            width,
-            height,
-            guidance_scale,
-            num_inference_steps,
-            randomize_seed,
-            num_images
+            input_image, text_prompt, use_enhancer, seed, randomize_seed,
+            width, height, guidance_scale, num_inference_steps
         ],
-        outputs=[result, seed],
-        api_name="stable_hamster",
-    )   
+        outputs=[output_image, final_prompt, used_seed]
+    )
 
-if __name__ == "__main__":
-    demo.queue(max_size=40).launch()
+demo.launch(debug=True)