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 							UpperCAmelCase				(     UpperCAmelCase      )->		Tuple:
  
  
  
  
  
  
  
       '''simple docstring'''
  
  
  
       if isinstance(UpperCAmelCase				,torch.Tensor      ):
            return image
       elif isinstance(UpperCAmelCase				,PIL.Image.Image      ):
            SCREAMING_SNAKE_CASE_	      =       [image]
  
       if isinstance(image[0]				,PIL.Image.Image      ):
            SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_	      =       image[0].size
            SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_	      =       (x - x % 64 for x in (w, h))  # resize to integer multiple of 32
  
            SCREAMING_SNAKE_CASE_	      =       [np.array(i.resize((w, h)      )      )[None, :] for i in image]
            SCREAMING_SNAKE_CASE_	      =       np.concatenate(UpperCAmelCase				,axis=0      )
            SCREAMING_SNAKE_CASE_	      =       np.array(UpperCAmelCase      ).astype(np.floataa      ) / 255.0
            SCREAMING_SNAKE_CASE_	      =       image.transpose(0				,3				,1				,2      )
            SCREAMING_SNAKE_CASE_	      =       2.0 * image - 1.0
            SCREAMING_SNAKE_CASE_	      =       torch.from_numpy(UpperCAmelCase      )
       elif isinstance(image[0]				,torch.Tensor      ):
            SCREAMING_SNAKE_CASE_	      =       torch.cat(UpperCAmelCase				,dim=0      )
  
       return image
  
  
  
  
  
  
  class 						snake_case			(       lowerCAmelCase__	):
     '''simple docstring'''
  
     def __init__(	self     :     int							,							lowerCAmelCase_     :     OnnxRuntimeModel							,							lowerCAmelCase_     :     OnnxRuntimeModel							,							lowerCAmelCase_     :     Any							,							lowerCAmelCase_     :     OnnxRuntimeModel							,							lowerCAmelCase_     :     DDPMScheduler							,							lowerCAmelCase_     :     Any							,							lowerCAmelCase_     :     int = 350							,							)					->					Optional[int]:
  
  
  
  
          """simple docstring"""
  
  
  
  
  
  
          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     :     List[str]							,							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							,							)					->					List[str]:
  
  
  
  
          """simple docstring"""
  
  
  
  
  
  
          self.check_inputs(lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_					)
  
          # 2. Define call parameters
          if prompt is not None and isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
               SCREAMING_SNAKE_CASE_	      =       1
          elif prompt is not None and isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
               SCREAMING_SNAKE_CASE_	      =       len(lowerCAmelCase_					)
          else:
               SCREAMING_SNAKE_CASE_	      =       prompt_embeds.shape[0]
  
          SCREAMING_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.
          SCREAMING_SNAKE_CASE_	      =       guidance_scale > 1.0
  
          # 3. Encode input prompt
          SCREAMING_SNAKE_CASE_	      =       self._encode_prompt(
              lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_							,							prompt_embeds=lowerCAmelCase_							,							negative_prompt_embeds=lowerCAmelCase_							,							)
  
          SCREAMING_SNAKE_CASE_	      =       ORT_TO_PT_TYPE[str(text_embeddings.dtype					)]
  
          # 4. Preprocess image
          SCREAMING_SNAKE_CASE_	      =       preprocess(lowerCAmelCase_					)
          SCREAMING_SNAKE_CASE_	      =       image.cpu()
  
          # 5. set timesteps
          self.scheduler.set_timesteps(lowerCAmelCase_							,							device=lowerCAmelCase_					)
          SCREAMING_SNAKE_CASE_	      =       self.scheduler.timesteps
  
          # 5. Add noise to image
          SCREAMING_SNAKE_CASE_	      =       torch.tensor([noise_level]							,							dtype=torch.long							,							device=lowerCAmelCase_					)
          SCREAMING_SNAKE_CASE_	      =       torch.randn(image.shape							,							generator=lowerCAmelCase_							,							device=lowerCAmelCase_							,							dtype=lowerCAmelCase_					)
          SCREAMING_SNAKE_CASE_	      =       self.low_res_scheduler.add_noise(lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_					)
  
          SCREAMING_SNAKE_CASE_	      =       2 if do_classifier_free_guidance else 1
          SCREAMING_SNAKE_CASE_	      =       np.concatenate([image] * batch_multiplier * num_images_per_prompt					)
          SCREAMING_SNAKE_CASE_	      =       np.concatenate([noise_level] * image.shape[0]					)
  
          # 6. Prepare latent variables
          SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_	      =       image.shape[2:]
          SCREAMING_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
          SCREAMING_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
          SCREAMING_SNAKE_CASE_	      =       self.prepare_extra_step_kwargs(lowerCAmelCase_							,							lowerCAmelCase_					)
  
          SCREAMING_SNAKE_CASE_	      =       next(
              (input.type for input in self.unet.model.get_inputs() if input.name == '''timestep''')							,							'''tensor(float)'''					)
          SCREAMING_SNAKE_CASE_	      =       ORT_TO_NP_TYPE[timestep_dtype]
  
          # 9. Denoising loop
          SCREAMING_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
                    SCREAMING_SNAKE_CASE_	      =       np.concatenate([latents] * 2					) if do_classifier_free_guidance else latents
  
                    # concat latents, mask, masked_image_latents in the channel dimension
                    SCREAMING_SNAKE_CASE_	      =       self.scheduler.scale_model_input(lowerCAmelCase_							,							lowerCAmelCase_					)
                    SCREAMING_SNAKE_CASE_	      =       np.concatenate([latent_model_input, image]							,							axis=1					)
  
                    # timestep to tensor
                    SCREAMING_SNAKE_CASE_	      =       np.array([t]							,							dtype=lowerCAmelCase_					)
  
                    # predict the noise residual
                    SCREAMING_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:
                         SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_	      =       np.split(lowerCAmelCase_							,							2					)
                    SCREAMING_SNAKE_CASE_	      =       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
  
                    # compute the previous noisy sample x_t -> x_t-1
                    SCREAMING_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
          SCREAMING_SNAKE_CASE_	      =       self.decode_latents(latents.float()					)
  
          # 11. Convert to PIL
          if output_type == "pil":
               SCREAMING_SNAKE_CASE_	      =       self.numpy_to_pil(lowerCAmelCase_					)
  
          if not return_dict:
               return (image,)
  
          return ImagePipelineOutput(images=lowerCAmelCase_					)
  
     def _lowercase (	self     :     Dict							,							lowerCAmelCase_     :     Optional[int]					)					->					str:
  
  
  
  
          """simple docstring"""
  
  
  
  
  
  
          SCREAMING_SNAKE_CASE_	      =       1 / 0.08_333 * latents
          SCREAMING_SNAKE_CASE_	      =       self.vae(latent_sample=lowerCAmelCase_					)[0]
          SCREAMING_SNAKE_CASE_	      =       np.clip(image / 2 + 0.5							,							0							,							1					)
          SCREAMING_SNAKE_CASE_	      =       image.transpose((0, 2, 3, 1)					)
          return image
  
  
  
  
     def _lowercase (	self     :     Dict							,							lowerCAmelCase_     :     Union[str, List[str]]							,							lowerCAmelCase_     :     int							,							lowerCAmelCase_     :     Optional[int]							,							lowerCAmelCase_     :     bool							,							lowerCAmelCase_     :     Optional[str]							,							lowerCAmelCase_     :     Optional[torch.FloatTensor] = None							,							lowerCAmelCase_     :     Optional[torch.FloatTensor] = None							,							)					->					Tuple:
  
  
  
  
          """simple docstring"""
  
  
  
  
  
  
          if prompt is not None and isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
               SCREAMING_SNAKE_CASE_	      =       1
          elif prompt is not None and isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
               SCREAMING_SNAKE_CASE_	      =       len(lowerCAmelCase_					)
          else:
               SCREAMING_SNAKE_CASE_	      =       prompt_embeds.shape[0]
  
          if prompt_embeds is None:
               SCREAMING_SNAKE_CASE_	      =       self.tokenizer(
                   lowerCAmelCase_							,							padding='''max_length'''							,							max_length=self.tokenizer.model_max_length							,							truncation=lowerCAmelCase_							,							return_tensors='''pt'''							,							)
               SCREAMING_SNAKE_CASE_	      =       text_inputs.input_ids
               SCREAMING_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_					):
                    SCREAMING_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
               SCREAMING_SNAKE_CASE_	      =       self.text_encoder(
                   input_ids=text_input_ids.int().to(lowerCAmelCase_					)							,							)
               SCREAMING_SNAKE_CASE_	      =       prompt_embeds[0]
  
          SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_	      =       prompt_embeds.shape
          # duplicate text embeddings for each generation per prompt, using mps friendly method
          SCREAMING_SNAKE_CASE_	      =       prompt_embeds.repeat(1							,							lowerCAmelCase_					)
          SCREAMING_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:
               SCREAMING_SNAKE_CASE_	      =       42
               if negative_prompt is None:
                    SCREAMING_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_					):
                    SCREAMING_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:
                    SCREAMING_SNAKE_CASE_	      =       negative_prompt
  
               SCREAMING_SNAKE_CASE_	      =       text_input_ids.shape[-1]
               SCREAMING_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
  
               SCREAMING_SNAKE_CASE_	      =       self.text_encoder(
                   input_ids=uncond_input.input_ids.int().to(lowerCAmelCase_					)							,							)
               SCREAMING_SNAKE_CASE_	      =       uncond_embeddings[0]
  
          if do_classifier_free_guidance:
               SCREAMING_SNAKE_CASE_	      =       uncond_embeddings.shape[1]
               # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
               SCREAMING_SNAKE_CASE_	      =       uncond_embeddings.repeat(1							,							lowerCAmelCase_					)
               SCREAMING_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
               SCREAMING_SNAKE_CASE_	      =       np.concatenate([uncond_embeddings, prompt_embeds]					)
  
          return prompt_embeds
  
  
  
