


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


__A :  List[str]    							=				logging.get_logger(__name__)






class        __snake_case	(       _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    lowercase       			= ['audio_values', 'audio_mask']

    def __init__(       self   :  List[str]   ,						lowerCamelCase   :  Any=20_48   ,						lowerCamelCase   :  str=1   ,						lowerCamelCase   :  str=[16, 16]   ,						lowerCamelCase   :  int=1_28   ,						lowerCamelCase   :  Optional[int]=4_41_00   ,						lowerCamelCase   :  Tuple=86   ,						lowerCamelCase   :  str=20_48   ,						lowerCamelCase   :  Any=0.0   ,						**lowerCamelCase   :  Any   ,						)       ->  str:
      super().__init__(
          feature_size=lowerCamelCase   ,						sampling_rate=lowerCamelCase   ,						padding_value=lowerCamelCase   ,						**lowerCamelCase   ,						)

      lowerCAmelCase_      :				List[Any]      =       spectrogram_length
      lowerCAmelCase_      :				Optional[int]      =       num_channels
      lowerCAmelCase_      :				List[str]      =       patch_size
      lowerCAmelCase_      :				Dict      =       feature_size // self.patch_size[1]
      lowerCAmelCase_      :				Optional[int]      =       n_fft
      lowerCAmelCase_      :				Any      =       sampling_rate // hop_length_to_sampling_rate
      lowerCAmelCase_      :				Optional[Any]      =       sampling_rate
      lowerCAmelCase_      :				int      =       padding_value
      lowerCAmelCase_      :				Tuple      =       mel_filter_bank(
          num_frequency_bins=1 + n_fft // 2   ,						num_mel_filters=lowerCamelCase   ,						min_frequency=0.0   ,						max_frequency=22_050.0   ,						sampling_rate=lowerCamelCase   ,						norm="""slaney"""   ,						mel_scale="""slaney"""   ,						).T

    def __lowercase       (       self   :  List[Any]   ,						lowerCamelCase   :  np.array		)       ->  np.ndarray:
      lowerCAmelCase_      :				List[Any]      =       spectrogram(
          lowerCamelCase   ,						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   ,						)
      lowerCAmelCase_      :				Any      =       log_spec[:, :-1]
      lowerCAmelCase_      :				List[Any]      =       log_spec - 20.0
      lowerCAmelCase_      :				Optional[Any]      =       np.clip(log_spec / 40.0   ,						-2.0   ,						0.0		) + 1.0
      return log_spec





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

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

      lowerCAmelCase_      :				Any      =       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}'		)
      lowerCAmelCase_      :				Optional[Any]      =       is_batched_numpy or (
          isinstance(lowerCamelCase   ,						(list, tuple)		) and (isinstance(raw_speech[0]   ,						(np.ndarray, tuple, list)		))
      )
      if is_batched:
        lowerCAmelCase_      :				Dict      =       [np.asarray([speech]   ,						dtype=np.floataa		).T for speech in raw_speech]
      elif not is_batched and not isinstance(lowerCamelCase   ,						np.ndarray		):
        lowerCAmelCase_      :				List[str]      =       np.asarray(lowerCamelCase   ,						dtype=np.floataa		)
      elif isinstance(lowerCamelCase   ,						np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
        lowerCAmelCase_      :				str      =       raw_speech.astype(np.floataa		)
      # always return batch
      if not is_batched:
        lowerCAmelCase_      :				Optional[int]      =       [np.asarray([raw_speech]		).T]

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

      # Create audio attention mask
      lowerCAmelCase_      :				Tuple      =       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:
        lowerCAmelCase_      :				Union[str, Any]      =       [
            (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
        ]
        lowerCAmelCase_      :				Optional[int]      =       np.array(lowerCamelCase		).astype(np.floataa		)

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

      # return as BatchFeature
      if return_attention_mask:
        lowerCAmelCase_      :				str      =       {"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
      else:
        lowerCAmelCase_      :				Optional[int]      =       {"""audio_values""": padded_audio_features}

      lowerCAmelCase_      :				List[str]      =       BatchFeature(data=lowerCamelCase   ,						tensor_type=lowerCamelCase		)
      return encoded_inputs







'''simple docstring'''






import unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible






class        __snake_case	(       unittest.TestCase):







    """simple docstring"""



    def __lowercase       (       self   :  Tuple		)       ->  Dict:
      lowerCAmelCase_      :				str      =       [
          """safety_checker/pytorch_model.bin""",
          """safety_checker/model.safetensors""",
          """vae/diffusion_pytorch_model.bin""",
          """vae/diffusion_pytorch_model.safetensors""",
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      self.assertTrue(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  List[Any]		)       ->  int:
      lowerCAmelCase_      :				Tuple      =       [
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      self.assertTrue(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  Optional[int]		)       ->  Optional[Any]:
      lowerCAmelCase_      :				int      =       [
          """safety_checker/pytorch_model.bin""",
          """safety_checker/model.safetensors""",
          """vae/diffusion_pytorch_model.bin""",
          """vae/diffusion_pytorch_model.safetensors""",
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
          """unet/diffusion_pytorch_model.bin""",
          # Removed: 'unet/diffusion_pytorch_model.safetensors',
      ]
      self.assertFalse(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  int		)       ->  List[Any]:
      lowerCAmelCase_      :				Dict      =       [
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
      ]
      self.assertTrue(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  str		)       ->  List[str]:
      lowerCAmelCase_      :				Union[str, Any]      =       [
          """safety_checker/pytorch_model.bin""",
          """safety_checker/model.safetensors""",
          """vae/diffusion_pytorch_model.bin""",
          """vae/diffusion_pytorch_model.safetensors""",
          """text_encoder/pytorch_model.bin""",
          # Removed: 'text_encoder/model.safetensors',
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      self.assertFalse(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  List[Any]		)       ->  Tuple:
      lowerCAmelCase_      :				Union[str, Any]      =       [
          """safety_checker/pytorch_model.fp16.bin""",
          """safety_checker/model.fp16.safetensors""",
          """vae/diffusion_pytorch_model.fp16.bin""",
          """vae/diffusion_pytorch_model.fp16.safetensors""",
          """text_encoder/pytorch_model.fp16.bin""",
          """text_encoder/model.fp16.safetensors""",
          """unet/diffusion_pytorch_model.fp16.bin""",
          """unet/diffusion_pytorch_model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				Union[str, Any]      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Optional[Any]		)       ->  List[str]:
      lowerCAmelCase_      :				str      =       [
          """unet/diffusion_pytorch_model.fp16.bin""",
          """unet/diffusion_pytorch_model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				Optional[int]      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Tuple		)       ->  Optional[Any]:
      # pass variant but use the non-variant filenames
      lowerCAmelCase_      :				Dict      =       [
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      lowerCAmelCase_      :				str      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Optional[int]		)       ->  List[str]:
      lowerCAmelCase_      :				str      =       [
          """safety_checker/pytorch_model.fp16.bin""",
          """safety_checker/model.fp16.safetensors""",
          """vae/diffusion_pytorch_model.fp16.bin""",
          """vae/diffusion_pytorch_model.fp16.safetensors""",
          """text_encoder/pytorch_model.fp16.bin""",
          """text_encoder/model.fp16.safetensors""",
          """unet/diffusion_pytorch_model.fp16.bin""",
          # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
      ]
      lowerCAmelCase_      :				List[str]      =       """fp16"""
      self.assertFalse(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Union[str, Any]		)       ->  Optional[int]:
      lowerCAmelCase_      :				str      =       [
          """text_encoder/pytorch_model.fp16.bin""",
          """text_encoder/model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				Optional[Any]      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  List[Any]		)       ->  List[Any]:
      # pass variant but use the non-variant filenames
      lowerCAmelCase_      :				Dict      =       [
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
      ]
      lowerCAmelCase_      :				Any      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)





    def __lowercase       (       self   :  Dict		)       ->  Any:
      lowerCAmelCase_      :				Optional[int]      =       [
          """safety_checker/pytorch_model.fp16.bin""",
          """safety_checker/model.fp16.safetensors""",
          """vae/diffusion_pytorch_model.fp16.bin""",
          """vae/diffusion_pytorch_model.fp16.safetensors""",
          """text_encoder/pytorch_model.fp16.bin""",
          # 'text_encoder/model.fp16.safetensors',
          """unet/diffusion_pytorch_model.fp16.bin""",
          """unet/diffusion_pytorch_model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				int      =       """fp16"""
      self.assertFalse(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)



