
'''simple docstring'''





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

from ...configuration_utils import PretrainedConfig
from ...feature_extraction_utils import FeatureExtractionMixin
from ...onnx import OnnxConfig
from ...onnx.utils import compute_effective_axis_dimension
from ...tokenization_utils_base import PreTrainedTokenizerBase
from ...utils import TensorType, logging


__lowerCAmelCase					:			Optional[int]			      =		logging.get_logger(__name__)

__lowerCAmelCase					:			int			      =		{
    "deepmind/language-perceiver": "https://huggingface.co/deepmind/language-perceiver/resolve/main/config.json",
    # See all Perceiver models at https://huggingface.co/models?filter=perceiver
}


class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      A__	:			int        =    '''perceiver'''




      def __init__(      self		:       List[Any]				,       _snake_case		:       Dict=256				,       _snake_case		:       Tuple=1280				,       _snake_case		:       List[str]=768				,       _snake_case		:       Tuple=1				,       _snake_case		:       Optional[Any]=26				,       _snake_case		:       Dict=8				,       _snake_case		:       Optional[int]=8				,       _snake_case		:       int=None				,       _snake_case		:       Tuple=None				,       _snake_case		:       Optional[Any]="kv"				,       _snake_case		:       Union[str, Any]=1				,       _snake_case		:       List[Any]=1				,       _snake_case		:       Any="gelu"				,       _snake_case		:       Optional[int]=0.1				,       _snake_case		:       Tuple=0.02				,       _snake_case		:       List[str]=1E-1_2				,       _snake_case		:       Optional[Any]=True				,       _snake_case		:       Tuple=262				,       _snake_case		:       Dict=2048				,       _snake_case		:       List[Any]=56				,       _snake_case		:       List[str]=[368, 496]				,       _snake_case		:       Optional[int]=16				,       _snake_case		:       str=1920				,       _snake_case		:       Dict=16				,       _snake_case		:       List[str]=[1, 16, 224, 224]				,       **_snake_case		:       Optional[Any]				,       ):
          super().__init__(**__a				)

          __lowercase		:			Optional[Any]	    =       num_latents
          __lowercase		:			Optional[Any]	    =       d_latents
          __lowercase		:			Union[str, Any]	    =       d_model
          __lowercase		:			Optional[Any]	    =       num_blocks
          __lowercase		:			List[str]	    =       num_self_attends_per_block
          __lowercase		:			Union[str, Any]	    =       num_self_attention_heads
          __lowercase		:			Tuple	    =       num_cross_attention_heads
          __lowercase		:			Optional[Any]	    =       qk_channels
          __lowercase		:			Union[str, Any]	    =       v_channels
          __lowercase		:			Optional[int]	    =       cross_attention_shape_for_attention
          __lowercase		:			int	    =       self_attention_widening_factor
          __lowercase		:			Dict	    =       cross_attention_widening_factor
          __lowercase		:			Union[str, Any]	    =       hidden_act
          __lowercase		:			Union[str, Any]	    =       attention_probs_dropout_prob
          __lowercase		:			List[str]	    =       initializer_range
          __lowercase		:			int	    =       layer_norm_eps
          __lowercase		:			Any	    =       use_query_residual
          # masked language modeling attributes
          __lowercase		:			Union[str, Any]	    =       vocab_size
          __lowercase		:			Optional[Any]	    =       max_position_embeddings
          # image classification attributes
          __lowercase		:			str	    =       image_size
          # flow attributes
          __lowercase		:			Optional[Any]	    =       train_size
          # multimodal autoencoding attributes
          __lowercase		:			int	    =       num_frames
          __lowercase		:			List[str]	    =       audio_samples_per_frame
          __lowercase		:			Optional[Any]	    =       samples_per_patch
          __lowercase		:			Tuple	    =       output_shape






