



"""simple docstring"""
import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


SCREAMING_SNAKE_CASE_				= logging.getLogger(__name__)

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




class 			snake_case_						(						a_				):






							def __init__(   self			,     *a_			,     a_=None			,     a_=None			,     a_=None			,     **a_      ):
									super().__init__(*a_			,     **a_      )
									a_     :      Tuple 						=     eval_examples
									a_     :      List[Any] 						=     post_process_function
									a_     :      Any 						=     quant_trainer_args
									a_     :      str 						=     1_2_8  # default number of calibration samples






							def snake_case_  (   self			,     a_=None      ):
									if calib_dataset is None and self.calib_dataset is None:
											raise ValueError("Trainer: calibration requires an calib_dataset."      )
									a_     :      str 						=     calib_dataset if calib_dataset is not None else self.calib_dataset

									a_     :      Dict 						=     self._remove_unused_columns(a_			,     description="Calibration"      )

									return DataLoader(
									    a_			,     batch_size=self.args.eval_batch_size			,     collate_fn=self.data_collator			,     drop_last=self.args.dataloader_drop_last			,     num_workers=self.args.dataloader_num_workers			,     pin_memory=self.args.dataloader_pin_memory			,     shuffle=a_			,     )






							def snake_case_  (   self			,     a_=None      ):
									a_     :      Union[str, Any] 						=     self.train_dataset if calib_dataset is None else calib_dataset
									a_     :      Dict 						=     self.get_calib_dataloader(a_      )

									a_     :      Optional[Any] 						=     self.model
									quant_trainer.configure_model(a_			,     self.quant_trainer_args			,     calib=a_      )
									model.eval()
									quant_trainer.enable_calibration(a_      )

									logger.info("***** Running calibration *****"      )
									logger.info(F"""  Num examples = {self.calib_num}"""      )
									logger.info(F"""  Batch size = {calib_dataloader.batch_size}"""      )

									for step, inputs in enumerate(a_      ):
											# Prediction step
											a_						,				a_						,				a_     :      Union[str, Any] 						=     self.prediction_step(a_			,     a_			,     prediction_loss_only=a_      )
											if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
													break

									quant_trainer.finish_calibration(a_			,     self.quant_trainer_args      )
									a_     :      Optional[Any] 						=     model






							def snake_case_  (   self			,     a_=None			,     a_=None			,     a_=None			,     a_ = "eval"      ):
									a_     :      Tuple 						=     self.eval_dataset if eval_dataset is None else eval_dataset
									a_     :      Union[str, Any] 						=     self.get_eval_dataloader(a_      )
									a_     :      Union[str, Any] 						=     self.eval_examples if eval_examples is None else eval_examples

									# Temporarily disable metric computation, we will do it in the loop here.
									a_     :      List[Any] 						=     self.compute_metrics
									a_     :      Dict 						=     None
									a_     :      Dict 						=     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
									try:
											a_     :      Optional[Any] 						=     eval_loop(
											    a_			,     description="Evaluation"			,     prediction_loss_only=True if compute_metrics is None else None			,     ignore_keys=a_			,     )
									finally:
											a_     :      Union[str, Any] 						=     compute_metrics

									if self.post_process_function is not None and self.compute_metrics is not None:
											a_     :      Union[str, Any] 						=     self.post_process_function(a_			,     a_			,     output.predictions      )
											a_     :      Union[str, Any] 						=     self.compute_metrics(a_      )

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

											self.log(a_      )
									else:
											a_     :      Optional[Any] 						=     {}

									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_     :      List[Any] 						=     self.callback_handler.on_evaluate(self.args			,     self.state			,     self.control			,     a_      )
									return metrics






							def snake_case_  (   self			,     a_			,     a_			,     a_=None			,     a_ = "test"      ):
									a_     :      int 						=     self.get_test_dataloader(a_      )

									# Temporarily disable metric computation, we will do it in the loop here.
									a_     :      Optional[int] 						=     self.compute_metrics
									a_     :      Optional[int] 						=     None
									a_     :      Any 						=     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
									try:
											a_     :      int 						=     eval_loop(
											    a_			,     description="Prediction"			,     prediction_loss_only=True if compute_metrics is None else None			,     ignore_keys=a_			,     )
									finally:
											a_     :      Tuple 						=     compute_metrics

									if self.post_process_function is None or self.compute_metrics is None:
											return output

									a_     :      List[str] 						=     self.post_process_function(a_			,     a_			,     output.predictions			,     "predict"      )
									a_     :      int 						=     self.compute_metrics(a_      )

									# Prefix all keys with metric_key_prefix + '_'
									for key in list(metrics.keys()      ):
											if not key.startswith(F"""{metric_key_prefix}_"""      ):
													a_     :      Optional[int] 						=     metrics.pop(a_      )

									return PredictionOutput(predictions=predictions.predictions			,     label_ids=predictions.label_ids			,     metrics=a_      )






							def snake_case_  (   self			,     a_="./"      ):
									a_     :      Any 						=     self.eval_dataset
									a_     :      Union[str, Any] 						=     self.get_eval_dataloader(a_      )

									a_     :      Any 						=     next(iter(a_      )      )

									# saving device - to make it consistent
									a_     :      str 						=     torch.device("cuda" if torch.cuda.is_available() else "cpu"      )

									# convert to tuple
									a_     :      Tuple 						=     tuple(v.to(a_      ) for k, v in batch.items()      )

									logger.info("Converting model to be onnx compatible"      )
									from pytorch_quantization.nn import TensorQuantizer

									a_     :      Optional[int] 						=     True

									a_     :      int 						=     self.model.to(a_      )

									model.eval()
									model.float()

									a_     :      int 						=     model.module if hasattr(a_			,     "module"      ) else model
									quant_trainer.configure_model(a_			,     self.quant_trainer_args      )

									a_     :      Optional[int] 						=     os.path.join(a_			,     "model.onnx"      )
									logger.info(F"""exporting model to {output_model_file}"""      )

									a_     :      Union[str, Any] 						=     {0: "batch_size", 1: "seq_len"}

									torch.onnx.export(
									    a_			,     a_			,     a_			,     export_params=a_			,     opset_version=1_3			,     do_constant_folding=a_			,     input_names=["input_ids", "attention_mask", "token_type_ids"]			,     output_names=["output_start_logits", "output_end_logits"]			,     dynamic_axes={
									        "input_ids": axes,
									        "attention_mask": axes,
									        "token_type_ids": axes,
									        "output_start_logits": axes,
									        "output_end_logits": axes,
									    }			,     verbose=a_			,     )
									logger.info("onnx export finished"      )




