






'''simple docstring'''







from typing import List, Optional, Union

import numpy as np

from ....audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ....feature_extraction_sequence_utils import SequenceFeatureExtractor
from ....feature_extraction_utils import BatchFeature
from ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


__lowerCAmelCase									=   logging.get_logger(__name__)





class 						SCREAMING_SNAKE_CASE       (       snake_case__							):
			snake_case__	=					['input_features', 'attention_mask']






			def __init__(			self   :      List[str]	,    __SCREAMING_SNAKE_CASE   :      List[str]=80	,    __SCREAMING_SNAKE_CASE   :      str=1_6000	,    __SCREAMING_SNAKE_CASE   :      Tuple=0.0	,    __SCREAMING_SNAKE_CASE   :      int=10	,    __SCREAMING_SNAKE_CASE   :      str=25	,    __SCREAMING_SNAKE_CASE   :      Any="hamming_window"	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]=3_2768.0	,    __SCREAMING_SNAKE_CASE   :      List[Any]=0.97	,    __SCREAMING_SNAKE_CASE   :      Optional[int]=1.0	,    __SCREAMING_SNAKE_CASE   :      str=True	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]=True	,    __SCREAMING_SNAKE_CASE   :      str=False	,    **__SCREAMING_SNAKE_CASE   :      Tuple	,    )      ->      Dict:
							super().__init__(feature_size=__SCREAMING_SNAKE_CASE	,    sampling_rate=__SCREAMING_SNAKE_CASE	,    padding_value=__SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       )

							a_		:	str									=   feature_size
							a_		:	List[Any]									=   sampling_rate
							a_		:	List[Any]									=   padding_value
							a_		:	Any									=   hop_length
							a_		:	Any									=   win_length
							a_		:	Tuple									=   frame_signal_scale
							a_		:	str									=   preemphasis_coeff
							a_		:	Optional[int]									=   mel_floor
							a_		:	Union[str, Any]									=   normalize_means
							a_		:	Dict									=   normalize_vars
							a_		:	Any									=   win_function
							a_		:	Optional[Any]									=   return_attention_mask

							a_		:	List[Any]									=   win_length * sampling_rate // 1000
							a_		:	str									=   hop_length * sampling_rate // 1000

							a_		:	Union[str, Any]									=   optimal_fft_length(self.sample_size       )
							a_		:	Optional[int]									=   (self.n_fft // 2) + 1






			def       SCREAMING_SNAKE_CASE   (			self   :      str	,    __SCREAMING_SNAKE_CASE   :      np.array       )      ->      Tuple:
							if self.win_function == "hamming_window":
											a_		:	Tuple									=   window_function(window_length=self.sample_size	,    name=self.win_function	,    periodic=__SCREAMING_SNAKE_CASE       )
							else:
											a_		:	Optional[int]									=   window_function(window_length=self.sample_size	,    name=self.win_function       )

							a_		:	Optional[Any]									=   mel_filter_bank(
							    num_frequency_bins=self.n_freqs	,    num_mel_filters=self.feature_size	,    min_frequency=0.0	,    max_frequency=self.sampling_rate / 2.0	,    sampling_rate=self.sampling_rate	,    )

							a_		:	Tuple									=   spectrogram(
							    one_waveform * self.frame_signal_scale	,    window=__SCREAMING_SNAKE_CASE	,    frame_length=self.sample_size	,    hop_length=self.sample_stride	,    fft_length=self.n_fft	,    center=__SCREAMING_SNAKE_CASE	,    preemphasis=self.preemphasis_coeff	,    mel_filters=__SCREAMING_SNAKE_CASE	,    mel_floor=self.mel_floor	,    log_mel='''log'''	,    )
							return msfc_features.T






			def       SCREAMING_SNAKE_CASE   (			self   :      int	,    __SCREAMING_SNAKE_CASE   :      str	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      int       )      ->      str:
							# make sure we normalize float32 arrays
							if self.normalize_means:
											a_		:	List[str]									=   x[:input_length].mean(axis=0       )
											a_		:	int									=   np.subtract(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       )
							if self.normalize_vars:
											a_		:	Optional[int]									=   x[:input_length].std(axis=0       )
											a_		:	Optional[int]									=   np.divide(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       )

							if input_length < x.shape[0]:
											a_		:	Tuple									=   padding_value

							# make sure array is in float32
							a_		:	int									=   x.astype(np.floataa       )

							return x






			def       SCREAMING_SNAKE_CASE   (			self   :      Union[str, Any]	,    __SCREAMING_SNAKE_CASE   :      List[np.ndarray]	,    __SCREAMING_SNAKE_CASE   :      Optional[np.ndarray] = None       )      ->      List[str]:
							a_		:	Union[str, Any]									=   attention_mask.sum(-1       ) if attention_mask is not None else [x.shape[0] for x in input_features]
							return [self._normalize_one(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    self.padding_value       ) for x, n in zip(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       )]







			def __call__(			self   :      Tuple	,    __SCREAMING_SNAKE_CASE   :      Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]	,    __SCREAMING_SNAKE_CASE   :      Union[bool, str, PaddingStrategy] = False	,    __SCREAMING_SNAKE_CASE   :      Optional[int] = None	,    __SCREAMING_SNAKE_CASE   :      bool = False	,    __SCREAMING_SNAKE_CASE   :      Optional[int] = None	,    __SCREAMING_SNAKE_CASE   :      Optional[bool] = None	,    __SCREAMING_SNAKE_CASE   :      Optional[Union[str, TensorType]] = None	,    __SCREAMING_SNAKE_CASE   :      Optional[int] = None	,    **__SCREAMING_SNAKE_CASE   :      int	,    )      ->      List[str]:

							if sampling_rate is not None:
											if sampling_rate != self.sampling_rate:
															raise ValueError(
															    f'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
															    f' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
															    f' {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.'''       )

							a_		:	Union[str, Any]									=   isinstance(__SCREAMING_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}'       )
							a_		:	int									=   is_batched_numpy or (
							    isinstance(__SCREAMING_SNAKE_CASE	,    (list, tuple)       ) and (isinstance(raw_speech[0]	,    (np.ndarray, tuple, list)       ))
							)

							if is_batched:
											a_		:	int									=   [np.asarray(__SCREAMING_SNAKE_CASE	,    dtype=np.floataa       ) for speech in raw_speech]
							elif not is_batched and not isinstance(__SCREAMING_SNAKE_CASE	,    np.ndarray       ):
											a_		:	int									=   np.asarray(__SCREAMING_SNAKE_CASE	,    dtype=np.floataa       )
							elif isinstance(__SCREAMING_SNAKE_CASE	,    np.ndarray       ) and raw_speech.dtype is np.dtype(np.floataa       ):
											a_		:	Any									=   raw_speech.astype(np.floataa       )

							# always return batch
							if not is_batched:
											a_		:	Union[str, Any]									=   [raw_speech]

							# extract fbank features
							a_		:	str									=   [self._extract_mfsc_features(__SCREAMING_SNAKE_CASE       ) for one_waveform in raw_speech]

							# convert into correct format for padding
							a_		:	Optional[Any]									=   BatchFeature({'''input_features''': features}       )

							a_		:	Dict									=   self.pad(
							    __SCREAMING_SNAKE_CASE	,    padding=__SCREAMING_SNAKE_CASE	,    max_length=__SCREAMING_SNAKE_CASE	,    truncation=__SCREAMING_SNAKE_CASE	,    pad_to_multiple_of=__SCREAMING_SNAKE_CASE	,    return_attention_mask=__SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE	,    )
							# make sure list is in array format
							a_		:	List[Any]									=   padded_inputs.get('''input_features'''       )
							if isinstance(input_features[0]	,    __SCREAMING_SNAKE_CASE       ):
											a_		:	Dict									=   [np.asarray(__SCREAMING_SNAKE_CASE	,    dtype=np.floataa       ) for feature in input_features]

							a_		:	Optional[Any]									=   padded_inputs.get('''attention_mask'''       )
							if attention_mask is not None:
											a_		:	Tuple									=   [np.asarray(__SCREAMING_SNAKE_CASE	,    dtype=np.intaa       ) for array in attention_mask]

							if self.normalize_means or self.normalize_vars:
											a_		:	List[Any]									=   (
											    np.array(__SCREAMING_SNAKE_CASE	,    dtype=np.intaa       )
											    if self._get_padding_strategies(__SCREAMING_SNAKE_CASE	,    max_length=__SCREAMING_SNAKE_CASE       ) is not PaddingStrategy.DO_NOT_PAD
											    and padding
											    else None
											)
											a_		:	str									=   self.normalize(
											    padded_inputs['''input_features''']	,    attention_mask=__SCREAMING_SNAKE_CASE       )

							if return_tensors is not None:
											a_		:	int									=   padded_inputs.convert_to_tensors(__SCREAMING_SNAKE_CASE       )

							return padded_inputs











