



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


A		=						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							:				int   ,	*_lowercase							:				List[Any]   ,	_lowercase							:				Union[str, Any]=None   ,	_lowercase							:				List[str]=None   ,	_lowercase							:				str=None   ,	**_lowercase							:				List[Any]		)							->					Any:
									super().__init__(*_lowercase   ,	**_lowercase		)
									snake_case   :			int						    = eval_examples
									snake_case   :			int						    = post_process_function
									snake_case   :			Optional[Any]						    = quant_trainer_args
									snake_case   :			List[Any]						    = 128  # default number of calibration samples



			def 				__lowercase					(			self							:				Any   ,	_lowercase							:				Any=None		)							->					Union[str, Any]:
									if calib_dataset is None and self.calib_dataset is None:
															raise ValueError("Trainer: calibration requires an calib_dataset."		)
									snake_case   :			Optional[int]						    = calib_dataset if calib_dataset is not None else self.calib_dataset

									snake_case   :			Optional[Any]						    = self._remove_unused_columns(_lowercase   ,	description="Calibration"		)

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



			def 				__lowercase					(			self							:				Dict   ,	_lowercase							:				Optional[int]=None		)							->					List[Any]:
									snake_case   :			List[str]						    = self.train_dataset if calib_dataset is None else calib_dataset
									snake_case   :			int						    = self.get_calib_dataloader(_lowercase		)

									snake_case   :			str						    = self.model
									quant_trainer.configure_model(_lowercase   ,	self.quant_trainer_args   ,	calib=_lowercase		)
									model.eval()
									quant_trainer.enable_calibration(_lowercase		)

									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(_lowercase		):
															# Prediction step
															snake_case    ,			snake_case    ,			snake_case   :			Optional[Any]						    = self.prediction_step(_lowercase   ,	_lowercase   ,	prediction_loss_only=_lowercase		)
															if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
																					break

									quant_trainer.finish_calibration(_lowercase   ,	self.quant_trainer_args		)
									snake_case   :			List[Any]						    = model



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

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

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

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

									snake_case   :			Dict						    = self.callback_handler.on_evaluate(self.args   ,	self.state   ,	self.control   ,	_lowercase		)
									return metrics



			def 				__lowercase					(			self							:				Union[str, Any]   ,	_lowercase							:				Optional[Any]   ,	_lowercase							:				Tuple   ,	_lowercase							:				List[str]=None   ,	_lowercase							:				str = "test"		)							->					List[str]:
									snake_case   :			str						    = self.get_test_dataloader(_lowercase		)

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

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

									snake_case   :			List[Any]						    = self.post_process_function(_lowercase   ,	_lowercase   ,	output.predictions   ,	"predict"		)
									snake_case   :			Any						    = self.compute_metrics(_lowercase		)

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

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



			def 				__lowercase					(			self							:				Optional[Any]   ,	_lowercase							:				List[str]="./"		)							->					Optional[Any]:
									snake_case   :			Any						    = self.eval_dataset
									snake_case   :			List[str]						    = self.get_eval_dataloader(_lowercase		)

									snake_case   :			List[str]						    = next(iter(_lowercase		)		)

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

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

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

									snake_case   :			Dict						    = True

									snake_case   :			Tuple						    = self.model.to(_lowercase		)

									model.eval()
									model.float()

									snake_case   :			List[str]						    = model.module if hasattr(_lowercase   ,	"module"		) else model
									quant_trainer.configure_model(_lowercase   ,	self.quant_trainer_args		)

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

									snake_case   :			List[str]						    = {0: "batch_size", 1: "seq_len"}

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








"""simple docstring"""







import importlib
import inspect
import os
import re


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_config_docstrings.py
A		=						'src/transformers'


# This is to make sure the transformers module imported is the one in the repo.
A		=						importlib.util.spec_from_file_location(
    'transformers',
    os.path.join(PATH_TO_TRANSFORMERS, '__init__.py'),
    submodule_search_locations=[PATH_TO_TRANSFORMERS],
)
A		=						spec.loader.load_module()

A		=						transformers.models.auto.configuration_auto.CONFIG_MAPPING

# Regex pattern used to find the checkpoint mentioned in the docstring of `config_class`.
# For example, `[bert-base-uncased](https://huggingface.co/bert-base-uncased)`
A		=						re.compile('\[(.+?)\]\((https://huggingface\.co/.+?)\)')


A		=						{
    'CLIPConfigMixin',
    'DecisionTransformerConfigMixin',
    'EncoderDecoderConfigMixin',
    'RagConfigMixin',
    'SpeechEncoderDecoderConfigMixin',
    'VisionEncoderDecoderConfigMixin',
    'VisionTextDualEncoderConfigMixin',
}

def  __SCREAMING_SNAKE_CASE		(			):
						"""simple docstring"""






						snake_case   :			Union[str, Any]						    = []

						for config_class in list(CONFIG_MAPPING.values() ):
												snake_case   :			List[str]						    = False

												# source code of `config_class`
												snake_case   :			Optional[int]						    = inspect.getsource(lowerCamelCase_ )
												snake_case   :			Optional[int]						    = _re_checkpoint.findall(lowerCamelCase_ )

												for checkpoint in checkpoints:
																		# Each `checkpoint` is a tuple of a checkpoint name and a checkpoint link.
																		# For example, `('bert-base-uncased', 'https://huggingface.co/bert-base-uncased')`
																		snake_case    ,			snake_case   :			Tuple						    = checkpoint

																		# verify the checkpoint name corresponds to the checkpoint link
																		snake_case   :			Any						    = f'''https://huggingface.co/{ckpt_name}'''
																		if ckpt_link == ckpt_link_from_name:
																								snake_case   :			List[str]						    = True
																								break

												snake_case   :			Any						    = config_class.__name__
												if not checkpoint_found and name not in CONFIG_CLASSES_TO_IGNORE_FOR_DOCSTRING_CHECKPOINT_CHECK:
																		configs_without_checkpoint.append(lowerCamelCase_ )

						if len(lowerCamelCase_ ) > 0:
												snake_case   :			List[str]						    = "\n".join(sorted(lowerCamelCase_ ) )
												raise ValueError(f'''The following configurations don\'t contain any valid checkpoint:\n{message}''' )


if __name__ == "__main__":
			check_config_docstrings_have_checkpoints()



