docstr: Create Stable Diffusion artworks and save the result into a DocumentArray.
  
  :param text_prompts: Phrase, sentence, or string of words and phrases describing what the image should look like.  The words will be analyzed by the AI and will guide the diffusion process toward the image(s) you describe.
  :param init_image: Initial image.
  :param strength: Control the noise strength added to initial image, value is in the interval [0.0, 1.0]. The closer to 1, the bigger change to the initial image.
  :param style: Image style, such as oil paintings, if specified, style will be used to construct prompts.
  :param artist: Artist style, if specified, style will be used to construct prompts.
  :param batch_size: This variable sets the number of still images you want SD to create for each prompt.
  :param num_inference_steps: The number of inference steps.
  :param guidance_scale: Increase the adherence to the conditional signal which in this case is text as well as overall sample quality.
  :param enable_fp16: Whether to use float16.
  :param use_gpu: whether to use gpu or not.
  :param output_dir: Output directory.
  :return: a DocumentArray object that has `n_batches` Documents
docstr_tok: ["Create","Stable","Diffusion","artworks","and","save","the","result","into","a","DocumentArray",".",":","param","text_prompts",":","Phrase",",","sentence",",","or","string","of","words","and","phrases","describing","what","the","image","should","look","like",".","The","words","will","be","analyzed","by","the","AI","and","will","guide","the","diffusion","process","toward","the","image","(","s",")","you","describe",".",":","param","init_image",":","Initial","image",".",":","param","strength",":","Control","the","noise","strength","added","to","initial","image",",","value","is","in","the","interval","[","0.0",",","1.0","]",".","The","closer","to","1",",","the","bigger","change","to","the","initial","image",".",":","param","style",":","Image","style",",","such","as","oil","paintings",",","if","specified",",","style","will","be","used","to","construct","prompts",".",":","param","artist",":","Artist","style",",","if","specified",",","style","will","be","used","to","construct","prompts",".",":","param","batch_size",":","This","variable","sets","the","number","of","still","images","you","want","SD","to","create","for","each","prompt",".",":","param","num_inference_steps",":","The","number","of","inference","steps",".",":","param","guidance_scale",":","Increase","the","adherence","to","the","conditional","signal","which","in","this","case","is","text","as","well","as","overall","sample","quality",".",":","param","enable_fp16",":","Whether","to","use","float16",".",":","param","use_gpu",":","whether","to","use","gpu","or","not",".",":","param","output_dir",":","Output","directory",".",":","return",":","a","DocumentArray","object","that","has","`","n_batches","`","Documents"]
code: def generate_image(
      self,
      text_prompts,
      init_image,
      strength: float = 0.8,
      style: Optional[str] = None,
      artist: Optional[str] = None,
      batch_size: Optional[int] = 1,
      num_inference_steps=50,
      guidance_scale=7.5,
      enable_fp16=False,
      seed=None,
      eta=0.0,
      display_rate=5,
      use_gpu=True,
      output_dir: Optional[str] = "stable_diffusion_img2img_out",
  ):
      """
      Create Stable Diffusion artworks and save the result into a DocumentArray.
  
      :param text_prompts: Phrase, sentence, or string of words and phrases describing what the image should look like.  The words will be analyzed by the AI and will guide the diffusion process toward the image(s) you describe.
      :param init_image: Initial image.
      :param strength: Control the noise strength added to initial image, value is in the interval [0.0, 1.0]. The closer to 1, the bigger change to the initial image.
      :param style: Image style, such as oil paintings, if specified, style will be used to construct prompts.
      :param artist: Artist style, if specified, style will be used to construct prompts.
      :param batch_size: This variable sets the number of still images you want SD to create for each prompt.
      :param num_inference_steps: The number of inference steps.
      :param guidance_scale: Increase the adherence to the conditional signal which in this case is text as well as overall sample quality.
      :param enable_fp16: Whether to use float16.
      :param use_gpu: whether to use gpu or not.
      :param output_dir: Output directory.
      :return: a DocumentArray object that has `n_batches` Documents
      """
      if seed:
          np.random.seed(seed)
          random.seed(seed)
          paddle.seed(seed)
  
      if use_gpu:
          try:
              _places = os.environ.get("CUDA_VISIBLE_DEVICES", None)
              if _places:
                  paddle.device.set_device("gpu:{}".format(0))
          except:
              raise RuntimeError(
                  "Environment Variable CUDA_VISIBLE_DEVICES is not set correctly. If you wanna use gpu, please set CUDA_VISIBLE_DEVICES as cuda_device_id."
              )
      else:
          paddle.device.set_device("cpu")
      paddle.disable_static()
  
      if not os.path.exists(output_dir):
          os.makedirs(output_dir, exist_ok=True)
  
      if isinstance(text_prompts, str):
          text_prompts = text_prompts.rstrip(",.，。")
          if style is not None:
              text_prompts += ",{}".format(style)
          if artist is not None:
              text_prompts += ",{},trending on artstation".format(
                  artist
              )
          text_prompts = [text_prompts]
      elif isinstance(text_prompts, list):
          for i, prompt in enumerate(
              text_prompts
          ):  # different from dd here, dd can have multiple prompts for one image with weight.
              text_prompts[i] = prompt.rstrip(",.，。")
              if style is not None:
                  text_prompts[i] += ",{}".format(style)
              if artist is not None:
                  text_prompts[
                      i
                  ] += ",{},trending on artstation".format(artist)
  
      if isinstance(init_image, str):
          init_image = preprocess(Image.open(init_image))
      else:
          init_image = preprocess(init_image)
  
      # set timesteps
      accepts_offset = "offset" in set(
          inspect.signature(
              self.scheduler.set_timesteps
          ).parameters.keys()
      )
      extra_set_kwargs = {}
      offset = 0
      if accepts_offset:
          offset = 1
          extra_set_kwargs["offset"] = 1
  
      self.scheduler.set_timesteps(
          num_inference_steps, **extra_set_kwargs
      )
  
      # encode the init image into latents and scale the latents
      init_latents = self.vae.encode(init_image).sample()
      init_latents = 0.18215 * init_latents
  
      # expand init_latents for batch_size
      init_latents = paddle.concat([init_latents] * batch_size)
  
      # get the original timestep using init_timestep
      init_timestep = int(num_inference_steps * strength) + offset
      init_timestep = min(init_timestep, num_inference_steps)
      if isinstance(self.scheduler, LMSDiscreteScheduler):
          timesteps = paddle.to_tensor(
              [num_inference_steps - init_timestep] * batch_size,
              dtype="int64",
          )
      else:
          timesteps = self.scheduler.timesteps[-init_timestep]
          timesteps = paddle.to_tensor(
              [timesteps] * batch_size, dtype="int64"
          )
  
      # add noise to latents using the timesteps
      noise = paddle.randn(init_latents.shape)
      init_latents = self.scheduler.add_noise(
          init_latents, noise, timesteps
      )
  
      # 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
  
      # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
      # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
      # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
      # and should be between [0, 1]
      accepts_eta = "eta" in set(
          inspect.signature(self.scheduler.step).parameters.keys()
      )
      extra_step_kwargs = {}
      if accepts_eta:
          extra_step_kwargs["eta"] = eta
  
      da_batches = DocumentArray()
  
      for prompt in text_prompts:
          d = Document(tags={"prompt": prompt})
          da_batches.append(d)
          for i in range(batch_size):
              d.chunks.append(
                  Document(tags={"prompt": prompt, "image idx": i})
              )
          d.chunks.append(
              Document(tags={"prompt": prompt, "image idx": "merged"})
          )
          with paddle.amp.auto_cast(enable=enable_fp16, level="O2"):
              prompts = [prompt] * batch_size
              text_input = tokenize(prompts)
              text_embeddings = self.text_encoder(text_input)
              if do_classifier_free_guidance:
                  uncond_input = tokenize([""] * batch_size)
                  uncond_embeddings = self.text_encoder(uncond_input)
                  text_embeddings = paddle.concat(
                      [uncond_embeddings, text_embeddings]
                  )
  
              latents = init_latents
  
              t_start = max(
                  num_inference_steps - init_timestep + offset, 0
              )
              for i, t in tqdm(
                  enumerate(self.scheduler.timesteps[t_start:])
              ):
                  t_index = t_start + i
                  # expand the latents if we are doing classifier-free guidance to avoid doing two forward passes.
                  latent_model_input = (
                      paddle.concat([latents] * 2)
                      if do_classifier_free_guidance
                      else latents
                  )
  
                  if isinstance(self.scheduler, LMSDiscreteScheduler):
                      sigma = self.scheduler.sigmas[t_index]
                      latent_model_input = latent_model_input / (
                          (sigma**2 + 1) ** 0.5
                      )
  
                  # predict the noise residual
                  noise_pred = self.unet(
                      latent_model_input,
                      t,
                      encoder_hidden_states=text_embeddings,
                  )["sample"]
  
                  # 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)
                      )
  
                  # compute the previous noisy sample x_t -> x_t-1
                  if isinstance(self.scheduler, LMSDiscreteScheduler):
                      latents = self.scheduler.step(
                          noise_pred,
                          t_index,
                          latents,
                          **extra_step_kwargs,
                      )["prev_sample"]
                  else:
                      latents = self.scheduler.step(
                          noise_pred, t, latents, **extra_step_kwargs
                      )["prev_sample"]
                  if i % display_rate == 0:
                      # vae decode
                      images = self.vae.decode(1 / 0.18215 * latents)
                      images = (images / 2 + 0.5).clip(0, 1)
                      merge_image = (
                          images.cpu()
                          .transpose([2, 0, 3, 1])
                          .flatten(1, 2)
                          .numpy()
                      )
                      merge_image = (
                          (merge_image * 255).round().astype(np.uint8)
                      )
                      merge_image = Image.fromarray(merge_image)
                      merge_image.save(
                          os.path.join(
                              output_dir, f"{prompt}-progress.png"
                          )
                      )
                      c = Document(tags={"step": i, "prompt": prompt})
                      c.load_pil_image_to_datauri(merge_image)
                      d.chunks[-1].chunks.append(c)
                      display.clear_output(wait=True)
                      display.display(merge_image)
                      images = (
                          images.cpu().transpose([0, 2, 3, 1]).numpy()
                      )
                      images = (images * 255).round().astype(np.uint8)
                      for j in range(images.shape[0]):
                          image = Image.fromarray(images[j])
                          c = Document(
                              tags={"step": i, "prompt": prompt}
                          )
                          c.load_pil_image_to_datauri(image)
                          d.chunks[j].chunks.append(c)
  
              # vae decode
              images = self.vae.decode(1 / 0.18215 * latents)
              images = (images / 2 + 0.5).clip(0, 1)
              merge_image = (
                  images.cpu()
                  .transpose([2, 0, 3, 1])
                  .flatten(1, 2)
                  .numpy()
              )
              merge_image = (merge_image * 255).round().astype(np.uint8)
              merge_image = Image.fromarray(merge_image)
              merge_image.save(
                  os.path.join(output_dir, f"{prompt}-merge.png")
              )
              display.clear_output(wait=True)
              display.display(merge_image)
              d.load_pil_image_to_datauri(merge_image)
              d.chunks[-1].load_pil_image_to_datauri(merge_image)
              images = images.cpu().transpose([0, 2, 3, 1]).numpy()
              images = (images * 255).round().astype(np.uint8)
              for j in range(images.shape[0]):
                  image = Image.fromarray(images[j])
                  image.save(
                      os.path.join(
                          output_dir, f"{prompt}-image-{j}.png"
                      )
                  )
                  d.chunks[j].load_pil_image_to_datauri(image)
      return da_batches
  
