
'''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


UpperCAmelCase__ :			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  (						SCREAMING_SNAKE_CASE__     ):





	def __init__(       self      :							Dict   ,				*__magic_name__      :							Optional[int]   ,				__magic_name__      :							Dict=None   ,				__magic_name__      :							Union[str, Any]=None   ,				__magic_name__      :							Any=None   ,				**__magic_name__      :							Union[str, Any]				):


				"""simple docstring"""

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





	def 				__SCREAMING_SNAKE_CASE	(       self      :							Any   ,				__magic_name__      :							int=None				):


				"""simple docstring"""

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

				lowerCAmelCase__										=				self._remove_unused_columns(__magic_name__   ,				description="Calibration"				)

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





	def 				__SCREAMING_SNAKE_CASE	(       self      :							Optional[int]   ,				__magic_name__      :							List[Any]=None				):


				"""simple docstring"""

				lowerCAmelCase__										=				self.train_dataset if calib_dataset is None else calib_dataset
				lowerCAmelCase__										=				self.get_calib_dataloader(__magic_name__				)

				lowerCAmelCase__										=				self.model
				quant_trainer.configure_model(__magic_name__   ,				self.quant_trainer_args   ,				calib=__magic_name__				)
				model.eval()
				quant_trainer.enable_calibration(__magic_name__				)

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

				quant_trainer.finish_calibration(__magic_name__   ,				self.quant_trainer_args				)
				lowerCAmelCase__										=				model





	def 				__SCREAMING_SNAKE_CASE	(       self      :							Union[str, Any]   ,				__magic_name__      :							Optional[Any]=None   ,				__magic_name__      :							Optional[int]=None   ,				__magic_name__      :							Dict=None   ,				__magic_name__      :							str = "eval"				):


				"""simple docstring"""

				lowerCAmelCase__										=				self.eval_dataset if eval_dataset is None else eval_dataset
				lowerCAmelCase__										=				self.get_eval_dataloader(__magic_name__				)
				lowerCAmelCase__										=				self.eval_examples if eval_examples is None else eval_examples

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

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

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

							self.log(__magic_name__				)
				else:
							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()				)

				lowerCAmelCase__										=				self.callback_handler.on_evaluate(self.args   ,				self.state   ,				self.control   ,				__magic_name__				)
				return metrics





	def 				__SCREAMING_SNAKE_CASE	(       self      :							Dict   ,				__magic_name__      :							Tuple   ,				__magic_name__      :							Dict   ,				__magic_name__      :							Any=None   ,				__magic_name__      :							str = "test"				):


				"""simple docstring"""

				lowerCAmelCase__										=				self.get_test_dataloader(__magic_name__				)

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

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

				lowerCAmelCase__										=				self.post_process_function(__magic_name__   ,				__magic_name__   ,				output.predictions   ,				"predict"				)
				lowerCAmelCase__										=				self.compute_metrics(__magic_name__				)

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

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







	def 				__SCREAMING_SNAKE_CASE	(       self      :							Tuple   ,				__magic_name__      :							Optional[Any]="./"				):


				"""simple docstring"""

				lowerCAmelCase__										=				self.eval_dataset
				lowerCAmelCase__										=				self.get_eval_dataloader(__magic_name__				)

				lowerCAmelCase__										=				next(iter(__magic_name__				)				)

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

				# convert to tuple
				lowerCAmelCase__										=				tuple(v.to(__magic_name__				) for k, v in batch.items()				)

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

				lowerCAmelCase__										=				True

				lowerCAmelCase__										=				self.model.to(__magic_name__				)

				model.eval()
				model.float()

				lowerCAmelCase__										=				model.module if hasattr(__magic_name__   ,				"module"				) else model
				quant_trainer.configure_model(__magic_name__   ,				self.quant_trainer_args				)

				lowerCAmelCase__										=				os.path.join(__magic_name__   ,				"model.onnx"				)
				logger.info(f"""exporting model to {output_model_file}"""				)

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

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








def 			lowerCAmelCase(      SCREAMING_SNAKE_CASE_			)->						str:



    """simple docstring"""






    if isinstance(SCREAMING_SNAKE_CASE_			,						SCREAMING_SNAKE_CASE_			):
        raise TypeError("'float' object cannot be interpreted as an integer"			)
    if isinstance(SCREAMING_SNAKE_CASE_			,						SCREAMING_SNAKE_CASE_			):
        raise TypeError("'str' object cannot be interpreted as an integer"			)

    if num == 0:
        return "0b0"

    UpperCamelCase_			  =						False

    if num < 0:
        UpperCamelCase_			  =						True
        UpperCamelCase_			  =						-num

    UpperCamelCase_			  =						[]
    while num > 0:
        binary.insert(0			,						num % 2			)
        num >>= 1

    if negative:
        return "-0b" + "".join(str(SCREAMING_SNAKE_CASE_			) for e in binary			)

    return "0b" + "".join(str(SCREAMING_SNAKE_CASE_			) for e in binary			)


if __name__ == "__main__":
 import doctest

 doctest.testmod()
