



import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
							import torch_xla.core.xla_model as xm
							import torch_xla.debug.metrics as met


class   __lowerCAmelCase  (		UpperCAmelCase_   ):
					"""simple docstring"""






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






								"""simple docstring"""


								super().__init__(*_snake_case					,       **_snake_case       )
								A__	  =     eval_examples
								A__	  =     post_process_function



					def      _a				(  self       :							int					,       _snake_case       :							Optional[Dataset] = None					,       _snake_case       :							int=None					,       _snake_case       :							Optional[List[str]] = None					,       _snake_case       :							str = "eval"					,       **_snake_case       :							Dict					,       ):






								"""simple docstring"""


								A__	  =     gen_kwargs.copy()
								A__	  =     (
								    gen_kwargs['max_length'] if gen_kwargs.get('max_length'       ) is not None else self.args.generation_max_length
								)
								A__	  =     (
								    gen_kwargs['num_beams'] if gen_kwargs.get('num_beams'       ) is not None else self.args.generation_num_beams
								)
								A__	  =     gen_kwargs

								A__	  =     self.eval_dataset if eval_dataset is None else eval_dataset
								A__	  =     self.get_eval_dataloader(_snake_case       )
								A__	  =     self.eval_examples if eval_examples is None else eval_examples

								# Temporarily disable metric computation, we will do it in the loop here.
								A__	  =     self.compute_metrics
								A__	  =     None
								A__	  =     time.time()
								A__	  =     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
								try:
											A__	  =     eval_loop(
											    _snake_case					,       description='Evaluation'					,       prediction_loss_only=True if compute_metrics is None else None					,       ignore_keys=_snake_case					,       metric_key_prefix=_snake_case					,       )
								finally:
											A__	  =     compute_metrics
								A__	  =     self.args.eval_batch_size * self.args.world_size
								if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
											start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
								output.metrics.update(
								    speed_metrics(
								        _snake_case					,       _snake_case					,       num_samples=output.num_samples					,       num_steps=math.ceil(output.num_samples / total_batch_size       )					,       )       )

								if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
											# Only the main node write the results by default
											A__	  =     self.post_process_function(_snake_case					,       _snake_case					,       _snake_case       )
											A__	  =     self.compute_metrics(_snake_case       )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()       ):
														if not key.startswith(F'''{metric_key_prefix}_'''       ):
																	A__	  =     metrics.pop(_snake_case       )

											metrics.update(output.metrics       )
								else:
											A__	  =     output.metrics

								if self.args.should_log:
											# Only the main node log the results by default
											self.log(_snake_case       )

								if self.args.tpu_metrics_debug or self.args.debug:
											# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
											xm.master_print(met.metrics_report()       )

								A__	  =     self.callback_handler.on_evaluate(self.args					,       self.state					,       self.control					,       _snake_case       )
								return metrics



					def      _a				(  self       :							Tuple					,       _snake_case       :							Union[str, Any]					,       _snake_case       :							Optional[int]					,       _snake_case       :							str=None					,       _snake_case       :							str = "test"					,       **_snake_case       :							List[str]       ):






								"""simple docstring"""


								A__	  =     gen_kwargs.copy()

								A__	  =     self.get_test_dataloader(_snake_case       )

								# Temporarily disable metric computation, we will do it in the loop here.
								A__	  =     self.compute_metrics
								A__	  =     None
								A__	  =     time.time()
								A__	  =     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
								try:
											A__	  =     eval_loop(
											    _snake_case					,       description='Prediction'					,       prediction_loss_only=True if compute_metrics is None else None					,       ignore_keys=_snake_case					,       metric_key_prefix=_snake_case					,       )
								finally:
											A__	  =     compute_metrics

								A__	  =     self.args.eval_batch_size * self.args.world_size
								if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
											start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
								output.metrics.update(
								    speed_metrics(
								        _snake_case					,       _snake_case					,       num_samples=output.num_samples					,       num_steps=math.ceil(output.num_samples / total_batch_size       )					,       )       )
								if self.post_process_function is None or self.compute_metrics is None:
											return output

								A__	  =     self.post_process_function(_snake_case					,       _snake_case					,       _snake_case					,       'predict'       )
								A__	  =     self.compute_metrics(_snake_case       )

								# Prefix all keys with metric_key_prefix + '_'
								for key in list(metrics.keys()       ):
											if not key.startswith(F'''{metric_key_prefix}_'''       ):
														A__	  =     metrics.pop(_snake_case       )
								metrics.update(output.metrics       )
								return PredictionOutput(predictions=predictions.predictions					,       label_ids=predictions.label_ids					,       metrics=_snake_case       )









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


