


"""simple docstring"""




import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


lowercase_     	=    logging.get_logger(__name__)  # pylint: disable=invalid-name
class    __lowerCAmelCase   (  __SCREAMING_SNAKE_CASE			):


		'''simple docstring'''



		def __init__(    self				,					_a				,					_a				,					_a				,					_a				,					_a				,					_a				,					_a				,					_a				,					_a				,					):
				super().__init__()

				if safety_checker is None:
						logger.warning(
						    f'''You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure'''
						    ''' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'''
						    ''' results in services or applications open to the public. Both the diffusers team and Hugging Face'''
						    ''' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'''
						    ''' it only for use-cases that involve analyzing network behavior or auditing its results. For more'''
						    ''' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .''' )

				self.register_modules(
				    speech_model=_a				,					speech_processor=_a				,					vae=_a				,					text_encoder=_a				,					tokenizer=_a				,					unet=_a				,					scheduler=_a				,					feature_extractor=_a				,					)

		def      __UpperCAmelCase			(    self				,					_a = "auto" ):






				if slice_size == "auto":
						__a	 =     self.unet.config.attention_head_dim // 2
				self.unet.set_attention_slice(_a )

		def      __UpperCAmelCase			(    self ):
				self.enable_attention_slicing(_a )





		@torch.no_grad()
		def __call__(    self				,					_a				,					_a=16_000				,					_a = 512				,					_a = 512				,					_a = 50				,					_a = 7.5				,					_a = None				,					_a = 1				,					_a = 0.0				,					_a = None				,					_a = None				,					_a = "pil"				,					_a = True				,					_a = None				,					_a = 1				,					**_a				,					):
				__a	 =     self.speech_processor.feature_extractor(
				    _a				,					return_tensors='''pt'''				,					sampling_rate=_a ).input_features.to(self.device )
				__a	 =     self.speech_model.generate(_a				,					max_length=480_000 )

				__a	 =     self.speech_processor.tokenizer.batch_decode(_a				,					skip_special_tokens=_a				,					normalize=_a )[
				    0
				]

				if isinstance(_a				,					_a ):
						__a	 =     1
				elif isinstance(_a				,					_a ):
						__a	 =     len(_a )
				else:
						raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(_a )}''' )

				if height % 8 != 0 or width % 8 != 0:
						raise ValueError(f'''`height` and `width` have to be divisible by 8 but are {height} and {width}.''' )

				if (callback_steps is None) or (
				    callback_steps is not None and (not isinstance(_a				,					_a ) or callback_steps <= 0)
				):
						raise ValueError(
						    f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
						    f''' {type(_a )}.''' )

				# get prompt text embeddings
				__a	 =     self.tokenizer(
				    _a				,					padding='''max_length'''				,					max_length=self.tokenizer.model_max_length				,					return_tensors='''pt'''				,					)
				__a	 =     text_inputs.input_ids

				if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
						__a	 =     self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :] )
						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}''' )
						__a	 =     text_input_ids[:, : self.tokenizer.model_max_length]
				__a	 =     self.text_encoder(text_input_ids.to(self.device ) )[0]

				# duplicate text embeddings for each generation per prompt, using mps friendly method
				__a	 =     text_embeddings.shape
				__a	 =     text_embeddings.repeat(1				,					_a				,					1 )
				__a	 =     text_embeddings.view(bs_embed * num_images_per_prompt				,					_a				,					-1 )

				# 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.
				__a	 =     guidance_scale > 1.0
				# get unconditional embeddings for classifier free guidance
				if do_classifier_free_guidance:
						__a	 =     42
						if negative_prompt is None:
								__a	 =     [""] * batch_size
						elif type(_a ) is not type(_a ):
								raise TypeError(
								    f'''`negative_prompt` should be the same type to `prompt`, but got {type(_a )} !='''
								    f''' {type(_a )}.''' )
						elif isinstance(_a				,					_a ):
								__a	 =     [negative_prompt]
						elif batch_size != len(_a ):
								raise ValueError(
								    f'''`negative_prompt`: {negative_prompt} has batch size {len(_a )}, but `prompt`:'''
								    f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
								    ''' the batch size of `prompt`.''' )
						else:
								__a	 =     negative_prompt

						__a	 =     text_input_ids.shape[-1]
						__a	 =     self.tokenizer(
						    _a				,					padding='''max_length'''				,					max_length=_a				,					truncation=_a				,					return_tensors='''pt'''				,					)
						__a	 =     self.text_encoder(uncond_input.input_ids.to(self.device ) )[0]

						# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
						__a	 =     uncond_embeddings.shape[1]
						__a	 =     uncond_embeddings.repeat(1				,					_a				,					1 )
						__a	 =     uncond_embeddings.view(batch_size * num_images_per_prompt				,					_a				,					-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
						__a	 =     torch.cat([uncond_embeddings, text_embeddings] )

				# get the initial random noise unless the user supplied it

				# Unlike in other pipelines, latents need to be generated in the target device
				# for 1-to-1 results reproducibility with the CompVis implementation.
				# However this currently doesn't work in `mps`.
				__a	 =     (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
				__a	 =     text_embeddings.dtype
				if latents is None:
						if self.device.type == "mps":
								# randn does not exist on mps
								__a	 =     torch.randn(_a				,					generator=_a				,					device='''cpu'''				,					dtype=_a ).to(
								    self.device )
						else:
								__a	 =     torch.randn(_a				,					generator=_a				,					device=self.device				,					dtype=_a )
				else:
						if latents.shape != latents_shape:
								raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}''' )
						__a	 =     latents.to(self.device )

				# set timesteps
				self.scheduler.set_timesteps(_a )

				# Some schedulers like PNDM have timesteps as arrays
				# It's more optimized to move all timesteps to correct device beforehand
				__a	 =     self.scheduler.timesteps.to(self.device )

				# scale the initial noise by the standard deviation required by the scheduler
				__a	 =     latents * self.scheduler.init_noise_sigma

				# 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]
				__a	 =     "eta" in set(inspect.signature(self.scheduler.step ).parameters.keys() )
				__a	 =     {}
				if accepts_eta:
						__a	 =     eta

				for i, t in enumerate(self.progress_bar(_a ) ):
						# expand the latents if we are doing classifier free guidance
						__a	 =     torch.cat([latents] * 2 ) if do_classifier_free_guidance else latents
						__a	 =     self.scheduler.scale_model_input(_a				,					_a )

						# predict the noise residual
						__a	 =     self.unet(_a				,					_a				,					encoder_hidden_states=_a ).sample

						# perform guidance
						if do_classifier_free_guidance:
								__a	 =     noise_pred.chunk(2 )
								__a	 =     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

						# compute the previous noisy sample x_t -> x_t-1
						__a	 =     self.scheduler.step(_a				,					_a				,					_a				,					**_a ).prev_sample

						# call the callback, if provided
						if callback is not None and i % callback_steps == 0:
								callback(_a				,					_a				,					_a )

				__a	 =     1 / 0.1_8215 * latents
				__a	 =     self.vae.decode(_a ).sample

				__a	 =     (image / 2 + 0.5).clamp(0				,					1 )

				# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
				__a	 =     image.cpu().permute(0				,					2				,					3				,					1 ).float().numpy()

				if output_type == "pil":
						__a	 =     self.numpy_to_pil(_a )

				if not return_dict:
						return image

				return StableDiffusionPipelineOutput(images=_a				,					nsfw_content_detected=_a )














'''simple docstring'''
import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


UpperCamelCase__			=      logging.get_logger(__name__)


UpperCamelCase__			=      {'vocab_file': 'vocab.json', 'merges_file': 'merges.txt'}

# See all BART models at https://huggingface.co/models?filter=bart
UpperCamelCase__			=      {
    'vocab_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/vocab.json',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/vocab.json',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json',
    },
    'merges_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/merges.txt',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/merges.txt',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt',
    },
}

UpperCamelCase__			=      {
    'facebook/bart-base': 1024,
    'facebook/bart-large': 1024,
    'facebook/bart-large-mnli': 1024,
    'facebook/bart-large-cnn': 1024,
    'facebook/bart-large-xsum': 1024,
    'yjernite/bart_eli5': 1024,
}




@lru_cache()
def 		__SCREAMING_SNAKE_CASE       (	):


   """simple docstring"""




   lowercase_						:     List[Any]							     =  (
       list(range(ord("!"				)    ,		ord("~"				) + 1				)				) + list(range(ord("¡"				)    ,		ord("¬"				) + 1				)				) + list(range(ord("®"				)    ,		ord("ÿ"				) + 1				)				)
   )
   lowercase_						:     str							     =  bs[:]
   lowercase_						:     Dict							     =  0
   for b in range(2**8				):
      if b not in bs:
         bs.append(_UpperCamelCase				)
         cs.append(2**8 + n				)
         n += 1
   lowercase_						:     Optional[int]							     =  [chr(_UpperCamelCase				) for n in cs]
   return dict(zip(_UpperCamelCase    ,		_UpperCamelCase				)				)




def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase				):


   """simple docstring"""




   lowercase_						:     Optional[int]							     =  set()
   lowercase_						:     Union[str, Any]							     =  word[0]
   for char in word[1:]:
      pairs.add((prev_char, char)				)
      lowercase_						:     Dict							     =  char
   return pairs





class 							_UpperCAmelCase    (		snake_case      ):
    __lowerCamelCase: Dict									=     VOCAB_FILES_NAMES
    __lowerCamelCase: int									=     PRETRAINED_VOCAB_FILES_MAP
    __lowerCamelCase: Dict									=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    __lowerCamelCase: str									=     ['input_ids', 'attention_mask']




    def __init__(      self	:							Tuple	,  a	:							Tuple	,  a	:							Tuple	,  a	:							int="replace"	,  a	:							Optional[int]="<s>"	,  a	:							Tuple="</s>"	,  a	:							Tuple="</s>"	,  a	:							Tuple="<s>"	,  a	:							Optional[Any]="<unk>"	,  a	:							Dict="<pad>"	,  a	:							List[str]="<mask>"	,  a	:							Tuple=False	,  **a	:							Optional[int]	,  ):






       '''simple docstring'''






       lowercase_						:     List[Any]							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else bos_token
       lowercase_						:     str							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else eos_token
       lowercase_						:     str							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else sep_token
       lowercase_						:     Any							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else cls_token
       lowercase_						:     Dict							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else unk_token
       lowercase_						:     List[Any]							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else pad_token

       # Mask token behave like a normal word, i.e. include the space before it
       lowercase_						:     Union[str, Any]							     =  AddedToken(a	,  lstrip=a	,  rstrip=a      ) if isinstance(a	,  a      ) else mask_token

       super().__init__(
           errors=a	,  bos_token=a	,  eos_token=a	,  unk_token=a	,  sep_token=a	,  cls_token=a	,  pad_token=a	,  mask_token=a	,  add_prefix_space=a	,  **a	,  )

       with open(a	,  encoding="utf-8"      ) as vocab_handle:
          lowercase_						:     Any							     =  json.load(a      )
       lowercase_						:     Union[str, Any]							     =  {v: k for k, v in self.encoder.items()}
       lowercase_						:     Dict							     =  errors  # how to handle errors in decoding
       lowercase_						:     Any							     =  bytes_to_unicode()
       lowercase_						:     List[Any]							     =  {v: k for k, v in self.byte_encoder.items()}
       with open(a	,  encoding="utf-8"      ) as merges_handle:
          lowercase_						:     Optional[Any]							     =  merges_handle.read().split("\n"      )[1:-1]
       lowercase_						:     Any							     =  [tuple(merge.split()      ) for merge in bpe_merges]
       lowercase_						:     List[str]							     =  dict(zip(a	,  range(len(a      )      )      )      )
       lowercase_						:     Optional[Any]							     =  {}
       lowercase_						:     Union[str, Any]							     =  add_prefix_space

       # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
       lowercase_						:     str							     =  re.compile(R"'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"      )




    @property
    def 					lowerCAmelCase__					(      self	:							Tuple      ):






       '''simple docstring'''






       return len(self.encoder      )




    def 					lowerCAmelCase__					(      self	:							Dict      ):






       '''simple docstring'''






       return dict(self.encoder	,  **self.added_tokens_encoder      )




    def 					lowerCAmelCase__					(      self	:							Any	,  a	:							int      ):






       '''simple docstring'''






       if token in self.cache:
          return self.cache[token]
       lowercase_						:     Optional[Any]							     =  tuple(a      )
       lowercase_						:     Tuple							     =  get_pairs(a      )

       if not pairs:
          return token

       while True:
          lowercase_						:     Any							     =  min(a	,  key=lambda a      : self.bpe_ranks.get(a	,  float("inf"      )      )      )
          if bigram not in self.bpe_ranks:
             break
          lowercase_     ,		lowercase_						:     Dict							     =  bigram
          lowercase_						:     List[Any]							     =  []
          lowercase_						:     Optional[Any]							     =  0
          while i < len(a      ):
             try:
                lowercase_						:     Union[str, Any]							     =  word.index(a	,  a      )
             except ValueError:
                new_word.extend(word[i:]      )
                break
             else:
                new_word.extend(word[i:j]      )
                lowercase_						:     Any							     =  j

             if word[i] == first and i < len(a      ) - 1 and word[i + 1] == second:
                new_word.append(first + second      )
                i += 2
             else:
                new_word.append(word[i]      )
                i += 1
          lowercase_						:     Any							     =  tuple(a      )
          lowercase_						:     List[str]							     =  new_word
          if len(a      ) == 1:
             break
          else:
             lowercase_						:     Union[str, Any]							     =  get_pairs(a      )
       lowercase_						:     List[str]							     =  " ".join(a      )
       lowercase_						:     Optional[int]							     =  word
       return word




    def 					lowerCAmelCase__					(      self	:							Any	,  a	:							str      ):






       '''simple docstring'''






       lowercase_						:     Dict							     =  []
       for token in re.findall(self.pat	,  a      ):
          lowercase_						:     Tuple							     =  "".join(
              self.byte_encoder[b] for b in token.encode("utf-8"      )      )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
          bpe_tokens.extend(bpe_token for bpe_token in self.bpe(a      ).split(" "      )      )
       return bpe_tokens




    def 					lowerCAmelCase__					(      self	:							Tuple	,  a	:							Dict      ):






       '''simple docstring'''






       return self.encoder.get(a	,  self.encoder.get(self.unk_token      )      )




    def 					lowerCAmelCase__					(      self	:							Tuple	,  a	:							str      ):






       '''simple docstring'''






       return self.decoder.get(a      )




    def 					lowerCAmelCase__					(      self	:							int	,  a	:							Optional[Any]      ):






       '''simple docstring'''






       lowercase_						:     int							     =  "".join(a      )
       lowercase_						:     str							     =  bytearray([self.byte_decoder[c] for c in text]      ).decode("utf-8"	,  errors=self.errors      )
       return text




    def 					lowerCAmelCase__					(      self	:							List[str]	,  a	:							str	,  a	:							Optional[str] = None      ):






       '''simple docstring'''






       if not os.path.isdir(a      ):
          logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""      )
          return
       lowercase_						:     Any							     =  os.path.join(
           a	,  (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]      )
       lowercase_						:     Optional[int]							     =  os.path.join(
           a	,  (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]      )

       with open(a	,  "w"	,  encoding="utf-8"      ) as f:
          f.write(json.dumps(self.encoder	,  indent=2	,  sort_keys=a	,  ensure_ascii=a      ) + "\n"      )

       lowercase_						:     Dict							     =  0
       with open(a	,  "w"	,  encoding="utf-8"      ) as writer:
          writer.write("#version: 0.2\n"      )
          for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	,  key=lambda a      : kv[1]      ):
             if index != token_index:
                logger.warning(
                    f"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
                    " Please check that the tokenizer is not corrupted!"      )
                lowercase_						:     Optional[Any]							     =  token_index
             writer.write(" ".join(a      ) + "\n"      )
             index += 1

       return vocab_file, merge_file




    def 					lowerCAmelCase__					(      self	:							List[str]	,  a	:							List[int]	,  a	:							Optional[List[int]] = None      ):






       '''simple docstring'''






       if token_ids_a is None:
          return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
       lowercase_						:     List[str]							     =  [self.cls_token_id]
       lowercase_						:     List[Any]							     =  [self.sep_token_id]
       return cls + token_ids_a + sep + sep + token_ids_a + sep




    def 					lowerCAmelCase__					(      self	:							int	,  a	:							List[int]	,  a	:							Optional[List[int]] = None	,  a	:							bool = False      ):






       '''simple docstring'''






       if already_has_special_tokens:
          return super().get_special_tokens_mask(
              token_ids_a=a	,  token_ids_a=a	,  already_has_special_tokens=a      )

       if token_ids_a is None:
          return [1] + ([0] * len(a      )) + [1]
       return [1] + ([0] * len(a      )) + [1, 1] + ([0] * len(a      )) + [1]




    def 					lowerCAmelCase__					(      self	:							Optional[int]	,  a	:							List[int]	,  a	:							Optional[List[int]] = None      ):






       '''simple docstring'''






       lowercase_						:     Union[str, Any]							     =  [self.sep_token_id]
       lowercase_						:     List[str]							     =  [self.cls_token_id]

       if token_ids_a is None:
          return len(cls + token_ids_a + sep      ) * [0]
       return len(cls + token_ids_a + sep + sep + token_ids_a + sep      ) * [0]





    def 					lowerCAmelCase__					(      self	:							str	,  a	:							Any	,  a	:							int=False	,  **a	:							List[Any]      ):






       '''simple docstring'''






       lowercase_						:     Any							     =  kwargs.pop("add_prefix_space"	,  self.add_prefix_space      )
       if (is_split_into_words or add_prefix_space) and (len(a      ) > 0 and not text[0].isspace()):
          lowercase_						:     str							     =  " " + text
       return (text, kwargs)




