
"""simple docstring"""

import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
      from .features import FeatureType


UpperCamelCase     ,    UpperCamelCase     ,    UpperCamelCase     :    Optional[Any]       =			False, False, False

@dataclass
class __lowerCAmelCase		:
     lowercase						       =						None
     lowercase						       =						True
     lowercase						       =						True
     lowercase						       =						None
     # Automatically constructed
     lowercase						       =						"dict"
     lowercase						       =						pa.struct({"bytes": pa.binary(), "path": pa.string()}      )
     lowercase						       =						field(default="Audio"      ,  init=__SCREAMING_SNAKE_CASE      ,  repr=__SCREAMING_SNAKE_CASE      )





     def __call__(     self				):

      '''simple docstring'''




      return self.pa_type





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase				):

      '''simple docstring'''




      try:
       import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
      except ImportError as err:
       raise ImportError('To support encoding audio data, please install \'soundfile\'.'				) from err
      if isinstance(__UpperCAmelCase     ,					__UpperCAmelCase				):
       return {"bytes": None, "path": value}
      elif isinstance(__UpperCAmelCase     ,					__UpperCAmelCase				):
       return {"bytes": value, "path": None}
      elif "array" in value:
       # convert the audio array to wav bytes
       __UpperCamelCase      =	BytesIO()
       sf.write(__UpperCAmelCase     ,					value['array']     ,					value['sampling_rate']     ,					format='wav'				)
       return {"bytes": buffer.getvalue(), "path": None}
      elif value.get('path'				) is not None and os.path.isfile(value['path']				):
       # we set "bytes": None to not duplicate the data if they're already available locally
       if value["path"].endswith('pcm'				):
        # "PCM" only has raw audio bytes
        if value.get('sampling_rate'				) is None:
         # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
         raise KeyError('To use PCM files, please specify a \'sampling_rate\' in Audio object'				)
        if value.get('bytes'				):
         # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
         __UpperCamelCase      =	np.frombuffer(value['bytes']     ,					dtype=np.intaa				).astype(np.floataa				) / 3_2767
        else:
         __UpperCamelCase      =	np.memmap(value['path']     ,					dtype='h'     ,					mode='r'				).astype(np.floataa				) / 3_2767

        __UpperCamelCase      =	BytesIO(bytes()				)
        sf.write(__UpperCAmelCase     ,					__UpperCAmelCase     ,					value['sampling_rate']     ,					format='wav'				)
        return {"bytes": buffer.getvalue(), "path": None}
       else:
        return {"bytes": None, "path": value.get('path'				)}
      elif value.get('bytes'				) is not None or value.get('path'				) is not None:
       # store the audio bytes, and path is used to infer the audio format using the file extension
       return {"bytes": value.get('bytes'				), "path": value.get('path'				)}
      else:
       raise ValueError(
           F'An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'				)





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase = None				):

      '''simple docstring'''




      if not self.decode:
       raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.'				)

      __UpperCamelCase  ,     __UpperCamelCase      =	(value['path'], BytesIO(value['bytes']				)) if value['bytes'] is not None else (value['path'], None)
      if path is None and file is None:
       raise ValueError(F'An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'				)

      try:
       import librosa
       import soundfile as sf
      except ImportError as err:
       raise ImportError('To support decoding audio files, please install \'librosa\' and \'soundfile\'.'				) from err

      __UpperCamelCase      =	xsplitext(__UpperCAmelCase				)[1][1:].lower() if path is not None else None
      if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
       raise RuntimeError(
           'Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '
           'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '				)
      elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
       raise RuntimeError(
           'Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '
           'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '				)

      if file is None:
       __UpperCamelCase      =	token_per_repo_id or {}
       __UpperCamelCase      =	path.split('::'				)[-1]
       try:
        __UpperCamelCase      =	string_to_dict(__UpperCAmelCase     ,					config.HUB_DATASETS_URL				)['repo_id']
        __UpperCamelCase      =	token_per_repo_id[repo_id]
       except (ValueError, KeyError):
        __UpperCamelCase      =	None

       with xopen(__UpperCAmelCase     ,					'rb'     ,					use_auth_token=__UpperCAmelCase				) as f:
        __UpperCamelCase  ,     __UpperCamelCase      =	sf.read(__UpperCAmelCase				)

      else:
       __UpperCamelCase  ,     __UpperCamelCase      =	sf.read(__UpperCAmelCase				)

      __UpperCamelCase      =	array.T
      if self.mono:
       __UpperCamelCase      =	librosa.to_mono(__UpperCAmelCase				)
      if self.sampling_rate and self.sampling_rate != sampling_rate:
       __UpperCamelCase      =	librosa.resample(__UpperCAmelCase     ,					orig_sr=__UpperCAmelCase     ,					target_sr=self.sampling_rate				)
       __UpperCamelCase      =	self.sampling_rate

      return {"path": path, "array": array, "sampling_rate": sampling_rate}





     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      from .features import Value

      if self.decode:
       raise ValueError('Cannot flatten a decoded Audio feature.'				)
      return {
          "bytes": Value('binary'				),
          "path": Value('string'				),
      }





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase				):

      '''simple docstring'''




      if pa.types.is_string(storage.type				):
       __UpperCamelCase      =	pa.array([None] * len(__UpperCAmelCase				)     ,					type=pa.binary()				)
       __UpperCamelCase      =	pa.StructArray.from_arrays([bytes_array, storage]     ,					['bytes', 'path']     ,					mask=storage.is_null()				)
      elif pa.types.is_binary(storage.type				):
       __UpperCamelCase      =	pa.array([None] * len(__UpperCAmelCase				)     ,					type=pa.string()				)
       __UpperCamelCase      =	pa.StructArray.from_arrays([storage, path_array]     ,					['bytes', 'path']     ,					mask=storage.is_null()				)
      elif pa.types.is_struct(storage.type				) and storage.type.get_all_field_indices('array'				):
       __UpperCamelCase      =	pa.array([Audio().encode_example(__UpperCAmelCase				) if x is not None else None for x in storage.to_pylist()]				)
      elif pa.types.is_struct(storage.type				):
       if storage.type.get_field_index('bytes'				) >= 0:
        __UpperCamelCase      =	storage.field('bytes'				)
       else:
        __UpperCamelCase      =	pa.array([None] * len(__UpperCAmelCase				)     ,					type=pa.binary()				)
       if storage.type.get_field_index('path'				) >= 0:
        __UpperCamelCase      =	storage.field('path'				)
       else:
        __UpperCamelCase      =	pa.array([None] * len(__UpperCAmelCase				)     ,					type=pa.string()				)
       __UpperCamelCase      =	pa.StructArray.from_arrays([bytes_array, path_array]     ,					['bytes', 'path']     ,					mask=storage.is_null()				)
      return array_cast(__UpperCAmelCase     ,					self.pa_type				)







     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase				):

      '''simple docstring'''




      @no_op_if_value_is_null
      def path_to_bytes(__UpperCAmelCase				):
       with xopen(__UpperCAmelCase     ,					'rb'				) as f:
        __UpperCamelCase      =	f.read()
       return bytes_

      __UpperCamelCase      =	pa.array(
          [
              (path_to_bytes(x['path']				) if x['bytes'] is None else x['bytes']) if x is not None else None
              for x in storage.to_pylist()
          ]     ,					type=pa.binary()     ,					)
      __UpperCamelCase      =	pa.array(
          [os.path.basename(__UpperCAmelCase				) if path is not None else None for path in storage.field('path'				).to_pylist()]     ,					type=pa.string()     ,					)
      __UpperCamelCase      =	pa.StructArray.from_arrays([bytes_array, path_array]     ,					['bytes', 'path']     ,					mask=bytes_array.is_null()				)
      return array_cast(__UpperCAmelCase     ,					self.pa_type				)





