code: 
  
  
  
  
  import os
  from typing import List, Union
  
  import tensorflow as tf
  from tensorflow_text import BertTokenizer as BertTokenizerLayer
  from tensorflow_text import FastBertTokenizer, ShrinkLongestTrimmer, case_fold_utfa, combine_segments, pad_model_inputs
  
  from .tokenization_bert import BertTokenizer
  class 							lowerCamelCase		( tf.keras.layers.Layer				):
  
  
  
  
  
  
  
     def __init__(							self    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = "longest"    ,      SCREAMING_SNAKE_CASE_ = True    ,      SCREAMING_SNAKE_CASE_ = 512    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = True    ,      SCREAMING_SNAKE_CASE_ = True    ,      SCREAMING_SNAKE_CASE_ = True    ,      **SCREAMING_SNAKE_CASE_    ,      ):
         super().__init__()
         if use_fast_bert_tokenizer:
             UpperCamelCase				:   Union[str, Any]          =							FastBertTokenizer(
                 SCREAMING_SNAKE_CASE_    ,      token_out_type=tf.intaa    ,      lower_case_nfd_strip_accents=SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ )
         else:
             UpperCamelCase				:   Any          =							tf.lookup.StaticVocabularyTable(
                 tf.lookup.KeyValueTensorInitializer(
                     keys=SCREAMING_SNAKE_CASE_    ,      key_dtype=tf.string    ,      values=tf.range(tf.size(SCREAMING_SNAKE_CASE_    ,      out_type=tf.intaa )    ,      dtype=tf.intaa )    ,      value_dtype=tf.intaa    ,      )    ,      num_oov_buckets=1    ,      )
             UpperCamelCase				:   Union[str, Any]          =							BertTokenizerLayer(
                 SCREAMING_SNAKE_CASE_    ,      token_out_type=tf.intaa    ,      lower_case=SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ )
  
         UpperCamelCase				:   str          =							vocab_list
         UpperCamelCase				:   Optional[int]          =							do_lower_case
         UpperCamelCase				:   Union[str, Any]          =							cls_token_id or vocab_list.index("""[CLS]""" )
         UpperCamelCase				:   Optional[int]          =							sep_token_id or vocab_list.index("""[SEP]""" )
         UpperCamelCase				:   Optional[int]          =							pad_token_id or vocab_list.index("""[PAD]""" )
         UpperCamelCase				:   Union[str, Any]          =							ShrinkLongestTrimmer(max_length - 3    ,      axis=1 )  # Allow room for special tokens
         UpperCamelCase				:   Any          =							max_length
         UpperCamelCase				:   str          =							padding
         UpperCamelCase				:   Any          =							truncation
         UpperCamelCase				:   str          =							pad_to_multiple_of
         UpperCamelCase				:   Dict          =							return_token_type_ids
         UpperCamelCase				:   Union[str, Any]          =							return_attention_mask
  
  
  
  
  
  
  
     @classmethod
     def       a_							(							cls    ,      SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ ):  # noqa: F821
         UpperCamelCase				:   Optional[Any]          =							kwargs.pop("""do_lower_case"""    ,      SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   Dict          =							tokenizer.do_lower_case if do_lower_case is None else do_lower_case
         UpperCamelCase				:   List[Any]          =							kwargs.pop("""cls_token_id"""    ,      SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   int          =							tokenizer.cls_token_id if cls_token_id is None else cls_token_id
         UpperCamelCase				:   Union[str, Any]          =							kwargs.pop("""sep_token_id"""    ,      SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   int          =							tokenizer.sep_token_id if sep_token_id is None else sep_token_id
         UpperCamelCase				:   int          =							kwargs.pop("""pad_token_id"""    ,      SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   List[Any]          =							tokenizer.pad_token_id if pad_token_id is None else pad_token_id
  
         UpperCamelCase				:   Tuple          =							tokenizer.get_vocab()
         UpperCamelCase				:   Tuple          =							sorted([(wordpiece, idx) for wordpiece, idx in vocab.items()]    ,      key=lambda SCREAMING_SNAKE_CASE_ : x[1] )
         UpperCamelCase				:   Dict          =							[entry[0] for entry in vocab]
         return cls(
             vocab_list=SCREAMING_SNAKE_CASE_    ,      do_lower_case=SCREAMING_SNAKE_CASE_    ,      cls_token_id=SCREAMING_SNAKE_CASE_    ,      sep_token_id=SCREAMING_SNAKE_CASE_    ,      pad_token_id=SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_    ,      )
  
  
  
  
  
  
  
     @classmethod
     def       a_							(							cls    ,      SCREAMING_SNAKE_CASE_    ,      *SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ ):
         try:
             UpperCamelCase				:   Tuple          =							BertTokenizer.from_pretrained(SCREAMING_SNAKE_CASE_    ,      *SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ )
         except:  # noqa: E722
             from .tokenization_bert_fast import BertTokenizerFast
  
             UpperCamelCase				:   Tuple          =							BertTokenizerFast.from_pretrained(SCREAMING_SNAKE_CASE_    ,      *SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ )
         return cls.from_tokenizer(SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ )
  
  
  
  
  
  
  
     def       a_							(							self    ,      SCREAMING_SNAKE_CASE_ ):
         if self.do_lower_case:
             UpperCamelCase				:   List[str]          =							case_fold_utfa(SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   Optional[Any]          =							self.tf_tokenizer.tokenize(SCREAMING_SNAKE_CASE_ )
         return tokens.merge_dims(1    ,      -1 )
  
  
  
  
  
  
  
     def       a_							(							self    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_=None    ,      SCREAMING_SNAKE_CASE_=None    ,      SCREAMING_SNAKE_CASE_=None    ,      SCREAMING_SNAKE_CASE_=None    ,      SCREAMING_SNAKE_CASE_=None    ,      SCREAMING_SNAKE_CASE_=None    ,      SCREAMING_SNAKE_CASE_=None    ,      ):
         if padding is None:
             UpperCamelCase				:   Optional[int]          =							self.padding
         if padding not in ("longest", "max_length"):
             raise ValueError("""Padding must be either 'longest' or 'max_length'!""" )
         if max_length is not None and text_pair is not None:
             # Because we have to instantiate a Trimmer to do it properly
             raise ValueError("""max_length cannot be overridden at call time when truncating paired texts!""" )
         if max_length is None:
             UpperCamelCase				:   List[Any]          =							self.max_length
         if truncation is None:
             UpperCamelCase				:   str          =							self.truncation
         if pad_to_multiple_of is None:
             UpperCamelCase				:   Optional[int]          =							self.pad_to_multiple_of
         if return_token_type_ids is None:
             UpperCamelCase				:   Any          =							self.return_token_type_ids
         if return_attention_mask is None:
             UpperCamelCase				:   Optional[Any]          =							self.return_attention_mask
         if not isinstance(SCREAMING_SNAKE_CASE_    ,      tf.Tensor ):
             UpperCamelCase				:   Tuple          =							tf.convert_to_tensor(SCREAMING_SNAKE_CASE_ )
         if text_pair is not None and not isinstance(SCREAMING_SNAKE_CASE_    ,      tf.Tensor ):
             UpperCamelCase				:   Optional[Any]          =							tf.convert_to_tensor(SCREAMING_SNAKE_CASE_ )
         if text_pair is not None:
             if text.shape.rank > 1:
                 raise ValueError("""text argument should not be multidimensional when a text pair is supplied!""" )
             if text_pair.shape.rank > 1:
                 raise ValueError("""text_pair should not be multidimensional!""" )
         if text.shape.rank == 2:
             UpperCamelCase				:   Tuple          =							text[:, 0], text[:, 1]
         UpperCamelCase				:   List[str]          =							self.unpaired_tokenize(SCREAMING_SNAKE_CASE_ )
         if text_pair is None:  # Unpaired text
             if truncation:
                 UpperCamelCase				:   Dict          =							text[:, : max_length - 2]  # Allow room for special tokens
             UpperCamelCase				:   Dict          =							combine_segments(
                 (text,)    ,      start_of_sequence_id=self.cls_token_id    ,      end_of_segment_id=self.sep_token_id )
         else:  # Paired text
             UpperCamelCase				:   List[str]          =							self.unpaired_tokenize(SCREAMING_SNAKE_CASE_ )
             if truncation:
                 UpperCamelCase				:   List[str]          =							self.paired_trimmer.trim([text, text_pair] )
             UpperCamelCase				:   List[str]          =							combine_segments(
                 (text, text_pair)    ,      start_of_sequence_id=self.cls_token_id    ,      end_of_segment_id=self.sep_token_id )
         if padding == "longest":
             UpperCamelCase				:   Dict          =							input_ids.bounding_shape(axis=1 )
             if pad_to_multiple_of is not None:
                 # No ceiling division in tensorflow, so we negate floordiv instead
                 UpperCamelCase				:   Union[str, Any]          =							pad_to_multiple_of * (-tf.math.floordiv(-pad_length    ,      SCREAMING_SNAKE_CASE_ ))
         else:
             UpperCamelCase				:   Dict          =							max_length
  
         UpperCamelCase				:   Any          =							pad_model_inputs(SCREAMING_SNAKE_CASE_    ,      max_seq_length=SCREAMING_SNAKE_CASE_    ,      pad_value=self.pad_token_id )
         UpperCamelCase				:   Any          =							{"""input_ids""": input_ids}
         if return_attention_mask:
             UpperCamelCase				:   Optional[Any]          =							attention_mask
         if return_token_type_ids:
             UpperCamelCase				:   int          =							pad_model_inputs(
                 SCREAMING_SNAKE_CASE_    ,      max_seq_length=SCREAMING_SNAKE_CASE_    ,      pad_value=self.pad_token_id )
             UpperCamelCase				:   List[Any]          =							token_type_ids
         return output
  
  
  
  
  
  
  
     def       a_							(							self ):
         return {
             "vocab_list": self.vocab_list,
             "do_lower_case": self.do_lower_case,
             "cls_token_id": self.cls_token_id,
             "sep_token_id": self.sep_token_id,
             "pad_token_id": self.pad_token_id,
         }
  
  
  
  
  
style_context: 
  
  
  
  
  """simple docstring"""
  
  
  
  
  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 :		Optional[Any]      							=    getLogger(__name__)
  
  
  __A :		Any      							=    4
  __A :		str      							=    7
  
  __A :		Any      							=    {
      '''float16''': torch.floataa,
      '''float32''': torch.floataa,
  }
  def 					A_			(		snake_case_     :       Optional[int]    ):
  
  
  
  
  
  
  
      '''simple docstring'''
      if isinstance(snake_case_						,torch.Tensor    ):
          return image
      elif isinstance(snake_case_						,PIL.Image.Image    ):
          UpperCamelCase				:   List[Any]          =							[image]
  
      if isinstance(image[0]						,PIL.Image.Image    ):
          UpperCamelCase       ,							UpperCamelCase				:   Any          =							image[0].size
          UpperCamelCase       ,							UpperCamelCase				:   List[str]          =							(x - x % 6_4 for x in (w, h))  # resize to integer multiple of 32
  
          UpperCamelCase				:   Tuple          =							[np.array(i.resize((w, h)    )    )[None, :] for i in image]
          UpperCamelCase				:   Union[str, Any]          =							np.concatenate(snake_case_						,axis=0    )
          UpperCamelCase				:   Union[str, Any]          =							np.array(snake_case_    ).astype(np.floataa    ) / 255.0
          UpperCamelCase				:   Dict          =							image.transpose(0						,3						,1						,2    )
          UpperCamelCase				:   Tuple          =							2.0 * image - 1.0
          UpperCamelCase				:   Tuple          =							torch.from_numpy(snake_case_    )
      elif isinstance(image[0]						,torch.Tensor    ):
          UpperCamelCase				:   Union[str, Any]          =							torch.cat(snake_case_						,dim=0    )
  
      return image
  
  
  class 							lowerCamelCase		( _UpperCAmelCase				):
  
  
  
  
  
  
  
     def __init__(							self    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ = 350    ,      ):
         super().__init__(
             vae=SCREAMING_SNAKE_CASE_    ,      text_encoder=SCREAMING_SNAKE_CASE_    ,      tokenizer=SCREAMING_SNAKE_CASE_    ,      unet=SCREAMING_SNAKE_CASE_    ,      low_res_scheduler=SCREAMING_SNAKE_CASE_    ,      scheduler=SCREAMING_SNAKE_CASE_    ,      safety_checker=SCREAMING_SNAKE_CASE_    ,      feature_extractor=SCREAMING_SNAKE_CASE_    ,      watermarker=SCREAMING_SNAKE_CASE_    ,      max_noise_level=SCREAMING_SNAKE_CASE_    ,      )
  
  
  
  
  
  
  
     def __call__(							self    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ = 75    ,      SCREAMING_SNAKE_CASE_ = 9.0    ,      SCREAMING_SNAKE_CASE_ = 20    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = 1    ,      SCREAMING_SNAKE_CASE_ = 0.0    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = "pil"    ,      SCREAMING_SNAKE_CASE_ = True    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = 1    ,      ):
         self.check_inputs(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ )
  
         # 2. Define call parameters
         if prompt is not None and isinstance(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ ):
             UpperCamelCase				:   int          =							1
         elif prompt is not None and isinstance(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ ):
             UpperCamelCase				:   Tuple          =							len(SCREAMING_SNAKE_CASE_ )
         else:
             UpperCamelCase				:   Dict          =							prompt_embeds.shape[0]
  
         UpperCamelCase				:   Optional[int]          =							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.
         UpperCamelCase				:   str          =							guidance_scale > 1.0
  
         # 3. Encode input prompt
         UpperCamelCase				:   Dict          =							self._encode_prompt(
             SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      prompt_embeds=SCREAMING_SNAKE_CASE_    ,      negative_prompt_embeds=SCREAMING_SNAKE_CASE_    ,      )
  
         UpperCamelCase				:   Optional[int]          =							ORT_TO_PT_TYPE[str(text_embeddings.dtype )]
  
         # 4. Preprocess image
         UpperCamelCase				:   List[Any]          =							preprocess(SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   Tuple          =							image.cpu()
  
         # 5. set timesteps
         self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE_    ,      device=SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   List[str]          =							self.scheduler.timesteps
  
         # 5. Add noise to image
         UpperCamelCase				:   Union[str, Any]          =							torch.tensor([noise_level]    ,      dtype=torch.long    ,      device=SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   List[Any]          =							torch.randn(image.shape    ,      generator=SCREAMING_SNAKE_CASE_    ,      device=SCREAMING_SNAKE_CASE_    ,      dtype=SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   Dict          =							self.low_res_scheduler.add_noise(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ )
  
         UpperCamelCase				:   Dict          =							2 if do_classifier_free_guidance else 1
         UpperCamelCase				:   List[Any]          =							np.concatenate([image] * batch_multiplier * num_images_per_prompt )
         UpperCamelCase				:   Any          =							np.concatenate([noise_level] * image.shape[0] )
  
         # 6. Prepare latent variables
         UpperCamelCase       ,							UpperCamelCase				:   Optional[int]          =							image.shape[2:]
         UpperCamelCase				:   Any          =							self.prepare_latents(
             batch_size * num_images_per_prompt    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      )
  
         # 7. Check that sizes of image and latents match
         UpperCamelCase				:   Any          =							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
         UpperCamelCase				:   Union[str, Any]          =							self.prepare_extra_step_kwargs(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ )
  
         UpperCamelCase				:   Optional[int]          =							next(
             (input.type for input in self.unet.model.get_inputs() if input.name == """timestep""")    ,      """tensor(float)""" )
         UpperCamelCase				:   Tuple          =							ORT_TO_NP_TYPE[timestep_dtype]
  
         # 9. Denoising loop
         UpperCamelCase				:   Optional[Any]          =							len(SCREAMING_SNAKE_CASE_ ) - num_inference_steps * self.scheduler.order
         with self.progress_bar(total=SCREAMING_SNAKE_CASE_ ) as progress_bar:
             for i, t in enumerate(SCREAMING_SNAKE_CASE_ ):
                 # expand the latents if we are doing classifier free guidance
                 UpperCamelCase				:   int          =							np.concatenate([latents] * 2 ) if do_classifier_free_guidance else latents
  
                 # concat latents, mask, masked_image_latents in the channel dimension
                 UpperCamelCase				:   Dict          =							self.scheduler.scale_model_input(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ )
                 UpperCamelCase				:   List[str]          =							np.concatenate([latent_model_input, image]    ,      axis=1 )
  
                 # timestep to tensor
                 UpperCamelCase				:   Any          =							np.array([t]    ,      dtype=SCREAMING_SNAKE_CASE_ )
  
                 # predict the noise residual
                 UpperCamelCase				:   List[Any]          =							self.unet(
                     sample=SCREAMING_SNAKE_CASE_    ,      timestep=SCREAMING_SNAKE_CASE_    ,      encoder_hidden_states=SCREAMING_SNAKE_CASE_    ,      class_labels=noise_level.astype(np.intaa )    ,      )[0]
  
                 # perform guidance
                 if do_classifier_free_guidance:
                     UpperCamelCase       ,							UpperCamelCase				:   Dict          =							np.split(SCREAMING_SNAKE_CASE_    ,      2 )
                 UpperCamelCase				:   Tuple          =							noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
  
                 # compute the previous noisy sample x_t -> x_t-1
                 UpperCamelCase				:   Tuple          =							self.scheduler.step(
                     torch.from_numpy(SCREAMING_SNAKE_CASE_ )    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      **SCREAMING_SNAKE_CASE_ ).prev_sample
  
                 # call the callback, if provided
                 if i == len(SCREAMING_SNAKE_CASE_ ) - 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(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ )
  
          # 10. Post-processing
         UpperCamelCase				:   List[str]          =							self.decode_latents(latents.float() )
  
         # 11. Convert to PIL
         if output_type == "pil":
             UpperCamelCase				:   str          =							self.numpy_to_pil(SCREAMING_SNAKE_CASE_ )
  
         if not return_dict:
             return (image,)
  
         return ImagePipelineOutput(images=SCREAMING_SNAKE_CASE_ )
  
  
  
  
  
  
  
     def       a_							(							self    ,      SCREAMING_SNAKE_CASE_ ):
         UpperCamelCase				:   List[Any]          =							1 / 0.08333 * latents
         UpperCamelCase				:   Union[str, Any]          =							self.vae(latent_sample=SCREAMING_SNAKE_CASE_ )[0]
         UpperCamelCase				:   Union[str, Any]          =							np.clip(image / 2 + 0.5    ,      0    ,      1 )
         UpperCamelCase				:   Any          =							image.transpose((0, 2, 3, 1) )
         return image
  
  
  
  
  
  
  
     def       a_							(							self    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ = None    ,      SCREAMING_SNAKE_CASE_ = None    ,      ):
         if prompt is not None and isinstance(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ ):
             UpperCamelCase				:   List[Any]          =							1
         elif prompt is not None and isinstance(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ ):
             UpperCamelCase				:   Dict          =							len(SCREAMING_SNAKE_CASE_ )
         else:
             UpperCamelCase				:   Dict          =							prompt_embeds.shape[0]
  
         if prompt_embeds is None:
             UpperCamelCase				:   List[Any]          =							self.tokenizer(
                 SCREAMING_SNAKE_CASE_    ,      padding="""max_length"""    ,      max_length=self.tokenizer.model_max_length    ,      truncation=SCREAMING_SNAKE_CASE_    ,      return_tensors="""pt"""    ,      )
             UpperCamelCase				:   int          =							text_inputs.input_ids
             UpperCamelCase				:   List[Any]          =							self.tokenizer(SCREAMING_SNAKE_CASE_    ,      padding="""longest"""    ,      return_tensors="""pt""" ).input_ids
  
             if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
                 SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ ):
                 UpperCamelCase				:   Dict          =							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
             UpperCamelCase				:   List[Any]          =							self.text_encoder(
                 input_ids=text_input_ids.int().to(SCREAMING_SNAKE_CASE_ )    ,      )
             UpperCamelCase				:   str          =							prompt_embeds[0]
  
         UpperCamelCase       ,							UpperCamelCase       ,							UpperCamelCase				:   Tuple          =							prompt_embeds.shape
         # duplicate text embeddings for each generation per prompt, using mps friendly method
         UpperCamelCase				:   Tuple          =							prompt_embeds.repeat(1    ,      SCREAMING_SNAKE_CASE_ )
         UpperCamelCase				:   str          =							prompt_embeds.reshape(bs_embed * num_images_per_prompt    ,      SCREAMING_SNAKE_CASE_    ,      -1 )
  
         # get unconditional embeddings for classifier free guidance
         if do_classifier_free_guidance and negative_prompt_embeds is None:
             UpperCamelCase				:   List[str]
             if negative_prompt is None:
                 UpperCamelCase				:   Optional[int]          =							[""""""] * batch_size
             elif type(SCREAMING_SNAKE_CASE_ ) is not type(SCREAMING_SNAKE_CASE_ ):
                 raise TypeError(
                     f'`negative_prompt` should be the same type to `prompt`, but got {type(SCREAMING_SNAKE_CASE_ )} !='
                     f' {type(SCREAMING_SNAKE_CASE_ )}.' )
             elif isinstance(SCREAMING_SNAKE_CASE_    ,      SCREAMING_SNAKE_CASE_ ):
                 UpperCamelCase				:   List[Any]          =							[negative_prompt]
             elif batch_size != len(SCREAMING_SNAKE_CASE_ ):
                 raise ValueError(
                     f'`negative_prompt`: {negative_prompt} has batch size {len(SCREAMING_SNAKE_CASE_ )}, but `prompt`:'
                     f' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
                     """ the batch size of `prompt`.""" )
             else:
                 UpperCamelCase				:   Optional[Any]          =							negative_prompt
  
             UpperCamelCase				:   Union[str, Any]          =							text_input_ids.shape[-1]
             UpperCamelCase				:   Tuple          =							self.tokenizer(
                 SCREAMING_SNAKE_CASE_    ,      padding="""max_length"""    ,      max_length=SCREAMING_SNAKE_CASE_    ,      truncation=SCREAMING_SNAKE_CASE_    ,      return_tensors="""pt"""    ,      )
  
             # if hasattr(uncond_input, "attention_mask"):
             #     attention_mask = uncond_input.attention_mask.to(device)
             # else:
             #     attention_mask = None
  
             UpperCamelCase				:   List[Any]          =							self.text_encoder(
                 input_ids=uncond_input.input_ids.int().to(SCREAMING_SNAKE_CASE_ )    ,      )
             UpperCamelCase				:   Tuple          =							uncond_embeddings[0]
  
         if do_classifier_free_guidance:
             UpperCamelCase				:   Dict          =							uncond_embeddings.shape[1]
             # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
             UpperCamelCase				:   Tuple          =							uncond_embeddings.repeat(1    ,      SCREAMING_SNAKE_CASE_ )
             UpperCamelCase				:   List[str]          =							uncond_embeddings.reshape(batch_size * num_images_per_prompt    ,      SCREAMING_SNAKE_CASE_    ,      -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
             UpperCamelCase				:   Any          =							np.concatenate([uncond_embeddings, prompt_embeds] )
  
         return prompt_embeds
  
  
  
  
  