




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


a		:	Optional[Any]       =      logging.get_logger(__name__)




class      UpperCamelCase       (	snake_case__       ):
  a     :List[str]          =					['input_values', 'attention_mask']



  def __init__(			self		,     __lowerCAmelCase=1		,     __lowerCAmelCase=1_6000		,     __lowerCAmelCase=128		,     __lowerCAmelCase=1024		,     __lowerCAmelCase=0.0		,     __lowerCAmelCase=True		,     __lowerCAmelCase=-4.2677393		,     __lowerCAmelCase=4.5689974		,     __lowerCAmelCase=False		,     **__lowerCAmelCase		,     ):







      '''simple docstring'''


      super().__init__(feature_size=__lowerCAmelCase		,     sampling_rate=__lowerCAmelCase		,     padding_value=__lowerCAmelCase		,     **__lowerCAmelCase )
      _a				:							Union[str, Any]							       =      num_mel_bins
      _a				:							Dict							       =      max_length
      _a				:							Union[str, Any]							       =      do_normalize
      _a				:							Optional[int]							       =      mean
      _a				:							Dict							       =      std
      _a				:							int							       =      return_attention_mask



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase		,     ):







      '''simple docstring'''


      _a				:							Optional[int]							       =      torch.from_numpy(__lowerCAmelCase ).unsqueeze(0 )
      _a				:							Union[str, Any]							       =      ta_kaldi.fbank(
          __lowerCAmelCase		,     htk_compat=__lowerCAmelCase		,     sample_frequency=self.sampling_rate		,     use_energy=__lowerCAmelCase		,     window_type="""hanning"""		,     num_mel_bins=self.num_mel_bins		,     dither=0.0		,     frame_shift=10		,     )

      _a				:							Tuple							       =      fbank.shape[0]
      _a				:							Optional[int]							       =      max_length - n_frames

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

      _a				:							List[Any]							       =      fbank.numpy()

      return fbank



  def     a_							(			self		,     __lowerCAmelCase ):







      '''simple docstring'''


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



  def __call__(			self		,     __lowerCAmelCase		,     __lowerCAmelCase = None		,     __lowerCAmelCase = None		,     **__lowerCAmelCase		,     ):







      '''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				:							Dict							       =      isinstance(__lowerCAmelCase		,     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				:							str							       =      is_batched_numpy or (
          isinstance(__lowerCAmelCase		,     (list, tuple) ) and (isinstance(raw_speech[0]		,     (np.ndarray, tuple, list) ))
      )

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

      # extract fbank features and pad/truncate to max_length
      _a				:							List[str]							       =      [self._extract_fbank_features(__lowerCAmelCase		,     max_length=self.max_length ) for waveform in raw_speech]

      # convert into BatchFeature
      _a				:							List[str]							       =      BatchFeature({"""input_values""": features} )

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

      # normalization
      if self.do_normalize:
          _a				:							Union[str, Any]							       =      [self.normalize(__lowerCAmelCase ) for feature in input_values]

      if return_tensors is not None:
          _a				:							Optional[int]							       =      padded_inputs.convert_to_tensors(__lowerCAmelCase )

      return padded_inputs



"""simple docstring"""







import random
import unittest

import torch

from diffusers import IFImgaImgPipeline
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 							__snake_case(					A_				,	A_				,	unittest.TestCase	):


    '''simple docstring'''





    _UpperCAmelCase       =			IFImgaImgPipeline
    _UpperCAmelCase       =			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"width", "height"}
    _UpperCAmelCase       =			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    _UpperCAmelCase       =			PipelineTesterMixin.required_optional_params - {"latents"}
    def 						_a							(  self	):




         '''simple docstring'''



         return self._get_dummy_components()
    def 						_a							(  self					,       __lowerCamelCase					,       __lowerCamelCase=0	):




         '''simple docstring'''



         if str(__lowerCamelCase	).startswith('mps'	):
              __A					:					Tuple								=		torch.manual_seed(__lowerCamelCase	)
         else:
              __A					:					int								=		torch.Generator(device=__lowerCamelCase	).manual_seed(__lowerCamelCase	)

         __A					:					str								=		floats_tensor((1, 3, 32, 32)					,       rng=random.Random(__lowerCamelCase	)	).to(__lowerCamelCase	)

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

         return inputs
    def 						_a							(  self	):




         '''simple docstring'''



         self._test_save_load_optional_components()
    @unittest.skipIf(
        torch_device != 'cuda' or not is_xformers_available()					,       reason='XFormers attention is only available with CUDA and `xformers` installed'					,       )
    def 						_a							(  self	):




         '''simple docstring'''



         self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3	)
    @unittest.skipIf(torch_device != 'cuda'					,       reason='float16 requires CUDA'	)
    def 						_a							(  self	):




         '''simple docstring'''



         super().test_save_load_floataa(expected_max_diff=1e-1	)
    @unittest.skipIf(torch_device != 'cuda'					,       reason='float16 requires CUDA'	)
    def 						_a							(  self	):




         '''simple docstring'''



         super().test_floataa_inference(expected_max_diff=1e-1	)
    def 						_a							(  self	):




         '''simple docstring'''



         self._test_attention_slicing_forward_pass(expected_max_diff=1e-2	)
    def 						_a							(  self	):




         '''simple docstring'''



         self._test_save_load_local()

    def 						_a							(  self	):




         '''simple docstring'''



         self._test_inference_batch_single_identical(
             expected_max_diff=1e-2					,       )



