

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


__A			:					List[str]						=			logging.get_logger(__name__)  # pylint: disable=invalid-name





class 			_SCREAMING_SNAKE_CASE (   __snake_case		):







   '''simple docstring'''




   def __init__(       self					:			Tuple     , __lowerCamelCase					:			WhisperForConditionalGeneration     , __lowerCamelCase					:			WhisperProcessor     , __lowerCamelCase					:			AutoencoderKL     , __lowerCamelCase					:			CLIPTextModel     , __lowerCamelCase					:			CLIPTokenizer     , __lowerCamelCase					:			UNetaDConditionModel     , __lowerCamelCase					:			Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]     , __lowerCamelCase					:			StableDiffusionSafetyChecker     , __lowerCamelCase					:			CLIPImageProcessor     , ):
       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=__lowerCamelCase     , speech_processor=__lowerCamelCase     , vae=__lowerCamelCase     , text_encoder=__lowerCamelCase     , tokenizer=__lowerCamelCase     , unet=__lowerCamelCase     , scheduler=__lowerCamelCase     , feature_extractor=__lowerCamelCase     , )





   def      _snake_case		(       self					:			Tuple     , __lowerCamelCase					:			Optional[Union[str, int]] = "auto"			):
       if slice_size == "auto":
           SCREAMING_SNAKE_CASE										=   self.unet.config.attention_head_dim // 2
       self.unet.set_attention_slice(__lowerCamelCase			)





   def      _snake_case		(       self					:			Any			):
       self.enable_attention_slicing(__lowerCamelCase			)





   @torch.no_grad()
   def __call__(       self					:			Optional[int]     , __lowerCamelCase					:			Optional[int]     , __lowerCamelCase					:			List[str]=16000     , __lowerCamelCase					:			int = 512     , __lowerCamelCase					:			int = 512     , __lowerCamelCase					:			int = 50     , __lowerCamelCase					:			float = 7.5     , __lowerCamelCase					:			Optional[Union[str, List[str]]] = None     , __lowerCamelCase					:			Optional[int] = 1     , __lowerCamelCase					:			float = 0.0     , __lowerCamelCase					:			Optional[torch.Generator] = None     , __lowerCamelCase					:			Optional[torch.FloatTensor] = None     , __lowerCamelCase					:			Optional[str] = "pil"     , __lowerCamelCase					:			bool = True     , __lowerCamelCase					:			Optional[Callable[[int, int, torch.FloatTensor], None]] = None     , __lowerCamelCase					:			int = 1     , **__lowerCamelCase					:			Dict     , ):
       SCREAMING_SNAKE_CASE										=   self.speech_processor.feature_extractor(
           __lowerCamelCase     , return_tensors="pt"     , sampling_rate=__lowerCamelCase			).input_features.to(self.device			)
       SCREAMING_SNAKE_CASE										=   self.speech_model.generate(__lowerCamelCase     , max_length=480000			)

       SCREAMING_SNAKE_CASE										=   self.speech_processor.tokenizer.batch_decode(__lowerCamelCase     , skip_special_tokens=__lowerCamelCase     , normalize=__lowerCamelCase			)[
           0
       ]

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

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

       # get prompt text embeddings
       SCREAMING_SNAKE_CASE										=   self.tokenizer(
           __lowerCamelCase     , padding="max_length"     , max_length=self.tokenizer.model_max_length     , return_tensors="pt"     , )
       SCREAMING_SNAKE_CASE										=   text_inputs.input_ids

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

       # duplicate text embeddings for each generation per prompt, using mps friendly method
       SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE										=   text_embeddings.shape
       SCREAMING_SNAKE_CASE										=   text_embeddings.repeat(1     , __lowerCamelCase     , 1			)
       SCREAMING_SNAKE_CASE										=   text_embeddings.view(bs_embed * num_images_per_prompt     , __lowerCamelCase     , -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.
       SCREAMING_SNAKE_CASE										=   guidance_scale > 1.0
       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance:
           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"     , )
           SCREAMING_SNAKE_CASE										=   self.text_encoder(uncond_input.input_ids.to(self.device			)			)[0]

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

       # set timesteps
       self.scheduler.set_timesteps(__lowerCamelCase			)

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

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

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

           # predict the noise residual
           SCREAMING_SNAKE_CASE										=   self.unet(__lowerCamelCase     , __lowerCamelCase     , encoder_hidden_states=__lowerCamelCase			).sample

           # perform guidance
           if do_classifier_free_guidance:
               SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE										=   noise_pred.chunk(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(__lowerCamelCase     , __lowerCamelCase     , __lowerCamelCase     , **__lowerCamelCase			).prev_sample

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

       SCREAMING_SNAKE_CASE										=   1 / 0.18_215 * latents
       SCREAMING_SNAKE_CASE										=   self.vae.decode(__lowerCamelCase			).sample

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

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

       if output_type == "pil":
           SCREAMING_SNAKE_CASE										=   self.numpy_to_pil(__lowerCamelCase			)

       if not return_dict:
           return image

       return StableDiffusionPipelineOutput(images=__lowerCamelCase     , nsfw_content_detected=__lowerCamelCase			)



import json
from typing import List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roberta import RobertaTokenizer


SCREAMING_SNAKE_CASE						:str			=  logging.get_logger(__name__)

SCREAMING_SNAKE_CASE						:Dict			=  {'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt''', '''tokenizer_file''': '''tokenizer.json'''}

SCREAMING_SNAKE_CASE						:str			=  {
    '''vocab_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/vocab.json''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/vocab.json''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/vocab.json''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/vocab.json''',
        '''roberta-base-openai-detector''': '''https://huggingface.co/roberta-base-openai-detector/resolve/main/vocab.json''',
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/vocab.json'''
        ),
    },
    '''merges_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/merges.txt''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/merges.txt''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/merges.txt''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/merges.txt''',
        '''roberta-base-openai-detector''': '''https://huggingface.co/roberta-base-openai-detector/resolve/main/merges.txt''',
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/merges.txt'''
        ),
    },
    '''tokenizer_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/tokenizer.json''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/tokenizer.json''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/tokenizer.json''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/tokenizer.json''',
        '''roberta-base-openai-detector''': (
            '''https://huggingface.co/roberta-base-openai-detector/resolve/main/tokenizer.json'''
        ),
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/tokenizer.json'''
        ),
    },
}

SCREAMING_SNAKE_CASE						:Tuple			=  {
    '''roberta-base''': 5_12,
    '''roberta-large''': 5_12,
    '''roberta-large-mnli''': 5_12,
    '''distilroberta-base''': 5_12,
    '''roberta-base-openai-detector''': 5_12,
    '''roberta-large-openai-detector''': 5_12,
}


class      __lowerCAmelCase  (			a       ):






  """simple docstring"""





  _SCREAMING_SNAKE_CASE		     =							VOCAB_FILES_NAMES
  _SCREAMING_SNAKE_CASE		     =							PRETRAINED_VOCAB_FILES_MAP
  _SCREAMING_SNAKE_CASE		     =							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  _SCREAMING_SNAKE_CASE		     =							['input_ids', 'attention_mask']
  _SCREAMING_SNAKE_CASE		     =							RobertaTokenizer



  def __init__(  self							:							Tuple     ,					_lowerCAmelCase							:							Union[str, Any]=None     ,					_lowerCAmelCase							:							Tuple=None     ,					_lowerCAmelCase							:							Any=None     ,					_lowerCAmelCase							:							Optional[Any]="replace"     ,					_lowerCAmelCase							:							str="<s>"     ,					_lowerCAmelCase							:							Tuple="</s>"     ,					_lowerCAmelCase							:							Any="</s>"     ,					_lowerCAmelCase							:							Tuple="<s>"     ,					_lowerCAmelCase							:							Any="<unk>"     ,					_lowerCAmelCase							:							Tuple="<pad>"     ,					_lowerCAmelCase							:							List[str]="<mask>"     ,					_lowerCAmelCase							:							str=False     ,					_lowerCAmelCase							:							Union[str, Any]=True     ,					**_lowerCAmelCase							:							int     ,					) ->			str:




     """simple docstring"""
     super().__init__(
         _lowerCAmelCase     ,					_lowerCAmelCase     ,					tokenizer_file=_lowerCAmelCase     ,					errors=_lowerCAmelCase     ,					bos_token=_lowerCAmelCase     ,					eos_token=_lowerCAmelCase     ,					sep_token=_lowerCAmelCase     ,					cls_token=_lowerCAmelCase     ,					unk_token=_lowerCAmelCase     ,					pad_token=_lowerCAmelCase     ,					mask_token=_lowerCAmelCase     ,					add_prefix_space=_lowerCAmelCase     ,					trim_offsets=_lowerCAmelCase     ,					**_lowerCAmelCase     ,					)

     snake_case_            =       json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()	)
     if pre_tok_state.get("add_prefix_space"     ,					_lowerCAmelCase	) != add_prefix_space:
        snake_case_            =       getattr(_lowerCAmelCase     ,					pre_tok_state.pop("type"	)	)
        snake_case_            =       add_prefix_space
        snake_case_            =       pre_tok_class(**_lowerCAmelCase	)

     snake_case_            =       add_prefix_space

     snake_case_            =       "post_processor"
     snake_case_            =       getattr(self.backend_tokenizer     ,					_lowerCAmelCase     ,					_lowerCAmelCase	)
     if tokenizer_component_instance:
        snake_case_            =       json.loads(tokenizer_component_instance.__getstate__()	)

        # The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
        if "sep" in state:
           snake_case_            =       tuple(state["sep"]	)
        if "cls" in state:
           snake_case_            =       tuple(state["cls"]	)

        snake_case_            =       False

        if state.get("add_prefix_space"     ,					_lowerCAmelCase	) != add_prefix_space:
           snake_case_            =       add_prefix_space
           snake_case_            =       True

        if state.get("trim_offsets"     ,					_lowerCAmelCase	) != trim_offsets:
           snake_case_            =       trim_offsets
           snake_case_            =       True

        if changes_to_apply:
           snake_case_            =       getattr(_lowerCAmelCase     ,					state.pop("type"	)	)
           snake_case_            =       component_class(**_lowerCAmelCase	)
           setattr(self.backend_tokenizer     ,					_lowerCAmelCase     ,					_lowerCAmelCase	)



  @property
  def        lowerCAmelCase__	(  self							:							Dict	) ->			str:




     """simple docstring"""
     if self._mask_token is None:
        if self.verbose:
           logger.error("Using mask_token, but it is not set yet."	)
        return None
     return str(self._mask_token	)



  @mask_token.setter
  def        lowerCAmelCase__	(  self							:							Tuple     ,					_lowerCAmelCase							:							Tuple	) ->			Optional[int]:




     """simple docstring"""
     snake_case_            =       AddedToken(_lowerCAmelCase     ,					lstrip=_lowerCAmelCase     ,					rstrip=_lowerCAmelCase	) if isinstance(_lowerCAmelCase     ,					_lowerCAmelCase	) else value
     snake_case_            =       value



  def        lowerCAmelCase__	(  self							:							Dict     ,					*_lowerCAmelCase							:							Optional[int]     ,					**_lowerCAmelCase							:							Optional[int]	) ->			BatchEncoding:




     """simple docstring"""
     snake_case_            =       kwargs.get("is_split_into_words"     ,					_lowerCAmelCase	)
     assert self.add_prefix_space or not is_split_into_words, (
         F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
         "to use it with pretokenized inputs."
     )

     return super()._batch_encode_plus(*_lowerCAmelCase     ,					**_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							Optional[Any]     ,					*_lowerCAmelCase							:							str     ,					**_lowerCAmelCase							:							Any	) ->			BatchEncoding:




     """simple docstring"""
     snake_case_            =       kwargs.get("is_split_into_words"     ,					_lowerCAmelCase	)

     assert self.add_prefix_space or not is_split_into_words, (
         F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
         "to use it with pretokenized inputs."
     )

     return super()._encode_plus(*_lowerCAmelCase     ,					**_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							int     ,					_lowerCAmelCase							:							str     ,					_lowerCAmelCase							:							Optional[str] = None	) ->			Tuple[str]:




     """simple docstring"""
     snake_case_            =       self._tokenizer.model.save(_lowerCAmelCase     ,					name=_lowerCAmelCase	)
     return tuple(_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							Any     ,					_lowerCAmelCase							:							int     ,					_lowerCAmelCase							:							List[Any]=None	) ->			Union[str, Any]:




     """simple docstring"""
     snake_case_            =       [self.bos_token_id] + token_ids_a + [self.eos_token_id]
     if token_ids_a is None:
        return output

     return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]



  def        lowerCAmelCase__	(  self							:							Union[str, Any]     ,					_lowerCAmelCase							:							List[int]     ,					_lowerCAmelCase							:							Optional[List[int]] = None	) ->			List[int]:




     """simple docstring"""
     snake_case_            =       [self.sep_token_id]
     snake_case_            =       [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]




