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					      =							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__							(					_UpperCamelCase				):



   def __init__(					self : List[Any]      ,						*A__ : Optional[int]      ,						A__ : Union[str, Any]=None      ,						A__ : Dict=None      ,						A__ : List[Any]=None      ,						**A__ : Optional[int] )				->  str:




         '''simple docstring'''




         super().__init__(*a_      ,						**a_ )
         snake_case_  :      List[str]  	=							eval_examples
         snake_case_  :      List[str]  	=							post_process_function
         snake_case_  :      int  	=							quant_trainer_args
         snake_case_  :      Union[str, Any]  	=							1_28  # default number of calibration samples



   def        UpperCAmelCase__							(					self : List[Any]      ,						A__ : str=None )				->  Union[str, Any]:




         '''simple docstring'''




         if calib_dataset is None and self.calib_dataset is None:
               raise ValueError("Trainer: calibration requires an calib_dataset." )
         snake_case_  :      Optional[Any]  	=							calib_dataset if calib_dataset is not None else self.calib_dataset

         snake_case_  :      str  	=							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        UpperCAmelCase__							(					self : Optional[Any]      ,						A__ : List[Any]=None )				->  Union[str, Any]:




         '''simple docstring'''




         snake_case_  :      Any  	=							self.train_dataset if calib_dataset is None else calib_dataset
         snake_case_  :      List[str]  	=							self.get_calib_dataloader(a_ )

         snake_case_  :      List[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
               snake_case_	,snake_case_	,snake_case_  :      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 )
         snake_case_  :      Any  	=							model



   def        UpperCAmelCase__							(					self : str      ,						A__ : Dict=None      ,						A__ : int=None      ,						A__ : int=None      ,						A__ : Optional[int] = "eval" )				->  Optional[Any]:




         '''simple docstring'''




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

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

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

         snake_case_  :      int  	=							self.callback_handler.on_evaluate(self.args      ,						self.state      ,						self.control      ,						a_ )
         return metrics



   def        UpperCAmelCase__							(					self : Any      ,						A__ : Dict      ,						A__ : Optional[int]      ,						A__ : Any=None      ,						A__ : Any = "test" )				->  Optional[Any]:




         '''simple docstring'''




         snake_case_  :      Any  	=							self.get_test_dataloader(a_ )

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

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

         snake_case_  :      Any  	=							self.post_process_function(a_      ,						a_      ,						output.predictions      ,						"predict" )
         snake_case_  :      Tuple  	=							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}_" ):
                     snake_case_  :      str  	=							metrics.pop(a_ )

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



   def        UpperCAmelCase__							(					self : Dict      ,						A__ : int="./" )				->  Any:




         '''simple docstring'''




         snake_case_  :      Dict  	=							self.eval_dataset
         snake_case_  :      Tuple  	=							self.get_eval_dataloader(a_ )

         snake_case_  :      List[Any]  	=							next(iter(a_ ) )

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

         # convert to tuple
         snake_case_  :      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

         snake_case_  :      List[str]  	=							True

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

         model.eval()
         model.float()

         snake_case_  :      int  	=							model.module if hasattr(a_      ,						"module" ) else model
         quant_trainer.configure_model(a_      ,						self.quant_trainer_args )

         snake_case_  :      Optional[Any]  	=							os.path.join(a_      ,						"model.onnx" )
         logger.info(f"exporting model to {output_model_file}" )

         snake_case_  :      str  	=							{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 typing import Optional

from torch import nn

from .transformer_ad import TransformeraDModel, TransformeraDModelOutput



class 							lowercase_						(		nn.Module			):

  '''simple docstring'''





  def __init__(  self ,			a_ = 1_6 ,			a_ = 8_8 ,			a_ = None ,			a_ = 1 ,			a_ = 0.0 ,			a_ = 3_2 ,			a_ = None ,			a_ = False ,			a_ = None ,			a_ = None ,			a_ = "geglu" ,			a_ = None ,			)				->					List[str]:


    """simple docstring"""
    super().__init__()
    UpperCAmelCase    				=						nn.ModuleList(
        [
            TransformeraDModel(
                num_attention_heads=a_ ,			attention_head_dim=a_ ,			in_channels=a_ ,			num_layers=a_ ,			dropout=a_ ,			norm_num_groups=a_ ,			cross_attention_dim=a_ ,			attention_bias=a_ ,			sample_size=a_ ,			num_vector_embeds=a_ ,			activation_fn=a_ ,			num_embeds_ada_norm=a_ ,			)
            for _ in range(2  )
        ]  )

    # Variables that can be set by a pipeline:

    # The ratio of transformer1 to transformer2's output states to be combined during inference
    UpperCAmelCase    				=						0.5

    # The shape of `encoder_hidden_states` is expected to be
    # `(batch_size, condition_lengths[0]+condition_lengths[1], num_features)`
    UpperCAmelCase    				=						[7_7, 2_5_7]

    # Which transformer to use to encode which condition.
    # E.g. `(1, 0)` means that we'll use `transformers[1](conditions[0])` and `transformers[0](conditions[1])`
    UpperCAmelCase    				=						[1, 0]





  def     snake_case_      (  self ,			a_ ,			a_ ,			a_=None ,			a_=None ,			a_=None ,			a_ = True ,			)				->					Optional[int]:


    """simple docstring"""
    UpperCAmelCase    				=						hidden_states

    UpperCAmelCase    				=						[]
    UpperCAmelCase    				=						0
    # attention_mask is not used yet
    for i in range(2  ):
      # for each of the two transformers, pass the corresponding condition tokens
      UpperCAmelCase    				=						encoder_hidden_states[:, tokens_start : tokens_start + self.condition_lengths[i]]
      UpperCAmelCase    				=						self.transformer_index_for_condition[i]
      UpperCAmelCase    				=						self.transformers[transformer_index](
          a_ ,			encoder_hidden_states=a_ ,			timestep=a_ ,			cross_attention_kwargs=a_ ,			return_dict=a_ ,			)[0]
      encoded_states.append(encoded_state - input_states  )
      tokens_start += self.condition_lengths[i]

    UpperCAmelCase    				=						encoded_states[0] * self.mix_ratio + encoded_states[1] * (1 - self.mix_ratio)
    UpperCAmelCase    				=						output_states + input_states

    if not return_dict:
      return (output_states,)

    return TransformeraDModelOutput(sample=a_  )


