



"""simple docstring"""






import inspect
import unittest

from transformers import DPTConfig
from transformers.file_utils import is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch
     from torch import nn

     from transformers import MODEL_MAPPING, DPTForDepthEstimation, DPTForSemanticSegmentation, DPTModel
     from transformers.models.dpt.modeling_dpt import DPT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from PIL import Image

     from transformers import DPTImageProcessor






class 	__snake_case	:




  def __init__(			self					:						List[Any]						, __lowerCAmelCase					:						Optional[int]						, __lowerCAmelCase					:						str=2						, __lowerCAmelCase					:						Optional[Any]=3_2						, __lowerCAmelCase					:						Tuple=1_6						, __lowerCAmelCase					:						List[Any]=3						, __lowerCAmelCase					:						Union[str, Any]=True						, __lowerCAmelCase					:						Optional[int]=True						, __lowerCAmelCase					:						int=3_2						, __lowerCAmelCase					:						Union[str, Any]=4						, __lowerCAmelCase					:						Any=[0, 1, 2, 3]						, __lowerCAmelCase					:						Dict=4						, __lowerCAmelCase					:						int=3_7						, __lowerCAmelCase					:						List[str]="gelu"						, __lowerCAmelCase					:						Any=0.1						, __lowerCAmelCase					:						List[str]=0.1						, __lowerCAmelCase					:						List[str]=0.02						, __lowerCAmelCase					:						int=3						, __lowerCAmelCase					:						List[str]=[1, 3_8_4, 2_4, 2_4]						, __lowerCAmelCase					:						List[str]=True						, __lowerCAmelCase					:						Any=None						, ):






         """simple docstring"""






         _lowerCamelCase						:    str         =			parent
         _lowerCamelCase						:    Dict         =			batch_size
         _lowerCamelCase						:    Any         =			image_size
         _lowerCamelCase						:    Any         =			patch_size
         _lowerCamelCase						:    List[Any]         =			num_channels
         _lowerCamelCase						:    Tuple         =			is_training
         _lowerCamelCase						:    Any         =			use_labels
         _lowerCamelCase						:    str         =			hidden_size
         _lowerCamelCase						:    Dict         =			num_hidden_layers
         _lowerCamelCase						:    Optional[int]         =			backbone_out_indices
         _lowerCamelCase						:    Any         =			num_attention_heads
         _lowerCamelCase						:    str         =			intermediate_size
         _lowerCamelCase						:    Union[str, Any]         =			hidden_act
         _lowerCamelCase						:    Optional[int]         =			hidden_dropout_prob
         _lowerCamelCase						:    Optional[int]         =			attention_probs_dropout_prob
         _lowerCamelCase						:    str         =			initializer_range
         _lowerCamelCase						:    Tuple         =			num_labels
         _lowerCamelCase						:    str         =			backbone_featmap_shape
         _lowerCamelCase						:    Optional[Any]         =			scope
         _lowerCamelCase						:    Optional[int]         =			is_hybrid
         # sequence length of DPT = num_patches + 1 (we add 1 for the [CLS] token)
         _lowerCamelCase						:    int         =			(image_size // patch_size) ** 2
         _lowerCamelCase						:    str         =			num_patches + 1




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         _lowerCamelCase						:    str         =			floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]		)

         _lowerCamelCase						:    Optional[int]         =			None
         if self.use_labels:
                _lowerCamelCase						:    List[Any]         =			ids_tensor([self.batch_size, self.image_size, self.image_size]						, self.num_labels		)

         _lowerCamelCase						:    Optional[Any]         =			self.get_config()

         return config, pixel_values, labels




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         _lowerCamelCase						:    Tuple         =			{
             '''global_padding''': '''same''',
             '''layer_type''': '''bottleneck''',
             '''depths''': [3, 4, 9],
             '''out_features''': ['''stage1''', '''stage2''', '''stage3'''],
             '''embedding_dynamic_padding''': True,
             '''hidden_sizes''': [9_6, 1_9_2, 3_8_4, 7_6_8],
             '''num_groups''': 2,
         }

         return DPTConfig(
             image_size=self.image_size						, patch_size=self.patch_size						, num_channels=self.num_channels						, hidden_size=self.hidden_size						, num_hidden_layers=self.num_hidden_layers						, backbone_out_indices=self.backbone_out_indices						, num_attention_heads=self.num_attention_heads						, intermediate_size=self.intermediate_size						, hidden_act=self.hidden_act						, hidden_dropout_prob=self.hidden_dropout_prob						, attention_probs_dropout_prob=self.attention_probs_dropout_prob						, is_decoder=__SCREAMING_SNAKE_CASE						, initializer_range=self.initializer_range						, is_hybrid=self.is_hybrid						, backbone_config=__SCREAMING_SNAKE_CASE						, backbone_featmap_shape=self.backbone_featmap_shape						, )




  def 						SCREAMING_SNAKE_CASE    (			self					:						int						, __lowerCAmelCase					:						Tuple						, __lowerCAmelCase					:						Union[str, Any]						, __lowerCAmelCase					:						Any		):






         """simple docstring"""






         _lowerCamelCase						:    int         =			DPTModel(config=__SCREAMING_SNAKE_CASE		)
         model.to(__SCREAMING_SNAKE_CASE		)
         model.eval()
         _lowerCamelCase						:    Optional[Any]         =			model(__SCREAMING_SNAKE_CASE		)
         self.parent.assertEqual(result.last_hidden_state.shape						, (self.batch_size, self.seq_length, self.hidden_size)		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[str]						, __lowerCAmelCase					:						Optional[int]						, __lowerCAmelCase					:						Dict						, __lowerCAmelCase					:						Tuple		):






         """simple docstring"""






         _lowerCamelCase						:    int         =			self.num_labels
         _lowerCamelCase						:    List[Any]         =			DPTForDepthEstimation(__SCREAMING_SNAKE_CASE		)
         model.to(__SCREAMING_SNAKE_CASE		)
         model.eval()
         _lowerCamelCase						:    Optional[int]         =			model(__SCREAMING_SNAKE_CASE		)
         self.parent.assertEqual(result.predicted_depth.shape						, (self.batch_size, self.image_size, self.image_size)		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[str]						, __lowerCAmelCase					:						Optional[int]						, __lowerCAmelCase					:						int						, __lowerCAmelCase					:						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			self.num_labels
         _lowerCamelCase						:    Optional[Any]         =			DPTForSemanticSegmentation(__SCREAMING_SNAKE_CASE		)
         model.to(__SCREAMING_SNAKE_CASE		)
         model.eval()
         _lowerCamelCase						:    str         =			model(__SCREAMING_SNAKE_CASE						, labels=__SCREAMING_SNAKE_CASE		)
         self.parent.assertEqual(
             result.logits.shape						, (self.batch_size, self.num_labels, self.image_size, self.image_size)		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Dict		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			self.prepare_config_and_inputs()
         _lowerCamelCase      ,   _lowerCamelCase      ,   _lowerCamelCase						:    Dict         =			config_and_inputs
         _lowerCamelCase						:    Optional[int]         =			{'''pixel_values''': pixel_values}
         return config, inputs_dict






@require_torch
class 	__snake_case	(     _A  ,					_A  ,					unittest.TestCase):
  snake_case__    :       str        =						(DPTModel, DPTForDepthEstimation, DPTForSemanticSegmentation) if is_torch_available() else ()
  snake_case__    :       int        =						(
      {
          "depth-estimation": DPTForDepthEstimation,
          "feature-extraction": DPTModel,
          "image-segmentation": DPTForSemanticSegmentation,
      }
      if is_torch_available()
      else {}
  )

  snake_case__    :       List[Any]        =						False
  snake_case__    :       Union[str, Any]        =						False
  snake_case__    :       str        =						False




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]		):






         """simple docstring"""






         _lowerCamelCase						:    Any         =			DPTModelTester(self		)
         _lowerCamelCase						:    List[Any]         =			ConfigTester(self						, config_class=__SCREAMING_SNAKE_CASE						, has_text_modality=__SCREAMING_SNAKE_CASE						, hidden_size=3_7		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Dict		):






         """simple docstring"""






         self.config_tester.run_common_tests()




  @unittest.skip(reason='''DPT does not use inputs_embeds'''		)
  def 						SCREAMING_SNAKE_CASE    (			self					:						List[str]		):






         """simple docstring"""






         pass




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase      ,   _lowerCamelCase						:    int         =			self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
                _lowerCamelCase						:    Union[str, Any]         =			model_class(__SCREAMING_SNAKE_CASE		)
                self.assertIsInstance(model.get_input_embeddings()						, (nn.Module)		)
                _lowerCamelCase						:    Optional[Any]         =			model.get_output_embeddings()
                self.assertTrue(x is None or isinstance(__SCREAMING_SNAKE_CASE						, nn.Linear		)		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						str		):






         """simple docstring"""






         _lowerCamelCase      ,   _lowerCamelCase						:    Tuple         =			self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
                _lowerCamelCase						:    Optional[int]         =			model_class(__SCREAMING_SNAKE_CASE		)
                _lowerCamelCase						:    int         =			inspect.signature(model.forward		)
                # signature.parameters is an OrderedDict => so arg_names order is deterministic
                _lowerCamelCase						:    List[Any]         =			[*signature.parameters.keys()]

                _lowerCamelCase						:    Union[str, Any]         =			['''pixel_values''']
                self.assertListEqual(arg_names[:1]						, __SCREAMING_SNAKE_CASE		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*__SCREAMING_SNAKE_CASE		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Tuple		):






         """simple docstring"""






         _lowerCamelCase						:    List[Any]         =			self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_depth_estimation(*__SCREAMING_SNAKE_CASE		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]		):






         """simple docstring"""






         _lowerCamelCase						:    Dict         =			self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_semantic_segmentation(*__SCREAMING_SNAKE_CASE		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[int]		):






         """simple docstring"""






         for model_class in self.all_model_classes:
                if model_class.__name__ == "DPTForDepthEstimation":
                       continue

                _lowerCamelCase      ,   _lowerCamelCase						:    Any         =			self.model_tester.prepare_config_and_inputs_for_common()
                _lowerCamelCase						:    List[Any]         =			True

                if model_class in get_values(__SCREAMING_SNAKE_CASE		):
                       continue

                _lowerCamelCase						:    int         =			model_class(__SCREAMING_SNAKE_CASE		)
                model.to(__SCREAMING_SNAKE_CASE		)
                model.train()
                _lowerCamelCase						:    Tuple         =			self._prepare_for_class(__SCREAMING_SNAKE_CASE						, __SCREAMING_SNAKE_CASE						, return_labels=__SCREAMING_SNAKE_CASE		)
                _lowerCamelCase						:    List[Any]         =			model(**__SCREAMING_SNAKE_CASE		).loss
                loss.backward()




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[str]		):






         """simple docstring"""






         for model_class in self.all_model_classes:
                if model_class.__name__ == "DPTForDepthEstimation":
                       continue

                _lowerCamelCase      ,   _lowerCamelCase						:    int         =			self.model_tester.prepare_config_and_inputs_for_common()
                _lowerCamelCase						:    List[Any]         =			False
                _lowerCamelCase						:    Dict         =			True

                if model_class in get_values(__SCREAMING_SNAKE_CASE		) or not model_class.supports_gradient_checkpointing:
                       continue
                _lowerCamelCase						:    Dict         =			model_class(__SCREAMING_SNAKE_CASE		)
                model.to(__SCREAMING_SNAKE_CASE		)
                model.gradient_checkpointing_enable()
                model.train()
                _lowerCamelCase						:    Union[str, Any]         =			self._prepare_for_class(__SCREAMING_SNAKE_CASE						, __SCREAMING_SNAKE_CASE						, return_labels=__SCREAMING_SNAKE_CASE		)
                _lowerCamelCase						:    List[Any]         =			model(**__SCREAMING_SNAKE_CASE		).loss
                loss.backward()




  def 						SCREAMING_SNAKE_CASE    (			self					:						int		):






         """simple docstring"""






         _lowerCamelCase      ,   _lowerCamelCase						:    int         =			self.model_tester.prepare_config_and_inputs_for_common()

         _lowerCamelCase						:    Dict         =			_config_zero_init(__SCREAMING_SNAKE_CASE		)
         for model_class in self.all_model_classes:
                _lowerCamelCase						:    Optional[int]         =			model_class(config=__SCREAMING_SNAKE_CASE		)
                # Skip the check for the backbone
                _lowerCamelCase						:    Optional[int]         =			[]
                for name, module in model.named_modules():
                       if module.__class__.__name__ == "DPTViTHybridEmbeddings":
                              _lowerCamelCase						:    Tuple         =			[f'''{name}.{key}''' for key in module.state_dict().keys()]
                              break

                for name, param in model.named_parameters():
                       if param.requires_grad:
                              if name in backbone_params:
                                     continue
                              self.assertIn(
                                  ((param.data.mean() * 1E9).round() / 1E9).item()						, [0.0, 1.0]						, msg=f'''Parameter {name} of model {model_class} seems not properly initialized'''						, )




  @unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''		)
  def 						SCREAMING_SNAKE_CASE    (			self					:						str		):






         """simple docstring"""






         pass




  @slow
  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         for model_name in DPT_PRETRAINED_MODEL_ARCHIVE_LIST[1:]:
                _lowerCamelCase						:    List[str]         =			DPTModel.from_pretrained(__SCREAMING_SNAKE_CASE		)
                self.assertIsNotNone(__SCREAMING_SNAKE_CASE		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase      ,   _lowerCamelCase						:    List[Any]         =			self.model_tester.prepare_config_and_inputs_for_common()
         _lowerCamelCase						:    Dict         =			'''add'''
         with self.assertRaises(__SCREAMING_SNAKE_CASE		):
                _lowerCamelCase						:    int         =			DPTForDepthEstimation(__SCREAMING_SNAKE_CASE		)






def 	snake_case_  (				):


       '''simple docstring'''





       _lowerCamelCase						:    Dict         =			Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''						)
       return image







