



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


snake_case    :						Optional[Any]  =						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 							lowerCAmelCase__  (	UpperCamelCase						):







					def __init__(       self      :     List[str]			,		*_A      :     Dict			,		_A      :     Optional[int]=None			,		_A      :     List[str]=None			,		_A      :     Optional[int]=None			,		**_A      :     str):
											super().__init__(*_A			,		**_A)
											A__	:							List[Any]						    = eval_examples
											A__	:							Optional[Any]						    = post_process_function
											A__	:							str						    = quant_trainer_args
											A__	:							List[str]						    = 128  # default number of calibration samples







					def  _lowercase			(       self      :     List[str]			,		_A      :     Union[str, Any]=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__	:							List[Any]						    = 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  _lowercase			(       self      :     Union[str, Any]			,		_A      :     Dict=None):
											A__	:							int						    = self.train_dataset if calib_dataset is None else calib_dataset
											A__	:							Tuple						    = self.get_calib_dataloader(_A)

											A__	:							Union[str, 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__	:							str						    = 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__	:							List[str]						    = model







					def  _lowercase			(       self      :     List[Any]			,		_A      :     Union[str, Any]=None			,		_A      :     Optional[int]=None			,		_A      :     Optional[int]=None			,		_A      :     str = "eval"):
											A__	:							Union[str, Any]						    = self.eval_dataset if eval_dataset is None else eval_dataset
											A__	:							str						    = self.get_eval_dataloader(_A)
											A__	:							Optional[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__	:							Optional[Any]						    = self.compute_metrics
											A__	:							List[Any]						    = None
											A__	:							int						    = self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																	A__	:							Dict						    = eval_loop(
																	    _A			,		description="Evaluation"			,		prediction_loss_only=True if compute_metrics is None else None			,		ignore_keys=_A			,		)
											finally:
																	A__	:							int						    = compute_metrics

											if self.post_process_function is not None and self.compute_metrics is not None:
																	A__	:							Optional[int]						    = self.post_process_function(_A			,		_A			,		output.predictions)
																	A__	:							str						    = 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__	:							int						    = {}

											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__	:							Tuple						    = self.callback_handler.on_evaluate(self.args			,		self.state			,		self.control			,		_A)
											return metrics







					def  _lowercase			(       self      :     Optional[Any]			,		_A      :     List[str]			,		_A      :     Any			,		_A      :     str=None			,		_A      :     str = "test"):
											A__	:							List[str]						    = self.get_test_dataloader(_A)

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

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

											A__	:							Optional[Any]						    = self.post_process_function(_A			,		_A			,		output.predictions			,		"predict")
											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__	:							Optional[int]						    = metrics.pop(_A)

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







					def  _lowercase			(       self      :     Dict			,		_A      :     List[str]="./"):
											A__	:							Any						    = self.eval_dataset
											A__	:							str						    = self.get_eval_dataloader(_A)

											A__	:							List[str]						    = next(iter(_A))

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

											# convert to tuple
											A__	:							Any						    = 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__	:							int						    = True

											A__	:							List[str]						    = self.model.to(_A)

											model.eval()
											model.float()

											A__	:							List[str]						    = 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__	:							Any						    = {0: "batch_size", 1: "seq_len"}

											torch.onnx.export(
											    _A			,		_A			,		_A			,		export_params=_A			,		opset_version=13			,		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")




from collections.abc import Callable

import numpy as np
def 	snake_case__       (					__lowercase							,     __lowercase							,     __lowercase							,     __lowercase							,     __lowercase       )		->		np.array:







						"""simple docstring"""

						A__	:							Any						    = int(np.ceil((x_end - xa) / step_size       )       )
						A__	:							Union[str, Any]						    = np.zeros((n + 1,)       )
						A__	:							Any						    = ya
						A__	:							Union[str, Any]						    = xa

						for k in range(__lowercase       ):
												A__	:							Any						    = y[k] + step_size * ode_func(__lowercase							,     y[k]       )
												A__	:							Any						    = y[k] + (
												    (step_size / 2) * (ode_func(__lowercase							,     y[k]       ) + ode_func(x + step_size							,     __lowercase       ))
												)
												x += step_size

						return y


if __name__ == "__main__":
		import doctest

		doctest.testmod()