



A_		:     List[Any]        =     '''ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'''


def 	UpperCAmelCase__      (      UpperCAmelCase__ :bytes					):






 '''simple docstring'''

 if not isinstance(UpperCAmelCase__ ,					UpperCAmelCase__					):
  a						=		F"""a bytes-like object is required, not '{data.__class__.__name__}'"""
  raise TypeError(UpperCAmelCase__					)

 a						=		"".join(bin(UpperCAmelCase__					)[2:].zfill(8					) for byte in data					)

 a						=		len(UpperCAmelCase__					) % 6 != 0

 if padding_needed:
  # The padding that will be added later
  a						=		b"=" * ((6 - len(UpperCAmelCase__					) % 6) // 2)

  # Append binary_stream with arbitrary binary digits (0's by default) to make its
  # length a multiple of 6.
  binary_stream += "0" * (6 - len(UpperCAmelCase__					) % 6)
 else:
  a						=		b""

 # Encode every 6 binary digits to their corresponding Base64 character
 return (
     "".join(
         B64_CHARSET[int(binary_stream[index : index + 6] ,					2					)]
         for index in range(0 ,					len(UpperCAmelCase__					) ,					6					)					).encode()
     + padding
 )


def 	UpperCAmelCase__      (      UpperCAmelCase__ :str					):






 '''simple docstring'''

 if not isinstance(UpperCAmelCase__ ,					UpperCAmelCase__					) and not isinstance(UpperCAmelCase__ ,					UpperCAmelCase__					):
  a						=		(
      "argument should be a bytes-like object or ASCII string, "
      F"""not '{encoded_data.__class__.__name__}'"""
  )
  raise TypeError(UpperCAmelCase__					)

 # In case encoded_data is a bytes-like object, make sure it contains only
 # ASCII characters so we convert it to a string object
 if isinstance(UpperCAmelCase__ ,					UpperCAmelCase__					):
  try:
   a						=		encoded_data.decode("utf-8"					)
  except UnicodeDecodeError:
   raise ValueError("base64 encoded data should only contain ASCII characters"					)

 a						=		encoded_data.count("="					)

 # Check if the encoded string contains non base64 characters
 if padding:
  assert all(
      char in B64_CHARSET for char in encoded_data[:-padding]					), "Invalid base64 character(s) found."
 else:
  assert all(
      char in B64_CHARSET for char in encoded_data					), "Invalid base64 character(s) found."

 # Check the padding
 assert len(UpperCAmelCase__					) % 4 == 0 and padding < 3, "Incorrect padding"

 if padding:
  # Remove padding if there is one
  a						=		encoded_data[:-padding]

  a						=		"".join(
      bin(B64_CHARSET.index(UpperCAmelCase__					)					)[2:].zfill(6					) for char in encoded_data					)[: -padding * 2]
 else:
  a						=		"".join(
      bin(B64_CHARSET.index(UpperCAmelCase__					)					)[2:].zfill(6					) for char in encoded_data					)

 a						=		[
     int(binary_stream[index : index + 8] ,					2					)
     for index in range(0 ,					len(UpperCAmelCase__					) ,					8					)
 ]

 return bytes(UpperCAmelCase__					)


if __name__ == "__main__":
     import doctest

     doctest.testmod()









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 		_lowercase		(       UpperCAmelCase__	):
   _UpperCAmelCase     	=			['''audio_values''', '''audio_mask''']

   def __init__(      self			:       List[Any]							,   __lowerCAmelCase			:       Dict=2048							,   __lowerCAmelCase			:       List[Any]=1							,   __lowerCAmelCase			:       Dict=[16, 16]							,   __lowerCAmelCase			:       str=128							,   __lowerCAmelCase			:       Optional[int]=4_4100							,   __lowerCAmelCase			:       int=86							,   __lowerCAmelCase			:       Optional[Any]=2048							,   __lowerCAmelCase			:       str=0.0							,   **__lowerCAmelCase			:       Optional[int]							,   )       ->				Union[str, Any]:


    """simple docstring"""






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

    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=__lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=2_2_0_5_0.0							,   sampling_rate=__lowerCAmelCase							,   norm="slaney"							,   mel_scale="slaney"							,   ).T

   def 		A							(      self			:       List[str]							,   __lowerCAmelCase			:       np.array							)       ->				np.ndarray:


    """simple docstring"""






    a						=		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=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			:       Union[str, 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[int]							,   )       ->				BatchFeature:


    """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(__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						=		is_batched_numpy or (
        isinstance(__lowerCAmelCase							,   (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(__lowerCAmelCase							,   np.ndarray							):
     a						=		np.asarray(__lowerCAmelCase							,   dtype=np.floataa							)
    elif isinstance(__lowerCAmelCase							,   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]							,   __lowerCAmelCase							):
     a						=		[np.asarray(__lowerCAmelCase							,   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(__lowerCAmelCase							).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(__lowerCAmelCase							), 1, max_time_len, self.feature_size]							).astype(np.floataa							)
    a						=		padded_audio_features * self.padding_value
    for i in range(len(__lowerCAmelCase							)							):
     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=__lowerCAmelCase							,   tensor_type=__lowerCAmelCase							)
    return encoded_inputs