@require_torch
@require_vision
@slow
class 	__snake_case	(     unittest.TestCase):




  def 						SCREAMING_SNAKE_CASE    (			self					:						str		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			DPTImageProcessor.from_pretrained('''Intel/dpt-hybrid-midas'''		)
         _lowerCamelCase						:    int         =			DPTForDepthEstimation.from_pretrained('''Intel/dpt-hybrid-midas'''		).to(__SCREAMING_SNAKE_CASE		)

         _lowerCamelCase						:    List[str]         =			prepare_img()
         _lowerCamelCase						:    Any         =			image_processor(images=__SCREAMING_SNAKE_CASE						, return_tensors='''pt'''		).to(__SCREAMING_SNAKE_CASE		)

         # forward pass
         with torch.no_grad():
                _lowerCamelCase						:    str         =			model(**__SCREAMING_SNAKE_CASE		)
                _lowerCamelCase						:    str         =			outputs.predicted_depth

         # verify the predicted depth
         _lowerCamelCase						:    Optional[Any]         =			torch.Size((1, 3_8_4, 3_8_4)		)
         self.assertEqual(predicted_depth.shape						, __SCREAMING_SNAKE_CASE		)

         _lowerCamelCase						:    List[Any]         =			torch.tensor(
             [[[5.64_37, 5.61_46, 5.65_11], [5.43_71, 5.56_49, 5.59_58], [5.52_15, 5.51_84, 5.52_93]]]		).to(__SCREAMING_SNAKE_CASE		)

         self.assertTrue(torch.allclose(outputs.predicted_depth[:3, :3, :3] / 1_0_0						, __SCREAMING_SNAKE_CASE						, atol=1E-4		)		)







"""simple docstring"""






from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin






@dataclass
class 	__snake_case	(     _lowercase):
  snake_case__    :       torch.FloatTensor






class 	__snake_case	(     _lowercase  ,					_lowercase):




  @register_to_config
  def __init__(			self					:						int						, __lowerCAmelCase					:						int = 3_2						, __lowerCAmelCase					:						int = 6_4						, __lowerCAmelCase					:						int = 2_0						, __lowerCAmelCase					:						int = 7_6_8						, __lowerCAmelCase					:						Tuple=7_7						, __lowerCAmelCase					:						Tuple=4						, __lowerCAmelCase					:						float = 0.0						, __lowerCAmelCase					:						str = "silu"						, __lowerCAmelCase					:						Optional[str] = None						, __lowerCAmelCase					:						Optional[str] = None						, __lowerCAmelCase					:						Optional[str] = "linear"						, __lowerCAmelCase					:						Optional[str] = "prd"						, __lowerCAmelCase					:						Optional[int] = None						, __lowerCAmelCase					:						Optional[int] = None						, __lowerCAmelCase					:						Optional[int] = None						, ):






         """simple docstring"""






         super().__init__()
         _lowerCamelCase						:    Any         =			num_attention_heads
         _lowerCamelCase						:    Optional[int]         =			attention_head_dim
         _lowerCamelCase						:    Dict         =			num_attention_heads * attention_head_dim
         _lowerCamelCase						:    Optional[Any]         =			additional_embeddings

         _lowerCamelCase						:    List[Any]         =			time_embed_dim or inner_dim
         _lowerCamelCase						:    Optional[Any]         =			embedding_proj_dim or embedding_dim
         _lowerCamelCase						:    Tuple         =			clip_embed_dim or embedding_dim

         _lowerCamelCase						:    int         =			Timesteps(__lowerCAmelCase						, __lowerCAmelCase						, 0		)
         _lowerCamelCase						:    str         =			TimestepEmbedding(__lowerCAmelCase						, __lowerCAmelCase						, out_dim=__lowerCAmelCase						, act_fn=__lowerCAmelCase		)

         _lowerCamelCase						:    Union[str, Any]         =			nn.Linear(__lowerCAmelCase						, __lowerCAmelCase		)

         if embedding_proj_norm_type is None:
                _lowerCamelCase						:    Dict         =			None
         elif embedding_proj_norm_type == "layer":
                _lowerCamelCase						:    Any         =			nn.LayerNorm(__lowerCAmelCase		)
         else:
                raise ValueError(f'''unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'''		)

         _lowerCamelCase						:    List[str]         =			nn.Linear(__lowerCAmelCase						, __lowerCAmelCase		)

         if encoder_hid_proj_type is None:
                _lowerCamelCase						:    Any         =			None
         elif encoder_hid_proj_type == "linear":
                _lowerCamelCase						:    List[str]         =			nn.Linear(__lowerCAmelCase						, __lowerCAmelCase		)
         else:
                raise ValueError(f'''unsupported encoder_hid_proj_type: {encoder_hid_proj_type}'''		)

         _lowerCamelCase						:    Optional[int]         =			nn.Parameter(torch.zeros(1						, num_embeddings + additional_embeddings						, __lowerCAmelCase		)		)

         if added_emb_type == "prd":
                _lowerCamelCase						:    Optional[Any]         =			nn.Parameter(torch.zeros(1						, 1						, __lowerCAmelCase		)		)
         elif added_emb_type is None:
                _lowerCamelCase						:    Union[str, Any]         =			None
         else:
                raise ValueError(
                    f'''`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.'''		)

         _lowerCamelCase						:    Tuple         =			nn.ModuleList(
             [
                 BasicTransformerBlock(
                     __lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, dropout=__lowerCAmelCase						, activation_fn='''gelu'''						, attention_bias=__lowerCAmelCase						, )
                 for d in range(__lowerCAmelCase		)
             ]		)

         if norm_in_type == "layer":
                _lowerCamelCase						:    Any         =			nn.LayerNorm(__lowerCAmelCase		)
         elif norm_in_type is None:
                _lowerCamelCase						:    Any         =			None
         else:
                raise ValueError(f'''Unsupported norm_in_type: {norm_in_type}.'''		)

         _lowerCamelCase						:    str         =			nn.LayerNorm(__lowerCAmelCase		)

         _lowerCamelCase						:    List[Any]         =			nn.Linear(__lowerCAmelCase						, __lowerCAmelCase		)

         _lowerCamelCase						:    Optional[int]         =			torch.full(
             [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]						, -1_00_00.0		)
         causal_attention_mask.triu_(1		)
         _lowerCamelCase						:    Dict         =			causal_attention_mask[None, ...]
         self.register_buffer('''causal_attention_mask'''						, __lowerCAmelCase						, persistent=__lowerCAmelCase		)

         _lowerCamelCase						:    int         =			nn.Parameter(torch.zeros(1						, __lowerCAmelCase		)		)
         _lowerCamelCase						:    Tuple         =			nn.Parameter(torch.zeros(1						, __lowerCAmelCase		)		)




  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def 						SCREAMING_SNAKE_CASE    (			self					:						int		):






         """simple docstring"""






         _lowerCamelCase						:    Any         =			{}

         def fn_recursive_add_processors(__lowerCAmelCase					:						str						, __lowerCAmelCase					:						torch.nn.Module						, __lowerCAmelCase					:						Dict[str, AttentionProcessor]		):
                if hasattr(__lowerCAmelCase						, '''set_processor'''		):
                       _lowerCamelCase						:    str         =			module.processor

                for sub_name, child in module.named_children():
                       fn_recursive_add_processors(f'''{name}.{sub_name}'''						, __lowerCAmelCase						, __lowerCAmelCase		)

                return processors

         for name, module in self.named_children():
                fn_recursive_add_processors(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase		)

         return processors




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any						, __lowerCAmelCase					:						Union[AttentionProcessor, Dict[str, AttentionProcessor]]		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			len(self.attn_processors.keys()		)

         if isinstance(__lowerCAmelCase						, __lowerCAmelCase		) and len(__lowerCAmelCase		) != count:
                raise ValueError(
                    f'''A dict of processors was passed, but the number of processors {len(__lowerCAmelCase		)} does not match the'''
                    f''' number of attention layers: {count}. Please make sure to pass {count} processor classes.'''		)

         def fn_recursive_attn_processor(__lowerCAmelCase					:						str						, __lowerCAmelCase					:						torch.nn.Module						, __lowerCAmelCase					:						Tuple		):
                if hasattr(__lowerCAmelCase						, '''set_processor'''		):
                       if not isinstance(__lowerCAmelCase						, __lowerCAmelCase		):
                              module.set_processor(__lowerCAmelCase		)
                       else:
                              module.set_processor(processor.pop(f'''{name}.processor'''		)		)

                for sub_name, child in module.named_children():
                       fn_recursive_attn_processor(f'''{name}.{sub_name}'''						, __lowerCAmelCase						, __lowerCAmelCase		)

         for name, module in self.named_children():
                fn_recursive_attn_processor(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]		):






         """simple docstring"""






         self.set_attn_processor(AttnProcessor()		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						str						, __lowerCAmelCase					:						List[str]						, __lowerCAmelCase					:						Union[torch.Tensor, float, int]						, __lowerCAmelCase					:						torch.FloatTensor						, __lowerCAmelCase					:						Optional[torch.FloatTensor] = None						, __lowerCAmelCase					:						Optional[torch.BoolTensor] = None						, __lowerCAmelCase					:						bool = True						, ):






         """simple docstring"""






         _lowerCamelCase						:    Optional[int]         =			hidden_states.shape[0]

         _lowerCamelCase						:    Union[str, Any]         =			timestep
         if not torch.is_tensor(__lowerCAmelCase		):
                _lowerCamelCase						:    Dict         =			torch.tensor([timesteps]						, dtype=torch.long						, device=hidden_states.device		)
         elif torch.is_tensor(__lowerCAmelCase		) and len(timesteps.shape		) == 0:
                _lowerCamelCase						:    str         =			timesteps[None].to(hidden_states.device		)

         # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
         _lowerCamelCase						:    Tuple         =			timesteps * torch.ones(__lowerCAmelCase						, dtype=timesteps.dtype						, device=timesteps.device		)

         _lowerCamelCase						:    Tuple         =			self.time_proj(__lowerCAmelCase		)

         # timesteps does not contain any weights and will always return f32 tensors
         # but time_embedding might be fp16, so we need to cast here.
         _lowerCamelCase						:    Union[str, Any]         =			timesteps_projected.to(dtype=self.dtype		)
         _lowerCamelCase						:    Any         =			self.time_embedding(__lowerCAmelCase		)

         if self.embedding_proj_norm is not None:
                _lowerCamelCase						:    str         =			self.embedding_proj_norm(__lowerCAmelCase		)

         _lowerCamelCase						:    Optional[Any]         =			self.embedding_proj(__lowerCAmelCase		)
         if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
                _lowerCamelCase						:    Optional[Any]         =			self.encoder_hidden_states_proj(__lowerCAmelCase		)
         elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
                raise ValueError('''`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set'''		)

         _lowerCamelCase						:    str         =			self.proj_in(__lowerCAmelCase		)

         _lowerCamelCase						:    Tuple         =			self.positional_embedding.to(hidden_states.dtype		)

         _lowerCamelCase						:    Tuple         =			[]
         _lowerCamelCase						:    Dict         =			0

         if encoder_hidden_states is not None:
                additional_embeds.append(__lowerCAmelCase		)
                additional_embeddings_len += encoder_hidden_states.shape[1]

         if len(proj_embeddings.shape		) == 2:
                _lowerCamelCase						:    str         =			proj_embeddings[:, None, :]

         if len(hidden_states.shape		) == 2:
                _lowerCamelCase						:    Dict         =			hidden_states[:, None, :]

         _lowerCamelCase						:    int         =			additional_embeds + [
             proj_embeddings,
             time_embeddings[:, None, :],
             hidden_states,
         ]

         if self.prd_embedding is not None:
                _lowerCamelCase						:    int         =			self.prd_embedding.to(hidden_states.dtype		).expand(__lowerCAmelCase						, -1						, -1		)
                additional_embeds.append(__lowerCAmelCase		)

         _lowerCamelCase						:    str         =			torch.cat(
             __lowerCAmelCase						, dim=1						, )

         # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
         _lowerCamelCase						:    Dict         =			additional_embeddings_len + proj_embeddings.shape[1] + 1
         if positional_embeddings.shape[1] < hidden_states.shape[1]:
                _lowerCamelCase						:    List[Any]         =			F.pad(
                    __lowerCAmelCase						, (
                        0,
                        0,
                        additional_embeddings_len,
                        self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
                    )						, value=0.0						, )

         _lowerCamelCase						:    Any         =			hidden_states + positional_embeddings

         if attention_mask is not None:
                _lowerCamelCase						:    Union[str, Any]         =			(1 - attention_mask.to(hidden_states.dtype		)) * -1_00_00.0
                _lowerCamelCase						:    Tuple         =			F.pad(__lowerCAmelCase						, (0, self.additional_embeddings)						, value=0.0		)
                _lowerCamelCase						:    Any         =			(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype		)
                _lowerCamelCase						:    int         =			attention_mask.repeat_interleave(self.config.num_attention_heads						, dim=0		)

         if self.norm_in is not None:
                _lowerCamelCase						:    List[str]         =			self.norm_in(__lowerCAmelCase		)

         for block in self.transformer_blocks:
                _lowerCamelCase						:    List[str]         =			block(__lowerCAmelCase						, attention_mask=__lowerCAmelCase		)

         _lowerCamelCase						:    Optional[Any]         =			self.norm_out(__lowerCAmelCase		)

         if self.prd_embedding is not None:
                _lowerCamelCase						:    Any         =			hidden_states[:, -1]
         else:
                _lowerCamelCase						:    Dict         =			hidden_states[:, additional_embeddings_len:]

         _lowerCamelCase						:    str         =			self.proj_to_clip_embeddings(__lowerCAmelCase		)

         if not return_dict:
                return (predicted_image_embedding,)

         return PriorTransformerOutput(predicted_image_embedding=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]						, __lowerCAmelCase					:						Tuple		):






         """simple docstring"""






         _lowerCamelCase						:    Dict         =			(prior_latents * self.clip_std) + self.clip_mean
         return prior_latents


