

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


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






class 			__magic_name__					(    lowerCAmelCase_   ):
			SCREAMING_SNAKE_CASE							     =		['input_values', 'attention_mask']





			def __init__(	self      ,						__snake_case=1      ,						__snake_case=1_6000      ,						__snake_case=128      ,						__snake_case=1024      ,						__snake_case=0.0      ,						__snake_case=True      ,						__snake_case=-4.267_7393      ,						__snake_case=4.568_9974      ,						__snake_case=False      ,						**__snake_case      ,						)     ->				Any:




						'''simple docstring'''
						super().__init__(feature_size=__a      ,						sampling_rate=__a      ,						padding_value=__a      ,						**__a      )
						__a						   =num_mel_bins
						__a						   =max_length
						__a						   =do_normalize
						__a						   =mean
						__a						   =std
						__a						   =return_attention_mask





			def 	__magic_name__					(	self      ,						__snake_case      ,						__snake_case      ,						)     ->				Union[str, Any]:




						'''simple docstring'''
						__a						   =torch.from_numpy(__a      ).unsqueeze(0      )
						__a						   =ta_kaldi.fbank(
						    __a      ,						htk_compat=__a      ,						sample_frequency=self.sampling_rate      ,						use_energy=__a      ,						window_type='hanning'      ,						num_mel_bins=self.num_mel_bins      ,						dither=0.0      ,						frame_shift=10      ,						)

						__a						   =fbank.shape[0]
						__a						   =max_length - n_frames

						# pad or truncate, depending on difference
						if difference > 0:
									__a						   =torch.nn.ZeroPadad((0, 0, 0, difference)      )
									__a						   =pad_module(__a      )
						elif difference < 0:
									__a						   =fbank[0:max_length, :]

						__a						   =fbank.numpy()

						return fbank





			def 	__magic_name__					(	self      ,						__snake_case      )     ->				int:




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





			def __call__(	self      ,						__snake_case      ,						__snake_case = None      ,						__snake_case = None      ,						**__snake_case      ,						)     ->				List[Any]:




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

						__a						   =isinstance(__a      ,						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(__a      ,						(list, tuple)      ) and (isinstance(raw_speech[0]      ,						(np.ndarray, tuple, list)      ))
						)

						if is_batched:
									__a						   =[np.asarray(__a      ,						dtype=np.floataa      ) for speech in raw_speech]
						elif not is_batched and not isinstance(__a      ,						np.ndarray      ):
									__a						   =np.asarray(__a      ,						dtype=np.floataa      )
						elif isinstance(__a      ,						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						   =[raw_speech]

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

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

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

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

						if return_tensors is not None:
									__a						   =padded_inputs.convert_to_tensors(__a      )

						return padded_inputs









import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_lowerCAmelCase      :    List[str]						       =    False






@nightly
@require_torch_gpu
class 			__magic_name__					(    unittest.TestCase   ):





			def 	__magic_name__					(	self      )     ->				Optional[Any]:




						'''simple docstring'''
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()





			def 	__magic_name__					(	self      )     ->				Tuple:




						'''simple docstring'''
						__a						   =VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'      )
						# remove text_unet
						pipe.remove_unused_weights()
						pipe.to(__snake_case      )
						pipe.set_progress_bar_config(disable=__snake_case      )

						__a						   =load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'      )

						__a						   =torch.manual_seed(0      )
						__a						   =pipe(
						    prompt='first prompt'      ,						image=__snake_case      ,						text_to_image_strength=0.75      ,						generator=__snake_case      ,						guidance_scale=7.5      ,						num_inference_steps=2      ,						output_type='numpy'      ,						).images

						with tempfile.TemporaryDirectory() as tmpdirname:
									pipe.save_pretrained(__snake_case      )
									__a						   =VersatileDiffusionDualGuidedPipeline.from_pretrained(__snake_case      )

						pipe.to(__snake_case      )
						pipe.set_progress_bar_config(disable=__snake_case      )

						__a						   =generator.manual_seed(0      )
						__a						   =pipe(
						    prompt='first prompt'      ,						image=__snake_case      ,						text_to_image_strength=0.75      ,						generator=__snake_case      ,						guidance_scale=7.5      ,						num_inference_steps=2      ,						output_type='numpy'      ,						).images

						assert np.abs(image - new_image      ).sum() < 1e-5, "Models don't have the same forward pass"





			def 	__magic_name__					(	self      )     ->				Optional[Any]:




						'''simple docstring'''
						__a						   =VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'      )
						pipe.remove_unused_weights()
						pipe.to(__snake_case      )
						pipe.set_progress_bar_config(disable=__snake_case      )

						__a						   ='cyberpunk 2077'
						__a						   =load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'      )
						__a						   =torch.manual_seed(0      )
						__a						   =pipe(
						    prompt=__snake_case      ,						image=__snake_case      ,						text_to_image_strength=0.75      ,						generator=__snake_case      ,						guidance_scale=7.5      ,						num_inference_steps=50      ,						output_type='numpy'      ,						).images

						__a						   =image[0, 253:256, 253:256, -1]

						assert image.shape == (1, 512, 512, 3)
						__a						   =np.array([0.0787, 0.0849, 0.0826, 0.0812, 0.0807, 0.0795, 0.0818, 0.0798, 0.0779]      )

						assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2






