





'''simple docstring'''







from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


snake_case_  :							Optional[int]            =	logging.get_logger(__name__)


class        __a   (lowerCamelCase  ):
							__a       : List[str]          =							["input_values", "attention_mask"]

							def __init__(				self				:  List[str]							,  __magic_name__				:  str=1							,  __magic_name__				:  List[str]=1_60_00							,  __magic_name__				:  Any=1_28							,  __magic_name__				:  List[Any]=10_24							,  __magic_name__				:  Optional[Any]=0.0							,  __magic_name__				:  Optional[Any]=True							,  __magic_name__				:  Optional[int]=-4.2_6_7_7_3_9_3							,  __magic_name__				:  List[Any]=4.5_6_8_9_9_7_4							,  __magic_name__				:  Dict=False							,  **__magic_name__				:  Tuple							,  )       ->							Optional[Any]:







													"""simple docstring"""
													super().__init__(feature_size=__magic_name__							,  sampling_rate=__magic_name__							,  padding_value=__magic_name__							,  **__magic_name__ )
													UpperCAmelCase_      :   Tuple      =	num_mel_bins
													UpperCAmelCase_      :   Union[str, Any]      =	max_length
													UpperCAmelCase_      :   Dict      =	do_normalize
													UpperCAmelCase_      :   int      =	mean
													UpperCAmelCase_      :   int      =	std
													UpperCAmelCase_      :   int      =	return_attention_mask

							def      UpperCAmelCase__						(				self				:  Dict							,  __magic_name__				:  np.ndarray							,  __magic_name__				:  int							,  )       ->							np.ndarray:







													"""simple docstring"""
													UpperCAmelCase_      :   str      =	torch.from_numpy(__magic_name__ ).unsqueeze(0 )
													UpperCAmelCase_      :   Dict      =	ta_kaldi.fbank(
													    __magic_name__							,  htk_compat=__magic_name__							,  sample_frequency=self.sampling_rate							,  use_energy=__magic_name__							,  window_type='''hanning'''							,  num_mel_bins=self.num_mel_bins							,  dither=0.0							,  frame_shift=10							,  )

													UpperCAmelCase_      :   int      =	fbank.shape[0]
													UpperCAmelCase_      :   int      =	max_length - n_frames

													# pad or truncate, depending on difference
													if difference > 0:
																			UpperCAmelCase_      :   Any      =	torch.nn.ZeroPadad((0, 0, 0, difference) )
																			UpperCAmelCase_      :   Optional[Any]      =	pad_module(__magic_name__ )
													elif difference < 0:
																			UpperCAmelCase_      :   List[str]      =	fbank[0:max_length, :]

													UpperCAmelCase_      :   Dict      =	fbank.numpy()

													return fbank

							def      UpperCAmelCase__						(				self				:  Any							,  __magic_name__				:  np.ndarray )       ->							np.ndarray:







													"""simple docstring"""
													return (input_values - (self.mean)) / (self.std * 2)



							def __call__(				self				:  Dict							,  __magic_name__				:  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,  __magic_name__				:  Optional[int] = None							,  __magic_name__				:  Optional[Union[str, TensorType]] = None							,  **__magic_name__				:  Union[str, Any]							,  )       ->							BatchFeature:







													"""simple docstring"""

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

													UpperCAmelCase_      :   Tuple      =	isinstance(__magic_name__							,  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_      :   Dict      =	is_batched_numpy or (
													    isinstance(__magic_name__							,  (list, tuple) ) and (isinstance(raw_speech[0]							,  (np.ndarray, tuple, list) ))
													)

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

													# always return batch
													if not is_batched:
																			UpperCAmelCase_      :   Dict      =	[raw_speech]

													# extract fbank features and pad/truncate to max_length
													UpperCAmelCase_      :   int      =	[self._extract_fbank_features(__magic_name__							,  max_length=self.max_length ) for waveform in raw_speech]

													# convert into BatchFeature
													UpperCAmelCase_      :   Tuple      =	BatchFeature({'''input_values''': features} )

													# make sure list is in array format
													UpperCAmelCase_      :   int      =	padded_inputs.get('''input_values''' )
													if isinstance(input_values[0]							,  __magic_name__ ):
																			UpperCAmelCase_      :   Any      =	[np.asarray(__magic_name__							,  dtype=np.floataa ) for feature in input_values]

													# normalization
													if self.do_normalize:
																			UpperCAmelCase_      :   Tuple      =	[self.normalize(__magic_name__ ) for feature in input_values]

													if return_tensors is not None:
																			UpperCAmelCase_      :   Any      =	padded_inputs.convert_to_tensors(__magic_name__ )

													return padded_inputs












'''simple docstring'''






import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor







class       lowerCamelCase       (    unittest.TestCase	):


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	bool = False							):





											'''simple docstring'''





											_snake_case: Any	       =		10

											if only_cross_attention:
																	_snake_case: Optional[int]	       =		12
											else:
																	# when only cross attention is not set, the cross attention dim must be the same as the query dim
																	_snake_case: Optional[int]	       =		query_dim

											return {
											    "query_dim": query_dim,
											    "cross_attention_dim": cross_attention_dim,
											    "heads": 2,
											    "dim_head": 4,
											    "added_kv_proj_dim": 6,
											    "norm_num_groups": 1,
											    "only_cross_attention": only_cross_attention,
											    "processor": AttnAddedKVProcessor(),
											}


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	List[Any]							,  __snake_case						:	Tuple							):





											'''simple docstring'''





											_snake_case: Optional[int]	       =		2

											_snake_case: Union[str, Any]	       =		torch.rand(__snake_case							,  __snake_case							,  3							,  2							)
											_snake_case: Optional[int]	       =		torch.rand(__snake_case							,  4							,  __snake_case							)
											_snake_case: int	       =		None

											return {
											    "hidden_states": hidden_states,
											    "encoder_hidden_states": encoder_hidden_states,
											    "attention_mask": attention_mask,
											}




					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							):





											'''simple docstring'''





											torch.manual_seed(0							)

											_snake_case: int	       =		self.get_constructor_arguments(only_cross_attention=__snake_case							)
											_snake_case: Union[str, Any]	       =		Attention(**__snake_case							)

											self.assertTrue(attn.to_k is not None							)
											self.assertTrue(attn.to_v is not None							)

											_snake_case: Optional[int]	       =		self.get_forward_arguments(
											    query_dim=constructor_args['query_dim']							,  added_kv_proj_dim=constructor_args['added_kv_proj_dim']							)

											_snake_case: str	       =		attn(**__snake_case							)

											# only self attention

											torch.manual_seed(0							)

											_snake_case: str	       =		self.get_constructor_arguments(only_cross_attention=__snake_case							)
											_snake_case: Optional[Any]	       =		Attention(**__snake_case							)

											self.assertTrue(attn.to_k is None							)
											self.assertTrue(attn.to_v is None							)

											_snake_case: Dict	       =		self.get_forward_arguments(
											    query_dim=constructor_args['query_dim']							,  added_kv_proj_dim=constructor_args['added_kv_proj_dim']							)

											_snake_case: str	       =		attn(**__snake_case							)

											self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()							)







