





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 					A__   (       __snake_case			):
      _UpperCAmelCase	:torch.FloatTensor



class 					A__   (       __snake_case       ,   __snake_case			):





      @register_to_config
      def __init__( self       ,							A_ = 32       ,							A_ = 64       ,							A_ = 20       ,							A_ = 768       ,							A_=77       ,							A_=4       ,							A_ = 0.0       ,							A_ = "silu"       ,							A_ = None       ,							A_ = None       ,							A_ = "linear"       ,							A_ = "prd"       ,							A_ = None       ,							A_ = None       ,							A_ = None       ,							):
       '''simple docstring'''




       super().__init__()
       UpperCamelCase	:					Dict       	=   num_attention_heads
       UpperCamelCase	:					Union[str, Any]       	=   attention_head_dim
       UpperCamelCase	:					Dict       	=   num_attention_heads * attention_head_dim
       UpperCamelCase	:					Any       	=   additional_embeddings

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

       UpperCamelCase	:					Optional[int]       	=   Timesteps(A_       ,							A_       ,							0   )
       UpperCamelCase	:					Optional[int]       	=   TimestepEmbedding(A_       ,							A_       ,							out_dim=A_       ,							act_fn=A_   )

       UpperCamelCase	:					List[Any]       	=   nn.Linear(A_       ,							A_   )

       if embedding_proj_norm_type is None:
        UpperCamelCase	:					Tuple       	=   None
       elif embedding_proj_norm_type == "layer":
        UpperCamelCase	:					int       	=   nn.LayerNorm(A_   )
       else:
        raise ValueError(F"""unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"""   )

       UpperCamelCase	:					str       	=   nn.Linear(A_       ,							A_   )

       if encoder_hid_proj_type is None:
        UpperCamelCase	:					Dict       	=   None
       elif encoder_hid_proj_type == "linear":
        UpperCamelCase	:					List[str]       	=   nn.Linear(A_       ,							A_   )
       else:
        raise ValueError(F"""unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"""   )

       UpperCamelCase	:					Tuple       	=   nn.Parameter(torch.zeros(1       ,							num_embeddings + additional_embeddings       ,							A_   )   )

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

       UpperCamelCase	:					List[Any]       	=   nn.ModuleList(
           [
               BasicTransformerBlock(
                   A_       ,							A_       ,							A_       ,							dropout=A_       ,							activation_fn="gelu"       ,							attention_bias=A_       ,							)
               for d in range(A_   )
           ]   )

       if norm_in_type == "layer":
        UpperCamelCase	:					List[Any]       	=   nn.LayerNorm(A_   )
       elif norm_in_type is None:
        UpperCamelCase	:					List[str]       	=   None
       else:
        raise ValueError(F"""Unsupported norm_in_type: {norm_in_type}."""   )

       UpperCamelCase	:					int       	=   nn.LayerNorm(A_   )

       UpperCamelCase	:					Optional[int]       	=   nn.Linear(A_       ,							A_   )

       UpperCamelCase	:					List[str]       	=   torch.full(
           [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]       ,							-1_00_00.0   )
       causal_attention_mask.triu_(1   )
       UpperCamelCase	:					Dict       	=   causal_attention_mask[None, ...]
       self.register_buffer("causal_attention_mask"       ,							A_       ,							persistent=A_   )

       UpperCamelCase	:					List[str]       	=   nn.Parameter(torch.zeros(1       ,							A_   )   )
       UpperCamelCase	:					Tuple       	=   nn.Parameter(torch.zeros(1       ,							A_   )   )





      @property
      # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   {}

       def fn_recursive_add_processors(A_       ,							A_       ,							A_   ):
        if hasattr(A_       ,							"set_processor"   ):
         UpperCamelCase	:					List[Any]       	=   module.processor

        for sub_name, child in module.named_children():
         fn_recursive_add_processors(F"""{name}.{sub_name}"""       ,							A_       ,							A_   )

        return processors

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

       return processors





      def        __UpperCamelCase( self       ,							A_   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   len(self.attn_processors.keys()   )

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

       def fn_recursive_attn_processor(A_       ,							A_       ,							A_   ):
        if hasattr(A_       ,							"set_processor"   ):
         if not isinstance(A_       ,							A_   ):
          module.set_processor(A_   )
         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}"""       ,							A_       ,							A_   )

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       self.set_attn_processor(AttnProcessor()   )





      def        __UpperCamelCase( self       ,							A_       ,							A_       ,							A_       ,							A_ = None       ,							A_ = None       ,							A_ = True       ,							):
       '''simple docstring'''




       UpperCamelCase	:					Any       	=   hidden_states.shape[0]

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

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       UpperCamelCase	:					Union[str, Any]       	=   timesteps * torch.ones(A_       ,							dtype=timesteps.dtype       ,							device=timesteps.device   )

       UpperCamelCase	:					Any       	=   self.time_proj(A_   )

       # timesteps does not contain any weights and will always return f32 tensors
       # but time_embedding might be fp16, so we need to cast here.
       UpperCamelCase	:					int       	=   timesteps_projected.to(dtype=self.dtype   )
       UpperCamelCase	:					Optional[int]       	=   self.time_embedding(A_   )

       if self.embedding_proj_norm is not None:
        UpperCamelCase	:					Any       	=   self.embedding_proj_norm(A_   )

       UpperCamelCase	:					Any       	=   self.embedding_proj(A_   )
       if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
        UpperCamelCase	:					Any       	=   self.encoder_hidden_states_proj(A_   )
       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"   )

       UpperCamelCase	:					str       	=   self.proj_in(A_   )

       UpperCamelCase	:					Optional[int]       	=   self.positional_embedding.to(hidden_states.dtype   )

       UpperCamelCase	:					Tuple       	=   []
       UpperCamelCase	:					Dict       	=   0

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

       if len(proj_embeddings.shape   ) == 2:
        UpperCamelCase	:					Tuple       	=   proj_embeddings[:, None, :]

       if len(hidden_states.shape   ) == 2:
        UpperCamelCase	:					Tuple       	=   hidden_states[:, None, :]

       UpperCamelCase	:					str       	=   additional_embeds + [
           proj_embeddings,
           time_embeddings[:, None, :],
           hidden_states,
       ]

       if self.prd_embedding is not None:
        UpperCamelCase	:					str       	=   self.prd_embedding.to(hidden_states.dtype   ).expand(A_       ,							-1       ,							-1   )
        additional_embeds.append(A_   )

       UpperCamelCase	:					Any       	=   torch.cat(
           A_       ,							dim=1       ,							)

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

       UpperCamelCase	:					List[Any]       	=   hidden_states + positional_embeddings

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

       if self.norm_in is not None:
        UpperCamelCase	:					Dict       	=   self.norm_in(A_   )

       for block in self.transformer_blocks:
        UpperCamelCase	:					Dict       	=   block(A_       ,							attention_mask=A_   )

       UpperCamelCase	:					Dict       	=   self.norm_out(A_   )

       if self.prd_embedding is not None:
        UpperCamelCase	:					str       	=   hidden_states[:, -1]
       else:
        UpperCamelCase	:					Union[str, Any]       	=   hidden_states[:, additional_embeddings_len:]

       UpperCamelCase	:					Dict       	=   self.proj_to_clip_embeddings(A_   )

       if not return_dict:
        return (predicted_image_embedding,)

       return PriorTransformerOutput(predicted_image_embedding=A_   )





      def        __UpperCamelCase( self       ,							A_   ):
       '''simple docstring'''




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







import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


__lowerCamelCase       :     str   =     random.Random()

if is_torch_available():
    import torch



def 		A_				(     _lowerCAmelCase							,   _lowerCAmelCase=1.0							,   _lowerCAmelCase=None							,   _lowerCAmelCase=None     )						-> Optional[Any]:
 if rng is None:
  UpperCamelCase	:					Optional[int]       	=   global_rng

 UpperCamelCase	:					Optional[Any]       	=   []
 for batch_idx in range(shape[0]     ):
  values.append([]     )
  for _ in range(shape[1]     ):
   values[-1].append(rng.random() * scale     )

 return values



class 					A__   (       unittest.TestCase			):





      def __init__( self       ,							A_       ,							A_=7       ,							A_=400       ,							A_=2000       ,							A_=1       ,							A_=0.0       ,							A_=1_6000       ,							A_=True       ,							A_=True       ,							):
       '''simple docstring'''




       UpperCamelCase	:					Tuple       	=   parent
       UpperCamelCase	:					List[Any]       	=   batch_size
       UpperCamelCase	:					List[Any]       	=   min_seq_length
       UpperCamelCase	:					List[str]       	=   max_seq_length
       UpperCamelCase	:					int       	=   (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
       UpperCamelCase	:					Union[str, Any]       	=   feature_size
       UpperCamelCase	:					List[str]       	=   padding_value
       UpperCamelCase	:					Optional[Any]       	=   sampling_rate
       UpperCamelCase	:					List[str]       	=   return_attention_mask
       UpperCamelCase	:					List[Any]       	=   do_normalize





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return {
           "feature_size": self.feature_size,
           "padding_value": self.padding_value,
           "sampling_rate": self.sampling_rate,
           "return_attention_mask": self.return_attention_mask,
           "do_normalize": self.do_normalize,
       }





      def        __UpperCamelCase( self       ,							A_=False       ,							A_=False   ):
       '''simple docstring'''




       def _flatten(A_   ):
        return list(itertools.chain(*A_   )   )

       if equal_length:
        UpperCamelCase	:					List[str]       	=   floats_list((self.batch_size, self.max_seq_length)   )
       else:
        # make sure that inputs increase in size
        UpperCamelCase	:					Dict       	=   [
            _flatten(floats_list((x, self.feature_size)   )   )
            for x in range(self.min_seq_length       ,							self.max_seq_length       ,							self.seq_length_diff   )
        ]

       if numpify:
        UpperCamelCase	:					Union[str, Any]       	=   [np.asarray(A_   ) for x in speech_inputs]

       return speech_inputs



@require_torch
@require_torchaudio
class 					A__   (       __snake_case       ,   unittest.TestCase			):
      _UpperCAmelCase	:Optional[Any]						    =   ASTFeatureExtractor





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Tuple       	=   ASTFeatureExtractionTester(self   )





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Optional[Any]       	=   self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
       # create three inputs of length 800, 1000, and 1200
       UpperCamelCase	:					Tuple       	=   [floats_list((1, x)   )[0] for x in range(800       ,							1400       ,							200   )]
       UpperCamelCase	:					Dict       	=   [np.asarray(A_   ) for speech_input in speech_inputs]

       # Test not batched input
       UpperCamelCase	:					Dict       	=   feat_extract(speech_inputs[0]       ,							return_tensors="np"   ).input_values
       UpperCamelCase	:					Union[str, Any]       	=   feat_extract(np_speech_inputs[0]       ,							return_tensors="np"   ).input_values
       self.assertTrue(np.allclose(A_       ,							A_       ,							atol=1e-3   )   )

       # Test batched
       UpperCamelCase	:					Any       	=   feat_extract(A_       ,							padding=A_       ,							return_tensors="np"   ).input_values
       UpperCamelCase	:					Any       	=   feat_extract(A_       ,							padding=A_       ,							return_tensors="np"   ).input_values
       for enc_seq_a, enc_seq_a in zip(A_       ,							A_   ):
        self.assertTrue(np.allclose(A_       ,							A_       ,							atol=1e-3   )   )

       # Test 2-D numpy arrays are batched.
       UpperCamelCase	:					Dict       	=   [floats_list((1, x)   )[0] for x in (800, 800, 800)]
       UpperCamelCase	:					int       	=   np.asarray(A_   )
       UpperCamelCase	:					Any       	=   feat_extract(A_       ,							return_tensors="np"   ).input_values
       UpperCamelCase	:					List[str]       	=   feat_extract(A_       ,							return_tensors="np"   ).input_values
       for enc_seq_a, enc_seq_a in zip(A_       ,							A_   ):
        self.assertTrue(np.allclose(A_       ,							A_       ,							atol=1e-3   )   )





      @require_torch
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       import torch

       UpperCamelCase	:					List[Any]       	=   self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
       UpperCamelCase	:					int       	=   np.random.rand(100   ).astype(np.floataa   )
       UpperCamelCase	:					str       	=   np_speech_inputs.tolist()

       for inputs in [py_speech_inputs, np_speech_inputs]:
        UpperCamelCase	:					List[Any]       	=   feature_extractor.pad([{"input_values": inputs}]       ,							return_tensors="np"   )
        self.assertTrue(np_processed.input_values.dtype == np.floataa   )
        UpperCamelCase	:					List[str]       	=   feature_extractor.pad([{"input_values": inputs}]       ,							return_tensors="pt"   )
        self.assertTrue(pt_processed.input_values.dtype == torch.floataa   )





      def        __UpperCamelCase( self       ,							A_   ):
       '''simple docstring'''




       from datasets import load_dataset

       UpperCamelCase	:					Dict       	=   load_dataset("hf-internal-testing/librispeech_asr_dummy"       ,							"clean"       ,							split="validation"   )
       # automatic decoding with librispeech
       UpperCamelCase	:					Any       	=   ds.sort("id"   ).select(range(A_   )   )[:num_samples]["audio"]

       return [x["array"] for x in speech_samples]





      @require_torch
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Any       	=   torch.tensor(
           [-0.98_94, -1.27_76, -0.90_66, -1.27_76, -0.93_49, -1.26_09, -1.03_86, -1.27_76,
            -1.15_61, -1.27_76, -1.20_52, -1.27_23, -1.21_90, -1.21_32, -1.27_76, -1.11_33,
            -1.19_53, -1.13_43, -1.15_84, -1.22_03, -1.17_70, -1.24_74, -1.23_81, -1.19_36,
            -0.92_70, -0.83_17, -0.80_49, -0.77_06, -0.75_65, -0.78_69]   )
       # fmt: on

       UpperCamelCase	:					List[Any]       	=   self._load_datasamples(1   )
       UpperCamelCase	:					Tuple       	=   ASTFeatureExtractor()
       UpperCamelCase	:					str       	=   feature_extractor(A_       ,							return_tensors="pt"   ).input_values
       self.assertEquals(input_values.shape       ,							(1, 1024, 128)   )
       self.assertTrue(torch.allclose(input_values[0, 0, :30]       ,							A_       ,							atol=1e-4   )   )