imports: import argparse
  import ast
  import base64
  import inspect
  import os
  import random
  import sys
  from functools import partial
  from io import BytesIO
  from typing import List
  from typing import Optional
  import numpy
  import paddle
  from docarray import Document
  from docarray import DocumentArray
  from IPython import display
  from PIL import Image
  from stable_diffusion_img2img.clip.clip.utils import build_model
  from stable_diffusion_img2img.clip.clip.utils import tokenize
  from stable_diffusion_img2img.diffusers import AutoencoderKL
  from stable_diffusion_img2img.diffusers import DDIMScheduler
  from stable_diffusion_img2img.diffusers import LMSDiscreteScheduler
  from stable_diffusion_img2img.diffusers import PNDMScheduler
  from stable_diffusion_img2img.diffusers import UNet2DConditionModel
  from stable_diffusion_img2img.utils import preprocess
  from tqdm.auto import tqdm
  import paddlehub
  from paddlehub.module.module import moduleinfo
  from paddlehub.module.module import runnable
  from paddlehub.module.module import serving
label_desc: Use image node_id 2 for calling the StableDiffusionImg2Img obj's underlying member method code with example usage: obj.generate_image(text_prompts, init_image, strength, style, artist, batch_size, num_inference_steps, guidance_scale, enable_fp16, seed, eta, display_rate, use_gpu, output_dir) and returns: da_batches