import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


_lowercase		:  Any			       =False






@skip_mps
class 						UpperCamelCase_  (  snake_case__  ,     snake_case__  ,     snake_case__  ,     unittest.TestCase							):
					_a      :						Optional[int]						    =					StableDiffusionAttendAndExcitePipeline
					_a      :						Union[str, Any]						    =					False
					_a      :						Dict						    =					TEXT_TO_IMAGE_PARAMS
					_a      :						List[str]						    =					TEXT_TO_IMAGE_BATCH_PARAMS.union({'token_indices'}							)
					_a      :						Optional[int]						    =					TEXT_TO_IMAGE_IMAGE_PARAMS
					_a      :						List[Any]						    =					TEXT_TO_IMAGE_IMAGE_PARAMS



					@classmethod
					def   __a     (		cls     :     Tuple						):
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					@classmethod
					def   __a     (		cls     :     Tuple						):
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					def   __a     (		self     :     Dict						):
							torch.manual_seed(0						)
							lowerCamelCase_	:     int 						=      UNetaDConditionModel(
							    block_out_channels=(32, 64)       ,      layers_per_block=1       ,      sample_size=32       ,      in_channels=4       ,      out_channels=4       ,      down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')       ,      up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')       ,      cross_attention_dim=32       ,      attention_head_dim=(2, 4)       ,      use_linear_projection=lowerCamelCase       ,      )
							lowerCamelCase_	:     Dict 						=      DDIMScheduler(
							    beta_start=0.00_085       ,      beta_end=0.012       ,      beta_schedule='scaled_linear'       ,      clip_sample=lowerCamelCase       ,      set_alpha_to_one=lowerCamelCase       ,      )
							torch.manual_seed(0						)
							lowerCamelCase_	:     int 						=      AutoencoderKL(
							    block_out_channels=[32, 64]       ,      in_channels=3       ,      out_channels=3       ,      down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']       ,      up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']       ,      latent_channels=4       ,      sample_size=1_28       ,      )
							torch.manual_seed(0						)
							lowerCamelCase_	:     List[Any] 						=      CLIPTextConfig(
							    bos_token_id=0       ,      eos_token_id=2       ,      hidden_size=32       ,      intermediate_size=37       ,      layer_norm_eps=1E-05       ,      num_attention_heads=4       ,      num_hidden_layers=5       ,      pad_token_id=1       ,      vocab_size=10_00       ,      hidden_act='gelu'       ,      projection_dim=5_12       ,      )
							lowerCamelCase_	:     Any 						=      CLIPTextModel(lowerCamelCase						)
							lowerCamelCase_	:     int 						=      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'						)

							lowerCamelCase_	:     Optional[Any] 						=      {
							    'unet': unet,
							    'scheduler': scheduler,
							    'vae': vae,
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							    'safety_checker': None,
							    'feature_extractor': None,
							}

							return components



					def   __a     (		self     :     List[str]       ,      lowerCamelCase     :     Tuple       ,      lowerCamelCase     :     List[Any]=0						):
							if str(lowerCamelCase						).startswith('mps'						):
									lowerCamelCase_	:     Union[str, Any] 						=      torch.manual_seed(lowerCamelCase						)
							else:
									lowerCamelCase_	:     Optional[Any] 						=      torch.Generator(device=lowerCamelCase						).manual_seed(lowerCamelCase						)
							lowerCamelCase_	:     Dict 						=      {
							    'prompt': 'a cat and a frog',
							    'token_indices': [2, 5],
							    'generator': generator,
							    'num_inference_steps': 1,
							    'guidance_scale': 6.0,
							    'output_type': 'numpy',
							    'max_iter_to_alter': 2,
							    'thresholds': {0: 0.7},
							}
							return inputs



					def   __a     (		self     :     Union[str, Any]						):
							lowerCamelCase_	:     List[Any] 						=      'cpu'

							lowerCamelCase_	:     List[Any] 						=      self.get_dummy_components()
							lowerCamelCase_	:     List[str] 						=      self.pipeline_class(**lowerCamelCase						)
							pipe.to(lowerCamelCase						)
							pipe.set_progress_bar_config(disable=lowerCamelCase						)

							lowerCamelCase_	:     List[Any] 						=      self.get_dummy_inputs(lowerCamelCase						)
							lowerCamelCase_	:     List[str] 						=      pipe(**lowerCamelCase						).images
							lowerCamelCase_	:     Optional[Any] 						=      image[0, -3:, -3:, -1]

							self.assertEqual(image.shape       ,      (1, 64, 64, 3)						)
							lowerCamelCase_	:     Dict 						=      np.array(
							    [0.63_905_364, 0.62_897_307, 0.48_599_017, 0.5_133_624, 0.5_550_048, 0.45_769_516, 0.50_326_973, 0.5_023_139, 0.45_384_496]						)
							lowerCamelCase_	:     Any 						=      np.abs(image_slice.flatten() - expected_slice						).max()
							self.assertLessEqual(lowerCamelCase       ,      1E-3						)



					def   __a     (		self     :     Tuple						):
							super().test_cpu_offload_forward_pass(expected_max_diff=5E-4						)



					def   __a     (		self     :     Dict						):
							# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
							self._test_inference_batch_consistent(batch_sizes=[1, 2]						)



					def   __a     (		self     :     List[str]						):
							self._test_inference_batch_single_identical(batch_size=2       ,      expected_max_diff=7E-4						)



					def   __a     (		self     :     str						):
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3						)



					def   __a     (		self     :     Optional[Any]						):
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4						)



					def   __a     (		self     :     Tuple						):
							super().test_save_load_local(expected_max_difference=5E-4						)





					def   __a     (		self     :     List[Any]						):
							super().test_save_load_optional_components(expected_max_difference=4E-4						)






