



'''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


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

class 	__magic_name__						(    SCREAMING_SNAKE_CASE_   ):







							def __init__(							self					:       Optional[int]	,_UpperCAmelCase					:       Tuple	,_UpperCAmelCase					:       Dict	,_UpperCAmelCase					:       int	,_UpperCAmelCase					:       Union[str, Any]	,_UpperCAmelCase					:       Any	,_UpperCAmelCase					:       Tuple	,_UpperCAmelCase					:       List[str]	,_UpperCAmelCase					:       str	,_UpperCAmelCase					:       Optional[Any]	,):
									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=__UpperCAmelCase	,speech_processor=__UpperCAmelCase	,vae=__UpperCAmelCase	,text_encoder=__UpperCAmelCase	,tokenizer=__UpperCAmelCase	,unet=__UpperCAmelCase	,scheduler=__UpperCAmelCase	,feature_extractor=__UpperCAmelCase	,)







							def 							__lowercase			(							self					:       Optional[Any]	,_UpperCAmelCase					:       Optional[Any] = "auto"		):
									if slice_size == "auto":
											_a     :      int        =							self.unet.config.attention_head_dim // 2
									self.unet.set_attention_slice(__UpperCAmelCase		)







							def 							__lowercase			(							self					:       List[str]		):
									self.enable_attention_slicing(__UpperCAmelCase		)







							@torch.no_grad()
							def __call__(							self					:       int	,_UpperCAmelCase					:       Union[str, Any]	,_UpperCAmelCase					:       List[Any]=16000	,_UpperCAmelCase					:       Optional[int] = 512	,_UpperCAmelCase					:       Tuple = 512	,_UpperCAmelCase					:       Optional[Any] = 50	,_UpperCAmelCase					:       Optional[int] = 7.5	,_UpperCAmelCase					:       Optional[Any] = None	,_UpperCAmelCase					:       List[Any] = 1	,_UpperCAmelCase					:       Tuple = 0.0	,_UpperCAmelCase					:       Any = None	,_UpperCAmelCase					:       Union[str, Any] = None	,_UpperCAmelCase					:       Any = "pil"	,_UpperCAmelCase					:       int = True	,_UpperCAmelCase					:       Union[str, Any] = None	,_UpperCAmelCase					:       Tuple = 1	,**_UpperCAmelCase					:       str	,):
									_a     :      Optional[int]        =							self.speech_processor.feature_extractor(
									    __UpperCAmelCase	,return_tensors='pt'	,sampling_rate=__UpperCAmelCase		).input_features.to(self.device		)
									_a     :      Any        =							self.speech_model.generate(__UpperCAmelCase	,max_length=480000		)

									_a     :      Union[str, Any]        =							self.speech_processor.tokenizer.batch_decode(__UpperCAmelCase	,skip_special_tokens=__UpperCAmelCase	,normalize=__UpperCAmelCase		)[
									    0
									]

									if isinstance(__UpperCAmelCase	,__UpperCAmelCase		):
											_a     :      Any        =							1
									elif isinstance(__UpperCAmelCase	,__UpperCAmelCase		):
											_a     :      Tuple        =							len(__UpperCAmelCase		)
									else:
											raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(__UpperCAmelCase		)}"""		)

									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(__UpperCAmelCase	,__UpperCAmelCase		) or callback_steps <= 0)
									):
											raise ValueError(
											    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
											    F""" {type(__UpperCAmelCase		)}."""		)

									# get prompt text embeddings
									_a     :      Union[str, Any]        =							self.tokenizer(
									    __UpperCAmelCase	,padding='max_length'	,max_length=self.tokenizer.model_max_length	,return_tensors='pt'	,)
									_a     :      List[str]        =							text_inputs.input_ids

									if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
											_a     :      Optional[int]        =							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     :      Optional[Any]        =							text_input_ids[:, : self.tokenizer.model_max_length]
									_a     :      Optional[int]        =							self.text_encoder(text_input_ids.to(self.device		)		)[0]

									# duplicate text embeddings for each generation per prompt, using mps friendly method
									_a     :      Optional[int]        =							text_embeddings.shape
									_a     :      Tuple        =							text_embeddings.repeat(1	,__UpperCAmelCase	,1		)
									_a     :      List[str]        =							text_embeddings.view(bs_embed * num_images_per_prompt	,__UpperCAmelCase	,-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     :      Optional[int]        =							guidance_scale > 1.0
									# get unconditional embeddings for classifier free guidance
									if do_classifier_free_guidance:
											_a     :      List[str]
											if negative_prompt is None:
													_a     :      Any        =							[""""""] * batch_size
											elif type(__UpperCAmelCase		) is not type(__UpperCAmelCase		):
													raise TypeError(
													    F"""`negative_prompt` should be the same type to `prompt`, but got {type(__UpperCAmelCase		)} !="""
													    F""" {type(__UpperCAmelCase		)}."""		)
											elif isinstance(__UpperCAmelCase	,__UpperCAmelCase		):
													_a     :      Any        =							[negative_prompt]
											elif batch_size != len(__UpperCAmelCase		):
													raise ValueError(
													    F"""`negative_prompt`: {negative_prompt} has batch size {len(__UpperCAmelCase		)}, but `prompt`:"""
													    F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
													    ' the batch size of `prompt`.'		)
											else:
													_a     :      List[Any]        =							negative_prompt

											_a     :      List[str]        =							text_input_ids.shape[-1]
											_a     :      Union[str, Any]        =							self.tokenizer(
											    __UpperCAmelCase	,padding='max_length'	,max_length=__UpperCAmelCase	,truncation=__UpperCAmelCase	,return_tensors='pt'	,)
											_a     :      Union[str, Any]        =							self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

											# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
											_a     :      Optional[int]        =							uncond_embeddings.shape[1]
											_a     :      int        =							uncond_embeddings.repeat(1	,__UpperCAmelCase	,1		)
											_a     :      Any        =							uncond_embeddings.view(batch_size * num_images_per_prompt	,__UpperCAmelCase	,-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     :      List[Any]        =							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     :      Tuple        =							(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
									_a     :      Tuple        =							text_embeddings.dtype
									if latents is None:
											if self.device.type == "mps":
													# randn does not exist on mps
													_a     :      Dict        =							torch.randn(__UpperCAmelCase	,generator=__UpperCAmelCase	,device='cpu'	,dtype=__UpperCAmelCase		).to(
													    self.device		)
											else:
													_a     :      int        =							torch.randn(__UpperCAmelCase	,generator=__UpperCAmelCase	,device=self.device	,dtype=__UpperCAmelCase		)
									else:
											if latents.shape != latents_shape:
													raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""		)
											_a     :      int        =							latents.to(self.device		)

									# set timesteps
									self.scheduler.set_timesteps(__UpperCAmelCase		)

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

									# scale the initial noise by the standard deviation required by the scheduler
									_a     :      Any        =							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     :      List[str]        =							"""eta""" in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
									_a     :      int        =							{}
									if accepts_eta:
											_a     :      List[str]        =							eta

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

											# predict the noise residual
											_a     :      List[str]        =							self.unet(__UpperCAmelCase	,__UpperCAmelCase	,encoder_hidden_states=__UpperCAmelCase		).sample

											# perform guidance
											if do_classifier_free_guidance:
													_a     :      Tuple        =							noise_pred.chunk(2		)
													_a     :      List[Any]        =							noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

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

									_a     :      Optional[int]        =							1 / 0.1_82_15 * latents
									_a     :      Optional[int]        =							self.vae.decode(__UpperCAmelCase		).sample

									_a     :      Any        =							(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     :      Any        =							image.cpu().permute(0	,2	,3	,1		).float().numpy()

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

									if not return_dict:
											return image

									return StableDiffusionPipelineOutput(images=__UpperCAmelCase	,nsfw_content_detected=__UpperCAmelCase		)









'''simple docstring'''





import json
import os
from typing import Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_lowerCAmelCase      						=	logging.get_logger(__name__)

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

_lowerCAmelCase      						=	{
    '''vocab_file''': {'''ctrl''': '''https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json'''},
    '''merges_file''': {'''ctrl''': '''https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt'''},
}

_lowerCAmelCase      						=	{
    '''ctrl''': 256,
}

_lowerCAmelCase      						=	{
    '''Pregnancy''': 16_8629,
    '''Christianity''': 7675,
    '''Explain''': 10_6423,
    '''Fitness''': 6_3440,
    '''Saving''': 6_3163,
    '''Ask''': 2_7171,
    '''Ass''': 9_5985,
    '''Joke''': 16_3509,
    '''Questions''': 4_5622,
    '''Thoughts''': 4_9605,
    '''Retail''': 5_2342,
    '''Feminism''': 16_4338,
    '''Writing''': 1_1992,
    '''Atheism''': 19_2263,
    '''Netflix''': 4_8616,
    '''Computing''': 3_9639,
    '''Opinion''': 4_3213,
    '''Alone''': 4_4967,
    '''Funny''': 5_8917,
    '''Gaming''': 4_0358,
    '''Human''': 4088,
    '''India''': 1331,
    '''Joker''': 7_7138,
    '''Diet''': 3_6206,
    '''Legal''': 1_1859,
    '''Norman''': 4939,
    '''Tip''': 7_2689,
    '''Weight''': 5_2343,
    '''Movies''': 4_6273,
    '''Running''': 2_3425,
    '''Science''': 2090,
    '''Horror''': 3_7793,
    '''Confession''': 6_0572,
    '''Finance''': 1_2250,
    '''Politics''': 1_6360,
    '''Scary''': 19_1985,
    '''Support''': 1_2654,
    '''Technologies''': 3_2516,
    '''Teenage''': 6_6160,
    '''Event''': 3_2769,
    '''Learned''': 6_7460,
    '''Notion''': 18_2770,
    '''Wikipedia''': 3_7583,
    '''Books''': 6665,
    '''Extract''': 7_6050,
    '''Confessions''': 10_2701,
    '''Conspiracy''': 7_5932,
    '''Links''': 6_3674,
    '''Narcissus''': 15_0425,
    '''Relationship''': 5_4766,
    '''Relationships''': 13_4796,
    '''Reviews''': 4_1671,
    '''News''': 4256,
    '''Translation''': 2_6820,
    '''multilingual''': 12_8406,
}
def    _SCREAMING_SNAKE_CASE  (       UpperCamelCase			):



    """simple docstring"""





    lowerCAmelCase__ :  Optional[int]							      =					set()
    lowerCAmelCase__ :  str							      =					word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)			)
        lowerCAmelCase__ :  List[Any]							      =					char

    lowerCAmelCase__ :  Optional[Any]							      =					set(UpperCamelCase			)
    return pairs






class        lowerCAmelCase_(				SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  __lowercase							:      Dict				=      VOCAB_FILES_NAMES
  __lowercase							:      Dict				=      PRETRAINED_VOCAB_FILES_MAP
  __lowercase							:      Optional[Any]				=      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  __lowercase							:      Any				=      CONTROL_CODES




  def __init__(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase="<unk>"  ,**__UpperCAmelCase      )    ->		Optional[int]:
      super().__init__(unk_token=__UpperCAmelCase  ,**__UpperCAmelCase      )

      with open(__UpperCAmelCase  ,encoding="""utf-8"""      ) as vocab_handle:
          lowerCAmelCase__ :  int							      =					json.load(__UpperCAmelCase      )
      lowerCAmelCase__ :  Any							      =					{v: k for k, v in self.encoder.items()}
      with open(__UpperCAmelCase  ,encoding="""utf-8"""      ) as merges_handle:
          lowerCAmelCase__ :  Union[str, Any]							      =					merges_handle.read().split("""\n"""      )[1:-1]
      lowerCAmelCase__ :  Optional[Any]							      =					[tuple(merge.split()      ) for merge in merges]
      lowerCAmelCase__ :  int							      =					dict(zip(__UpperCAmelCase  ,range(len(__UpperCAmelCase      )      )      )      )
      lowerCAmelCase__ :  List[Any]							      =					{}




  @property
  def 							UpperCAmelCase_	(    self      )    ->		Optional[Any]:
      return len(self.encoder      )




  def 							UpperCAmelCase_	(    self      )    ->		Optional[Any]:
      return dict(self.encoder  ,**self.added_tokens_encoder      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		int:
      if token in self.cache:
          return self.cache[token]
      lowerCAmelCase__ :  Optional[Any]							      =					tuple(__UpperCAmelCase      )
      lowerCAmelCase__ :  List[str]							      =					tuple(list(word[:-1]      ) + [word[-1] + """</w>"""]      )
      lowerCAmelCase__ :  Any							      =					get_pairs(__UpperCAmelCase      )

      if not pairs:
          return token

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

              if word[i] == first and i < len(__UpperCAmelCase      ) - 1 and word[i + 1] == second:
                  new_word.append(first + second      )
                  i += 2
              else:
                  new_word.append(word[i]      )
                  i += 1
          lowerCAmelCase__ :  Tuple							      =					tuple(__UpperCAmelCase      )
          lowerCAmelCase__ :  Tuple							      =					new_word
          if len(__UpperCAmelCase      ) == 1:
              break
          else:
              lowerCAmelCase__ :  Dict							      =					get_pairs(__UpperCAmelCase      )
      lowerCAmelCase__ :  List[Any]							      =					"""@@ """.join(__UpperCAmelCase      )
      lowerCAmelCase__ :  Optional[Any]							      =					word[:-4]
      lowerCAmelCase__ :  Optional[Any]							      =					word
      return word




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Dict:
      lowerCAmelCase__ :  Optional[int]							      =					[]

      lowerCAmelCase__ :  Any							      =					re.findall(R"""\S+\n?"""  ,__UpperCAmelCase      )

      for token in words:
          split_tokens.extend(list(self.bpe(__UpperCAmelCase      ).split(""" """      )      )      )
      return split_tokens




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Optional[Any]:
      return self.encoder.get(__UpperCAmelCase  ,self.encoder.get(self.unk_token      )      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Tuple:
      return self.decoder.get(__UpperCAmelCase  ,self.unk_token      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Tuple:
      lowerCAmelCase__ :  Any							      =					""" """.join(__UpperCAmelCase      ).replace("""@@ """  ,""""""      ).strip()
      return out_string






  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase = None      )    ->		Tuple[str]:
      if not os.path.isdir(__UpperCAmelCase      ):
          logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""      )
          return
      lowerCAmelCase__ :  List[Any]							      =					os.path.join(
          __UpperCAmelCase  ,(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]      )
      lowerCAmelCase__ :  Optional[int]							      =					os.path.join(
          __UpperCAmelCase  ,(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]      )

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

      lowerCAmelCase__ :  int							      =					0
      with open(__UpperCAmelCase  ,"""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 __UpperCAmelCase      : 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!"""      )
                  lowerCAmelCase__ :  Dict							      =					token_index
              writer.write(""" """.join(__UpperCAmelCase      ) + """\n"""      )
              index += 1

      return vocab_file, merge_file

  # def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
  #     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
  #     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
  #     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
  #     return ''.join(tokens_generated_so_far)