"""simple docstring"""
import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
					import torch

if is_datasets_available():
					from datasets import load_dataset

SCREAMING_SNAKE_CASE_				= random.Random()



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=1.0, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None	)			->      Tuple:
		if rng is None:
				a_     :      Any 						=     global_rng

		a_     :      Dict 						=     []
		for batch_idx in range(shape[0]	):
				values.append([]	)
				for _ in range(shape[1]	):
						values[-1].append(rng.random() * scale	)

		return values




class 			snake_case_						(						unittest.TestCase				):






							def __init__(   self			,     a_			,     a_=7			,     a_=4_0_0			,     a_=2_0_0_0			,     a_=2_0_4_8			,     a_=1_2_8			,     a_=1			,     a_=5_1_2			,     a_=3_0			,     a_=4_4_1_0_0			,     ):
									a_     :      Optional[Any] 						=     parent
									a_     :      Tuple 						=     batch_size
									a_     :      Union[str, Any] 						=     min_seq_length
									a_     :      Dict 						=     max_seq_length
									a_     :      Union[str, Any] 						=     (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
									a_     :      int 						=     spectrogram_length
									a_     :      Optional[Any] 						=     feature_size
									a_     :      Optional[int] 						=     num_audio_channels
									a_     :      List[str] 						=     hop_length
									a_     :      Optional[int] 						=     chunk_length
									a_     :      List[str] 						=     sampling_rate






							def snake_case_  (   self      ):
									return {
									    "spectrogram_length": self.spectrogram_length,
									    "feature_size": self.feature_size,
									    "num_audio_channels": self.num_audio_channels,
									    "hop_length": self.hop_length,
									    "chunk_length": self.chunk_length,
									    "sampling_rate": self.sampling_rate,
									}






							def snake_case_  (   self			,     a_=False			,     a_=False      ):
									def _flatten(a_      ):
											return list(itertools.chain(*a_      )      )

									if equal_length:
											a_     :      Tuple 						=     [floats_list((self.max_seq_length, self.feature_size)      ) for _ in range(self.batch_size      )]
									else:
											# make sure that inputs increase in size
											a_     :      Optional[Any] 						=     [
											    floats_list((x, self.feature_size)      )
											    for x in range(self.min_seq_length			,     self.max_seq_length			,     self.seq_length_diff      )
											]
									if numpify:
											a_     :      Tuple 						=     [np.asarray(a_      ) for x in speech_inputs]
									return speech_inputs




@require_torch
@require_torchaudio
class 			snake_case_						(						a_     ,unittest.TestCase				):
							__lowerCAmelCase    					=					TvltFeatureExtractor






							def snake_case_  (   self      ):
									a_     :      Dict 						=     TvltFeatureExtractionTester(self      )






							def snake_case_  (   self      ):
									a_     :      Optional[Any] 						=     self.feature_extraction_class(**self.feat_extract_dict      )
									self.assertTrue(hasattr(a_			,     "spectrogram_length"      )      )
									self.assertTrue(hasattr(a_			,     "feature_size"      )      )
									self.assertTrue(hasattr(a_			,     "num_audio_channels"      )      )
									self.assertTrue(hasattr(a_			,     "hop_length"      )      )
									self.assertTrue(hasattr(a_			,     "chunk_length"      )      )
									self.assertTrue(hasattr(a_			,     "sampling_rate"      )      )






							def snake_case_  (   self      ):
									a_     :      List[Any] 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									with tempfile.TemporaryDirectory() as tmpdirname:
											a_     :      List[Any] 						=     feat_extract_first.save_pretrained(a_      )[0]
											check_json_file_has_correct_format(a_      )
											a_     :      Tuple 						=     self.feature_extraction_class.from_pretrained(a_      )

									a_     :      Dict 						=     feat_extract_first.to_dict()
									a_     :      List[str] 						=     feat_extract_second.to_dict()
									a_     :      List[Any] 						=     dict_first.pop("mel_filters"      )
									a_     :      str 						=     dict_second.pop("mel_filters"      )
									self.assertTrue(np.allclose(a_			,     a_      )      )
									self.assertEqual(a_			,     a_      )






							def snake_case_  (   self      ):
									a_     :      List[str] 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									with tempfile.TemporaryDirectory() as tmpdirname:
											a_     :      Dict 						=     os.path.join(a_			,     "feat_extract.json"      )
											feat_extract_first.to_json_file(a_      )
											a_     :      str 						=     self.feature_extraction_class.from_json_file(a_      )

									a_     :      Dict 						=     feat_extract_first.to_dict()
									a_     :      str 						=     feat_extract_second.to_dict()
									a_     :      int 						=     dict_first.pop("mel_filters"      )
									a_     :      List[Any] 						=     dict_second.pop("mel_filters"      )
									self.assertTrue(np.allclose(a_			,     a_      )      )
									self.assertEqual(a_			,     a_      )






							def snake_case_  (   self      ):
									# Initialize feature_extractor
									a_     :      str 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									# create three inputs of length 800, 1000, and 1200
									a_     :      Optional[Any] 						=     [floats_list((1, x)      )[0] for x in range(8_0_0			,     1_4_0_0			,     2_0_0      )]
									a_     :      Union[str, Any] 						=     [np.asarray(a_      ) for speech_input in speech_inputs]

									# Test not batched input
									a_     :      Dict 						=     feature_extractor(np_speech_inputs[0]			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0      ).audio_values

									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )

									# Test batched
									a_     :      int 						=     feature_extractor(a_			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0      ).audio_values

									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )

									# Test audio masking
									a_     :      Optional[int] 						=     feature_extractor(
									    a_			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0			,     mask_audio=a_      ).audio_values

									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )

									# Test 2-D numpy arrays are batched.
									a_     :      List[str] 						=     [floats_list((1, x)      )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
									a_     :      Union[str, Any] 						=     np.asarray(a_      )
									a_     :      Dict 						=     feature_extractor(a_			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0      ).audio_values
									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )






							def snake_case_  (   self			,     a_      ):
									a_     :      Optional[int] 						=     load_dataset("hf-internal-testing/librispeech_asr_dummy"			,     "clean"			,     split="validation"      )
									# automatic decoding with librispeech
									a_     :      List[str] 						=     ds.sort("id"      ).select(range(a_      )      )[:num_samples]["audio"]

									return [x["array"] for x in speech_samples]






							def snake_case_  (   self      ):
									a_     :      List[str] 						=     self._load_datasamples(1      )
									a_     :      Any 						=     TvltFeatureExtractor()
									a_     :      Union[str, Any] 						=     feature_extractor(a_			,     return_tensors="pt"      ).audio_values

									self.assertEquals(audio_values.shape			,     (1, 1, 1_9_2, 1_2_8)      )

									a_     :      Dict 						=     torch.tensor([[-0.3_032, -0.2_708], [-0.4_434, -0.4_007]]      )
									self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]			,     a_			,     atol=1e-4      )      )