


import gc
import inspect
import unittest

import torch
from parameterized import parameterized

from diffusers import PriorTransformer
from diffusers.utils import floats_tensor, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin


enable_full_determinism()







class 	__lowercase    (__SCREAMING_SNAKE_CASE	,       unittest.TestCase  ):





							"""simple docstring"""







							_UpperCAmelCase      =						PriorTransformer
							_UpperCAmelCase      =						"""hidden_states"""






							@property
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     Tuple					=					4
										SCREAMING_SNAKE_CASE_		:     int					=					8
										SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					7

										SCREAMING_SNAKE_CASE_		:     List[str]					=					floats_tensor((batch_size, embedding_dim)       ).to(lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					floats_tensor((batch_size, embedding_dim)       ).to(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     str					=					floats_tensor((batch_size, num_embeddings, embedding_dim)       ).to(lowerCAmelCase__       )

										return {
										    "hidden_states": hidden_states,
										    "timestep": 2,
										    "proj_embedding": proj_embedding,
										    "encoder_hidden_states": encoder_hidden_states,
										}






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__=0       ):





										"""simple docstring"""




										torch.manual_seed(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					4
										SCREAMING_SNAKE_CASE_		:     Tuple					=					8
										SCREAMING_SNAKE_CASE_		:     List[str]					=					7

										SCREAMING_SNAKE_CASE_		:     str					=					torch.randn((batch_size, embedding_dim)       ).to(lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		:     Dict					=					torch.randn((batch_size, embedding_dim)       ).to(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					torch.randn((batch_size, num_embeddings, embedding_dim)       ).to(lowerCAmelCase__       )

										return {
										    "hidden_states": hidden_states,
										    "timestep": 2,
										    "proj_embedding": proj_embedding,
										    "encoder_hidden_states": encoder_hidden_states,
										}






							@property
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return (4, 8)






							@property
							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										return (4, 8)






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					{
										    'num_attention_heads': 2,
										    'attention_head_dim': 4,
										    'num_layers': 2,
										    'embedding_dim': 8,
										    'num_embeddings': 7,
										    'additional_embeddings': 4,
										}
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					self.dummy_input
										return init_dict, inputs_dict






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		,      SCREAMING_SNAKE_CASE_		:     Tuple					=					PriorTransformer.from_pretrained(
										    'hf-internal-testing/prior-dummy'			,						output_loading_info=lowerCAmelCase__       )
										self.assertIsNotNone(lowerCAmelCase__       )
										self.assertEqual(len(loading_info['missing_keys']       )			,						0       )

										model.to(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					model(**self.dummy_input       )[0]

										assert hidden_states is not None, "Make sure output is not None"






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		,      SCREAMING_SNAKE_CASE_		:     Tuple					=					self.prepare_init_args_and_inputs_for_common()

										SCREAMING_SNAKE_CASE_		:     List[Any]					=					self.model_class(**lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Dict					=					inspect.signature(model.forward       )
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										SCREAMING_SNAKE_CASE_		:     Any					=					[*signature.parameters.keys()]

										SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					['hidden_states', 'timestep']
										self.assertListEqual(arg_names[:2]			,						lowerCAmelCase__       )






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     str					=					PriorTransformer.from_pretrained('hf-internal-testing/prior-dummy'       )
										SCREAMING_SNAKE_CASE_		:     Tuple					=					model.to(lowerCAmelCase__       )

										if hasattr(lowerCAmelCase__			,						'set_default_attn_processor'       ):
													model.set_default_attn_processor()

										SCREAMING_SNAKE_CASE_		:     Union[str, Any]					=					self.get_dummy_seed_input()

										with torch.no_grad():
													SCREAMING_SNAKE_CASE_		:     List[str]					=					model(**lowerCAmelCase__       )[0]

										SCREAMING_SNAKE_CASE_		:     List[str]					=					output[0, :5].flatten().cpu()
										print(lowerCAmelCase__       )

										# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
										# the expected output slices are not the same for CPU and GPU.
										SCREAMING_SNAKE_CASE_		:     List[str]					=					torch.tensor([-1.3_436, -0.2_870, 0.7_538, 0.4_368, -0.0_239]       )
										self.assertTrue(torch_all_close(lowerCAmelCase__			,						lowerCAmelCase__			,						rtol=1E-2       )       )







@slow
class 	__lowercase    (unittest.TestCase  ):





							"""simple docstring"""







							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__=1			,						lowerCAmelCase__=7_6_8			,						lowerCAmelCase__=7_7			,						lowerCAmelCase__=0       ):





										"""simple docstring"""




										torch.manual_seed(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     List[str]					=					batch_size
										SCREAMING_SNAKE_CASE_		:     List[str]					=					embedding_dim
										SCREAMING_SNAKE_CASE_		:     Dict					=					num_embeddings

										SCREAMING_SNAKE_CASE_		:     Tuple					=					torch.randn((batch_size, embedding_dim)       ).to(lowerCAmelCase__       )

										SCREAMING_SNAKE_CASE_		:     Any					=					torch.randn((batch_size, embedding_dim)       ).to(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					torch.randn((batch_size, num_embeddings, embedding_dim)       ).to(lowerCAmelCase__       )

										return {
										    "hidden_states": hidden_states,
										    "timestep": 2,
										    "proj_embedding": proj_embedding,
										    "encoder_hidden_states": encoder_hidden_states,
										}






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




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






							@parameterized.expand(
							    [
							        # fmt: off
							        [1_3, [-0.5_861,  0.1_283, -0.0_931,  0.0_882,  0.4_476,  0.1_329, -0.0_498,  0.0_640]],
							        [3_7, [-0.4_913,  0.0_110, -0.0_483,  0.0_541,  0.4_954, -0.0_170,  0.0_354,  0.1_651]],
							        # fmt: on
							    ]       )
							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__			,						lowerCAmelCase__       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     int					=					PriorTransformer.from_pretrained('kandinsky-community/kandinsky-2-1-prior'			,						subfolder='prior'       )
										model.to(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     Optional[int]					=					self.get_dummy_seed_input(seed=lowerCAmelCase__       )

										with torch.no_grad():
													SCREAMING_SNAKE_CASE_		:     str					=					model(**lowerCAmelCase__       )[0]

										assert list(sample.shape       ) == [1, 7_6_8]

										SCREAMING_SNAKE_CASE_		:     Optional[Any]					=					sample[0, :8].flatten().cpu()
										print(lowerCAmelCase__       )
										SCREAMING_SNAKE_CASE_		:     List[Any]					=					torch.tensor(lowerCAmelCase__       )

										assert torch_all_close(lowerCAmelCase__			,						lowerCAmelCase__			,						atol=1E-3       )









'''simple docstring'''






from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


lowercase				=logging.get_logger(__name__)






class    __magic_name__				(						lowerCAmelCase							):
       UpperCAmelCase		 =["audio_values", "audio_mask"]






       def __init__(     self    ,				snake_case=2_0_4_8    ,				snake_case=1    ,				snake_case=[1_6, 1_6]    ,				snake_case=1_2_8    ,				snake_case=4_4_1_0_0    ,				snake_case=8_6    ,				snake_case=2_0_4_8    ,				snake_case=0.0    ,				**snake_case    ,				)		->    Optional[int]:




          '''simple docstring'''
          super().__init__(
              feature_size=snake_case    ,				sampling_rate=snake_case    ,				padding_value=snake_case    ,				**snake_case    ,				)

          _UpperCAmelCase  :  Dict				       =spectrogram_length
          _UpperCAmelCase  :  Any				       =num_channels
          _UpperCAmelCase  :  Optional[int]				       =patch_size
          _UpperCAmelCase  :  List[str]				       =feature_size // self.patch_size[1]
          _UpperCAmelCase  :  Optional[int]				       =n_fft
          _UpperCAmelCase  :  int				       =sampling_rate // hop_length_to_sampling_rate
          _UpperCAmelCase  :  int				       =sampling_rate
          _UpperCAmelCase  :  List[Any]				       =padding_value
          _UpperCAmelCase  :  Optional[int]				       =mel_filter_bank(
              num_frequency_bins=1 + n_fft // 2    ,				num_mel_filters=snake_case    ,				min_frequency=0.0    ,				max_frequency=2_20_50.0    ,				sampling_rate=snake_case    ,				norm='slaney'    ,				mel_scale='slaney'    ,				).T






       def      lowerCAmelCase						(     self    ,				snake_case)		->    np.ndarray:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       =spectrogram(
              snake_case    ,				window_function(self.n_fft    ,				'hann')    ,				frame_length=self.n_fft    ,				hop_length=self.hop_length    ,				power=2.0    ,				mel_filters=self.mel_filters.T    ,				log_mel='dB'    ,				db_range=80.0    ,				)
          _UpperCAmelCase  :  Any				       =log_spec[:, :-1]
          _UpperCAmelCase  :  List[str]				       =log_spec - 20.0
          _UpperCAmelCase  :  Union[str, Any]				       =np.clip(log_spec / 40.0    ,				-2.0    ,				0.0) + 1.0
          return log_spec






       def __call__(     self    ,				snake_case    ,				snake_case = None    ,				snake_case = True    ,				snake_case = None    ,				snake_case = False    ,				snake_case = False    ,				**snake_case    ,				)		->    BatchFeature:




          '''simple docstring'''

          if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                raise ValueError(
                    'This feature extractor is set to support sampling rate'
                    f" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"
                    f" with {self.sampling_rate} and not {sampling_rate}.")
          else:
             logger.warning(
                 'It is strongly recommended to pass the `sampling_rate` argument to this function. '
                 'Failing to do so can result in silent errors that might be hard to debug.')

          _UpperCAmelCase  :  Tuple				       =isinstance(snake_case    ,				np.ndarray) and len(raw_speech.shape) > 1
          if is_batched_numpy and len(raw_speech.shape) > 2:
             raise ValueError(f"Only mono-channel audio is supported for input to {self}")
          _UpperCAmelCase  :  str				       =is_batched_numpy or (
              isinstance(snake_case    ,				(list, tuple)) and (isinstance(raw_speech[0]    ,				(np.ndarray, tuple, list)))
          )
          if is_batched:
             _UpperCAmelCase  :  List[str]				       =[np.asarray([speech]    ,				dtype=np.floataa).T for speech in raw_speech]
          elif not is_batched and not isinstance(snake_case    ,				np.ndarray):
             _UpperCAmelCase  :  List[Any]				       =np.asarray(snake_case    ,				dtype=np.floataa)
          elif isinstance(snake_case    ,				np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
             _UpperCAmelCase  :  List[Any]				       =raw_speech.astype(np.floataa)
          # always return batch
          if not is_batched:
             _UpperCAmelCase  :  Dict				       =[np.asarray([raw_speech]).T]

          # Convert audio signals to log mel spectrograms, truncate by time axis
          _UpperCAmelCase  :  List[Any]				       =[
              self._np_extract_fbank_features(waveform.squeeze()).T[: self.spectrogram_length] for waveform in raw_speech
          ]
          if isinstance(audio_features[0]    ,				snake_case):
             _UpperCAmelCase  :  Dict				       =[np.asarray(snake_case    ,				dtype=np.floataa) for feature in audio_features]

          # Create audio attention mask
          _UpperCAmelCase  :  Tuple				       =max(
              [ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len for feature in audio_features])  # The maximum number of audio patches in a batch
          if return_attention_mask:
             _UpperCAmelCase  :  Any				       =[
                 (ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len) * [1]
                 + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len) * [0]
                 for feature in audio_features
             ]
             _UpperCAmelCase  :  int				       =np.array(snake_case).astype(np.floataa)

          # convert into correct format for padding
          _UpperCAmelCase  :  List[Any]				       =max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
          _UpperCAmelCase  :  List[str]				       =np.ones([len(snake_case), 1, max_time_len, self.feature_size]).astype(np.floataa)
          _UpperCAmelCase  :  Any				       =padded_audio_features * self.padding_value
          for i in range(len(snake_case)):
             _UpperCAmelCase  :  List[str]				       =audio_features[i]
             _UpperCAmelCase  :  Dict				       =feature

          # return as BatchFeature
          if return_attention_mask:
             _UpperCAmelCase  :  List[str]				       ={'audio_values': padded_audio_features, 'audio_mask': audio_mask}
          else:
             _UpperCAmelCase  :  Optional[int]				       ={'audio_values': padded_audio_features}

          _UpperCAmelCase  :  Any				       =BatchFeature(data=snake_case    ,				tensor_type=snake_case)
          return encoded_inputs

