






"""simple docstring"""





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


UpperCamelCase_						:					Tuple      				=			logging.get_logger(__name__)

UpperCamelCase_						:					Dict      				=			{
    "facebook/mbart-large-cc25": "https://huggingface.co/facebook/mbart-large-cc25/resolve/main/config.json",
    # See all MBART models at https://huggingface.co/models?filter=mbart
}


class   _lowercase			(  SCREAMING_SNAKE_CASE_    ):
      _a :  str										= '''mbart'''
      _a :  List[str]										= ['''past_key_values''']
      _a :  str										= {'''num_attention_heads''': '''encoder_attention_heads''', '''hidden_size''': '''d_model'''}


      def __init__(     self					:						Tuple				, a					:						str=5_0_2_6_5				, a					:						str=1_0_2_4				, a					:						str=1_2				, a					:						Tuple=4_0_9_6				, a					:						Union[str, Any]=1_6				, a					:						Any=1_2				, a					:						Dict=4_0_9_6				, a					:						List[str]=1_6				, a					:						int=0.0				, a					:						List[str]=0.0				, a					:						List[Any]=True				, a					:						Optional[int]=True				, a					:						Any="gelu"				, a					:						List[str]=1_0_2_4				, a					:						str=0.1				, a					:						Dict=0.0				, a					:						Dict=0.0				, a					:						Tuple=0.0_2				, a					:						Union[str, Any]=0.0				, a					:						Optional[Any]=False				, a					:						Optional[Any]=1				, a					:						Dict=0				, a					:						Union[str, Any]=2				, a					:						List[Any]=2				, **a					:						int				, ):





            """simple docstring"""







            __snake_case   :     Optional[Any]	      =vocab_size
            __snake_case   :     int	      =max_position_embeddings
            __snake_case   :     int	      =d_model
            __snake_case   :     Optional[int]	      =encoder_ffn_dim
            __snake_case   :     int	      =encoder_layers
            __snake_case   :     List[str]	      =encoder_attention_heads
            __snake_case   :     List[str]	      =decoder_ffn_dim
            __snake_case   :     int	      =decoder_layers
            __snake_case   :     int	      =decoder_attention_heads
            __snake_case   :     Dict	      =dropout
            __snake_case   :     Optional[Any]	      =attention_dropout
            __snake_case   :     Tuple	      =activation_dropout
            __snake_case   :     str	      =activation_function
            __snake_case   :     List[str]	      =init_std
            __snake_case   :     List[Any]	      =encoder_layerdrop
            __snake_case   :     Optional[int]	      =decoder_layerdrop
            __snake_case   :     str	      =classifier_dropout
            __snake_case   :     Optional[Any]	      =use_cache
            __snake_case   :     Any	      =encoder_layers
            __snake_case   :     int	      =scale_embedding  # scale factor will be sqrt(d_model) if True
            super().__init__(
                pad_token_id=SCREAMING_SNAKE_CASE_				, bos_token_id=SCREAMING_SNAKE_CASE_				, eos_token_id=SCREAMING_SNAKE_CASE_				, is_encoder_decoder=SCREAMING_SNAKE_CASE_				, forced_eos_token_id=SCREAMING_SNAKE_CASE_				, **SCREAMING_SNAKE_CASE_				, )



