



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


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 			snake_case_						(						a_				):






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






       def snake_case_  (   self			,     a_=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_     :      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 snake_case_  (   self			,     a_=None      ):
         a_     :      Union[str, Any] 						=     self.train_dataset if calib_dataset is None else calib_dataset
         a_     :      Dict 						=     self.get_calib_dataloader(a_      )

         a_     :      Optional[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_     :      Union[str, 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      )
         a_     :      Optional[Any] 						=     model






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

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

           self.log(a_      )
         else:
           a_     :      Optional[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()      )

         a_     :      List[Any] 						=     self.callback_handler.on_evaluate(self.args			,     self.state			,     self.control			,     a_      )
         return metrics






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

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

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

         a_     :      List[str] 						=     self.post_process_function(a_			,     a_			,     output.predictions			,     "predict"      )
         a_     :      int 						=     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 snake_case_  (   self			,     a_="./"      ):
         a_     :      Any 						=     self.eval_dataset
         a_     :      Union[str, Any] 						=     self.get_eval_dataloader(a_      )

         a_     :      Any 						=     next(iter(a_      )      )

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

         # convert to tuple
         a_     :      Tuple 						=     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_     :      Optional[int] 						=     True

         a_     :      int 						=     self.model.to(a_      )

         model.eval()
         model.float()

         a_     :      int 						=     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_     :      Union[str, Any] 						=     {0: "batch_size", 1: "seq_len"}

         torch.onnx.export(
             a_			,     a_			,     a_			,     export_params=a_			,     opset_version=1_3			,     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 unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionImageVariationPipeline
from diffusers.utils.testing_utils import load_image, require_torch_gpu, slow, torch_device


SCREAMING_SNAKE_CASE_				= False




class 			snake_case_						(						unittest.TestCase				):
							pass




@slow
@require_torch_gpu
class 			snake_case_						(						unittest.TestCase				):






							def snake_case_  (   self      ):
									a_     :      Tuple 						=     VersatileDiffusionImageVariationPipeline.from_pretrained("shi-labs/versatile-diffusion"      )
									pipe.to(a_      )
									pipe.set_progress_bar_config(disable=a_      )

									a_     :      Optional[Any] 						=     load_image(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"      )
									a_     :      Optional[Any] 						=     torch.manual_seed(0      )
									a_     :      Dict 						=     pipe(
									    image=a_			,     generator=a_			,     guidance_scale=7.5			,     num_inference_steps=5_0			,     output_type="numpy"			,     ).images

									a_     :      List[Any] 						=     image[0, 2_5_3:2_5_6, 2_5_3:2_5_6, -1]

									assert image.shape == (1, 5_1_2, 5_1_2, 3)
									a_     :      Any 						=     np.array([0.0_441, 0.0_469, 0.0_507, 0.0_575, 0.0_632, 0.0_650, 0.0_865, 0.0_909, 0.0_945]      )

									assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2