@require_torch_gpu
@slow
class 						UpperCamelCase_  (  unittest.TestCase							):



					@classmethod
					def   __a     (		cls     :     Union[str, Any]						):
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					@classmethod
					def   __a     (		cls     :     List[Any]						):
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					def   __a     (		self     :     List[str]						):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()





					def   __a     (		self     :     str						):
							lowerCamelCase_	:     Optional[Any] 						=      torch.manual_seed(51						)

							lowerCamelCase_	:     Any 						=      StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    'CompVis/stable-diffusion-v1-4'       ,      safety_checker=lowerCamelCase       ,      torch_dtype=torch.floataa						)
							pipe.to('cuda'						)

							lowerCamelCase_	:     List[Any] 						=      'a painting of an elephant with glasses'
							lowerCamelCase_	:     Tuple 						=      [5, 7]

							lowerCamelCase_	:     List[str] 						=      pipe(
							    prompt=lowerCamelCase       ,      token_indices=lowerCamelCase       ,      guidance_scale=7.5       ,      generator=lowerCamelCase       ,      num_inference_steps=5       ,      max_iter_to_alter=5       ,      output_type='numpy'       ,      ).images[0]

							lowerCamelCase_	:     Optional[Any] 						=      load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'						)
							assert np.abs((expected_image - image).max()						) < 5E-1