class       lowerCamelCase       (    unittest.TestCase	):


					def  SCREAMING_SNAKE_CASE_		(       self						:	Tuple							):





											'''simple docstring'''





											_snake_case: str	       =		DiffusionPipeline.from_pretrained('hf-internal-testing/tiny-stable-diffusion-pipe'							,  safety_checker=__snake_case							)

											_snake_case: int	       =		pipe(
											    'foo'							,  num_inference_steps=2							,  generator=torch.Generator('cpu'							).manual_seed(0							)							,  output_type='np'							,  ).images

											# the initial conversion succeeds
											_snake_case: Dict	       =		DiffusionPipeline.from_pretrained(
											    'hf-internal-testing/tiny-stable-diffusion-pipe'							,  device_map='sequential'							,  safety_checker=__snake_case							)

											_snake_case: Any	       =		pipe(
											    'foo'							,  num_inference_steps=2							,  generator=torch.Generator('cpu'							).manual_seed(0							)							,  output_type='np'							,  ).images

											with tempfile.TemporaryDirectory() as tmpdir:
																	# save the converted model
																	pipe.save_pretrained(__snake_case							)

																	# can also load the converted weights
																	_snake_case: Union[str, Any]	       =		DiffusionPipeline.from_pretrained(__snake_case							,  device_map='sequential'							,  safety_checker=__snake_case							)

											_snake_case: Union[str, Any]	       =		pipe(
											    'foo'							,  num_inference_steps=2							,  generator=torch.Generator('cpu'							).manual_seed(0							)							,  output_type='np'							,  ).images

											self.assertTrue(np.allclose(__snake_case							,  __snake_case							,  atol=1e-5							)							)
											self.assertTrue(np.allclose(__snake_case							,  __snake_case							,  atol=1e-5							)							)





