






'''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


lowerCamelCase_												=					logging.get_logger(__name__)

class    _UpperCAmelCase			(     snake_case_						):




   """simple docstring"""

   snake_case    				=						['''audio_values''', '''audio_mask''']

   def __init__(		self :	Union[str, Any]      ,      __UpperCAmelCase :	Optional[Any]=2048      ,      __UpperCAmelCase :	Any=1      ,      __UpperCAmelCase :	Optional[Any]=[16, 16]      ,      __UpperCAmelCase :	int=128      ,      __UpperCAmelCase :	Union[str, Any]=44100      ,      __UpperCAmelCase :	Optional[int]=86      ,      __UpperCAmelCase :	List[Any]=2048      ,      __UpperCAmelCase :	Dict=0.0      ,      **__UpperCAmelCase :	List[str]      ,      ):





         '''simple docstring'''





         super().__init__(
             feature_size=__UpperCAmelCase      ,      sampling_rate=__UpperCAmelCase      ,      padding_value=__UpperCAmelCase      ,      **__UpperCAmelCase      ,      )

         _A			     =	spectrogram_length
         _A			     =	num_channels
         _A			     =	patch_size
         _A			     =	feature_size // self.patch_size[1]
         _A			     =	n_fft
         _A			     =	sampling_rate // hop_length_to_sampling_rate
         _A			     =	sampling_rate
         _A			     =	padding_value
         _A			     =	mel_filter_bank(
             num_frequency_bins=1 + n_fft // 2      ,      num_mel_filters=__UpperCAmelCase      ,      min_frequency=0.0      ,      max_frequency=22050.0      ,      sampling_rate=__UpperCAmelCase      ,      norm="slaney"      ,      mel_scale="slaney"      ,      ).T

   def 						lowerCAmelCase (		self :	Optional[int]      ,      __UpperCAmelCase :	np.array		):





         '''simple docstring'''





         _A			     =	spectrogram(
             __UpperCAmelCase      ,      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      ,      )
         _A			     =	log_spec[:, :-1]
         _A			     =	log_spec - 20.0
         _A			     =	np.clip(log_spec / 40.0      ,      -2.0      ,      0.0		) + 1.0
         return log_spec

   def __call__(		self :	List[Any]      ,      __UpperCAmelCase :	Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,      __UpperCAmelCase :	Optional[Union[str, TensorType]] = None      ,      __UpperCAmelCase :	Optional[bool] = True      ,      __UpperCAmelCase :	Optional[int] = None      ,      __UpperCAmelCase :	bool = False      ,      __UpperCAmelCase :	bool = False      ,      **__UpperCAmelCase :	Dict      ,      ):





         '''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."		)

         _A			     =	isinstance(__UpperCAmelCase      ,      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			     =	is_batched_numpy or (
             isinstance(__UpperCAmelCase      ,      (list, tuple)		) and (isinstance(raw_speech[0]      ,      (np.ndarray, tuple, list)		))
         )
         if is_batched:
               _A			     =	[np.asarray([speech]      ,      dtype=np.floataa		).T for speech in raw_speech]
         elif not is_batched and not isinstance(__UpperCAmelCase      ,      np.ndarray		):
               _A			     =	np.asarray(__UpperCAmelCase      ,      dtype=np.floataa		)
         elif isinstance(__UpperCAmelCase      ,      np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
               _A			     =	raw_speech.astype(np.floataa		)
         # always return batch
         if not is_batched:
               _A			     =	[np.asarray([raw_speech]		).T]

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

         # Create audio attention mask
         _A			     =	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:
               _A			     =	[
                   (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
               ]
               _A			     =	np.array(__UpperCAmelCase		).astype(np.floataa		)

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

         # return as BatchFeature
         if return_attention_mask:
               _A			     =	{"audio_values": padded_audio_features, "audio_mask": audio_mask}
         else:
               _A			     =	{"audio_values": padded_audio_features}

         _A			     =	BatchFeature(data=__UpperCAmelCase      ,      tensor_type=__UpperCAmelCase		)
         return encoded_inputs













import random
import unittest

import torch

from diffusers import IFImgaImgSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin



@skip_mps
class   lowerCamelCase      (_snake_case ,  _snake_case ,  unittest.TestCase			):







		'''simple docstring'''



		_snake_case    :    Union[str, Any]       						=  IFImgaImgSuperResolutionPipeline
		_snake_case    :    Union[str, Any]       						=  TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''width''', '''height'''}
		_snake_case    :    List[Any]       						=  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''original_image'''}			)
		_snake_case    :    List[str]       						=  PipelineTesterMixin.required_optional_params - {'''latents'''}



		def 					__UpperCAmelCase			(							self   )	->							Optional[Any]:
							return self._get_superresolution_dummy_components()



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase=0   )	->							Any:
							if str(_UpperCamelCase   ).startswith('mps'   ):
												UpperCAmelCase_    :      List[Any]            =						torch.manual_seed(_UpperCamelCase   )
							else:
												UpperCAmelCase_    :      int            =						torch.Generator(device=_UpperCamelCase   ).manual_seed(_UpperCamelCase   )

							UpperCAmelCase_    :      List[Any]            =						floats_tensor((1, 3, 3_2, 3_2)				,		rng=random.Random(_UpperCamelCase   )   ).to(_UpperCamelCase   )
							UpperCAmelCase_    :      Dict            =						floats_tensor((1, 3, 1_6, 1_6)				,		rng=random.Random(_UpperCamelCase   )   ).to(_UpperCamelCase   )

							UpperCAmelCase_    :      Tuple            =						{
							    'prompt': 'A painting of a squirrel eating a burger',
							    'image': image,
							    'original_image': original_image,
							    'generator': generator,
							    'num_inference_steps': 2,
							    'output_type': 'numpy',
							}

							return inputs



		@unittest.skipIf(
		    torch_device != 'cuda' or not is_xformers_available()				,		reason='XFormers attention is only available with CUDA and `xformers` installed'				,		)
		def 					__UpperCAmelCase			(							self   )	->							Any:
							self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3   )



		def 					__UpperCAmelCase			(							self   )	->							Dict:
							self._test_save_load_optional_components()



		@unittest.skipIf(torch_device != 'cuda'				,		reason='float16 requires CUDA'   )
		def 					__UpperCAmelCase			(							self   )	->							str:
							# Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
							super().test_save_load_floataa(expected_max_diff=1E-1   )



		def 					__UpperCAmelCase			(							self   )	->							List[Any]:
							self._test_attention_slicing_forward_pass(expected_max_diff=1E-2   )



		def 					__UpperCAmelCase			(							self   )	->							Union[str, Any]:
							self._test_save_load_local()



		def 					__UpperCAmelCase			(							self   )	->							Dict:
							self._test_inference_batch_single_identical(
							    expected_max_diff=1E-2				,		)






