





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


lowerCAmelCase									=   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	(   _A							):



	def __init__( self		,   *a_		,   a_=None		,   a_=None		,   a_=None		,   **a_						):
			super().__init__(*a_		,   **a_						)
			lowerCAmelCase  :     Tuple       						= eval_examples
			lowerCAmelCase  :     List[Any]       						= post_process_function
			lowerCAmelCase  :     List[Any]       						= quant_trainer_args
			lowerCAmelCase  :     Dict       						= 128  # default number of calibration samples



	def 	_lowerCamelCase     ( self		,   a_=None						):
			if calib_dataset is None and self.calib_dataset is None:
					raise ValueError("Trainer: calibration requires an calib_dataset."						)
			lowerCAmelCase  :     List[str]       						= calib_dataset if calib_dataset is not None else self.calib_dataset

			lowerCAmelCase  :     Union[str, 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 	_lowerCamelCase     ( self		,   a_=None						):
			lowerCAmelCase  :     str       						= self.train_dataset if calib_dataset is None else calib_dataset
			lowerCAmelCase  :     str       						= self.get_calib_dataloader(a_						)

			lowerCAmelCase  :     Tuple       						= 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
					lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase  :     Tuple       						= 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						)
			lowerCAmelCase  :     Union[str, Any]       						= model



	def 	_lowerCamelCase     ( self		,   a_=None		,   a_=None		,   a_=None		,   a_ = "eval"						):
			lowerCAmelCase  :     Optional[Any]       						= self.eval_dataset if eval_dataset is None else eval_dataset
			lowerCAmelCase  :     Optional[Any]       						= self.get_eval_dataloader(a_						)
			lowerCAmelCase  :     Union[str, Any]       						= self.eval_examples if eval_examples is None else eval_examples

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

			if self.post_process_function is not None and self.compute_metrics is not None:
					lowerCAmelCase  :     Optional[Any]       						= self.post_process_function(a_		,   a_		,   output.predictions						)
					lowerCAmelCase  :     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}_'''						):
									lowerCAmelCase  :     Any       						= metrics.pop(a_						)

					self.log(a_						)
			else:
					lowerCAmelCase  :     str       						= {}

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



	def 	_lowerCamelCase     ( self		,   a_		,   a_		,   a_=None		,   a_ = "test"						):
			lowerCAmelCase  :     int       						= self.get_test_dataloader(a_						)

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

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

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

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





	def 	_lowerCamelCase     ( self		,   a_="./"						):
			lowerCAmelCase  :     Optional[Any]       						= self.eval_dataset
			lowerCAmelCase  :     Optional[int]       						= self.get_eval_dataloader(a_						)

			lowerCAmelCase  :     Union[str, Any]       						= next(iter(a_						)						)

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

			# convert to tuple
			lowerCAmelCase  :     str       						= tuple(v.to(a_						) for k, v in batch.items()						)

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

			lowerCAmelCase  :     Any       						= True

			lowerCAmelCase  :     Dict       						= self.model.to(a_						)

			model.eval()
			model.float()

			lowerCAmelCase  :     Union[str, Any]       						= model.module if hasattr(a_		,   "module"						) else model
			quant_trainer.configure_model(a_		,   self.quant_trainer_args						)

			lowerCAmelCase  :     Any       						= os.path.join(a_		,   "model.onnx"						)
			logger.info(F'''exporting model to {output_model_file}'''						)

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






import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType
class 		lowerCamelCase	(   _A							):
	snake_case_   = ["image_processor", "tokenizer"]
	snake_case_   = "LayoutLMv3ImageProcessor"
	snake_case_   = ("LayoutLMv3Tokenizer", "LayoutLMv3TokenizerFast")



	def __init__( self		,   a_=None		,   a_=None		,   **a_						):
			lowerCAmelCase  :     Union[str, Any]       						= None
			if "feature_extractor" in kwargs:
					warnings.warn(
					    "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
					    " instead."		,   a_		,   )
					lowerCAmelCase  :     Tuple       						= kwargs.pop("feature_extractor"						)

			lowerCAmelCase  :     str       						= image_processor if image_processor is not None else feature_extractor
			if image_processor is None:
					raise ValueError("You need to specify an `image_processor`."						)
			if tokenizer is None:
					raise ValueError("You need to specify a `tokenizer`."						)

			super().__init__(a_		,   a_						)



	def __call__( self		,   a_		,   a_ = None		,   a_ = None		,   a_ = None		,   a_ = None		,   a_ = True		,   a_ = False		,   a_ = None		,   a_ = None		,   a_ = 0		,   a_ = None		,   a_ = None		,   a_ = None		,   a_ = False		,   a_ = False		,   a_ = False		,   a_ = False		,   a_ = True		,   a_ = None		,   **a_		,   ):
			# verify input
			if self.image_processor.apply_ocr and (boxes is not None):
					raise ValueError(
					    "You cannot provide bounding boxes if you initialized the image processor with apply_ocr set to True."						)

			if self.image_processor.apply_ocr and (word_labels is not None):
					raise ValueError(
					    "You cannot provide word labels if you initialized the image processor with apply_ocr set to True."						)

			# first, apply the image processor
			lowerCAmelCase  :     str       						= self.image_processor(images=a_		,   return_tensors=a_						)

			# second, apply the tokenizer
			if text is not None and self.image_processor.apply_ocr and text_pair is None:
					if isinstance(a_		,   a_						):
							lowerCAmelCase  :     Union[str, Any]       						= [text]  # add batch dimension (as the image processor always adds a batch dimension)
					lowerCAmelCase  :     int       						= features["words"]

			lowerCAmelCase  :     List[Any]       						= self.tokenizer(
			    text=text if text is not None else features["words"]		,   text_pair=text_pair if text_pair is not None else None		,   boxes=boxes if boxes is not None else features["boxes"]		,   word_labels=a_		,   add_special_tokens=a_		,   padding=a_		,   truncation=a_		,   max_length=a_		,   stride=a_		,   pad_to_multiple_of=a_		,   return_token_type_ids=a_		,   return_attention_mask=a_		,   return_overflowing_tokens=a_		,   return_special_tokens_mask=a_		,   return_offsets_mapping=a_		,   return_length=a_		,   verbose=a_		,   return_tensors=a_		,   **a_		,   )

			# add pixel values
			lowerCAmelCase  :     Tuple       						= features.pop("pixel_values"						)
			if return_overflowing_tokens is True:
					lowerCAmelCase  :     Optional[int]       						= self.get_overflowing_images(a_		,   encoded_inputs["overflow_to_sample_mapping"]						)
			lowerCAmelCase  :     str       						= images

			return encoded_inputs



	def 	_lowerCamelCase     ( self		,   a_		,   a_						):
			# in case there's an overflow, ensure each `input_ids` sample is mapped to its corresponding image
			lowerCAmelCase  :     Optional[Any]       						= []
			for sample_idx in overflow_to_sample_mapping:
					images_with_overflow.append(images[sample_idx]						)

			if len(a_						) != len(a_						):
					raise ValueError(
					    "Expected length of images to be the same as the length of `overflow_to_sample_mapping`, but got"
					    F''' {len(a_						)} and {len(a_						)}'''						)

			return images_with_overflow



	def 	_lowerCamelCase     ( self		,   *a_		,   **a_						):
			return self.tokenizer.batch_decode(*a_		,   **a_						)



	def 	_lowerCamelCase     ( self		,   *a_		,   **a_						):
			return self.tokenizer.decode(*a_		,   **a_						)



	@property
	def 	_lowerCamelCase     ( self						):
			return ["input_ids", "bbox", "attention_mask", "pixel_values"]



	@property
	def 	_lowerCamelCase     ( self						):
			warnings.warn(
			    "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."		,   a_		,   )
			return self.image_processor_class





	@property
	def 	_lowerCamelCase     ( self						):
			warnings.warn(
			    "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."		,   a_		,   )
			return self.image_processor



