





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     A     (SCREAMING_SNAKE_CASE       ):
			'''simple docstring'''




			def __init__(     self  :       Any				,      *__lowerCAmelCase  :       str				,      __lowerCAmelCase  :       str=None				,      __lowerCAmelCase  :       int=None				,      **__lowerCAmelCase  :       Optional[Any]  )      ->      Optional[int]:







					"""simple docstring"""






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






			def   a_			(     self  :       Union[str, Any]				,      __lowerCAmelCase  :       Optional[Dataset] = None				,      __lowerCAmelCase  :       int=None				,      __lowerCAmelCase  :       Optional[List[str]] = None				,      __lowerCAmelCase  :       str = "eval"				,      **__lowerCAmelCase  :       str				,      )      ->      Dict[str, float]:







					"""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(__lowerCAmelCase  )
					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(
							    __lowerCAmelCase				,      description="""Evaluation"""				,      prediction_loss_only=True if compute_metrics is None else None				,      ignore_keys=__lowerCAmelCase				,      metric_key_prefix=__lowerCAmelCase				,      )
					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(
					        __lowerCAmelCase				,      __lowerCAmelCase				,      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(__lowerCAmelCase				,      __lowerCAmelCase				,      __lowerCAmelCase  )
							A__     =  self.compute_metrics(__lowerCAmelCase  )

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

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

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

					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				,      __lowerCAmelCase  )
					return metrics







			def   a_			(     self  :       Optional[Any]				,      __lowerCAmelCase  :       List[str]				,      __lowerCAmelCase  :       Optional[Any]				,      __lowerCAmelCase  :       Optional[Any]=None				,      __lowerCAmelCase  :       str = "test"				,      **__lowerCAmelCase  :       Dict  )      ->      Optional[Any]:







					"""simple docstring"""






					A__     =  gen_kwargs.copy()

					A__     =  self.get_test_dataloader(__lowerCAmelCase  )

					# 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(
							    __lowerCAmelCase				,      description="""Prediction"""				,      prediction_loss_only=True if compute_metrics is None else None				,      ignore_keys=__lowerCAmelCase				,      metric_key_prefix=__lowerCAmelCase				,      )
					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(
					        __lowerCAmelCase				,      __lowerCAmelCase				,      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(__lowerCAmelCase				,      __lowerCAmelCase				,      __lowerCAmelCase				,      """predict"""  )
					A__     =  self.compute_metrics(__lowerCAmelCase  )

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










import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer


class     A     (SCREAMING_SNAKE_CASE       ):
			'''simple docstring'''




			__lowerCamelCase      :			Dict							     =       '''Wav2Vec2FeatureExtractor'''
			__lowerCamelCase      :			str							     =       '''AutoTokenizer'''






			def __init__(     self  :       Union[str, Any]				,      __lowerCAmelCase  :       List[Any]				,      __lowerCAmelCase  :       Optional[Any]  )      ->      Tuple:







					"""simple docstring"""






					super().__init__(__lowerCAmelCase				,      __lowerCAmelCase  )
					A__     =  self.feature_extractor
					A__     =  False






			@classmethod
			def   a_			(     cls  :       Union[str, Any]				,      __lowerCAmelCase  :       Optional[Any]				,      **__lowerCAmelCase  :       Union[str, Any]  )      ->      int:







					"""simple docstring"""






					try:
							return super().from_pretrained(__lowerCAmelCase				,      **__lowerCAmelCase  )
					except OSError:
							warnings.warn(
							    f'Loading a tokenizer inside {cls.__name__} from a config that does not'
							    """ include a `tokenizer_class` attribute is deprecated and will be """
							    """removed in v5. Please add `'tokenizer_class': 'Wav2Vec2CTCTokenizer'`"""
							    """ attribute to either your `config.json` or `tokenizer_config.json` """
							    """file to suppress this warning: """				,      __lowerCAmelCase				,      )

							A__     =  WavaVecaFeatureExtractor.from_pretrained(__lowerCAmelCase				,      **__lowerCAmelCase  )
							A__     =  WavaVecaCTCTokenizer.from_pretrained(__lowerCAmelCase				,      **__lowerCAmelCase  )

							return cls(feature_extractor=__lowerCAmelCase				,      tokenizer=__lowerCAmelCase  )






			def __call__(     self  :       Any				,      *__lowerCAmelCase  :       Union[str, Any]				,      **__lowerCAmelCase  :       List[Any]  )      ->      Optional[Any]:







					"""simple docstring"""






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

					if "raw_speech" in kwargs:
							warnings.warn("""Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."""  )
							A__     =  kwargs.pop("""raw_speech"""  )
					else:
							A__     =  kwargs.pop("""audio"""				,      __lowerCAmelCase  )
					A__     =  kwargs.pop("""sampling_rate"""				,      __lowerCAmelCase  )
					A__     =  kwargs.pop("""text"""				,      __lowerCAmelCase  )
					if len(__lowerCAmelCase  ) > 0:
							A__     =  args[0]
							A__     =  args[1:]

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

					if audio is not None:
							A__     =  self.feature_extractor(__lowerCAmelCase				,      *__lowerCAmelCase				,      sampling_rate=__lowerCAmelCase				,      **__lowerCAmelCase  )
					if text is not None:
							A__     =  self.tokenizer(__lowerCAmelCase				,      **__lowerCAmelCase  )

					if text is None:
							return inputs
					elif audio is None:
							return encodings
					else:
							A__     =  encodings["""input_ids"""]
							return inputs






			def   a_			(     self  :       List[str]				,      *__lowerCAmelCase  :       Optional[Any]				,      **__lowerCAmelCase  :       Tuple  )      ->      List[str]:







					"""simple docstring"""






					if self._in_target_context_manager:
							return self.current_processor.pad(*__lowerCAmelCase				,      **__lowerCAmelCase  )

					A__     =  kwargs.pop("""input_features"""				,      __lowerCAmelCase  )
					A__     =  kwargs.pop("""labels"""				,      __lowerCAmelCase  )
					if len(__lowerCAmelCase  ) > 0:
							A__     =  args[0]
							A__     =  args[1:]

					if input_features is not None:
							A__     =  self.feature_extractor.pad(__lowerCAmelCase				,      *__lowerCAmelCase				,      **__lowerCAmelCase  )
					if labels is not None:
							A__     =  self.tokenizer.pad(__lowerCAmelCase				,      **__lowerCAmelCase  )

					if labels is None:
							return input_features
					elif input_features is None:
							return labels
					else:
							A__     =  labels["""input_ids"""]
							return input_features






			def   a_			(     self  :       Optional[int]				,      *__lowerCAmelCase  :       List[str]				,      **__lowerCAmelCase  :       List[Any]  )      ->      Dict:







					"""simple docstring"""






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






			def   a_			(     self  :       Any				,      *__lowerCAmelCase  :       List[str]				,      **__lowerCAmelCase  :       Union[str, Any]  )      ->      List[str]:







					"""simple docstring"""






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







			@contextmanager
			def   a_			(     self  :       Union[str, Any]  )      ->      int:







					"""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 audio inputs, or in a separate call."""  )
					A__     =  True
					A__     =  self.tokenizer
					yield
					A__     =  self.feature_extractor
					A__     =  False



