Create app.py

app.py

@@ -0,0 +1,420 @@
+import gradio as gr
+import torch
+
+from diffusers import StableDiffusionXLPipeline
+from diffusers.pipelines.stable_diffusion_xl import StableDiffusionXLPipelineOutput
+import torch
+from PIL import Image
+
+import diffusers
+from share_btn import community_icon_html, loading_icon_html, share_js
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+pipe = StableDiffusionXLPipeline.from_pretrained(
+    "stabilityai/stable-diffusion-xl-base-1.0",
+    torch_dtype=torch.float32,
+    variants="fp32",
+    use_safetensor=True,
+)
+pipe.to("cuda")
+
+@torch.no_grad()
+def call(
+        pipe,
+        prompt: Union[str, List[str]] = None,
+        prompt2: Union[str, List[str]] = None,
+        height: Optional[int] = None,
+        width: Optional[int] = None,
+        num_inference_steps: int = 50,
+        denoising_end: Optional[float] = None,
+        guidance_scale: float = 5.0,
+        guidance_scale2: float = 5.0,
+        negative_prompt: Optional[Union[str, List[str]]] = None,
+        negative_prompt2: Optional[Union[str, List[str]]] = None,
+        num_images_per_prompt: Optional[int] = 1,
+        eta: float = 0.0,
+        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
+        latents: Optional[torch.FloatTensor] = None,
+        prompt_embeds: Optional[torch.FloatTensor] = None,
+        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
+        pooled_prompt_embeds: Optional[torch.FloatTensor] = None,
+        negative_pooled_prompt_embeds: Optional[torch.FloatTensor] = None,
+        output_type: Optional[str] = "pil",
+        return_dict: bool = True,
+        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
+        callback_steps: int = 1,
+        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
+        guidance_rescale: float = 0.0,
+        original_size: Optional[Tuple[int, int]] = None,
+        crops_coords_top_left: Tuple[int, int] = (0, 0),
+        target_size: Optional[Tuple[int, int]] = None,
+        negative_original_size: Optional[Tuple[int, int]] = None,
+        negative_crops_coords_top_left: Tuple[int, int] = (0, 0),
+        negative_target_size: Optional[Tuple[int, int]] = None,
+    ):
+        # 0. Default height and width to unet
+        height = height or pipe.default_sample_size * pipe.vae_scale_factor
+        width = width or pipe.default_sample_size * pipe.vae_scale_factor
+
+        original_size = original_size or (height, width)
+        target_size = target_size or (height, width)
+
+        # 1. Check inputs. Raise error if not correct
+        pipe.check_inputs(
+            prompt,
+            None,
+            height,
+            width,
+            callback_steps,
+            negative_prompt,
+            None,
+            prompt_embeds,
+            negative_prompt_embeds,
+            pooled_prompt_embeds,
+            negative_pooled_prompt_embeds,
+        )
+
+        # 2. Define call parameters
+        if prompt is not None and isinstance(prompt, str):
+            batch_size = 1
+        elif prompt is not None and isinstance(prompt, list):
+            batch_size = len(prompt)
+        else:
+            batch_size = prompt_embeds.shape[0]
+
+        device = pipe._execution_device
+
+        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
+        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
+        # corresponds to doing no classifier free guidance.
+        do_classifier_free_guidance = guidance_scale > 1.0
+
+        # 3. Encode input prompt
+        text_encoder_lora_scale = (
+            cross_attention_kwargs.get("scale", None) if cross_attention_kwargs is not None else None
+        )
+
+        (
+            prompt_embeds,
+            negative_prompt_embeds,
+            pooled_prompt_embeds,
+            negative_pooled_prompt_embeds,
+        ) = pipe.encode_prompt(
+            prompt=prompt,
+            device=device,
+            num_images_per_prompt=num_images_per_prompt,
+            do_classifier_free_guidance=do_classifier_free_guidance,
+            negative_prompt=negative_prompt,
+            prompt_embeds=None,
+            negative_prompt_embeds=None,
+            pooled_prompt_embeds=None,
+            negative_pooled_prompt_embeds=None,
+            lora_scale=text_encoder_lora_scale,
+        )
+
+        (
+            prompt2_embeds,
+            negative_prompt2_embeds,
+            pooled_prompt2_embeds,
+            negative_pooled_prompt2_embeds,
+        ) = pipe.encode_prompt(
+            prompt=prompt2,
+            device=device,
+            num_images_per_prompt=num_images_per_prompt,
+            do_classifier_free_guidance=do_classifier_free_guidance,
+            negative_prompt=negative_prompt2,
+            prompt_embeds=None,
+            negative_prompt_embeds=None,
+            pooled_prompt_embeds=None,
+            negative_pooled_prompt_embeds=None,
+            lora_scale=text_encoder_lora_scale,
+        )
+
+        # 4. Prepare timesteps
+        pipe.scheduler.set_timesteps(num_inference_steps, device=device)
+
+        timesteps = pipe.scheduler.timesteps
+
+        # 5. Prepare latent variables
+        num_channels_latents = pipe.unet.config.in_channels
+        latents = pipe.prepare_latents(
+            batch_size * num_images_per_prompt,
+            num_channels_latents,
+            height,
+            width,
+            prompt_embeds.dtype,
+            device,
+            generator,
+            latents,
+        )
+
+        # 6. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
+        extra_step_kwargs = pipe.prepare_extra_step_kwargs(generator, eta)
+
+        # 7. Prepare added time ids & embeddings
+        add_text_embeds = pooled_prompt_embeds
+        add_text2_embeds = pooled_prompt2_embeds
+
+        add_time_ids = pipe._get_add_time_ids(
+            original_size, crops_coords_top_left, target_size, dtype=prompt_embeds.dtype
+        )
+        add_time2_ids = pipe._get_add_time_ids(
+            original_size, crops_coords_top_left, target_size, dtype=prompt2_embeds.dtype
+        )
+
+        if negative_original_size is not None and negative_target_size is not None:
+            negative_add_time_ids = pipe._get_add_time_ids(
+                negative_original_size,
+                negative_crops_coords_top_left,
+                negative_target_size,
+                dtype=prompt_embeds.dtype,
+            )
+        else:
+            negative_add_time_ids = add_time_ids
+            negative_add_time2_ids = add_time2_ids
+
+        if do_classifier_free_guidance:
+            prompt_embeds = torch.cat([negative_prompt_embeds, prompt_embeds], dim=0)
+            add_text_embeds = torch.cat([negative_pooled_prompt_embeds, add_text_embeds], dim=0)
+            add_time_ids = torch.cat([negative_add_time_ids, add_time_ids], dim=0)
+
+            prompt2_embeds = torch.cat([negative_prompt2_embeds, prompt2_embeds], dim=0)
+            add_text2_embeds = torch.cat([negative_pooled_prompt2_embeds, add_text2_embeds], dim=0)
+            add_time2_ids = torch.cat([negative_add_time2_ids, add_time2_ids], dim=0)
+
+        prompt_embeds = prompt_embeds.to(device)
+        add_text_embeds = add_text_embeds.to(device)
+        add_time_ids = add_time_ids.to(device).repeat(batch_size * num_images_per_prompt, 1)
+
+        prompt2_embeds = prompt2_embeds.to(device)
+        add_text2_embeds = add_text2_embeds.to(device)
+        add_time2_ids = add_time2_ids.to(device).repeat(batch_size * num_images_per_prompt, 1)
+
+        # 8. Denoising loop
+        num_warmup_steps = max(len(timesteps) - num_inference_steps * pipe.scheduler.order, 0)
+
+        # 7.1 Apply denoising_end
+        if denoising_end is not None and isinstance(denoising_end, float) and denoising_end > 0 and denoising_end < 1:
+            discrete_timestep_cutoff = int(
+                round(
+                    pipe.scheduler.config.num_train_timesteps
+                    - (denoising_end * pipe.scheduler.config.num_train_timesteps)
+                )
+            )
+            num_inference_steps = len(list(filter(lambda ts: ts >= discrete_timestep_cutoff, timesteps)))
+            timesteps = timesteps[:num_inference_steps]
+
+        with pipe.progress_bar(total=num_inference_steps) as progress_bar:
+            for i, t in enumerate(timesteps):
+                if i % 2 == 0:
+                  # expand the latents if we are doing classifier-free guidance
+                  latent_model_input = torch.cat([latents] * 2) if do_classifier_free_guidance else latents
+
+                  latent_model_input = pipe.scheduler.scale_model_input(latent_model_input, t)
+
+                  # predict the noise residual
+                  added_cond_kwargs = {"text_embeds": add_text_embeds, "time_ids": add_time_ids}
+                  noise_pred = pipe.unet(
+                      latent_model_input,
+                      t,
+                      encoder_hidden_states=prompt_embeds,
+                      cross_attention_kwargs=cross_attention_kwargs,
+                      added_cond_kwargs=added_cond_kwargs,
+                      return_dict=False,
+                  )[0]
+
+                  # perform guidance
+                  if do_classifier_free_guidance:
+                      noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)
+                      noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
+                else:
+                  # expand the latents if we are doing classifier free guidance
+                  latent_model_input2 = torch.cat([latents.flip(2)] * 2) if do_classifier_free_guidance else latents
+                  latent_model_input2 = pipe.scheduler.scale_model_input(latent_model_input2, t)
+
+                  # predict the noise residual
+                  added_cond2_kwargs = {"text_embeds": add_text2_embeds, "time_ids": add_time2_ids}
+                  noise_pred2 = pipe.unet(
+                      latent_model_input2,
+                      t,
+                      encoder_hidden_states=prompt2_embeds,
+                      cross_attention_kwargs=cross_attention_kwargs,
+                      added_cond_kwargs=added_cond2_kwargs,
+                      return_dict=False,
+                  )[0]
+
+                  # perform guidance
+                  if do_classifier_free_guidance:
+                      noise_pred2_uncond, noise_pred2_text = noise_pred2.chunk(2)
+                      noise_pred2 = noise_pred2_uncond + guidance_scale2 * (noise_pred2_text - noise_pred2_uncond)
+
+                noise_pred = noise_pred if i % 2 == 0 else noise_pred2.flip(2)
+
+                # compute the previous noisy sample x_t -> x_t-1
+                latents = pipe.scheduler.step(noise_pred, t, latents, **extra_step_kwargs, return_dict=False)[0]
+
+                # call the callback, if provided
+                if i == len(timesteps) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % pipe.scheduler.order == 0):
+                    progress_bar.update()
+                    if callback is not None and i % callback_steps == 0:
+                        callback(i, t, latents)
+
+        if not output_type == "latent":
+            # make sure the VAE is in float32 mode, as it overflows in float16
+            needs_upcasting = pipe.vae.dtype == torch.float16 and pipe.vae.config.force_upcast
+
+            if needs_upcasting:
+                pipe.upcast_vae()
+                latents = latents.to(next(iter(pipe.vae.post_quant_conv.parameters())).dtype)
+
+            image = pipe.vae.decode(latents / pipe.vae.config.scaling_factor, return_dict=False)[0]
+
+            # cast back to fp16 if needed
+            if needs_upcasting:
+                pipe.vae.to(dtype=torch.float16)
+        else:
+            image = latents
+
+        if not output_type == "latent":
+            # apply watermark if available
+            if pipe.watermark is not None:
+                image = pipe.watermark.apply_watermark(image)
+
+            image = pipe.image_processor.postprocess(image, output_type=output_type)
+
+        # Offload all models
+        pipe.maybe_free_model_hooks()
+
+        if not return_dict:
+            return (image,)
+
+        return StableDiffusionXLPipelineOutput(images=image)
+
+def read_content(file_path: str) -> str:
+    """read the content of target file
+    """
+    with open(file_path, 'r', encoding='utf-8') as f:
+        content = f.read()
+
+    return content
+
+def predict(dict, prompt="", negative_prompt="", guidance_scale=7.5, steps=20, strength=1.0, scheduler="EulerDiscreteScheduler"):
+    if negative_prompt == "":
+        negative_prompt = None
+    scheduler_class_name = scheduler.split("-")[0]
+
+    add_kwargs = {}
+    if len(scheduler.split("-")) > 1:
+        add_kwargs["use_karras"] = True
+    if len(scheduler.split("-")) > 2:
+        add_kwargs["algorithm_type"] = "sde-dpmsolver++"
+
+    scheduler = getattr(diffusers, scheduler_class_name)
+    pipe.scheduler = scheduler.from_pretrained("stabilityai/stable-diffusion-xl-base-1.0", subfolder="scheduler", **add_kwargs)
+    
+    init_image = dict["image"].convert("RGB").resize((1024, 1024))
+    mask = dict["mask"].convert("RGB").resize((1024, 1024))
+    
+    output = pipe(prompt = prompt, negative_prompt=negative_prompt, image=init_image, mask_image=mask, guidance_scale=guidance_scale, num_inference_steps=int(steps), strength=strength)
+    
+    return output.images[0], gr.update(visible=True)
+
+
+css = '''
+.gradio-container{max-width: 1100px !important}
+#image_upload{min-height:400px}
+#image_upload [data-testid="image"], #image_upload [data-testid="image"] > div{min-height: 400px}
+#mask_radio .gr-form{background:transparent; border: none}
+#word_mask{margin-top: .75em !important}
+#word_mask textarea:disabled{opacity: 0.3}
+.footer {margin-bottom: 45px;margin-top: 35px;text-align: center;border-bottom: 1px solid #e5e5e5}
+.footer>p {font-size: .8rem; display: inline-block; padding: 0 10px;transform: translateY(10px);background: white}
+.dark .footer {border-color: #303030}
+.dark .footer>p {background: #0b0f19}
+.acknowledgments h4{margin: 1.25em 0 .25em 0;font-weight: bold;font-size: 115%}
+#image_upload .touch-none{display: flex}
+@keyframes spin {
+    from {
+        transform: rotate(0deg);
+    }
+    to {
+        transform: rotate(360deg);
+    }
+}
+#share-btn-container {padding-left: 0.5rem !important; padding-right: 0.5rem !important; background-color: #000000; justify-content: center; align-items: center; border-radius: 9999px !important; max-width: 13rem; margin-left: auto;}
+div#share-btn-container > div {flex-direction: row;background: black;align-items: center}
+#share-btn-container:hover {background-color: #060606}
+#share-btn {all: initial; color: #ffffff;font-weight: 600; cursor:pointer; font-family: 'IBM Plex Sans', sans-serif; margin-left: 0.5rem !important; padding-top: 0.5rem !important; padding-bottom: 0.5rem !important;right:0;}
+#share-btn * {all: unset}
+#share-btn-container div:nth-child(-n+2){width: auto !important;min-height: 0px !important;}
+#share-btn-container .wrap {display: none !important}
+#share-btn-container.hidden {display: none!important}
+#prompt input{width: calc(100% - 160px);border-top-right-radius: 0px;border-bottom-right-radius: 0px;}
+#run_button{position:absolute;margin-top: 11px;right: 0;margin-right: 0.8em;border-bottom-left-radius: 0px;
+    border-top-left-radius: 0px;}
+#prompt-container{margin-top:-18px;}
+#prompt-container .form{border-top-left-radius: 0;border-top-right-radius: 0}
+#image_upload{border-bottom-left-radius: 0px;border-bottom-right-radius: 0px}
+'''
+
+image_blocks = gr.Blocks(css=css, elem_id="total-container")
+with image_blocks as demo:
+    gr.HTML(read_content("header.html"))
+    with gr.Row():
+                with gr.Column():
+                    image = gr.Image(source='upload', tool='sketch', elem_id="image_upload", type="pil", label="Upload",height=400)
+                    with gr.Row(elem_id="prompt-container", mobile_collapse=False, equal_height=True):
+                        with gr.Row():
+                            prompt = gr.Textbox(placeholder="Your prompt (what you want in place of what is erased)", show_label=False, elem_id="prompt")
+                            btn = gr.Button("Inpaint!", elem_id="run_button")
+                    
+                    with gr.Accordion(label="Advanced Settings", open=False):
+                        with gr.Row(mobile_collapse=False, equal_height=True):
+                            guidance_scale = gr.Number(value=7.5, minimum=1.0, maximum=20.0, step=0.1, label="guidance_scale")
+                            steps = gr.Number(value=20, minimum=10, maximum=30, step=1, label="steps")
+                            strength = gr.Number(value=0.99, minimum=0.01, maximum=0.99, step=0.01, label="strength")
+                            negative_prompt = gr.Textbox(label="negative_prompt", placeholder="Your negative prompt", info="what you don't want to see in the image")
+                        with gr.Row(mobile_collapse=False, equal_height=True):
+                            schedulers = ["DEISMultistepScheduler", "HeunDiscreteScheduler", "EulerDiscreteScheduler", "DPMSolverMultistepScheduler", "DPMSolverMultistepScheduler-Karras", "DPMSolverMultistepScheduler-Karras-SDE"]
+                            scheduler = gr.Dropdown(label="Schedulers", choices=schedulers, value="EulerDiscreteScheduler")
+                        
+                with gr.Column():
+                    image_out = gr.Image(label="Output", elem_id="output-img", height=400)
+                    with gr.Group(elem_id="share-btn-container", visible=False) as share_btn_container:
+                        community_icon = gr.HTML(community_icon_html)
+                        loading_icon = gr.HTML(loading_icon_html)
+                        share_button = gr.Button("Share to community", elem_id="share-btn",visible=True)
+            
+
+    btn.click(fn=predict, inputs=[image, prompt, negative_prompt, guidance_scale, steps, strength, scheduler], outputs=[image_out, share_btn_container], api_name='run')
+    prompt.submit(fn=predict, inputs=[image, prompt, negative_prompt, guidance_scale, steps, strength, scheduler], outputs=[image_out, share_btn_container])
+    share_button.click(None, [], [], _js=share_js)
+
+    gr.Examples(
+                examples=[
+                    ["./imgs/aaa (8).png"],
+                    ["./imgs/download (1).jpeg"],
+                    ["./imgs/0_oE0mLhfhtS_3Nfm2.png"],
+                    ["./imgs/02_HubertyBlog-1-1024x1024.jpg"],
+                    ["./imgs/jdn_jacques_de_nuce-1024x1024.jpg"],
+                    ["./imgs/c4ca473acde04280d44128ad8ee09e8a.jpg"],
+                    ["./imgs/canam-electric-motorcycles-scaled.jpg"],
+                    ["./imgs/e8717ce80b394d1b9a610d04a1decd3a.jpeg"],
+                    ["./imgs/Nature___Mountains_Big_Mountain_018453_31.jpg"],
+                    ["./imgs/Multible-sharing-room_ccexpress-2-1024x1024.jpeg"],
+                ],
+                fn=predict,
+                inputs=[image],
+                cache_examples=False,
+    )
+    gr.HTML(
+        """
+            <div class="footer">
+                <p>Model by <a href="https://huggingface.co/diffusers" style="text-decoration: underline;" target="_blank">Diffusers</a> - Gradio Demo by 🤗 Hugging Face
+                </p>
+            </div>
+        """
+    )
+
+image_blocks.queue(max_size=25).launch()