


'''simple docstring'''







from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


lowerCAmelCase     :  int											=    logging.get_logger(__name__)


if is_vision_available():
       import PIL
class 						UpperCamelCase__       (			_a						):



     """simple docstring"""






     __magic_name__           =     ['''pixel_values''']







     def __init__(			self   ,						snake_case__ = True   ,						snake_case__ = None   ,						snake_case__ = PILImageResampling.BICUBIC   ,						snake_case__ = True   ,						snake_case__ = None   ,						snake_case__ = True   ,						snake_case__ = 1 / 255   ,						snake_case__ = True   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = True   ,						**snake_case__   ,						):

            '''simple docstring'''

            super().__init__(**snake_case__ )
            _lowerCAmelCase     :				Tuple       =							size if size is not None else {'shortest_edge': 224}
            _lowerCAmelCase     :				Optional[Any]       =							get_size_dict(snake_case__   ,						default_to_square=snake_case__ )
            _lowerCAmelCase     :				Optional[int]       =							crop_size if crop_size is not None else {'height': 224, 'width': 224}
            _lowerCAmelCase     :				Tuple       =							get_size_dict(snake_case__   ,						default_to_square=snake_case__   ,						param_name='crop_size' )

            _lowerCAmelCase     :				Tuple       =							do_resize
            _lowerCAmelCase     :				int       =							size
            _lowerCAmelCase     :				Any       =							resample
            _lowerCAmelCase     :				Dict       =							do_center_crop
            _lowerCAmelCase     :				int       =							crop_size
            _lowerCAmelCase     :				Tuple       =							do_rescale
            _lowerCAmelCase     :				Tuple       =							rescale_factor
            _lowerCAmelCase     :				List[Any]       =							do_normalize
            _lowerCAmelCase     :				str       =							image_mean if image_mean is not None else OPENAI_CLIP_MEAN
            _lowerCAmelCase     :				Optional[int]       =							image_std if image_std is not None else OPENAI_CLIP_STD
            _lowerCAmelCase     :				int       =							do_convert_rgb







     def a						(			self   ,						snake_case__   ,						snake_case__   ,						snake_case__ = PILImageResampling.BICUBIC   ,						snake_case__ = None   ,						**snake_case__   ,						):

            '''simple docstring'''

            _lowerCAmelCase     :				str       =							get_size_dict(snake_case__   ,						default_to_square=snake_case__ )
            if "shortest_edge" not in size:
                   raise ValueError(F'The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}' )
            _lowerCAmelCase     :				Tuple       =							get_resize_output_image_size(snake_case__   ,						size=size['shortest_edge']   ,						default_to_square=snake_case__ )
            return resize(snake_case__   ,						size=snake_case__   ,						resample=snake_case__   ,						data_format=snake_case__   ,						**snake_case__ )







     def a						(			self   ,						snake_case__   ,						snake_case__   ,						snake_case__ = None   ,						**snake_case__   ,						):

            '''simple docstring'''

            _lowerCAmelCase     :				int       =							get_size_dict(snake_case__ )
            if "height" not in size or "width" not in size:
                   raise ValueError(F'The `size` parameter must contain the keys (height, width). Got {size.keys()}' )
            return center_crop(snake_case__   ,						size=(size['height'], size['width'])   ,						data_format=snake_case__   ,						**snake_case__ )







     def a						(			self   ,						snake_case__   ,						snake_case__   ,						snake_case__ = None   ,						**snake_case__   ,						):

            '''simple docstring'''

            return rescale(snake_case__   ,						scale=snake_case__   ,						data_format=snake_case__   ,						**snake_case__ )







     def a						(			self   ,						snake_case__   ,						snake_case__   ,						snake_case__   ,						snake_case__ = None   ,						**snake_case__   ,						):

            '''simple docstring'''

            return normalize(snake_case__   ,						mean=snake_case__   ,						std=snake_case__   ,						data_format=snake_case__   ,						**snake_case__ )







     def a						(			self   ,						snake_case__   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = None   ,						snake_case__ = ChannelDimension.FIRST   ,						**snake_case__   ,						):

            '''simple docstring'''

            _lowerCAmelCase     :				List[Any]       =							do_resize if do_resize is not None else self.do_resize
            _lowerCAmelCase     :				str       =							size if size is not None else self.size
            _lowerCAmelCase     :				Tuple       =							get_size_dict(snake_case__   ,						param_name='size'   ,						default_to_square=snake_case__ )
            _lowerCAmelCase     :				List[str]       =							resample if resample is not None else self.resample
            _lowerCAmelCase     :				Dict       =							do_center_crop if do_center_crop is not None else self.do_center_crop
            _lowerCAmelCase     :				List[str]       =							crop_size if crop_size is not None else self.crop_size
            _lowerCAmelCase     :				List[str]       =							get_size_dict(snake_case__   ,						param_name='crop_size'   ,						default_to_square=snake_case__ )
            _lowerCAmelCase     :				str       =							do_rescale if do_rescale is not None else self.do_rescale
            _lowerCAmelCase     :				Optional[int]       =							rescale_factor if rescale_factor is not None else self.rescale_factor
            _lowerCAmelCase     :				Union[str, Any]       =							do_normalize if do_normalize is not None else self.do_normalize
            _lowerCAmelCase     :				int       =							image_mean if image_mean is not None else self.image_mean
            _lowerCAmelCase     :				List[str]       =							image_std if image_std is not None else self.image_std
            _lowerCAmelCase     :				int       =							do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

            _lowerCAmelCase     :				List[str]       =							make_list_of_images(snake_case__ )

            if not valid_images(snake_case__ ):
                   raise ValueError(
                       'Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '
                       'torch.Tensor, tf.Tensor or jax.ndarray.' )

            if do_resize and size is None:
                   raise ValueError('Size must be specified if do_resize is True.' )

            if do_center_crop and crop_size is None:
                   raise ValueError('Crop size must be specified if do_center_crop is True.' )

            if do_rescale and rescale_factor is None:
                   raise ValueError('Rescale factor must be specified if do_rescale is True.' )

            if do_normalize and (image_mean is None or image_std is None):
                   raise ValueError('Image mean and std must be specified if do_normalize is True.' )

            # PIL RGBA images are converted to RGB
            if do_convert_rgb:
                   _lowerCAmelCase     :				str       =							[convert_to_rgb(snake_case__ ) for image in images]

            # All transformations expect numpy arrays.
            _lowerCAmelCase     :				str       =							[to_numpy_array(snake_case__ ) for image in images]

            if do_resize:
                   _lowerCAmelCase     :				Dict       =							[self.resize(image=snake_case__   ,						size=snake_case__   ,						resample=snake_case__ ) for image in images]

            if do_center_crop:
                   _lowerCAmelCase     :				List[Any]       =							[self.center_crop(image=snake_case__   ,						size=snake_case__ ) for image in images]

            if do_rescale:
                   _lowerCAmelCase     :				int       =							[self.rescale(image=snake_case__   ,						scale=snake_case__ ) for image in images]

            if do_normalize:
                   _lowerCAmelCase     :				Optional[Any]       =							[self.normalize(image=snake_case__   ,						mean=snake_case__   ,						std=snake_case__ ) for image in images]

            _lowerCAmelCase     :				Any       =							[to_channel_dimension_format(snake_case__   ,						snake_case__ ) for image in images]

            _lowerCAmelCase     :				Optional[Any]       =							{'pixel_values': images}
            return BatchFeature(data=snake_case__   ,						tensor_type=snake_case__ )









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