SCREAMING_SNAKE_CASE__   			=   logging.get_logger(__name__)


class   __lowerCAmelCase  (		UpperCAmelCase_   ):
					"""simple docstring"""






					A__					:		Optional[Any]      							=     ["input_values", "attention_mask"]



					def __init__(  self       :							str					,       _snake_case       :							int = 1					,       _snake_case       :							int = 1_60_00					,       _snake_case       :							float = 0.0					,       _snake_case       :							bool = False					,       _snake_case       :							int = 80					,       _snake_case       :							int = 16					,       _snake_case       :							int = 64					,       _snake_case       :							str = "hann_window"					,       _snake_case       :							float = 1.0					,       _snake_case       :							float = 80					,       _snake_case       :							float = 76_00					,       _snake_case       :							float = 1E-10					,       _snake_case       :							int = 2					,       _snake_case       :							bool = True					,       **_snake_case       :							Union[str, Any]					,       ):






								"""simple docstring"""


								super().__init__(feature_size=_snake_case					,       sampling_rate=_snake_case					,       padding_value=_snake_case					,       **_snake_case       )
								A__	  =     do_normalize
								A__	  =     return_attention_mask

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

								A__	  =     win_length * sampling_rate // 10_00
								A__	  =     hop_length * sampling_rate // 10_00
								A__	  =     optimal_fft_length(self.sample_size       )
								A__	  =     (self.n_fft // 2) + 1

								A__	  =     window_function(window_length=self.sample_size					,       name=self.win_function					,       periodic=_snake_case       )

								A__	  =     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'					,       _snake_case					,       )
								if reduction_factor != 2.0:
											warnings.warn(
											    'The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'					,       _snake_case					,       )



					@staticmethod
					# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
					def      _a				(  _snake_case       :							List[np.ndarray]					,       _snake_case       :							List[np.ndarray]					,       _snake_case       :							float = 0.0       ):






								"""simple docstring"""


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

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

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

								return normed_input_values



					def      _a				(  self       :							Tuple					,       _snake_case       :							np.ndarray					,       ):






								"""simple docstring"""


								A__	  =     spectrogram(
								    _snake_case					,       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       :							List[str]					,       _snake_case       :							Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None					,       _snake_case       :							Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None					,       _snake_case       :							Union[bool, str, PaddingStrategy] = False					,       _snake_case       :							Optional[int] = None					,       _snake_case       :							bool = False					,       _snake_case       :							Optional[int] = None					,       _snake_case       :							Optional[bool] = None					,       _snake_case       :							Optional[Union[str, TensorType]] = None					,       _snake_case       :							Optional[int] = None					,       **_snake_case       :							Tuple					,       ):






								"""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:
											A__	  =     self._process_audio(
											    _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       **_snake_case					,       )
								else:
											A__	  =     None

								if audio_target is not None:
											A__	  =     self._process_audio(
											    _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       _snake_case					,       **_snake_case					,       )

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

								return inputs



					def      _a				(  self       :							Tuple					,       _snake_case       :							Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]					,       _snake_case       :							bool = False					,       _snake_case       :							Union[bool, str, PaddingStrategy] = False					,       _snake_case       :							Optional[int] = None					,       _snake_case       :							bool = False					,       _snake_case       :							Optional[int] = None					,       _snake_case       :							Optional[bool] = None					,       _snake_case       :							Optional[Union[str, TensorType]] = None					,       **_snake_case       :							Tuple					,       ):






								"""simple docstring"""


								A__	  =     isinstance(_snake_case					,       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}'''       )
								A__	  =     is_batched_numpy or (
								    isinstance(_snake_case					,       (list, tuple)       ) and (isinstance(speech[0]					,       (np.ndarray, tuple, list)       ))
								)

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

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

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

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

								A__	  =     self.pad(
								    _snake_case					,       padding=_snake_case					,       max_length=_snake_case					,       truncation=_snake_case					,       pad_to_multiple_of=_snake_case					,       return_attention_mask=_snake_case					,       **_snake_case					,       )

								A__	  =     feature_size_hack

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

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

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

								if return_tensors is not None:
											A__	  =     padded_inputs.convert_to_tensors(_snake_case       )

								return padded_inputs



					def      _a				(  self       :							Optional[Any]       ):






								"""simple docstring"""


								A__	  =     super().to_dict()

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

								return output




