






"""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_     :				Dict         =				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__       (    __snake_case				):






						'''simple docstring'''

						def __init__(		self  :						List[Any]		,			*lowercase_  :						Any		,			lowercase_  :						Optional[Any]=None		,			lowercase_  :						Any=None		,			lowercase_  :						Optional[Any]=None		,			**lowercase_  :						Tuple):




										'''simple docstring'''


										super().__init__(*A_		,			**A_)
										SCREAMING_SNAKE_CASE_   :						Tuple							=	eval_examples
										SCREAMING_SNAKE_CASE_   :						Dict							=	post_process_function
										SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	quant_trainer_args
										SCREAMING_SNAKE_CASE_   :						Any							=	128  # default number of calibration samples
						def _SCREAMING_SNAKE_CASE				(		self  :						Tuple		,			lowercase_  :						Any=None):




										'''simple docstring'''


										if calib_dataset is None and self.calib_dataset is None:
														raise ValueError('''Trainer: calibration requires an calib_dataset.''')
										SCREAMING_SNAKE_CASE_   :						Any							=	calib_dataset if calib_dataset is not None else self.calib_dataset

										SCREAMING_SNAKE_CASE_   :						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 _SCREAMING_SNAKE_CASE				(		self  :						int		,			lowercase_  :						List[Any]=None):




										'''simple docstring'''


										SCREAMING_SNAKE_CASE_   :						Any							=	self.train_dataset if calib_dataset is None else calib_dataset
										SCREAMING_SNAKE_CASE_   :						Dict							=	self.get_calib_dataloader(A_)

										SCREAMING_SNAKE_CASE_   :						Optional[int]							=	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
														SCREAMING_SNAKE_CASE_   :						List[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)
										SCREAMING_SNAKE_CASE_   :						str							=	model
						def _SCREAMING_SNAKE_CASE				(		self  :						List[Any]		,			lowercase_  :						Dict=None		,			lowercase_  :						Tuple=None		,			lowercase_  :						List[str]=None		,			lowercase_  :						Tuple = "eval"):




										'''simple docstring'''


										SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	self.eval_dataset if eval_dataset is None else eval_dataset
										SCREAMING_SNAKE_CASE_   :						Any							=	self.get_eval_dataloader(A_)
										SCREAMING_SNAKE_CASE_   :						List[Any]							=	self.eval_examples if eval_examples is None else eval_examples

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

										if self.post_process_function is not None and self.compute_metrics is not None:
														SCREAMING_SNAKE_CASE_   :						List[str]							=	self.post_process_function(A_		,			A_		,			output.predictions)
														SCREAMING_SNAKE_CASE_   :						Dict							=	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}_'):
																						SCREAMING_SNAKE_CASE_   :						Optional[int]							=	metrics.pop(A_)

														self.log(A_)
										else:
														SCREAMING_SNAKE_CASE_   :						List[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())

										SCREAMING_SNAKE_CASE_   :						Tuple							=	self.callback_handler.on_evaluate(self.args		,			self.state		,			self.control		,			A_)
										return metrics
						def _SCREAMING_SNAKE_CASE				(		self  :						int		,			lowercase_  :						Dict		,			lowercase_  :						Optional[Any]		,			lowercase_  :						str=None		,			lowercase_  :						int = "test"):




										'''simple docstring'''


										SCREAMING_SNAKE_CASE_   :						Optional[int]							=	self.get_test_dataloader(A_)

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

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

										SCREAMING_SNAKE_CASE_   :						List[str]							=	self.post_process_function(A_		,			A_		,			output.predictions		,			'''predict''')
										SCREAMING_SNAKE_CASE_   :						Optional[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}_'):
																		SCREAMING_SNAKE_CASE_   :						Tuple							=	metrics.pop(A_)

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



						def _SCREAMING_SNAKE_CASE				(		self  :						List[Any]		,			lowercase_  :						int="./"):




										'''simple docstring'''


										SCREAMING_SNAKE_CASE_   :						str							=	self.eval_dataset
										SCREAMING_SNAKE_CASE_   :						Dict							=	self.get_eval_dataloader(A_)

										SCREAMING_SNAKE_CASE_   :						List[str]							=	next(iter(A_))

										# saving device - to make it consistent
										SCREAMING_SNAKE_CASE_   :						Tuple							=	torch.device('''cuda''' if torch.cuda.is_available() else '''cpu''')

										# convert to tuple
										SCREAMING_SNAKE_CASE_   :						Union[str, 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

										SCREAMING_SNAKE_CASE_   :						str							=	True

										SCREAMING_SNAKE_CASE_   :						int							=	self.model.to(A_)

										model.eval()
										model.float()

										SCREAMING_SNAKE_CASE_   :						Tuple							=	model.module if hasattr(A_		,			'''module''') else model
										quant_trainer.configure_model(A_		,			self.quant_trainer_args)

										SCREAMING_SNAKE_CASE_   :						int							=	os.path.join(A_		,			'''model.onnx''')
										logger.info(F'exporting model to {output_model_file}')

										SCREAMING_SNAKE_CASE_   :						List[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''')










"""simple docstring"""






from __future__ import annotations



def       _A    (__a					,       __a					,       __a							)				-> int | float:
    """simple docstring"""






    if len(__a							) == 0:
        raise ValueError('''find_max() arg is an empty sequence'''							)
    if (
        left >= len(__a							)
        or left < -len(__a							)
        or right >= len(__a							)
        or right < -len(__a							)
    ):
        raise IndexError('''list index out of range'''							)
    if left == right:
        return nums[left]
    SCREAMING_SNAKE_CASE_   :						str							=	(left + right) >> 1  # the middle
    SCREAMING_SNAKE_CASE_   :						int							=	find_max(__a					,       __a					,       __a							)  # find max in range[left, mid]
    SCREAMING_SNAKE_CASE_   :						str							=	find_max(__a					,       mid + 1					,       __a							)  # find max in range[mid + 1, right]

    return left_max if left_max >= right_max else right_max


if __name__ == "__main__":
  import doctest

  doctest.testmod(verbose=True)