_SCREAMING_SNAKE_CASE   =  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 					SCREAMING_SNAKE_CASE_  (       _a ):
       """simple docstring"""


       def __init__(				self	:Dict,      *snake_case	:int,      snake_case	:Any=None,      snake_case	:str=None,      snake_case	:int=None,      **snake_case	:List[str]):



          """simple docstring"""

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





       def 				UpperCamelCase__						(				self	:Dict,      snake_case	:List[str]=None):



          """simple docstring"""

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

          _lowercase              =self._remove_unused_columns(snake_case,      description='Calibration')

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





       def 				UpperCamelCase__						(				self	:Any,      snake_case	:Union[str, Any]=None):



          """simple docstring"""

          _lowercase              =self.train_dataset if calib_dataset is None else calib_dataset
          _lowercase              =self.get_calib_dataloader(snake_case)

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

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

          quant_trainer.finish_calibration(snake_case,      self.quant_trainer_args)
          _lowercase              =model





       def 				UpperCamelCase__						(				self	:Optional[int],      snake_case	:Union[str, Any]=None,      snake_case	:Optional[int]=None,      snake_case	:Union[str, Any]=None,      snake_case	:str = "eval"):



          """simple docstring"""

          _lowercase              =self.eval_dataset if eval_dataset is None else eval_dataset
          _lowercase              =self.get_eval_dataloader(snake_case)
          _lowercase              =self.eval_examples if eval_examples is None else eval_examples

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

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

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

             self.log(snake_case)
          else:
             _lowercase              ={}

          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())

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





       def 				UpperCamelCase__						(				self	:Optional[int],      snake_case	:str,      snake_case	:Optional[Any],      snake_case	:Any=None,      snake_case	:str = "test"):



          """simple docstring"""

          _lowercase              =self.get_test_dataloader(snake_case)

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

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

          _lowercase              =self.post_process_function(snake_case,      snake_case,      output.predictions,      'predict')
          _lowercase              =self.compute_metrics(snake_case)

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

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





       def 				UpperCamelCase__						(				self	:List[str],      snake_case	:Dict="./"):



          """simple docstring"""

          _lowercase              =self.eval_dataset
          _lowercase              =self.get_eval_dataloader(snake_case)

          _lowercase              =next(iter(snake_case))

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

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

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

          _lowercase              =True

          _lowercase              =self.model.to(snake_case)

          model.eval()
          model.float()

          _lowercase              =model.module if hasattr(snake_case,      'module') else model
          quant_trainer.configure_model(snake_case,      self.quant_trainer_args)

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

          _lowercase              ={0: 'batch_size', 1: 'seq_len'}

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