"""simple docstring"""

import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCamelCase     :    int       =			logging.get_logger(__name__)

class __lowerCAmelCase		(				__SCREAMING_SNAKE_CASE      ):
     lowercase						       =						["input_values", "attention_mask"]





     def __init__(     self     ,					__UpperCAmelCase = 1     ,					__UpperCAmelCase = 1_6000     ,					__UpperCAmelCase = 0.0     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = 80     ,					__UpperCAmelCase = 16     ,					__UpperCAmelCase = 64     ,					__UpperCAmelCase = "hann_window"     ,					__UpperCAmelCase = 1.0     ,					__UpperCAmelCase = 80     ,					__UpperCAmelCase = 7600     ,					__UpperCAmelCase = 1E-10     ,					__UpperCAmelCase = 2     ,					__UpperCAmelCase = True     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      super().__init__(feature_size=__UpperCAmelCase     ,					sampling_rate=__UpperCAmelCase     ,					padding_value=__UpperCAmelCase     ,					**__UpperCAmelCase				)
      __UpperCamelCase      =	do_normalize
      __UpperCamelCase      =	return_attention_mask

      __UpperCamelCase      =	num_mel_bins
      __UpperCamelCase      =	hop_length
      __UpperCamelCase      =	win_length
      __UpperCamelCase      =	win_function
      __UpperCamelCase      =	frame_signal_scale
      __UpperCamelCase      =	fmin
      __UpperCamelCase      =	fmax
      __UpperCamelCase      =	mel_floor
      __UpperCamelCase      =	reduction_factor

      __UpperCamelCase      =	win_length * sampling_rate // 1000
      __UpperCamelCase      =	hop_length * sampling_rate // 1000
      __UpperCamelCase      =	optimal_fft_length(self.sample_size				)
      __UpperCamelCase      =	(self.n_fft // 2) + 1

      __UpperCamelCase      =	window_function(window_length=self.sample_size     ,					name=self.win_function     ,					periodic=__UpperCAmelCase				)

      __UpperCamelCase      =	mel_filter_bank(
          num_frequency_bins=self.n_freqs     ,					num_mel_filters=self.num_mel_bins     ,					min_frequency=self.fmin     ,					max_frequency=self.fmax     ,					sampling_rate=self.sampling_rate     ,					norm='slaney'     ,					mel_scale='slaney'     ,					)

      if frame_signal_scale != 1.0:
       warnings.warn(
           'The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers'     ,					__UpperCAmelCase     ,					)
      if reduction_factor != 2.0:
       warnings.warn(
           'The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'     ,					__UpperCAmelCase     ,					)





     @staticmethod
     # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
     def 						UpperCAmelCase (     __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase = 0.0				):

      '''simple docstring'''




      if attention_mask is not None:
       __UpperCamelCase      =	np.array(__UpperCAmelCase     ,					np.intaa				)
       __UpperCamelCase      =	[]

       for vector, length in zip(__UpperCAmelCase     ,					attention_mask.sum(-1				)				):
        __UpperCamelCase      =	(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7				)
        if length < normed_slice.shape[0]:
         __UpperCamelCase      =	padding_value

        normed_input_values.append(__UpperCAmelCase				)
      else:
       __UpperCamelCase      =	[(x - x.mean()) / np.sqrt(x.var() + 1E-7				) for x in input_values]

      return normed_input_values





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	spectrogram(
          __UpperCAmelCase     ,					window=self.window     ,					frame_length=self.sample_size     ,					hop_length=self.sample_stride     ,					fft_length=self.n_fft     ,					mel_filters=self.mel_filters     ,					mel_floor=self.mel_floor     ,					log_mel='log10'     ,					)
      return log_mel_spec.T





     def __call__(     self     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      if audio is None and audio_target is None:
       raise ValueError('You must provide either `audio` or `audio_target` values.'				)

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

      if audio is not None:
       __UpperCamelCase      =	self._process_audio(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)
      else:
       __UpperCamelCase      =	None

      if audio_target is not None:
       __UpperCamelCase      =	self._process_audio(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)

       if inputs is None:
        return inputs_target
       else:
        __UpperCamelCase      =	inputs_target['input_values']
        __UpperCamelCase      =	inputs_target.get('attention_mask'				)
        if decoder_attention_mask is not None:
         __UpperCamelCase      =	decoder_attention_mask

      return inputs





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	isinstance(__UpperCAmelCase     ,					np.ndarray				) and len(speech.shape				) > 1
      if is_batched_numpy and len(speech.shape				) > 2:
       raise ValueError(F'Only mono-channel audio is supported for input to {self}'				)
      __UpperCamelCase      =	is_batched_numpy or (
          isinstance(__UpperCAmelCase     ,					(list, tuple)				) and (isinstance(speech[0]     ,					(np.ndarray, tuple, list)				))
      )

      if is_batched:
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				) for speech in speech]
      elif not is_batched and not isinstance(__UpperCAmelCase     ,					np.ndarray				):
       __UpperCamelCase      =	np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				)
      elif isinstance(__UpperCAmelCase     ,					np.ndarray				) and speech.dtype is np.dtype(np.floataa				):
       __UpperCamelCase      =	speech.astype(np.floataa				)

      # always return batch
      if not is_batched:
       __UpperCamelCase      =	[speech]

      # needed to make pad() work on spectrogram inputs
      __UpperCamelCase      =	self.feature_size

      # convert into correct format for padding
      if is_target:
       __UpperCamelCase      =	[self._extract_mel_features(__UpperCAmelCase				) for waveform in speech]
       __UpperCamelCase      =	BatchFeature({'input_values': features}				)
       __UpperCamelCase      =	self.num_mel_bins
      else:
       __UpperCamelCase      =	BatchFeature({'input_values': speech}				)

      __UpperCamelCase      =	self.pad(
          __UpperCAmelCase     ,					padding=__UpperCAmelCase     ,					max_length=__UpperCAmelCase     ,					truncation=__UpperCAmelCase     ,					pad_to_multiple_of=__UpperCAmelCase     ,					return_attention_mask=__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)

      __UpperCamelCase      =	feature_size_hack

      # convert input values to correct format
      __UpperCamelCase      =	padded_inputs['input_values']
      if not isinstance(input_values[0]     ,					np.ndarray				):
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				) for array in input_values]
      elif (
          not isinstance(__UpperCAmelCase     ,					np.ndarray				)
          and isinstance(input_values[0]     ,					np.ndarray				)
          and input_values[0].dtype is np.dtype(np.floataa				)
      ):
       __UpperCamelCase      =	[array.astype(np.floataa				) for array in input_values]
      elif isinstance(__UpperCAmelCase     ,					np.ndarray				) and input_values.dtype is np.dtype(np.floataa				):
       __UpperCamelCase      =	input_values.astype(np.floataa				)

      # convert attention_mask to correct format
      __UpperCamelCase      =	padded_inputs.get('attention_mask'				)
      if attention_mask is not None:
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.intaa				) for array in attention_mask]

      # zero-mean and unit-variance normalization
      if not is_target and self.do_normalize:
       __UpperCamelCase      =	(
           attention_mask
           if self._get_padding_strategies(__UpperCAmelCase     ,					max_length=__UpperCAmelCase				) is not PaddingStrategy.DO_NOT_PAD
           else None
       )
       __UpperCamelCase      =	self.zero_mean_unit_var_norm(
           padded_inputs['input_values']     ,					attention_mask=__UpperCAmelCase     ,					padding_value=self.padding_value				)

      if return_tensors is not None:
       __UpperCamelCase      =	padded_inputs.convert_to_tensors(__UpperCAmelCase				)

      return padded_inputs







     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      __UpperCamelCase      =	super().to_dict()

      # Don't serialize these as they are derived from the other properties.
      __UpperCamelCase      =	['window', 'mel_filters', 'sample_size', 'sample_stride', 'n_fft', 'n_freqs']
      for name in names:
       if name in output:
        del output[name]

      return output



