



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

import numpy as np
import torch

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


__snake_case		:Optional[int]   	=  logging.get_logger(__name__)

class 					_A (			__UpperCAmelCase					):
       UpperCamelCase__						:					str		      =		['''input_features''', '''is_longer''']



       def __init__(	self							:  int  ,	__SCREAMING_SNAKE_CASE							:  int=64  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=48_000  ,	__SCREAMING_SNAKE_CASE							:  Any=480  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]=10  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]=1_024  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=0.0  ,	__SCREAMING_SNAKE_CASE							:  Any=False  ,	__SCREAMING_SNAKE_CASE							:  float = 0  ,	__SCREAMING_SNAKE_CASE							:  float = 14_000  ,	__SCREAMING_SNAKE_CASE							:  int = None  ,	__SCREAMING_SNAKE_CASE							:  str = "fusion"  ,	__SCREAMING_SNAKE_CASE							:  str = "repeatpad"  ,	**__SCREAMING_SNAKE_CASE							:  Dict  ,	):


           '''simple docstring'''

           super().__init__(
               feature_size=lowerCAmelCase_  ,	sampling_rate=lowerCAmelCase_  ,	padding_value=lowerCAmelCase_  ,	return_attention_mask=lowerCAmelCase_  ,	**lowerCAmelCase_  ,	)
           __a		 =       top_db
           __a		 =       truncation
           __a		 =       padding
           __a		 =       fft_window_size
           __a		 =       (fft_window_size >> 1) + 1
           __a		 =       hop_length
           __a		 =       max_length_s
           __a		 =       max_length_s * sampling_rate
           __a		 =       sampling_rate
           __a		 =       frequency_min
           __a		 =       frequency_max
           __a		 =       mel_filter_bank(
               num_frequency_bins=self.nb_frequency_bins  ,	num_mel_filters=lowerCAmelCase_  ,	min_frequency=lowerCAmelCase_  ,	max_frequency=lowerCAmelCase_  ,	sampling_rate=lowerCAmelCase_  ,	norm=lowerCAmelCase_  ,	mel_scale='''htk'''  ,	)
           __a		 =       mel_filter_bank(
               num_frequency_bins=self.nb_frequency_bins  ,	num_mel_filters=lowerCAmelCase_  ,	min_frequency=lowerCAmelCase_  ,	max_frequency=lowerCAmelCase_  ,	sampling_rate=lowerCAmelCase_  ,	norm='''slaney'''  ,	mel_scale='''slaney'''  ,	)



       def 	_lowerCamelCase    (	self							:  Optional[int]):


           '''simple docstring'''

           __a		 =       copy.deepcopy(self.__dict__)
           __a		 =       self.__class__.__name__
           if "mel_filters" in output:
               del output["mel_filters"]
           if "mel_filters_slaney" in output:
               del output["mel_filters_slaney"]
           return output



       def 	_lowerCamelCase    (	self							:  Optional[Any]  ,	__SCREAMING_SNAKE_CASE							:  np.array  ,	__SCREAMING_SNAKE_CASE							:  Optional[np.array] = None):


           '''simple docstring'''

           __a		 =       spectrogram(
               lowerCAmelCase_  ,	window_function(self.fft_window_size  ,	'''hann''')  ,	frame_length=self.fft_window_size  ,	hop_length=self.hop_length  ,	power=2.0  ,	mel_filters=lowerCAmelCase_  ,	log_mel='''dB'''  ,	)
           return log_mel_spectrogram.T



       def 	_lowerCamelCase    (	self							:  List[str]  ,	__SCREAMING_SNAKE_CASE							:  List[str]  ,	__SCREAMING_SNAKE_CASE							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  Optional[Any]):


           '''simple docstring'''

           __a		 =       np.array_split(list(range(0  ,	total_frames - chunk_frames + 1))  ,	3)
           if len(ranges[1]) == 0:
               # if the audio is too short, we just use the first chunk
               __a		 =       [0]
           if len(ranges[2]) == 0:
               # if the audio is too short, we just use the first chunk
               __a		 =       [0]
           # randomly choose index for each part
           __a		 =       np.random.choice(ranges[0])
           __a		 =       np.random.choice(ranges[1])
           __a		 =       np.random.choice(ranges[2])

           __a		 =       mel[idx_front : idx_front + chunk_frames, :]
           __a		 =       mel[idx_middle : idx_middle + chunk_frames, :]
           __a		 =       mel[idx_back : idx_back + chunk_frames, :]

           __a		 =       torch.tensor(mel[None, None, :])
           __a		 =       torch.nn.functional.interpolate(
               lowerCAmelCase_  ,	size=[chunk_frames, 64]  ,	mode='''bilinear'''  ,	align_corners=lowerCAmelCase_)
           __a		 =       mel_shrink[0][0].numpy()
           __a		 =       np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]  ,	axis=0)
           return mel_fusion



       def 	_lowerCamelCase    (	self							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  np.array  ,	__SCREAMING_SNAKE_CASE							:  List[str]  ,	__SCREAMING_SNAKE_CASE							:  List[str]  ,	__SCREAMING_SNAKE_CASE							:  Union[str, Any]):


           '''simple docstring'''

           if waveform.shape[0] > max_length:
               if truncation == "rand_trunc":
                   __a		 =       True
                   # random crop to max_length (for compatibility) -> this should be handled by self.pad
                   __a		 =       len(lowerCAmelCase_) - max_length
                   __a		 =       np.random.randint(0  ,	overflow + 1)
                   __a		 =       waveform[idx : idx + max_length]
                   __a		 =       self._np_extract_fbank_features(lowerCAmelCase_  ,	self.mel_filters_slaney)[None, :]
               elif truncation == "fusion":
                   __a		 =       self._np_extract_fbank_features(lowerCAmelCase_  ,	self.mel_filters)
                   __a		 =       max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                   __a		 =       mel.shape[0]
                   if chunk_frames == total_frames:
                       # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                       # In this case, we just use the whole audio.
                       __a		 =       np.stack([mel, mel, mel, mel]  ,	axis=0)
                       __a		 =       False
                   else:
                       __a		 =       self._random_mel_fusion(lowerCAmelCase_  ,	lowerCAmelCase_  ,	lowerCAmelCase_)
                       __a		 =       True
               else:
                   raise NotImplementedError(F'data_truncating {truncation} not implemented')

           else:
               __a		 =       False
               # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
               if waveform.shape[0] < max_length:
                   if padding == "repeat":
                       __a		 =       int(max_length / len(lowerCAmelCase_))
                       __a		 =       np.stack(np.tile(lowerCAmelCase_  ,	n_repeat + 1))[:max_length]
                   if padding == "repeatpad":
                       __a		 =       int(max_length / len(lowerCAmelCase_))
                       __a		 =       np.stack(np.tile(lowerCAmelCase_  ,	lowerCAmelCase_))
                   __a		 =       np.pad(lowerCAmelCase_  ,	(0, max_length - waveform.shape[0])  ,	mode='''constant'''  ,	constant_values=0)

               if truncation == "fusion":
                   __a		 =       self._np_extract_fbank_features(lowerCAmelCase_  ,	self.mel_filters)
                   __a		 =       np.stack([input_mel, input_mel, input_mel, input_mel]  ,	axis=0)
               else:
                   __a		 =       self._np_extract_fbank_features(lowerCAmelCase_  ,	self.mel_filters_slaney)[None, :]

           return input_mel, longer





       def __call__(	self							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]  ,	__SCREAMING_SNAKE_CASE							:  str = None  ,	__SCREAMING_SNAKE_CASE							:  Optional[str] = None  ,	__SCREAMING_SNAKE_CASE							:  Optional[int] = None  ,	__SCREAMING_SNAKE_CASE							:  Optional[int] = None  ,	__SCREAMING_SNAKE_CASE							:  Optional[Union[str, TensorType]] = None  ,	**__SCREAMING_SNAKE_CASE							:  List[str]  ,	):


           '''simple docstring'''

           __a		 =       truncation if truncation is not None else self.truncation
           __a		 =       padding if padding else self.padding

           if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                   raise ValueError(
                       F'The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'
                       F' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'
                       F' was sampled 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(lowerCAmelCase_  ,	dtype=np.floataa) 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(lowerCAmelCase_)]

           # convert to mel spectrogram, truncate and pad if needed.
           __a		 =       [
               self._get_input_mel(lowerCAmelCase_  ,	max_length if max_length else self.nb_max_samples  ,	lowerCAmelCase_  ,	lowerCAmelCase_)
               for waveform in raw_speech
           ]

           __a		 =       []
           __a		 =       []
           for mel, longer in padded_inputs:
               input_mel.append(lowerCAmelCase_)
               is_longer.append(lowerCAmelCase_)

           if truncation == "fusion" and sum(lowerCAmelCase_) == 0:
               # if no audio is longer than 10s, then randomly select one audio to be longer
               __a		 =       np.random.randint(0  ,	len(lowerCAmelCase_))
               __a		 =       True

           if isinstance(input_mel[0]  ,	lowerCAmelCase_):
               __a		 =       [np.asarray(lowerCAmelCase_  ,	dtype=np.floataa) for feature in input_mel]

           # is_longer is a list of bool
           __a		 =       [[longer] for longer in is_longer]

           __a		 =       {'''input_features''': input_mel, '''is_longer''': is_longer}
           __a		 =       BatchFeature(lowerCAmelCase_)

           if return_tensors is not None:
               __a		 =       input_features.convert_to_tensors(lowerCAmelCase_)

           return input_features










# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
 import jax
 import jaxlib

__snake_case		:str   	=  get_logger()

__snake_case		:Optional[dict]   	=  None

class 					_A (			TensorFormatter[Mapping, '''jax.Array''', Mapping]					):



       def __init__(	self							:  str  ,	__SCREAMING_SNAKE_CASE							:  int=None  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=None  ,	**__SCREAMING_SNAKE_CASE							:  Any):


           '''simple docstring'''

           super().__init__(features=__SCREAMING_SNAKE_CASE)
           import jax
           from jaxlib.xla_client import Device

           if isinstance(__SCREAMING_SNAKE_CASE  ,	__SCREAMING_SNAKE_CASE):
               raise ValueError(
                   F'Expected {device} to be a `str` not {type(__SCREAMING_SNAKE_CASE)}, as `jaxlib.xla_extension.Device` '
                   '''is not serializable neither with `pickle` nor with `dill`. Instead you can surround '''
                   '''the device with `str()` to get its string identifier that will be internally mapped '''
                   '''to the actual `jaxlib.xla_extension.Device`.''')
           __a		 =       device if isinstance(__SCREAMING_SNAKE_CASE  ,	__SCREAMING_SNAKE_CASE) else str(jax.devices()[0])
           # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
           # with `pickle` nor with `dill`, so we need to use a global variable instead
           global DEVICE_MAPPING
           if DEVICE_MAPPING is None:
               __a		 =       self._map_devices_to_str()
           if self.device not in list(DEVICE_MAPPING.keys()):
               logger.warning(
                   F'Device with string identifier {self.device} not listed among the available '
                   F'devices: {list(DEVICE_MAPPING.keys())}, so falling back to the default '
                   F'device: {str(jax.devices()[0])}.')
               __a		 =       str(jax.devices()[0])
           __a		 =       jnp_array_kwargs



       @staticmethod
       def 	_lowerCamelCase    (	):


           '''simple docstring'''

           import jax

           return {str(__SCREAMING_SNAKE_CASE): device for device in jax.devices()}



       def 	_lowerCamelCase    (	self							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  Tuple):


           '''simple docstring'''

           import jax
           import jax.numpy as jnp

           if isinstance(__SCREAMING_SNAKE_CASE  ,	__SCREAMING_SNAKE_CASE) and column:
               if all(
                   isinstance(__SCREAMING_SNAKE_CASE  ,	jax.Array) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column):
                   return jnp.stack(__SCREAMING_SNAKE_CASE  ,	axis=0)
           return column



       def 	_lowerCamelCase    (	self							:  str  ,	__SCREAMING_SNAKE_CASE							:  str):


           '''simple docstring'''

           import jax
           import jax.numpy as jnp

           if isinstance(__SCREAMING_SNAKE_CASE  ,	(str, bytes, type(__SCREAMING_SNAKE_CASE))):
               return value
           elif isinstance(__SCREAMING_SNAKE_CASE  ,	(np.character, np.ndarray)) and np.issubdtype(value.dtype  ,	np.character):
               return value.tolist()

           __a		 =       {}

           if isinstance(__SCREAMING_SNAKE_CASE  ,	(np.number, np.ndarray)) and np.issubdtype(value.dtype  ,	np.integer):
               # the default int precision depends on the jax config
               # see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
               if jax.config.jax_enable_xaa:
                   __a		 =       {'''dtype''': jnp.intaa}
               else:
                   __a		 =       {'''dtype''': jnp.intaa}
           elif isinstance(__SCREAMING_SNAKE_CASE  ,	(np.number, np.ndarray)) and np.issubdtype(value.dtype  ,	np.floating):
               __a		 =       {'''dtype''': jnp.floataa}
           elif config.PIL_AVAILABLE and "PIL" in sys.modules:
               import PIL.Image

               if isinstance(__SCREAMING_SNAKE_CASE  ,	PIL.Image.Image):
                   __a		 =       np.asarray(__SCREAMING_SNAKE_CASE)

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
           global DEVICE_MAPPING
           if DEVICE_MAPPING is None:
               __a		 =       self._map_devices_to_str()

           with jax.default_device(DEVICE_MAPPING[self.device]):
               # calling jnp.array on a np.ndarray does copy the data
               # see https://github.com/google/jax/issues/4486
               return jnp.array(__SCREAMING_SNAKE_CASE  ,	**{**default_dtype, **self.jnp_array_kwargs})



       def 	_lowerCamelCase    (	self							:  Tuple  ,	__SCREAMING_SNAKE_CASE							:  str):


           '''simple docstring'''

           import jax

           # support for torch, tf, jax etc.
           if config.TORCH_AVAILABLE and "torch" in sys.modules:
               import torch

               if isinstance(__SCREAMING_SNAKE_CASE  ,	torch.Tensor):
                   return self._tensorize(data_struct.detach().cpu().numpy()[()])
           if hasattr(__SCREAMING_SNAKE_CASE  ,	'''__array__''') and not isinstance(__SCREAMING_SNAKE_CASE  ,	jax.Array):
               __a		 =       data_struct.__array__()
           # support for nested types like struct of list of struct
           if isinstance(__SCREAMING_SNAKE_CASE  ,	np.ndarray):
               if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
                   return self._consolidate([self.recursive_tensorize(__SCREAMING_SNAKE_CASE) for substruct in data_struct])
           elif isinstance(__SCREAMING_SNAKE_CASE  ,	(list, tuple)):
               return self._consolidate([self.recursive_tensorize(__SCREAMING_SNAKE_CASE) for substruct in data_struct])
           return self._tensorize(__SCREAMING_SNAKE_CASE)



       def 	_lowerCamelCase    (	self							:  Tuple  ,	__SCREAMING_SNAKE_CASE							:  dict):


           '''simple docstring'''

           return map_nested(self._recursive_tensorize  ,	__SCREAMING_SNAKE_CASE  ,	map_list=__SCREAMING_SNAKE_CASE)



       def 	_lowerCamelCase    (	self							:  str  ,	__SCREAMING_SNAKE_CASE							:  pa.Table):


           '''simple docstring'''

           __a		 =       self.numpy_arrow_extractor().extract_row(__SCREAMING_SNAKE_CASE)
           __a		 =       self.python_features_decoder.decode_row(__SCREAMING_SNAKE_CASE)
           return self.recursive_tensorize(__SCREAMING_SNAKE_CASE)



       def 	_lowerCamelCase    (	self							:  Tuple  ,	__SCREAMING_SNAKE_CASE							:  pa.Table):


           '''simple docstring'''

           __a		 =       self.numpy_arrow_extractor().extract_column(__SCREAMING_SNAKE_CASE)
           __a		 =       self.python_features_decoder.decode_column(__SCREAMING_SNAKE_CASE  ,	pa_table.column_names[0])
           __a		 =       self.recursive_tensorize(__SCREAMING_SNAKE_CASE)
           __a		 =       self._consolidate(__SCREAMING_SNAKE_CASE)
           return column





       def 	_lowerCamelCase    (	self							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  pa.Table):


           '''simple docstring'''

           __a		 =       self.numpy_arrow_extractor().extract_batch(__SCREAMING_SNAKE_CASE)
           __a		 =       self.python_features_decoder.decode_batch(__SCREAMING_SNAKE_CASE)
           __a		 =       self.recursive_tensorize(__SCREAMING_SNAKE_CASE)
           for column_name in batch:
               __a		 =       self._consolidate(batch[column_name])
           return batch





