code: 
  from logging import getLogger
  from typing import Any, Callable, List, Optional, Union
  
  import numpy as np
  import PIL
  import torch
  
  from ...schedulers import DDPMScheduler
  from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
  from ..pipeline_utils import ImagePipelineOutput
  from . import StableDiffusionUpscalePipeline
  
  
  A           =   getLogger(__name__)
  
  
  A           =   4
  A           =   7
  
  A           =   {
      "float16": torch.floataa,
      "float32": torch.floataa,
  }
  def   snake_case_       (							_a     :							List[str]			):
        if isinstance(_a ,					torch.Tensor			):
              return image
        elif isinstance(_a ,					PIL.Image.Image			):
              snake_case_						       =							[image]
  
        if isinstance(image[0] ,					PIL.Image.Image			):
              snake_case_       ,	snake_case_						       =							image[0].size
              snake_case_       ,	snake_case_						       =							(x - x % 64 for x in (w, h))  # resize to integer multiple of 32
  
              snake_case_						       =							[np.array(i.resize((w, h)			)			)[None, :] for i in image]
              snake_case_						       =							np.concatenate(_a ,					axis=0			)
              snake_case_						       =							np.array(_a			).astype(np.floataa			) / 255.0
              snake_case_						       =							image.transpose(0 ,					3 ,					1 ,					2			)
              snake_case_						       =							2.0 * image - 1.0
              snake_case_						       =							torch.from_numpy(_a			)
        elif isinstance(image[0] ,					torch.Tensor			):
              snake_case_						       =							torch.cat(_a ,					dim=0			)
  
        return image
  
  
  
  
  class  lowerCAmelCase__   (       UpperCamelCase       ):
  
  
  
    '''simple docstring'''
  
  
  
  
  
  
    def __init__(self  : int	, lowerCAmelCase_  : OnnxRuntimeModel	, lowerCAmelCase_  : OnnxRuntimeModel	, lowerCAmelCase_  : Any	, lowerCAmelCase_  : OnnxRuntimeModel	, lowerCAmelCase_  : DDPMScheduler	, lowerCAmelCase_  : Any	, lowerCAmelCase_  : int = 3_50	, )   ->       Union[str, Any]:
          super().__init__(
              vae=lowerCAmelCase_	, text_encoder=lowerCAmelCase_	, tokenizer=lowerCAmelCase_	, unet=lowerCAmelCase_	, low_res_scheduler=lowerCAmelCase_	, scheduler=lowerCAmelCase_	, safety_checker=lowerCAmelCase_	, feature_extractor=lowerCAmelCase_	, watermarker=lowerCAmelCase_	, max_noise_level=lowerCAmelCase_	, )
  
    def __call__(self  : Dict	, lowerCAmelCase_  : Union[str, List[str]]	, lowerCAmelCase_  : Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]	, lowerCAmelCase_  : int = 75	, lowerCAmelCase_  : float = 9.0	, lowerCAmelCase_  : int = 20	, lowerCAmelCase_  : Optional[Union[str, List[str]]] = None	, lowerCAmelCase_  : Optional[int] = 1	, lowerCAmelCase_  : float = 0.0	, lowerCAmelCase_  : Optional[Union[torch.Generator, List[torch.Generator]]] = None	, lowerCAmelCase_  : Optional[torch.FloatTensor] = None	, lowerCAmelCase_  : Optional[np.ndarray] = None	, lowerCAmelCase_  : Optional[np.ndarray] = None	, lowerCAmelCase_  : Optional[str] = "pil"	, lowerCAmelCase_  : bool = True	, lowerCAmelCase_  : Optional[Callable[[int, int, torch.FloatTensor], None]] = None	, lowerCAmelCase_  : Optional[int] = 1	, )   ->       Dict:
          self.check_inputs(lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_    )
  
          # 2. Define call parameters
          if prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
                snake_case_						       =							1
          elif prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
                snake_case_						       =							len(lowerCAmelCase_    )
          else:
                snake_case_						       =							prompt_embeds.shape[0]
  
          snake_case_						       =							self._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.
          snake_case_						       =							guidance_scale > 1.0
  
          # 3. Encode input prompt
          snake_case_						       =							self._encode_prompt(
              lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, prompt_embeds=lowerCAmelCase_	, negative_prompt_embeds=lowerCAmelCase_	, )
  
          snake_case_						       =							ORT_TO_PT_TYPE[str(text_embeddings.dtype    )]
  
          # 4. Preprocess image
          snake_case_						       =							preprocess(lowerCAmelCase_    )
          snake_case_						       =							image.cpu()
  
          # 5. set timesteps
          self.scheduler.set_timesteps(lowerCAmelCase_	, device=lowerCAmelCase_    )
          snake_case_						       =							self.scheduler.timesteps
  
          # 5. Add noise to image
          snake_case_						       =							torch.tensor([noise_level]	, dtype=torch.long	, device=lowerCAmelCase_    )
          snake_case_						       =							torch.randn(image.shape	, generator=lowerCAmelCase_	, device=lowerCAmelCase_	, dtype=lowerCAmelCase_    )
          snake_case_						       =							self.low_res_scheduler.add_noise(lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_    )
  
          snake_case_						       =							2 if do_classifier_free_guidance else 1
          snake_case_						       =							np.concatenate([image] * batch_multiplier * num_images_per_prompt    )
          snake_case_						       =							np.concatenate([noise_level] * image.shape[0]    )
  
          # 6. Prepare latent variables
          snake_case_       ,	snake_case_						       =							image.shape[2:]
          snake_case_						       =							self.prepare_latents(
              batch_size * num_images_per_prompt	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, )
  
          # 7. Check that sizes of image and latents match
          snake_case_						       =							image.shape[1]
          if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
                raise ValueError(
                    "Incorrect configuration settings! The config of `pipeline.unet` expects"
                    f''' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'''
                    f''' `num_channels_image`: {num_channels_image} '''
                    f''' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'''
                    " `pipeline.unet` or your `image` input."    )
  
          # 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
          snake_case_						       =							self.prepare_extra_step_kwargs(lowerCAmelCase_	, lowerCAmelCase_    )
  
          snake_case_						       =							next(
              (input.type for input in self.unet.model.get_inputs() if input.name == "timestep")	, "tensor(float)"    )
          snake_case_						       =							ORT_TO_NP_TYPE[timestep_dtype]
  
          # 9. Denoising loop
          snake_case_						       =							len(lowerCAmelCase_    ) - num_inference_steps * self.scheduler.order
          with self.progress_bar(total=lowerCAmelCase_    ) as progress_bar:
                for i, t in enumerate(lowerCAmelCase_    ):
                      # expand the latents if we are doing classifier free guidance
                      snake_case_						       =							np.concatenate([latents] * 2    ) if do_classifier_free_guidance else latents
  
                      # concat latents, mask, masked_image_latents in the channel dimension
                      snake_case_						       =							self.scheduler.scale_model_input(lowerCAmelCase_	, lowerCAmelCase_    )
                      snake_case_						       =							np.concatenate([latent_model_input, image]	, axis=1    )
  
                      # timestep to tensor
                      snake_case_						       =							np.array([t]	, dtype=lowerCAmelCase_    )
  
                      # predict the noise residual
                      snake_case_						       =							self.unet(
                          sample=lowerCAmelCase_	, timestep=lowerCAmelCase_	, encoder_hidden_states=lowerCAmelCase_	, class_labels=noise_level.astype(np.intaa    )	, )[0]
  
                      # perform guidance
                      if do_classifier_free_guidance:
                            snake_case_       ,	snake_case_						       =							np.split(lowerCAmelCase_	, 2    )
                      snake_case_						       =							noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
  
                      # compute the previous noisy sample x_t -> x_t-1
                      snake_case_						       =							self.scheduler.step(
                          torch.from_numpy(lowerCAmelCase_    )	, lowerCAmelCase_	, lowerCAmelCase_	, **lowerCAmelCase_    ).prev_sample
  
                      # call the callback, if provided
                      if i == len(lowerCAmelCase_    ) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
                            progress_bar.update()
                            if callback is not None and i % callback_steps == 0:
                                  callback(lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_    )
  
          # 10. Post-processing
          snake_case_						       =							self.decode_latents(latents.float()    )
  
          # 11. Convert to PIL
          if output_type == "pil":
                snake_case_						       =							self.numpy_to_pil(lowerCAmelCase_    )
  
          if not return_dict:
                return (image,)
  
          return ImagePipelineOutput(images=lowerCAmelCase_    )
  
    def 		_lowerCamelCase	(self  : int	, lowerCAmelCase_  : str    )   ->       int:
          snake_case_						       =							1 / 0.08_333 * latents
          snake_case_						       =							self.vae(latent_sample=lowerCAmelCase_    )[0]
          snake_case_						       =							np.clip(image / 2 + 0.5	, 0	, 1    )
          snake_case_						       =							image.transpose((0, 2, 3, 1)    )
          return image
  
  
    def 		_lowerCamelCase	(self  : List[Any]	, lowerCAmelCase_  : Union[str, List[str]]	, lowerCAmelCase_  : str	, lowerCAmelCase_  : Optional[int]	, lowerCAmelCase_  : bool	, lowerCAmelCase_  : Optional[str]	, lowerCAmelCase_  : Optional[torch.FloatTensor] = None	, lowerCAmelCase_  : Optional[torch.FloatTensor] = None	, )   ->       Tuple:
          if prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
                snake_case_						       =							1
          elif prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
                snake_case_						       =							len(lowerCAmelCase_    )
          else:
                snake_case_						       =							prompt_embeds.shape[0]
  
          if prompt_embeds is None:
                snake_case_						       =							self.tokenizer(
                    lowerCAmelCase_	, padding="max_length"	, max_length=self.tokenizer.model_max_length	, truncation=lowerCAmelCase_	, return_tensors="pt"	, )
                snake_case_						       =							text_inputs.input_ids
                snake_case_						       =							self.tokenizer(lowerCAmelCase_	, padding="longest"	, return_tensors="pt"    ).input_ids
  
                if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
                    lowerCAmelCase_	, lowerCAmelCase_    ):
                      snake_case_						       =							self.tokenizer.batch_decode(
                          untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]    )
                      logger.warning(
                          "The following part of your input was truncated because CLIP can only handle sequences up to"
                          f''' {self.tokenizer.model_max_length} tokens: {removed_text}'''    )
  
                # no positional arguments to text_encoder
                snake_case_						       =							self.text_encoder(
                    input_ids=text_input_ids.int().to(lowerCAmelCase_    )	, )
                snake_case_						       =							prompt_embeds[0]
  
          snake_case_       ,	snake_case_       ,	snake_case_						       =							prompt_embeds.shape
          # duplicate text embeddings for each generation per prompt, using mps friendly method
          snake_case_						       =							prompt_embeds.repeat(1	, lowerCAmelCase_    )
          snake_case_						       =							prompt_embeds.reshape(bs_embed * num_images_per_prompt	, lowerCAmelCase_	, -1    )
  
          # get unconditional embeddings for classifier free guidance
          if do_classifier_free_guidance and negative_prompt_embeds is None:
                snake_case_						       =							42
                if negative_prompt is None:
                      snake_case_						       =							[""] * batch_size
                elif type(lowerCAmelCase_    ) is not type(lowerCAmelCase_    ):
                      raise TypeError(
                          f'''`negative_prompt` should be the same type to `prompt`, but got {type(lowerCAmelCase_    )} !='''
                          f''' {type(lowerCAmelCase_    )}.'''    )
                elif isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
                      snake_case_						       =							[negative_prompt]
                elif batch_size != len(lowerCAmelCase_    ):
                      raise ValueError(
                          f'''`negative_prompt`: {negative_prompt} has batch size {len(lowerCAmelCase_    )}, but `prompt`:'''
                          f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                          " the batch size of `prompt`."    )
                else:
                      snake_case_						       =							negative_prompt
  
                snake_case_						       =							text_input_ids.shape[-1]
                snake_case_						       =							self.tokenizer(
                    lowerCAmelCase_	, padding="max_length"	, max_length=lowerCAmelCase_	, truncation=lowerCAmelCase_	, return_tensors="pt"	, )
  
                # if hasattr(uncond_input, "attention_mask"):
                #     attention_mask = uncond_input.attention_mask.to(device)
                # else:
                #     attention_mask = None
  
                snake_case_						       =							self.text_encoder(
                    input_ids=uncond_input.input_ids.int().to(lowerCAmelCase_    )	, )
                snake_case_						       =							uncond_embeddings[0]
  
          if do_classifier_free_guidance:
                snake_case_						       =							uncond_embeddings.shape[1]
                # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                snake_case_						       =							uncond_embeddings.repeat(1	, lowerCAmelCase_    )
                snake_case_						       =							uncond_embeddings.reshape(batch_size * num_images_per_prompt	, lowerCAmelCase_	, -1    )
  
                # For classifier free guidance, we need to do two forward passes.
                # Here we concatenate the unconditional and text embeddings into a single batch
                # to avoid doing two forward passes
                snake_case_						       =							np.concatenate([uncond_embeddings, prompt_embeds]    )
  
          return prompt_embeds
  
  
style_context: 
  '''simple docstring'''
  
  
  
  
  
  def   snake_case_       (							_a     :							int = 1_000			):
        return sum(i for i in range(_a			) if i % 3 == 0 or i % 5 == 0			)
  
  
  if __name__ == "__main__":
       print(F"""{solution() = }""")
  
  