docstr: Function invoked when calling the pipeline for generation.
  
  
  Args:
      prompt (`Optional[Union[str, List[str]]]`, defaults to None):
          The prompt or prompts to guide the image generation. If not defined, one has to pass `prompt_embeds`.
          instead.
      image (`Union[np.ndarray, PIL.Image.Image]`):
          `Image`, or tensor representing an image batch which will be upscaled.
      strength (`float`, defaults to 0.8):
          Conceptually, indicates how much to transform the reference `image`. Must be between 0 and 1. `image`
          will be used as a starting point, adding more noise to it the larger the `strength`. The number of
          denoising steps depends on the amount of noise initially added. When `strength` is 1, added noise will
          be maximum and the denoising process will run for the full number of iterations specified in
          `num_inference_steps`. A value of 1, therefore, essentially ignores `image`.
      num_inference_steps (`int`, defaults to 50):
          The number of denoising steps. More denoising steps usually lead to a higher quality image at the
          expense of slower inference.
      guidance_scale (`float`, defaults to 7.5):
          Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
          `guidance_scale` is defined as `w` of equation 2. of [Imagen
          Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
          1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
          usually at the expense of lower image quality.
      negative_prompt (`Optional[Union[str, list]]`):
          The prompt or prompts not to guide the image generation. If not defined, one has to pass
          `negative_prompt_embeds`. instead. Ignored when not using guidance (i.e., ignored if `guidance_scale`
          is less than `1`).
      num_images_per_prompt (`int`, defaults to 1):
          The number of images to generate per prompt.
      eta (`float`, defaults to 0.0):
          Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
          [`schedulers.DDIMScheduler`], will be ignored for others.
      generator (`Optional[np.random.RandomState]`, defaults to `None`)::
          A np.random.RandomState to make generation deterministic.
      prompt_embeds (`Optional[np.ndarray]`, defaults to `None`):
          Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
          provided, text embeddings will be generated from `prompt` input argument.
      negative_prompt_embeds (`Optional[np.ndarray]`, defaults to `None`):
          Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
          weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
          argument.
      output_type (`str`, defaults to `"pil"`):
          The output format of the generate image. Choose between
          [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
      return_dict (`bool`, defaults to `True`):
          Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
          plain tuple.
      callback (Optional[Callable], defaults to `None`):
          A function that will be called every `callback_steps` steps during inference. The function will be
          called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
      callback_steps (`int`, defaults to 1):
          The frequency at which the `callback` function will be called. If not specified, the callback will be
          called at every step.
  
  
  Returns:
      [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
      [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
      When returning a tuple, the first element is a list with the generated images, and the second element is a
      list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
      (nsfw) content, according to the `safety_checker`.
code: def __call__(
      self,
      prompt: Optional[Union[str, List[str]]] = None,
      image: Union[np.ndarray, PIL.Image.Image] = None,
      strength: float = 0.8,
      num_inference_steps: int = 50,
      guidance_scale: float = 7.5,
      negative_prompt: Optional[Union[str, List[str]]] = None,
      num_images_per_prompt: int = 1,
      eta: float = 0.0,
      generator: Optional[np.random.RandomState] = None,
      prompt_embeds: Optional[np.ndarray] = None,
      negative_prompt_embeds: Optional[np.ndarray] = None,
      output_type: str = "pil",
      return_dict: bool = True,
      callback: Optional[Callable[[int, int, np.ndarray], None]] = None,
      callback_steps: int = 1,
  ):
      r"""
      Function invoked when calling the pipeline for generation.
  
  
      Args:
          prompt (`Optional[Union[str, List[str]]]`, defaults to None):
              The prompt or prompts to guide the image generation. If not defined, one has to pass `prompt_embeds`.
              instead.
          image (`Union[np.ndarray, PIL.Image.Image]`):
              `Image`, or tensor representing an image batch which will be upscaled.
          strength (`float`, defaults to 0.8):
              Conceptually, indicates how much to transform the reference `image`. Must be between 0 and 1. `image`
              will be used as a starting point, adding more noise to it the larger the `strength`. The number of
              denoising steps depends on the amount of noise initially added. When `strength` is 1, added noise will
              be maximum and the denoising process will run for the full number of iterations specified in
              `num_inference_steps`. A value of 1, therefore, essentially ignores `image`.
          num_inference_steps (`int`, defaults to 50):
              The number of denoising steps. More denoising steps usually lead to a higher quality image at the
              expense of slower inference.
          guidance_scale (`float`, defaults to 7.5):
              Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
              `guidance_scale` is defined as `w` of equation 2. of [Imagen
              Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
              1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
              usually at the expense of lower image quality.
          negative_prompt (`Optional[Union[str, list]]`):
              The prompt or prompts not to guide the image generation. If not defined, one has to pass
              `negative_prompt_embeds`. instead. Ignored when not using guidance (i.e., ignored if `guidance_scale`
              is less than `1`).
          num_images_per_prompt (`int`, defaults to 1):
              The number of images to generate per prompt.
          eta (`float`, defaults to 0.0):
              Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
              [`schedulers.DDIMScheduler`], will be ignored for others.
          generator (`Optional[np.random.RandomState]`, defaults to `None`)::
              A np.random.RandomState to make generation deterministic.
          prompt_embeds (`Optional[np.ndarray]`, defaults to `None`):
              Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
              provided, text embeddings will be generated from `prompt` input argument.
          negative_prompt_embeds (`Optional[np.ndarray]`, defaults to `None`):
              Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
              weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
              argument.
          output_type (`str`, defaults to `"pil"`):
              The output format of the generate image. Choose between
              [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
          return_dict (`bool`, defaults to `True`):
              Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
              plain tuple.
          callback (Optional[Callable], defaults to `None`):
              A function that will be called every `callback_steps` steps during inference. The function will be
              called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
          callback_steps (`int`, defaults to 1):
              The frequency at which the `callback` function will be called. If not specified, the callback will be
              called at every step.
  
  
      Returns:
          [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
          [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
          When returning a tuple, the first element is a list with the generated images, and the second element is a
          list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
          (nsfw) content, according to the `safety_checker`.
      """
  
      # check inputs. Raise error if not correct
      self.check_inputs(
          prompt,
          strength,
          callback_steps,
          negative_prompt,
          prompt_embeds,
          negative_prompt_embeds,
      )
  
      # 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]
  
      if generator is None:
          generator = np.random
  
      # set timesteps
      self.scheduler.set_timesteps(num_inference_steps)
  
      image = self.image_processor.preprocess(image)
  
      # 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
  
      prompt_embeds = self._encode_prompt(
          prompt,
          num_images_per_prompt,
          do_classifier_free_guidance,
          negative_prompt,
          prompt_embeds=prompt_embeds,
          negative_prompt_embeds=negative_prompt_embeds,
      )
  
      latents_dtype = prompt_embeds.dtype
      image = image.astype(latents_dtype)
      # encode the init image into latents and scale the latents
      init_latents = self.vae_encoder(sample=image)[0]
  
      scaling_factor = self.vae_decoder.config.get(
          "scaling_factor", 0.18215
      )
      init_latents = scaling_factor * init_latents
  
      if isinstance(prompt, str):
          prompt = [prompt]
      if (
          len(prompt) > init_latents.shape[0]
          and len(prompt) % init_latents.shape[0] == 0
      ):
          # expand init_latents for batch_size
          deprecation_message = (
              f"You have passed {len(prompt)} text prompts (`prompt`), but only {init_latents.shape[0]} initial"
              " images (`image`). Initial images are now duplicating to match the number of text prompts. Note"
              " that this behavior is deprecated and will be removed in a version 1.0.0. Please make sure to update"
              " your script to pass as many initial images as text prompts to suppress this warning."
          )
          deprecate(
              "len(prompt) != len(image)",
              "1.0.0",
              deprecation_message,
              standard_warn=False,
          )
          additional_image_per_prompt = (
              len(prompt) // init_latents.shape[0]
          )
          init_latents = np.concatenate(
              [init_latents]
              * additional_image_per_prompt
              * num_images_per_prompt,
              axis=0,
          )
      elif (
          len(prompt) > init_latents.shape[0]
          and len(prompt) % init_latents.shape[0] != 0
      ):
          raise ValueError(
              f"Cannot duplicate `image` of batch size {init_latents.shape[0]} to {len(prompt)} text prompts."
          )
      else:
          init_latents = np.concatenate(
              [init_latents] * num_images_per_prompt, axis=0
          )
  
      # get the original timestep using init_timestep
      offset = self.scheduler.config.get("steps_offset", 0)
      init_timestep = int(num_inference_steps * strength) + offset
      init_timestep = min(init_timestep, num_inference_steps)
  
      timesteps = self.scheduler.timesteps.numpy()[-init_timestep]
      timesteps = np.array(
          [timesteps] * batch_size * num_images_per_prompt
      )
  
      # add noise to latents using the timesteps
      noise = generator.randn(*init_latents.shape).astype(latents_dtype)
      init_latents = self.scheduler.add_noise(
          torch.from_numpy(init_latents),
          torch.from_numpy(noise),
          torch.from_numpy(timesteps),
      )
      init_latents = init_latents.numpy()
  
      # 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
  
      latents = init_latents
  
      t_start = max(num_inference_steps - init_timestep + offset, 0)
      timesteps = self.scheduler.timesteps[t_start:].numpy()
  
      # Adapted from diffusers to extend it for other runtimes than ORT
      timestep_dtype = self.unet.input_dtype.get("timestep", np.float32)
  
      num_warmup_steps = (
          len(timesteps) - num_inference_steps * self.scheduler.order
      )
      for i, t in enumerate(self.progress_bar(timesteps)):
          # expand the latents if we are doing classifier free guidance
          latent_model_input = (
              np.concatenate([latents] * 2)
              if do_classifier_free_guidance
              else latents
          )
          latent_model_input = self.scheduler.scale_model_input(
              torch.from_numpy(latent_model_input), t
          )
          latent_model_input = latent_model_input.cpu().numpy()
  
          # predict the noise residual
          timestep = np.array([t], dtype=timestep_dtype)
          noise_pred = self.unet(
              sample=latent_model_input,
              timestep=timestep,
              encoder_hidden_states=prompt_embeds,
          )[0]
  
          # perform guidance
          if do_classifier_free_guidance:
              noise_pred_uncond, noise_pred_text = np.split(
                  noise_pred, 2
              )
              noise_pred = noise_pred_uncond + guidance_scale * (
                  noise_pred_text - noise_pred_uncond
              )
  
          # compute the previous noisy sample x_t -> x_t-1
          scheduler_output = self.scheduler.step(
              torch.from_numpy(noise_pred),
              t,
              torch.from_numpy(latents),
              **extra_step_kwargs,
          )
          latents = scheduler_output.prev_sample.numpy()
  
          # call the callback, if provided
          if i == len(timesteps) - 1 or (
              (i + 1) > num_warmup_steps
              and (i + 1) % self.scheduler.order == 0
          ):
              if callback is not None and i % callback_steps == 0:
                  callback(i, t, latents)
  
      if output_type == "latent":
          image = latents
          has_nsfw_concept = None
      else:
          latents /= scaling_factor
          # it seems likes there is a strange result for using half-precision vae decoder if batchsize>1
          image = np.concatenate(
              [
                  self.vae_decoder(latent_sample=latents[i : i + 1])[0]
                  for i in range(latents.shape[0])
              ]
          )
          image, has_nsfw_concept = self.run_safety_checker(image)
  
      if has_nsfw_concept is None:
          do_denormalize = [True] * image.shape[0]
      else:
          do_denormalize = [
              not has_nsfw for has_nsfw in has_nsfw_concept
          ]
  
      image = self.image_processor.postprocess(
          image, output_type=output_type, do_denormalize=do_denormalize
      )
  
      if not return_dict:
          return (image, has_nsfw_concept)
  
      return StableDiffusionPipelineOutput(
          images=image, nsfw_content_detected=has_nsfw_concept
      )
  
file_path: optimum/optimum/pipelines/diffusers/pipeline_stable_diffusion_img2img.py
imports: import inspect
  from typing import Callable, List, Optional, Union
  import numpy
  import PIL
  import torch
  from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
  from diffusers.utils import deprecate
  from .pipeline_stable_diffusion import StableDiffusionPipelineMixin
label_desc: Use image node_id 2 for calling the StableDiffusionImg2ImgPipelineMixin obj's underlying member method code with example usage: obj.__call__(prompt, image, strength, num_inference_steps, guidance_scale, negative_prompt, num_images_per_prompt, eta, generator, prompt_embeds, negative_prompt_embeds, output_type, return_dict, callback, callback_steps) and returns: StableDiffusionPipelineOutput, image,  has_nsfw_concept