class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      @property
      def 		snake_case_			(      self		:       Union[str, Any]				):
          if self.task == "multiple-choice":
              __lowercase		:			List[str]	    =       {0: '''batch''', 1: '''choice''', 2: '''sequence'''}
          else:
              __lowercase		:			List[str]	    =       {0: '''batch''', 1: '''sequence'''}
          return OrderedDict(
              [
                  ('''inputs''', dynamic_axis),
                  ('''attention_mask''', dynamic_axis),
              ]				)




      @property
      def 		snake_case_			(      self		:       str				):
          return 1E-4




      def 		snake_case_			(      self		:       List[str]				,       _snake_case		:       Optional[int]				,       _snake_case		:       List[str] = -1				,       _snake_case		:       List[Any] = -1				,       _snake_case		:       Any = -1				,       _snake_case		:       Optional[int] = False				,       _snake_case		:       Union[str, Any] = None				,       _snake_case		:       Union[str, Any] = 3				,       _snake_case		:       Optional[Any] = 40				,       _snake_case		:       Dict = 40				,       ):
          # copied from `transformers.onnx.config.OnnxConfig` and slightly altered/simplified

          if isinstance(__a				,       __a				):
              # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
              __lowercase		:			Any	    =       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
              __lowercase		:			int	    =       preprocessor.num_special_tokens_to_add(__a				)
              __lowercase		:			Optional[Any]	    =       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
              __lowercase		:			Dict	    =       [''' '''.join(['''a''']				) * seq_length] * batch_size
              __lowercase		:			Optional[int]	    =       dict(preprocessor(__a				,       return_tensors=__a				)				)
              __lowercase		:			Optional[int]	    =       inputs.pop('''input_ids'''				)
              return inputs
          elif isinstance(__a				,       __a				) and preprocessor.model_input_names[0] == "pixel_values":
              # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
              __lowercase		:			Optional[Any]	    =       compute_effective_axis_dimension(__a				,       fixed_dimension=OnnxConfig.default_fixed_batch				)
              __lowercase		:			Union[str, Any]	    =       self._generate_dummy_images(__a				,       __a				,       __a				,       __a				)
              __lowercase		:			List[Any]	    =       dict(preprocessor(images=__a				,       return_tensors=__a				)				)
              __lowercase		:			Optional[Any]	    =       inputs.pop('''pixel_values'''				)
              return inputs
          else:
              raise ValueError(
                  '''Unable to generate dummy inputs for the model. Please provide a tokenizer or a preprocessor.'''				)




import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np


