
from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


lowercase_											=	logging.get_logger(__name__)

lowercase_											=	{
    """google/bigbird-pegasus-large-arxiv""": (
        """https://huggingface.co/google/bigbird-pegasus-large-arxiv/resolve/main/config.json"""
    ),
    """google/bigbird-pegasus-large-pubmed""": (
        """https://huggingface.co/google/bigbird-pegasus-large-pubmed/resolve/main/config.json"""
    ),
    """google/bigbird-pegasus-large-bigpatent""": (
        """https://huggingface.co/google/bigbird-pegasus-large-bigpatent/resolve/main/config.json"""
    ),
    # See all BigBirdPegasus models at https://huggingface.co/models?filter=bigbird_pegasus
}





class   __UpperCamelCase				(    lowerCAmelCase__   ):






     """simple docstring"""







     lowerCAmelCase_     =	'''bigbird_pegasus'''
     lowerCAmelCase_     =	['''past_key_values''']
     lowerCAmelCase_     =	{
         '''num_attention_heads''': '''encoder_attention_heads''',
         '''hidden_size''': '''d_model''',
         '''attention_probs_dropout_prob''': '''attention_dropout''',
     }



     def __init__(							self    :    Union[str, Any]      ,		_A    :    List[Any]=9_6103      ,		_A    :    List[str]=4096      ,		_A    :    str=16      ,		_A    :    int=4096      ,		_A    :    Union[str, Any]=16      ,		_A    :    List[Any]=16      ,		_A    :    List[Any]=4096      ,		_A    :    List[str]=16      ,		_A    :    Any=0.0      ,		_A    :    Optional[Any]=0.0      ,		_A    :    str=True      ,		_A    :    List[str]=True      ,		_A    :    int="gelu_new"      ,		_A    :    str=1024      ,		_A    :    Dict=0.1      ,		_A    :    Tuple=0.0      ,		_A    :    Any=0.0      ,		_A    :    Optional[int]=0.02      ,		_A    :    str=2      ,		_A    :    Tuple=0.0      ,		_A    :    List[Any]=True      ,		_A    :    List[Any]=0      ,		_A    :    List[Any]=2      ,		_A    :    Any=1      ,		_A    :    str="block_sparse"      ,		_A    :    Union[str, Any]=64      ,		_A    :    Any=3      ,		_A    :    int=False      ,		**_A    :    List[Any]      ,		):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					Tuple			=						vocab_size
       __SCREAMING_SNAKE_CASE     :					Optional[int]			=						max_position_embeddings
       __SCREAMING_SNAKE_CASE     :					List[Any]			=						d_model
       __SCREAMING_SNAKE_CASE     :					Tuple			=						encoder_ffn_dim
       __SCREAMING_SNAKE_CASE     :					List[Any]			=						encoder_layers
       __SCREAMING_SNAKE_CASE     :					str			=						encoder_attention_heads
       __SCREAMING_SNAKE_CASE     :					str			=						decoder_ffn_dim
       __SCREAMING_SNAKE_CASE     :					Dict			=						decoder_layers
       __SCREAMING_SNAKE_CASE     :					str			=						decoder_attention_heads
       __SCREAMING_SNAKE_CASE     :					Tuple			=						dropout
       __SCREAMING_SNAKE_CASE     :					int			=						attention_dropout
       __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						activation_dropout
       __SCREAMING_SNAKE_CASE     :					Tuple			=						activation_function
       __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						init_std
       __SCREAMING_SNAKE_CASE     :					List[Any]			=						encoder_layerdrop
       __SCREAMING_SNAKE_CASE     :					List[str]			=						decoder_layerdrop
       __SCREAMING_SNAKE_CASE     :					int			=						classifier_dropout
       __SCREAMING_SNAKE_CASE     :					Tuple			=						use_cache
       __SCREAMING_SNAKE_CASE     :					List[str]			=						encoder_layers
       __SCREAMING_SNAKE_CASE     :					Optional[int]			=						scale_embedding  # scale factor will be sqrt(d_model) if True

       # extra config
       __SCREAMING_SNAKE_CASE     :					Optional[int]			=						attention_type
       __SCREAMING_SNAKE_CASE     :					Dict			=						block_size
       __SCREAMING_SNAKE_CASE     :					Optional[int]			=						num_random_blocks
       __SCREAMING_SNAKE_CASE     :					Optional[int]			=						use_bias

       super().__init__(
           pad_token_id=_A      ,		bos_token_id=_A      ,		eos_token_id=_A      ,		is_encoder_decoder=_A      ,		decoder_start_token_id=_A      ,		**_A      ,		)