style_context: 
  
  
  import unittest
  
  from transformers import load_tool
  from transformers.tools.agent_types import AGENT_TYPE_MAPPING
  
  from .test_tools_common import ToolTesterMixin, output_types
  
  class 						snake_case			(       unittest.TestCase	, lowerCAmelCase__	):
  			'''simple docstring'''
  
  			def _lowercase (	self     :     List[Any]					)					->					Optional[Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       load_tool('''translation'''					)
  								self.tool.setup()
  								SCREAMING_SNAKE_CASE_	      =       load_tool('''translation'''							,							remote=lowerCAmelCase_					)
  
  			def _lowercase (	self     :     Tuple					)					->					str:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       self.tool('''Hey, what\'s up?'''							,							src_lang='''English'''							,							tgt_lang='''French'''					)
  								self.assertEqual(lowerCAmelCase_							,							'''- Hé, comment ça va?'''					)
  
  			def _lowercase (	self     :     Optional[Any]					)					->					int:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       self.remote_tool('''Hey, what\'s up?'''							,							src_lang='''English'''							,							tgt_lang='''French'''					)
  								self.assertEqual(lowerCAmelCase_							,							'''- Hé, comment ça va?'''					)
  
  			def _lowercase (	self     :     Optional[int]					)					->					str:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       self.tool(text='''Hey, what\'s up?'''							,							src_lang='''English'''							,							tgt_lang='''French'''					)
  								self.assertEqual(lowerCAmelCase_							,							'''- Hé, comment ça va?'''					)
  
  			def _lowercase (	self     :     str					)					->					List[str]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       self.remote_tool(text='''Hey, what\'s up?'''							,							src_lang='''English'''							,							tgt_lang='''French'''					)
  								self.assertEqual(lowerCAmelCase_							,							'''- Hé, comment ça va?'''					)
  
  			def _lowercase (	self     :     Optional[int]					)					->					Dict:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       ['''Hey, what\'s up?''', '''English''', '''Spanish''']
  								SCREAMING_SNAKE_CASE_	      =       self.tool(*lowerCAmelCase_					)
  
  								# There is a single output
  								if len(self.tool.outputs					) == 1:
  													SCREAMING_SNAKE_CASE_	      =       [outputs]
  
  								self.assertListEqual(output_types(lowerCAmelCase_					)							,							self.tool.outputs					)
  
  			def _lowercase (	self     :     Optional[int]					)					->					List[str]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       ['''Hey, what\'s up?''', '''English''', '''Spanish''']
  								SCREAMING_SNAKE_CASE_	      =       self.tool(*lowerCAmelCase_					)
  
  								if not isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
  													SCREAMING_SNAKE_CASE_	      =       [outputs]
  
  								self.assertEqual(len(lowerCAmelCase_					)							,							len(self.tool.outputs					)					)
  
  								for output, output_type in zip(lowerCAmelCase_							,							self.tool.outputs					):
  													SCREAMING_SNAKE_CASE_	      =       AGENT_TYPE_MAPPING[output_type]
  													self.assertTrue(isinstance(lowerCAmelCase_							,							lowerCAmelCase_					)					)
  
  
  
  
  			def _lowercase (	self     :     Tuple					)					->					int:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       ['''Hey, what\'s up?''', '''English''', '''Spanish''']
  
  								SCREAMING_SNAKE_CASE_	      =       []
  
  								for _input, input_type in zip(lowerCAmelCase_							,							self.tool.inputs					):
  													if isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
  																		_inputs.append([AGENT_TYPE_MAPPING[_input_type](_input					) for _input_type in input_type]					)
  													else:
  																		_inputs.append(AGENT_TYPE_MAPPING[input_type](_input					)					)
  
          # Should not raise an error
  								SCREAMING_SNAKE_CASE_	      =       self.tool(*lowerCAmelCase_					)
  
  								if not isinstance(lowerCAmelCase_							,							lowerCAmelCase_					):
  													SCREAMING_SNAKE_CASE_	      =       [outputs]
  
  								self.assertEqual(len(lowerCAmelCase_					)							,							len(self.tool.outputs					)					)
  
  
  