class   _lowercase			(  SCREAMING_SNAKE_CASE_    ):


      @property
      def _UpperCamelCase    (     self					:						Any    ):





            """simple docstring"""







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

                  if self.use_past:
                        __snake_case   :     Optional[Any]	      ={0: '''batch'''}
                        __snake_case   :     str	      ={0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
                  else:
                        __snake_case   :     Union[str, Any]	      ={0: '''batch''', 1: '''decoder_sequence'''}
                        __snake_case   :     str	      ={0: '''batch''', 1: '''decoder_sequence'''}

                  if self.use_past:
                        self.fill_with_past_key_values_(SCREAMING_SNAKE_CASE_				, direction='''inputs'''    )
            elif self.task == "causal-lm":
                  # TODO: figure this case out.
                  __snake_case   :     List[Any]	      =OrderedDict(
                      [
                          ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                          ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                      ]    )
                  if self.use_past:
                        __snake_case      ,					__snake_case   :     Optional[Any]	      =self.num_layers
                        for i in range(SCREAMING_SNAKE_CASE_    ):
                              __snake_case   :     Optional[Any]	      ={0: '''batch''', 2: '''past_sequence + sequence'''}
                              __snake_case   :     Tuple	      ={0: '''batch''', 2: '''past_sequence + sequence'''}
            else:
                  __snake_case   :     str	      =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					:						Tuple    ):





            """simple docstring"""







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


      def _UpperCamelCase    (     self					:						Union[str, Any]				, a					:						Dict				, a					:						Any = -1				, a					:						List[Any] = -1				, a					:						Any = False				, a					:						List[str] = None				, ):





            """simple docstring"""







            __snake_case   :     Tuple	      =self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )

            # Generate decoder inputs
            __snake_case   :     int	      =seq_length if not self.use_past else 1
            __snake_case   :     str	      =self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )
            __snake_case   :     List[str]	      ={f'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
            __snake_case   :     int	      =dict(**SCREAMING_SNAKE_CASE_				, **SCREAMING_SNAKE_CASE_    )

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

                  __snake_case   :     str	      =torch.cat(
                      [common_inputs['''decoder_attention_mask'''], torch.ones(SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )]				, dim=1    )

                  __snake_case   :     Tuple	      =[]
                  # If the number of encoder and decoder layers are present in the model configuration, both are considered
                  __snake_case      ,					__snake_case   :     Tuple	      =self.num_layers
                  __snake_case   :     Union[str, Any]	      =min(SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )
                  __snake_case   :     Optional[int]	      =max(SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    ) - min_num_layers
                  __snake_case   :     str	      ='''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

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


      def _UpperCamelCase    (     self					:						List[str]				, a					:						Dict				, a					:						Optional[Any] = -1				, a					:						Any = -1				, a					:						Union[str, Any] = False				, a					:						Optional[int] = None				, ):





            """simple docstring"""







            __snake_case   :     List[Any]	      =self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )

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

                  __snake_case   :     Tuple	      =common_inputs['''attention_mask'''].dtype
                  __snake_case   :     Optional[Any]	      =torch.cat(
                      [common_inputs['''attention_mask'''], torch.ones(SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, dtype=SCREAMING_SNAKE_CASE_    )]				, dim=1    )
                  __snake_case   :     Tuple	      =[
                      (torch.zeros(SCREAMING_SNAKE_CASE_    ), torch.zeros(SCREAMING_SNAKE_CASE_    )) for _ in range(SCREAMING_SNAKE_CASE_    )
                  ]
            return common_inputs


      def _UpperCamelCase    (     self					:						List[str]				, a					:						List[str]				, a					:						Union[str, Any] = -1				, a					:						Union[str, Any] = -1				, a					:						Dict = False				, a					:						Any = None				, ):





            """simple docstring"""







            __snake_case   :     Optional[Any]	      =compute_effective_axis_dimension(
                SCREAMING_SNAKE_CASE_				, 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
            __snake_case   :     int	      =tokenizer.num_special_tokens_to_add(SCREAMING_SNAKE_CASE_    )
            __snake_case   :     Optional[Any]	      =compute_effective_axis_dimension(
                SCREAMING_SNAKE_CASE_				, fixed_dimension=OnnxConfig.default_fixed_sequence				, num_token_to_add=SCREAMING_SNAKE_CASE_    )

            # Generate dummy inputs according to compute batch and sequence
            __snake_case   :     Optional[int]	      =[''' '''.join([tokenizer.unk_token]    ) * seq_length] * batch_size
            __snake_case   :     Tuple	      =dict(tokenizer(SCREAMING_SNAKE_CASE_				, return_tensors=SCREAMING_SNAKE_CASE_    )    )
            return common_inputs


      def _UpperCamelCase    (     self					:						Tuple				, a					:						Union[str, Any]				, a					:						Optional[Any] = -1				, a					:						str = -1				, a					:						Dict = False				, a					:						int = None				, ):





            """simple docstring"""







            if self.task in ["default", "seq2seq-lm"]:
                  __snake_case   :     Any	      =self._generate_dummy_inputs_for_default_and_seqaseq_lm(
                      SCREAMING_SNAKE_CASE_				, batch_size=SCREAMING_SNAKE_CASE_				, seq_length=SCREAMING_SNAKE_CASE_				, is_pair=SCREAMING_SNAKE_CASE_				, framework=SCREAMING_SNAKE_CASE_    )

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

            return common_inputs







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





            """simple docstring"""







            if self.task in ["default", "seq2seq-lm"]:
                  __snake_case   :     Optional[int]	      =super()._flatten_past_key_values_(SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )
            else:
                  __snake_case   :     Dict	      =super(SCREAMING_SNAKE_CASE_				, self    )._flatten_past_key_values_(
                      SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_				, SCREAMING_SNAKE_CASE_    )




import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


lowercase__				:				List[str]						=      False






@nightly
@require_torch_gpu
class 							SCREAMING_SNAKE_CASE__     (				unittest.TestCase     ):

							"""simple docstring"""






							def      A__		(	self      )-> List[Any]:





													'''simple docstring'''






													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()



							def      A__		(	self      )-> Dict:





													'''simple docstring'''






													__UpperCamelCase  	=      VersatileDiffusionDualGuidedPipeline.from_pretrained('''shi-labs/versatile-diffusion'''      )
													# remove text_unet
													pipe.remove_unused_weights()
													pipe.to(SCREAMING_SNAKE_CASE_      )
													pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

													__UpperCamelCase  	=      load_image(
													    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''      )

													__UpperCamelCase  	=      torch.manual_seed(0      )
													__UpperCamelCase  	=      pipe(
													    prompt='''first prompt'''  ,						image=SCREAMING_SNAKE_CASE_  ,						text_to_image_strength=0.7_5  ,						generator=SCREAMING_SNAKE_CASE_  ,						guidance_scale=7.5  ,						num_inference_steps=2  ,						output_type='''numpy'''  ,						).images

													with tempfile.TemporaryDirectory() as tmpdirname:
																			pipe.save_pretrained(SCREAMING_SNAKE_CASE_      )
																			__UpperCamelCase  	=      VersatileDiffusionDualGuidedPipeline.from_pretrained(SCREAMING_SNAKE_CASE_      )

													pipe.to(SCREAMING_SNAKE_CASE_      )
													pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

													__UpperCamelCase  	=      generator.manual_seed(0      )
													__UpperCamelCase  	=      pipe(
													    prompt='''first prompt'''  ,						image=SCREAMING_SNAKE_CASE_  ,						text_to_image_strength=0.7_5  ,						generator=SCREAMING_SNAKE_CASE_  ,						guidance_scale=7.5  ,						num_inference_steps=2  ,						output_type='''numpy'''  ,						).images

													assert np.abs(image - new_image      ).sum() < 1E-5, "Models don't have the same forward pass"



							def      A__		(	self      )-> int:





													'''simple docstring'''






													__UpperCamelCase  	=      VersatileDiffusionDualGuidedPipeline.from_pretrained('''shi-labs/versatile-diffusion'''      )
													pipe.remove_unused_weights()
													pipe.to(SCREAMING_SNAKE_CASE_      )
													pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

													__UpperCamelCase  	=      '''cyberpunk 2077'''
													__UpperCamelCase  	=      load_image(
													    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''      )
													__UpperCamelCase  	=      torch.manual_seed(0      )
													__UpperCamelCase  	=      pipe(
													    prompt=SCREAMING_SNAKE_CASE_  ,						image=SCREAMING_SNAKE_CASE_  ,						text_to_image_strength=0.7_5  ,						generator=SCREAMING_SNAKE_CASE_  ,						guidance_scale=7.5  ,						num_inference_steps=50  ,						output_type='''numpy'''  ,						).images

													__UpperCamelCase  	=      image[0, 253:256, 253:256, -1]

													assert image.shape == (1, 512, 512, 3)
													__UpperCamelCase  	=      np.array([0.0_7_8_7, 0.0_8_4_9, 0.0_8_2_6, 0.0_8_1_2, 0.0_8_0_7, 0.0_7_9_5, 0.0_8_1_8, 0.0_7_9_8, 0.0_7_7_9]      )

													assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1E-2




