


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


__UpperCamelCase    :						str     =							logging.get_logger(__name__)



class 							_UpperCamelCase (						_UpperCAmelCase						):

   '''simple docstring'''







   a_				:							Any    = ['''audio_values''', '''audio_mask''']

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






         '''simple docstring'''



         super().__init__(
             feature_size=A_       ,  sampling_rate=A_       ,  padding_value=A_       ,  **A_       ,  )

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

   def   _snake_case    (  self  : Optional[int]       ,  _lowerCamelCase  : np.array							):






         '''simple docstring'''



         __lowerCamelCase     :      str       						=						spectrogram(
             A_       ,  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     :      Optional[Any]       						=						log_spec[:, :-1]
         __lowerCamelCase     :      int       						=						log_spec - 20.0
         __lowerCamelCase     :      str       						=						np.clip(log_spec / 40.0       ,  -2.0       ,  0.0							) + 1.0
         return log_spec


   def __call__(  self  : List[str]       ,  _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  : Union[str, Any]       ,  ):






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

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

         # Convert audio signals to log mel spectrograms, truncate by time axis
         __lowerCamelCase     :      Optional[int]       						=						[
             self._np_extract_fbank_features(waveform.squeeze()							).T[: self.spectrogram_length] for waveform in raw_speech
         ]
         if isinstance(audio_features[0]       ,  A_							):
               __lowerCamelCase     :      Optional[Any]       						=						[np.asarray(A_       ,  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     :      Optional[int]       						=						[
                   (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     :      Any       						=						np.array(A_							).astype(np.floataa							)

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

         # return as BatchFeature
         if return_attention_mask:
               __lowerCamelCase     :      str       						=						{"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
         else:
               __lowerCamelCase     :      List[str]       						=						{"""audio_values""": padded_audio_features}

         __lowerCamelCase     :      Dict       						=						BatchFeature(data=A_       ,  tensor_type=A_							)
         return encoded_inputs





import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin



class 							_UpperCamelCase (						A						):

   '''simple docstring'''







   a_				:							List[Any]    = ["image_processor", "tokenizer"]
   a_				:							Tuple    = "AutoImageProcessor"
   a_				:							Tuple    = "AutoTokenizer"

   def __init__(  self  : Union[str, Any]       ,  _lowerCamelCase  : List[Any]=None       ,  _lowerCamelCase  : Any=None       ,  **_lowerCamelCase  : List[str]							):






         '''simple docstring'''



         __lowerCamelCase     :      Tuple       						=						None
         if "feature_extractor" in kwargs:
               warnings.warn(
                   """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
                   """ instead."""       ,  _lowerCamelCase       ,  )
               __lowerCamelCase     :      List[Any]       						=						kwargs.pop("""feature_extractor"""							)

         __lowerCamelCase     :      Optional[Any]       						=						image_processor if image_processor is not None else feature_extractor
         if image_processor is None:
               raise ValueError("""You need to specify an `image_processor`."""							)
         if tokenizer is None:
               raise ValueError("""You need to specify a `tokenizer`."""							)

         super().__init__(_lowerCamelCase       ,  _lowerCamelCase							)
         __lowerCamelCase     :      List[Any]       						=						self.image_processor
         __lowerCamelCase     :      Any       						=						False

   def __call__(  self  : Union[str, Any]       ,  *_lowerCamelCase  : str       ,  **_lowerCamelCase  : Dict							):






         '''simple docstring'''



         if self._in_target_context_manager:
               return self.current_processor(*_lowerCamelCase       ,  **_lowerCamelCase							)

         __lowerCamelCase     :      List[str]       						=						kwargs.pop("""images"""       ,  _lowerCamelCase							)
         __lowerCamelCase     :      List[str]       						=						kwargs.pop("""text"""       ,  _lowerCamelCase							)
         if len(_lowerCamelCase							) > 0:
               __lowerCamelCase     :      Tuple       						=						args[0]
               __lowerCamelCase     :      List[Any]       						=						args[1:]

         if images is None and text is None:
               raise ValueError("""You need to specify either an `images` or `text` input to process."""							)

         if images is not None:
               __lowerCamelCase     :      Tuple       						=						self.image_processor(_lowerCamelCase       ,  *_lowerCamelCase       ,  **_lowerCamelCase							)
         if text is not None:
               __lowerCamelCase     :      int       						=						self.tokenizer(_lowerCamelCase       ,  **_lowerCamelCase							)

         if text is None:
               return inputs
         elif images is None:
               return encodings
         else:
               __lowerCamelCase     :      Optional[int]       						=						encodings["""input_ids"""]
               return inputs

   def   _snake_case    (  self  : List[Any]       ,  *_lowerCamelCase  : Tuple       ,  **_lowerCamelCase  : Union[str, Any]							):






         '''simple docstring'''



         return self.tokenizer.batch_decode(*_lowerCamelCase       ,  **_lowerCamelCase							)

   def   _snake_case    (  self  : List[str]       ,  *_lowerCamelCase  : Optional[int]       ,  **_lowerCamelCase  : int							):






         '''simple docstring'''



         return self.tokenizer.decode(*_lowerCamelCase       ,  **_lowerCamelCase							)

   @contextmanager
   def   _snake_case    (  self  : Optional[Any]							):






         '''simple docstring'''



         warnings.warn(
             """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
             """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
             """your images inputs, or in a separate call."""							)
         __lowerCamelCase     :      Any       						=						True
         __lowerCamelCase     :      int       						=						self.tokenizer
         yield
         __lowerCamelCase     :      Dict       						=						self.image_processor
         __lowerCamelCase     :      List[Any]       						=						False

   def   _snake_case    (  self  : List[str]       ,  _lowerCamelCase  : str       ,  _lowerCamelCase  : List[str]=False       ,  _lowerCamelCase  : List[Any]=None							):






         '''simple docstring'''



         if added_vocab is None:
               __lowerCamelCase     :      List[str]       						=						self.tokenizer.get_added_vocab()

         __lowerCamelCase     :      List[Any]       						=						{}

         while tokens:
               __lowerCamelCase     :      Dict       						=						re.search(R"""<s_(.*?)>"""       ,  _lowerCamelCase       ,  re.IGNORECASE							)
               if start_token is None:
                     break
               __lowerCamelCase     :      Dict       						=						start_token.group(1							)
               __lowerCamelCase     :      List[str]       						=						re.search(RF"""</s_{key}>"""       ,  _lowerCamelCase       ,  re.IGNORECASE							)
               __lowerCamelCase     :      Optional[int]       						=						start_token.group()
               if end_token is None:
                     __lowerCamelCase     :      int       						=						tokens.replace(_lowerCamelCase       ,  """"""							)
               else:
                     __lowerCamelCase     :      Dict       						=						end_token.group()
                     __lowerCamelCase     :      Optional[Any]       						=						re.escape(_lowerCamelCase							)
                     __lowerCamelCase     :      List[str]       						=						re.escape(_lowerCamelCase							)
                     __lowerCamelCase     :      Any       						=						re.search(F"""{start_token_escaped}(.*?){end_token_escaped}"""       ,  _lowerCamelCase       ,  re.IGNORECASE							)
                     if content is not None:
                           __lowerCamelCase     :      Union[str, Any]       						=						content.group(1							).strip()
                           if r"<s_" in content and r"</s_" in content:  # non-leaf node
                                 __lowerCamelCase     :      Union[str, Any]       						=						self.tokenajson(_lowerCamelCase       ,  is_inner_value=_lowerCamelCase       ,  added_vocab=_lowerCamelCase							)
                                 if value:
                                       if len(_lowerCamelCase							) == 1:
                                             __lowerCamelCase     :      Optional[int]       						=						value[0]
                                       __lowerCamelCase     :      Union[str, Any]       						=						value
                           else:  # leaf nodes
                                 __lowerCamelCase     :      List[Any]       						=						[]
                                 for leaf in content.split(R"""<sep/>"""							):
                                       __lowerCamelCase     :      Optional[Any]       						=						leaf.strip()
                                       if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
                                             __lowerCamelCase     :      str       						=						leaf[1:-2]  # for categorical special tokens
                                       output[key].append(_lowerCamelCase							)
                                 if len(output[key]							) == 1:
                                       __lowerCamelCase     :      Tuple       						=						output[key][0]

                     __lowerCamelCase     :      Tuple       						=						tokens[tokens.find(_lowerCamelCase							) + len(_lowerCamelCase							) :].strip()
                     if tokens[:6] == r"<sep/>":  # non-leaf nodes
                           return [output] + self.tokenajson(tokens[6:]       ,  is_inner_value=_lowerCamelCase       ,  added_vocab=_lowerCamelCase							)

         if len(_lowerCamelCase							):
               return [output] if is_inner_value else output
         else:
               return [] if is_inner_value else {"text_sequence": tokens}

   @property
   def   _snake_case    (  self  : Tuple							):






         '''simple docstring'''



         warnings.warn(
             """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""       ,  _lowerCamelCase       ,  )
         return self.image_processor_class


   @property
   def   _snake_case    (  self  : List[Any]							):






         '''simple docstring'''



         warnings.warn(
             """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""       ,  _lowerCamelCase       ,  )
         return self.image_processor