class 						__lowerCAmelCase	:



      """simple docstring"""







      def 		snake_case_			(      self		:       str				):
          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Optional[Any]	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Tuple	    =       UNetaDConditionModel(
              sample_size=32				,       layers_per_block=1				,       block_out_channels=[32, 64]				,       down_block_types=[
                  '''ResnetDownsampleBlock2D''',
                  '''SimpleCrossAttnDownBlock2D''',
              ]				,       mid_block_type='''UNetMidBlock2DSimpleCrossAttn'''				,       up_block_types=['''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D''']				,       in_channels=3				,       out_channels=6				,       cross_attention_dim=32				,       encoder_hid_dim=32				,       attention_head_dim=8				,       addition_embed_type='''text'''				,       addition_embed_type_num_heads=2				,       cross_attention_norm='''group_norm'''				,       resnet_time_scale_shift='''scale_shift'''				,       act_fn='''gelu'''				,       )
          unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          torch.manual_seed(0				)
          __lowercase		:			List[str]	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       thresholding=_snake_case				,       dynamic_thresholding_ratio=0.95				,       sample_max_value=1.0				,       prediction_type='''epsilon'''				,       variance_type='''learned_range'''				,       )

          torch.manual_seed(0				)
          __lowercase		:			str	    =       IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }




      def 		snake_case_			(      self		:       Optional[int]				):
          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Tuple	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Dict	    =       UNetaDConditionModel(
              sample_size=32				,       layers_per_block=[1, 2]				,       block_out_channels=[32, 64]				,       down_block_types=[
                  '''ResnetDownsampleBlock2D''',
                  '''SimpleCrossAttnDownBlock2D''',
              ]				,       mid_block_type='''UNetMidBlock2DSimpleCrossAttn'''				,       up_block_types=['''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D''']				,       in_channels=6				,       out_channels=6				,       cross_attention_dim=32				,       encoder_hid_dim=32				,       attention_head_dim=8				,       addition_embed_type='''text'''				,       addition_embed_type_num_heads=2				,       cross_attention_norm='''group_norm'''				,       resnet_time_scale_shift='''scale_shift'''				,       act_fn='''gelu'''				,       class_embed_type='''timestep'''				,       mid_block_scale_factor=1.4_14				,       time_embedding_act_fn='''gelu'''				,       time_embedding_dim=32				,       )
          unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          torch.manual_seed(0				)
          __lowercase		:			Dict	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       thresholding=_snake_case				,       dynamic_thresholding_ratio=0.95				,       sample_max_value=1.0				,       prediction_type='''epsilon'''				,       variance_type='''learned_range'''				,       )

          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       )

          torch.manual_seed(0				)
          __lowercase		:			Any	    =       IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "image_noising_scheduler": image_noising_scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }




      def 		snake_case_			(      self		:       Any				):
          __lowercase		:			Tuple	    =       self.get_dummy_components()
          __lowercase		:			Any	    =       self.pipeline_class(**_snake_case				)
          pipe.to(_snake_case				)
          pipe.set_progress_bar_config(disable=_snake_case				)

          __lowercase		:			List[str]	    =       self.get_dummy_inputs(_snake_case				)

          __lowercase		:			List[Any]	    =       inputs['''prompt''']
          __lowercase		:			int	    =       inputs['''generator''']
          __lowercase		:			Dict	    =       inputs['''num_inference_steps''']
          __lowercase		:			Optional[int]	    =       inputs['''output_type''']

          if "image" in inputs:
              __lowercase		:			Tuple	    =       inputs['''image''']
          else:
              __lowercase		:			Dict	    =       None

          if "mask_image" in inputs:
              __lowercase		:			Tuple	    =       inputs['''mask_image''']
          else:
              __lowercase		:			List[Any]	    =       None

          if "original_image" in inputs:
              __lowercase		:			List[Any]	    =       inputs['''original_image''']
          else:
              __lowercase		:			Optional[int]	    =       None

          __lowercase    ,      __lowercase		:			Optional[int]	    =       pipe.encode_prompt(_snake_case				)

          # inputs with prompt converted to embeddings
          __lowercase		:			int	    =       {
              '''prompt_embeds''': prompt_embeds,
              '''negative_prompt_embeds''': negative_prompt_embeds,
              '''generator''': generator,
              '''num_inference_steps''': num_inference_steps,
              '''output_type''': output_type,
          }

          if image is not None:
              __lowercase		:			Tuple	    =       image

          if mask_image is not None:
              __lowercase		:			List[str]	    =       mask_image

          if original_image is not None:
              __lowercase		:			List[Any]	    =       original_image

          # set all optional components to None
          for optional_component in pipe._optional_components:
              setattr(_snake_case				,       _snake_case				,       _snake_case				)

          __lowercase		:			List[str]	    =       pipe(**_snake_case				)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_snake_case				)
              __lowercase		:			List[Any]	    =       self.pipeline_class.from_pretrained(_snake_case				)
              pipe_loaded.to(_snake_case				)
              pipe_loaded.set_progress_bar_config(disable=_snake_case				)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          for optional_component in pipe._optional_components:
              self.assertTrue(
                  getattr(_snake_case				,       _snake_case				) is None				,       F'`{optional_component}` did not stay set to None after loading.'				,       )

          __lowercase		:			int	    =       self.get_dummy_inputs(_snake_case				)

          __lowercase		:			Union[str, Any]	    =       inputs['''generator''']
          __lowercase		:			Any	    =       inputs['''num_inference_steps''']
          __lowercase		:			Tuple	    =       inputs['''output_type''']

          # inputs with prompt converted to embeddings
          __lowercase		:			List[str]	    =       {
              '''prompt_embeds''': prompt_embeds,
              '''negative_prompt_embeds''': negative_prompt_embeds,
              '''generator''': generator,
              '''num_inference_steps''': num_inference_steps,
              '''output_type''': output_type,
          }

          if image is not None:
              __lowercase		:			Dict	    =       image

          if mask_image is not None:
              __lowercase		:			Tuple	    =       mask_image

          if original_image is not None:
              __lowercase		:			Any	    =       original_image

          __lowercase		:			List[str]	    =       pipe_loaded(**_snake_case				)[0]

          __lowercase		:			Optional[int]	    =       np.abs(to_np(_snake_case				) - to_np(_snake_case				)				).max()
          self.assertLess(_snake_case				,       1E-4				)




      def 		snake_case_			(      self		:       Optional[int]				):
          __lowercase		:			Union[str, Any]	    =       self.get_dummy_components()
          __lowercase		:			Dict	    =       self.pipeline_class(**_snake_case				)
          pipe.to(_snake_case				)
          pipe.set_progress_bar_config(disable=_snake_case				)

          __lowercase		:			Dict	    =       self.get_dummy_inputs(_snake_case				)
          __lowercase		:			Any	    =       pipe(**_snake_case				)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_snake_case				)
              __lowercase		:			Dict	    =       self.pipeline_class.from_pretrained(_snake_case				)
              pipe_loaded.to(_snake_case				)
              pipe_loaded.set_progress_bar_config(disable=_snake_case				)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          __lowercase		:			Optional[Any]	    =       self.get_dummy_inputs(_snake_case				)
          __lowercase		:			str	    =       pipe_loaded(**_snake_case				)[0]

          __lowercase		:			int	    =       np.abs(to_np(_snake_case				) - to_np(_snake_case				)				).max()
          self.assertLess(_snake_case				,       1E-4				)


