


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


A			:						List[str]        =  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 				A       (			UpperCAmelCase__				):
					'''simple docstring'''






					def __init__(self       :		Dict ,   *_UpperCAmelCase       :		Optional[int] ,   _UpperCAmelCase       :		Union[str, Any]=None ,   _UpperCAmelCase       :		Dict=None ,   _UpperCAmelCase       :		Optional[int]=None ,   **_UpperCAmelCase       :		List[Any]						)		->  Optional[Any]:




							"""simple docstring"""


							super().__init__(*_UpperCAmelCase ,   **_UpperCAmelCase						)
							lowercase__	     =						eval_examples
							lowercase__	     =						post_process_function
							lowercase__	     =						quant_trainer_args
							lowercase__	     =						128  # default number of calibration samples

					def     lowerCamelCase__	(self       :		int ,   _UpperCAmelCase       :		Dict=None						)		->  List[Any]:




							"""simple docstring"""


							if calib_dataset is None and self.calib_dataset is None:
									raise ValueError("""Trainer: calibration requires an calib_dataset."""						)
							lowercase__	     =						calib_dataset if calib_dataset is not None else self.calib_dataset

							lowercase__	     =						self._remove_unused_columns(_UpperCAmelCase ,   description="""Calibration"""						)

							return DataLoader(
							    _UpperCAmelCase ,   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=_UpperCAmelCase ,   )

					def     lowerCamelCase__	(self       :		Tuple ,   _UpperCAmelCase       :		Dict=None						)		->  Optional[int]:




							"""simple docstring"""


							lowercase__	     =						self.train_dataset if calib_dataset is None else calib_dataset
							lowercase__	     =						self.get_calib_dataloader(_UpperCAmelCase						)

							lowercase__	     =						self.model
							quant_trainer.configure_model(_UpperCAmelCase ,   self.quant_trainer_args ,   calib=_UpperCAmelCase						)
							model.eval()
							quant_trainer.enable_calibration(_UpperCAmelCase						)

							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(_UpperCAmelCase						):
									# Prediction step
									lowercase__     ,      lowercase__     ,      lowercase__	     =						self.prediction_step(_UpperCAmelCase ,   _UpperCAmelCase ,   prediction_loss_only=_UpperCAmelCase						)
									if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
											break

							quant_trainer.finish_calibration(_UpperCAmelCase ,   self.quant_trainer_args						)
							lowercase__	     =						model

					def     lowerCamelCase__	(self       :		Optional[int] ,   _UpperCAmelCase       :		str=None ,   _UpperCAmelCase       :		Optional[int]=None ,   _UpperCAmelCase       :		List[Any]=None ,   _UpperCAmelCase       :		str = "eval"						)		->  List[Any]:




							"""simple docstring"""


							lowercase__	     =						self.eval_dataset if eval_dataset is None else eval_dataset
							lowercase__	     =						self.get_eval_dataloader(_UpperCAmelCase						)
							lowercase__	     =						self.eval_examples if eval_examples is None else eval_examples

							# Temporarily disable metric computation, we will do it in the loop here.
							lowercase__	     =						self.compute_metrics
							lowercase__	     =						None
							lowercase__	     =						self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
							try:
									lowercase__	     =						eval_loop(
									    _UpperCAmelCase ,   description="""Evaluation""" ,   prediction_loss_only=True if compute_metrics is None else None ,   ignore_keys=_UpperCAmelCase ,   )
							finally:
									lowercase__	     =						compute_metrics

							if self.post_process_function is not None and self.compute_metrics is not None:
									lowercase__	     =						self.post_process_function(_UpperCAmelCase ,   _UpperCAmelCase ,   output.predictions						)
									lowercase__	     =						self.compute_metrics(_UpperCAmelCase						)

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

									self.log(_UpperCAmelCase						)
							else:
									lowercase__	     =						{}

							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()						)

							lowercase__	     =						self.callback_handler.on_evaluate(self.args ,   self.state ,   self.control ,   _UpperCAmelCase						)
							return metrics

					def     lowerCamelCase__	(self       :		Union[str, Any] ,   _UpperCAmelCase       :		Tuple ,   _UpperCAmelCase       :		List[Any] ,   _UpperCAmelCase       :		Any=None ,   _UpperCAmelCase       :		str = "test"						)		->  int:




							"""simple docstring"""


							lowercase__	     =						self.get_test_dataloader(_UpperCAmelCase						)

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

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

							lowercase__	     =						self.post_process_function(_UpperCAmelCase ,   _UpperCAmelCase ,   output.predictions ,   """predict"""						)
							lowercase__	     =						self.compute_metrics(_UpperCAmelCase						)

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

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


					def     lowerCamelCase__	(self       :		List[Any] ,   _UpperCAmelCase       :		List[str]="./"						)		->  List[str]:




							"""simple docstring"""


							lowercase__	     =						self.eval_dataset
							lowercase__	     =						self.get_eval_dataloader(_UpperCAmelCase						)

							lowercase__	     =						next(iter(_UpperCAmelCase						)						)

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

							# convert to tuple
							lowercase__	     =						tuple(v.to(_UpperCAmelCase						) for k, v in batch.items()						)

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

							lowercase__	     =						True

							lowercase__	     =						self.model.to(_UpperCAmelCase						)

							model.eval()
							model.float()

							lowercase__	     =						model.module if hasattr(_UpperCAmelCase ,   """module"""						) else model
							quant_trainer.configure_model(_UpperCAmelCase ,   self.quant_trainer_args						)

							lowercase__	     =						os.path.join(_UpperCAmelCase ,   """model.onnx"""						)
							logger.info(f'''exporting model to {output_model_file}'''						)

							lowercase__	     =						{0: """batch_size""", 1: """seq_len"""}

							torch.onnx.export(
							    _UpperCAmelCase ,   _UpperCAmelCase ,   _UpperCAmelCase ,   export_params=_UpperCAmelCase ,   opset_version=13 ,   do_constant_folding=_UpperCAmelCase ,   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=_UpperCAmelCase ,   )
							logger.info("""onnx export finished"""						)





# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.speechta import SpeechTaForTextToSpeech, SpeechTaHifiGan, SpeechTaProcessor
from ..utils import is_datasets_available
from .base import PipelineTool


if is_datasets_available():
				from datasets import load_dataset







class 	__lowercase    (__SCREAMING_SNAKE_CASE  ):





							"""simple docstring"""







							_UpperCAmelCase      =						"""microsoft/speecht5_tts"""
							_UpperCAmelCase      =						(
							    """This is a tool that reads an English text out loud. It takes an input named `text` which should contain the """
							    """text to read (in English) and returns a waveform object containing the sound."""
							)
							_UpperCAmelCase      =						"""text_reader"""
							_UpperCAmelCase      =						SpeechTaProcessor
							_UpperCAmelCase      =						SpeechTaForTextToSpeech
							_UpperCAmelCase      =						SpeechTaHifiGan

							_UpperCAmelCase      =						["""text"""]
							_UpperCAmelCase      =						["""audio"""]






							def 					UpperCamelCase__     (      self       ):





										"""simple docstring"""




										if self.post_processor is None:
													SCREAMING_SNAKE_CASE_		:     List[Any]					=					'microsoft/speecht5_hifigan'
										super().setup()






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__			,						lowerCAmelCase__=None       ):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_		:     int					=					self.pre_processor(text=lowerCAmelCase__			,						return_tensors='pt'			,						truncation=lowerCAmelCase__       )

										if speaker_embeddings is None:
													if not is_datasets_available():
																raise ImportError('Datasets needs to be installed if not passing speaker embeddings.'       )

													SCREAMING_SNAKE_CASE_		:     List[Any]					=					load_dataset('Matthijs/cmu-arctic-xvectors'			,						split='validation'       )
													SCREAMING_SNAKE_CASE_		:     Optional[int]					=					torch.tensor(embeddings_dataset[7_3_0_5]['xvector']       ).unsqueeze(0       )

										return {"input_ids": inputs["input_ids"], "speaker_embeddings": speaker_embeddings}






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__       ):





										"""simple docstring"""




										with torch.no_grad():
													return self.model.generate_speech(**lowerCAmelCase__       )






							def 					UpperCamelCase__     (      self			,						lowerCAmelCase__       ):





										"""simple docstring"""




										with torch.no_grad():
													return self.post_processor(lowerCAmelCase__       ).cpu().detach()