class   __UpperCamelCase				(    lowerCAmelCase__   ):






     """simple docstring"""







     @property
     def 				UpperCAmelCase__    (							self    :    int				):







       """simple docstring"""




       if self.task in ["default", "seq2seq-lm"]:
         __SCREAMING_SNAKE_CASE     :					List[str]			=						OrderedDict(
             [
                 ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                 ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
             ]				)

         if self.use_past:
           __SCREAMING_SNAKE_CASE     :					Optional[int]			=						{0: '''batch'''}
           __SCREAMING_SNAKE_CASE     :					Any			=						{0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
         else:
           __SCREAMING_SNAKE_CASE     :					List[Any]			=						{0: '''batch''', 1: '''decoder_sequence'''}
           __SCREAMING_SNAKE_CASE     :					int			=						{0: '''batch''', 1: '''decoder_sequence'''}

         if self.use_past:
           self.fill_with_past_key_values_(_A      ,		direction='''inputs'''				)
       elif self.task == "causal-lm":
         # TODO: figure this case out.
         __SCREAMING_SNAKE_CASE     :					Dict			=						OrderedDict(
             [
                 ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                 ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
             ]				)
         if self.use_past:
           __SCREAMING_SNAKE_CASE     :					Dict			=						self.num_layers
           for i in range(_A				):
             __SCREAMING_SNAKE_CASE     :					Dict			=						{0: '''batch''', 2: '''past_sequence + sequence'''}
             __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						{0: '''batch''', 2: '''past_sequence + sequence'''}
       else:
         __SCREAMING_SNAKE_CASE     :					int			=						OrderedDict(
             [
                 ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                 ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                 ('''decoder_input_ids''', {0: '''batch''', 1: '''decoder_sequence'''}),
                 ('''decoder_attention_mask''', {0: '''batch''', 1: '''decoder_sequence'''}),
             ]				)

       return common_inputs



     @property
     def 				UpperCAmelCase__    (							self    :    Optional[Any]				):







       """simple docstring"""




       if self.task in ["default", "seq2seq-lm"]:
         __SCREAMING_SNAKE_CASE     :					Any			=						super().outputs
       else:
         __SCREAMING_SNAKE_CASE     :					Any			=						super(_A      ,		self				).outputs
         if self.use_past:
           __SCREAMING_SNAKE_CASE     :					Union[str, Any]			=						self.num_layers
           for i in range(_A				):
             __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						{0: '''batch''', 2: '''past_sequence + sequence'''}
             __SCREAMING_SNAKE_CASE     :					Optional[int]			=						{0: '''batch''', 2: '''past_sequence + sequence'''}
       return common_outputs



     def 				UpperCAmelCase__    (							self    :    Tuple      ,		_A    :    PreTrainedTokenizer      ,		_A    :    int = -1      ,		_A    :    int = -1      ,		_A    :    bool = False      ,		_A    :    Optional[TensorType] = None      ,		):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					List[str]			=						self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
           _A      ,		_A      ,		_A      ,		_A      ,		_A				)

       # Generate decoder inputs
       __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						seq_length if not self.use_past else 1
       __SCREAMING_SNAKE_CASE     :					Tuple			=						self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
           _A      ,		_A      ,		_A      ,		_A      ,		_A				)
       __SCREAMING_SNAKE_CASE     :					int			=						{F'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
       __SCREAMING_SNAKE_CASE     :					Union[str, Any]			=						dict(**_A      ,		**_A				)

       if self.use_past:
         if not is_torch_available():
           raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''				)
         else:
           import torch
         __SCREAMING_SNAKE_CASE     :					int			=						common_inputs['''input_ids'''].shape
         __SCREAMING_SNAKE_CASE     :					List[str]			=						common_inputs['''decoder_input_ids'''].shape[1]
         __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						self.num_attention_heads
         __SCREAMING_SNAKE_CASE     :					List[Any]			=						(
             batch,
             num_encoder_attention_heads,
             encoder_seq_length,
             self._config.hidden_size // num_encoder_attention_heads,
         )
         __SCREAMING_SNAKE_CASE     :					List[Any]			=						decoder_seq_length + 3
         __SCREAMING_SNAKE_CASE     :					List[str]			=						(
             batch,
             num_decoder_attention_heads,
             decoder_past_length,
             self._config.hidden_size // num_decoder_attention_heads,
         )

         __SCREAMING_SNAKE_CASE     :					List[str]			=						torch.cat(
             [common_inputs['''decoder_attention_mask'''], torch.ones(_A      ,		_A				)]      ,		dim=1				)

         __SCREAMING_SNAKE_CASE     :					List[str]			=						[]
         # If the number of encoder and decoder layers are present in the model configuration, both are considered
         __SCREAMING_SNAKE_CASE     :					str			=						self.num_layers
         __SCREAMING_SNAKE_CASE     :					Dict			=						min(_A      ,		_A				)
         __SCREAMING_SNAKE_CASE     :					List[str]			=						max(_A      ,		_A				) - min_num_layers
         __SCREAMING_SNAKE_CASE     :					Tuple			=						'''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

         for _ in range(_A				):
           common_inputs["past_key_values"].append(
               (
                   torch.zeros(_A				),
                   torch.zeros(_A				),
                   torch.zeros(_A				),
                   torch.zeros(_A				),
               )				)
         # TODO: test this.
         __SCREAMING_SNAKE_CASE     :					str			=						encoder_shape if remaining_side_name == '''encoder''' else decoder_shape
         for _ in range(_A      ,		_A				):
           common_inputs["past_key_values"].append((torch.zeros(_A				), torch.zeros(_A				))				)
       return common_inputs



     def 				UpperCAmelCase__    (							self    :    Optional[Any]      ,		_A    :    PreTrainedTokenizer      ,		_A    :    int = -1      ,		_A    :    int = -1      ,		_A    :    bool = False      ,		_A    :    Optional[TensorType] = None      ,		):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					str			=						self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
           _A      ,		_A      ,		_A      ,		_A      ,		_A				)

       if self.use_past:
         if not is_torch_available():
           raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''				)
         else:
           import torch
         __SCREAMING_SNAKE_CASE     :					str			=						common_inputs['''input_ids'''].shape
         # Not using the same length for past_key_values
         __SCREAMING_SNAKE_CASE     :					int			=						seqlen + 2
         __SCREAMING_SNAKE_CASE     :					Any			=						self.num_layers
         __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						self.num_attention_heads
         __SCREAMING_SNAKE_CASE     :					List[Any]			=						(
             batch,
             num_encoder_attention_heads,
             past_key_values_length,
             self._config.hidden_size // num_encoder_attention_heads,
         )

         __SCREAMING_SNAKE_CASE     :					Optional[int]			=						common_inputs['''attention_mask'''].dtype
         __SCREAMING_SNAKE_CASE     :					Union[str, Any]			=						torch.cat(
             [common_inputs['''attention_mask'''], torch.ones(_A      ,		_A      ,		dtype=_A				)]      ,		dim=1				)
         __SCREAMING_SNAKE_CASE     :					Optional[int]			=						[
             (torch.zeros(_A				), torch.zeros(_A				)) for _ in range(_A				)
         ]
       return common_inputs



     def 				UpperCAmelCase__    (							self    :    List[Any]      ,		_A    :    PreTrainedTokenizer      ,		_A    :    int = -1      ,		_A    :    int = -1      ,		_A    :    bool = False      ,		_A    :    Optional[TensorType] = None      ,		):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					List[str]			=						compute_effective_axis_dimension(
           _A      ,		fixed_dimension=OnnxConfig.default_fixed_batch      ,		num_token_to_add=0				)

       # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
       __SCREAMING_SNAKE_CASE     :					Optional[int]			=						tokenizer.num_special_tokens_to_add(_A				)
       __SCREAMING_SNAKE_CASE     :					int			=						compute_effective_axis_dimension(
           _A      ,		fixed_dimension=OnnxConfig.default_fixed_sequence      ,		num_token_to_add=_A				)

       # Generate dummy inputs according to compute batch and sequence
       __SCREAMING_SNAKE_CASE     :					List[Any]			=						[''' '''.join([tokenizer.unk_token]				) * seq_length] * batch_size
       __SCREAMING_SNAKE_CASE     :					Tuple			=						dict(tokenizer(_A      ,		return_tensors=_A				)				)
       return common_inputs



     def 				UpperCAmelCase__    (							self    :    str      ,		_A    :    PreTrainedTokenizer      ,		_A    :    int = -1      ,		_A    :    int = -1      ,		_A    :    bool = False      ,		_A    :    Optional[TensorType] = None      ,		):







       """simple docstring"""




       if self.task in ["default", "seq2seq-lm"]:
         __SCREAMING_SNAKE_CASE     :					List[str]			=						self._generate_dummy_inputs_for_default_and_seqaseq_lm(
             _A      ,		batch_size=_A      ,		seq_length=_A      ,		is_pair=_A      ,		framework=_A				)

       elif self.task == "causal-lm":
         __SCREAMING_SNAKE_CASE     :					List[Any]			=						self._generate_dummy_inputs_for_causal_lm(
             _A      ,		batch_size=_A      ,		seq_length=_A      ,		is_pair=_A      ,		framework=_A				)
       else:
         __SCREAMING_SNAKE_CASE     :					List[Any]			=						self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
             _A      ,		batch_size=_A      ,		seq_length=_A      ,		is_pair=_A      ,		framework=_A				)

       return common_inputs




     def 				UpperCAmelCase__    (							self    :    str      ,		_A    :    Any      ,		_A    :    List[str]      ,		_A    :    str      ,		_A    :    Dict				):







       """simple docstring"""




       if self.task in ["default", "seq2seq-lm"]:
         __SCREAMING_SNAKE_CASE     :					int			=						super()._flatten_past_key_values_(_A      ,		_A      ,		_A      ,		_A				)
       else:
         __SCREAMING_SNAKE_CASE     :					List[str]			=						super(_A      ,		self				)._flatten_past_key_values_(
             _A      ,		_A      ,		_A      ,		_A				)


from typing import Union

import torch
from PIL import Image
from torchvision import transforms as tfms
from tqdm.auto import tqdm
from transformers import CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)





class   __UpperCamelCase				(    lowerCAmelCase__   ):






     """simple docstring"""







     def __init__(							self    :    Optional[int]      ,		_A    :    AutoencoderKL      ,		_A    :    CLIPTextModel      ,		_A    :    CLIPTokenizer      ,		_A    :    UNetaDConditionModel      ,		_A    :    Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler]      ,		):







       """simple docstring"""




       super().__init__()

       self.register_modules(vae=_A      ,		text_encoder=_A      ,		tokenizer=_A      ,		unet=_A      ,		scheduler=_A				)



     def 				UpperCAmelCase__    (							self    :    Optional[int]      ,		_A    :    Dict				):







       """simple docstring"""




       with torch.no_grad():
         __SCREAMING_SNAKE_CASE     :					Dict			=						self.vae.encode(tfms.ToTensor()(_A				).unsqueeze(0				).to(self.device				) * 2 - 1				)
         __SCREAMING_SNAKE_CASE     :					str			=						0.1_82_15 * latent.latent_dist.sample()
       return latent



     def 				UpperCAmelCase__    (							self    :    Any      ,		_A    :    List[Any]				):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					List[str]			=						(1 / 0.1_82_15) * latent
       with torch.no_grad():
         __SCREAMING_SNAKE_CASE     :					Any			=						self.vae.decode(_A				).sample
       __SCREAMING_SNAKE_CASE     :					Any			=						(img / 2 + 0.5).clamp(0      ,		1				)
       __SCREAMING_SNAKE_CASE     :					List[str]			=						img.detach().cpu().permute(0      ,		2      ,		3      ,		1				).numpy()
       __SCREAMING_SNAKE_CASE     :					List[str]			=						(img * 255).round().astype('''uint8'''				)
       return Image.fromarray(img[0]				)



     def 				UpperCAmelCase__    (							self    :    Dict      ,		_A    :    List[Any]				):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					Any			=						self.tokenizer(
           _A      ,		padding='''max_length'''      ,		max_length=self.tokenizer.model_max_length      ,		truncation=_A      ,		return_tensors='''pt'''      ,		)

       __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						self.text_encoder(text_input.input_ids.to(self.device				)				)[0]

       __SCREAMING_SNAKE_CASE     :					Tuple			=						self.tokenizer(
           ''''''      ,		padding='''max_length'''      ,		max_length=self.tokenizer.model_max_length      ,		truncation=_A      ,		return_tensors='''pt'''      ,		)

       __SCREAMING_SNAKE_CASE     :					str			=						self.text_encoder(uncond_input.input_ids.to(self.device				)				)[0]

       return torch.cat([uncond_embedding, text_embedding]				)




     def __call__(							self    :    Union[str, Any]      ,		_A    :    Image.Image      ,		_A    :    str      ,		_A    :    float = 0.3      ,		_A    :    float = 0.6      ,		_A    :    float = 0.5      ,		_A    :    int = 42      ,		_A    :    int = 50      ,		_A    :    float = 7.5      ,		):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						steps - int(kmin * steps				)
       __SCREAMING_SNAKE_CASE     :					List[Any]			=						steps - int(kmax * steps				)

       __SCREAMING_SNAKE_CASE     :					List[str]			=						self.prep_text(_A				)

       self.scheduler.set_timesteps(_A				)

       __SCREAMING_SNAKE_CASE,						__SCREAMING_SNAKE_CASE     :					Dict			=						img.size
       __SCREAMING_SNAKE_CASE     :					Dict			=						self.encode(_A				)

       torch.manual_seed(_A				)
       __SCREAMING_SNAKE_CASE     :					Union[str, Any]			=						torch.randn(
           (1, self.unet.config.in_channels, height // 8, width // 8)      ,		).to(self.device				)

       __SCREAMING_SNAKE_CASE     :					Dict			=						self.scheduler.add_noise(
           _A      ,		_A      ,		timesteps=self.scheduler.timesteps[tmax]      ,		)

       __SCREAMING_SNAKE_CASE     :					List[str]			=						torch.cat([latents] * 2				)

       __SCREAMING_SNAKE_CASE     :					List[Any]			=						self.scheduler.scale_model_input(_A      ,		self.scheduler.timesteps[tmax]				)

       with torch.no_grad():
         __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						self.unet(
             _A      ,		self.scheduler.timesteps[tmax]      ,		encoder_hidden_states=_A      ,		).sample

       __SCREAMING_SNAKE_CASE,						__SCREAMING_SNAKE_CASE     :					List[str]			=						pred.chunk(2				)
       __SCREAMING_SNAKE_CASE     :					str			=						pred_uncond + guidance_scale * (pred_text - pred_uncond)

       __SCREAMING_SNAKE_CASE     :					Any			=						self.scheduler.step(_A      ,		self.scheduler.timesteps[tmax]      ,		_A				).prev_sample

       for i, t in enumerate(tqdm(self.scheduler.timesteps				)				):
         if i > tmax:
           if i < tmin:  # layout generation phase
             __SCREAMING_SNAKE_CASE     :					List[str]			=						self.scheduler.add_noise(
                 _A      ,		_A      ,		timesteps=_A      ,		)

             __SCREAMING_SNAKE_CASE     :					Union[str, Any]			=						(mix_factor * latents) + (
                 1 - mix_factor
             ) * orig_latents  # interpolating between layout noise and conditionally generated noise to preserve layout sematics
             __SCREAMING_SNAKE_CASE     :					Optional[Any]			=						torch.cat([input] * 2				)

           else:  # content generation phase
             __SCREAMING_SNAKE_CASE     :					List[str]			=						torch.cat([latents] * 2				)

           __SCREAMING_SNAKE_CASE     :					str			=						self.scheduler.scale_model_input(_A      ,		_A				)

           with torch.no_grad():
             __SCREAMING_SNAKE_CASE     :					int			=						self.unet(
                 _A      ,		_A      ,		encoder_hidden_states=_A      ,		).sample

           __SCREAMING_SNAKE_CASE,						__SCREAMING_SNAKE_CASE     :					Optional[int]			=						pred.chunk(2				)
           __SCREAMING_SNAKE_CASE     :					Optional[int]			=						pred_uncond + guidance_scale * (pred_text - pred_uncond)

           __SCREAMING_SNAKE_CASE     :					Any			=						self.scheduler.step(_A      ,		_A      ,		_A				).prev_sample

       return self.decode(_A				)
