






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


SCREAMING_SNAKE_CASE_      =			logging.get_logger(__name__)


class  lowerCAmelCase_						(       lowercase_      ):





  """simple docstring"""



  a_       :Optional[int]												=['''audio_values''', '''audio_mask''']

  def __init__(					self     :							int				,							SCREAMING_SNAKE_CASE__     :							Optional[Any]=2_0_4_8				,							SCREAMING_SNAKE_CASE__     :							Dict=1				,							SCREAMING_SNAKE_CASE__     :							str=[1_6, 1_6]				,							SCREAMING_SNAKE_CASE__     :							str=1_2_8				,							SCREAMING_SNAKE_CASE__     :							Union[str, Any]=4_4_1_0_0				,							SCREAMING_SNAKE_CASE__     :							Dict=8_6				,							SCREAMING_SNAKE_CASE__     :							Any=2_0_4_8				,							SCREAMING_SNAKE_CASE__     :							List[str]=0.0				,							**SCREAMING_SNAKE_CASE__     :							Tuple				,							):





       '''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=2_2_0_5_0.0				,							sampling_rate=UpperCamelCase__				,							norm="""slaney"""				,							mel_scale="""slaney"""				,							).T

  def __a						(					self     :							Any				,							SCREAMING_SNAKE_CASE__     :							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=8_0.0				,							)
       __a       =   log_spec[:, :-1]
       __a       =   log_spec - 2_0.0
       __a       =   np.clip(log_spec / 4_0.0				,							-2.0				,							0.0						) + 1.0
       return log_spec


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





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










'''simple docstring'''







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


SCREAMING_SNAKE_CASE_      =			{
    'configuration_bridgetower': [
        'BRIDGETOWER_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'BridgeTowerConfig',
        'BridgeTowerTextConfig',
        'BridgeTowerVisionConfig',
    ],
    'processing_bridgetower': ['BridgeTowerProcessor'],
}

try:
 if not is_vision_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 SCREAMING_SNAKE_CASE_      =			['BridgeTowerImageProcessor']

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 SCREAMING_SNAKE_CASE_      =			[
     'BRIDGETOWER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'BridgeTowerForContrastiveLearning',
     'BridgeTowerForImageAndTextRetrieval',
     'BridgeTowerForMaskedLM',
     'BridgeTowerModel',
     'BridgeTowerPreTrainedModel',
 ]


if TYPE_CHECKING:
 from .configuration_bridgetower import (
     BRIDGETOWER_PRETRAINED_CONFIG_ARCHIVE_MAP,
     BridgeTowerConfig,
     BridgeTowerTextConfig,
     BridgeTowerVisionConfig,
 )
 from .processing_bridgetower import BridgeTowerProcessor

 try:
  if not is_vision_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .image_processing_bridgetower import BridgeTowerImageProcessor

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_bridgetower import (
      BRIDGETOWER_PRETRAINED_MODEL_ARCHIVE_LIST,
      BridgeTowerForContrastiveLearning,
      BridgeTowerForImageAndTextRetrieval,
      BridgeTowerForMaskedLM,
      BridgeTowerModel,
      BridgeTowerPreTrainedModel,
  )


else:
 import sys

 SCREAMING_SNAKE_CASE_      =			_LazyModule(__name__, globals()['__file__'], _import_structure)






