
"""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__      :	Tuple          =    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   _lowercase						(       lowerCAmelCase_				):


     '''simple docstring'''





     def __init__(      self ,			*__UpperCamelCase ,			__UpperCamelCase=None ,			__UpperCamelCase=None ,			__UpperCamelCase=None ,			**__UpperCamelCase  )->					Dict:
          super().__init__(*__UpperCamelCase ,			**__UpperCamelCase  )
          UpperCAmelCase__							: Tuple 							=						eval_examples
          UpperCAmelCase__							: List[Any] 							=						post_process_function
          UpperCAmelCase__							: Any 							=						quant_trainer_args
          UpperCAmelCase__							: Any 							=						1_28  # default number of calibration samples



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase=None  )->					Union[str, Any]:
          if calib_dataset is None and self.calib_dataset is None:
               raise ValueError("Trainer: calibration requires an calib_dataset."  )
          UpperCAmelCase__							: Union[str, Any] 							=						calib_dataset if calib_dataset is not None else self.calib_dataset

          UpperCAmelCase__							: List[Any] 							=						self._remove_unused_columns(__UpperCamelCase ,			description="Calibration"  )

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



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase=None  )->					Union[str, Any]:
          UpperCAmelCase__							: Optional[Any] 							=						self.train_dataset if calib_dataset is None else calib_dataset
          UpperCAmelCase__							: Union[str, Any] 							=						self.get_calib_dataloader(__UpperCamelCase  )

          UpperCAmelCase__							: List[str] 							=						self.model
          quant_trainer.configure_model(__UpperCamelCase ,			self.quant_trainer_args ,			calib=__UpperCamelCase  )
          model.eval()
          quant_trainer.enable_calibration(__UpperCamelCase  )

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

          quant_trainer.finish_calibration(__UpperCamelCase ,			self.quant_trainer_args  )
          UpperCAmelCase__							: Optional[int] 							=						model



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase=None ,			__UpperCamelCase=None ,			__UpperCamelCase=None ,			__UpperCamelCase = "eval"  )->					Dict:
          UpperCAmelCase__							: Dict 							=						self.eval_dataset if eval_dataset is None else eval_dataset
          UpperCAmelCase__							: str 							=						self.get_eval_dataloader(__UpperCamelCase  )
          UpperCAmelCase__							: 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.
          UpperCAmelCase__							: Tuple 							=						self.compute_metrics
          UpperCAmelCase__							: Any 							=						None
          UpperCAmelCase__							: Tuple 							=						self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
          try:
               UpperCAmelCase__							: int 							=						eval_loop(
                   __UpperCamelCase ,			description="Evaluation" ,			prediction_loss_only=True if compute_metrics is None else None ,			ignore_keys=__UpperCamelCase ,			)
          finally:
               UpperCAmelCase__							: Dict 							=						compute_metrics

          if self.post_process_function is not None and self.compute_metrics is not None:
               UpperCAmelCase__							: Any 							=						self.post_process_function(__UpperCamelCase ,			__UpperCamelCase ,			output.predictions  )
               UpperCAmelCase__							: Dict 							=						self.compute_metrics(__UpperCamelCase  )

               # Prefix all keys with metric_key_prefix + '_'
               for key in list(metrics.keys()  ):
                    if not key.startswith(F"{metric_key_prefix}_"  ):
                         UpperCAmelCase__							: List[Any] 							=						metrics.pop(__UpperCamelCase  )

               self.log(__UpperCamelCase  )
          else:
               UpperCAmelCase__							: 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()  )

          UpperCAmelCase__							: Union[str, Any] 							=						self.callback_handler.on_evaluate(self.args ,			self.state ,			self.control ,			__UpperCamelCase  )
          return metrics



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase ,			__UpperCamelCase ,			__UpperCamelCase=None ,			__UpperCamelCase = "test"  )->					Union[str, Any]:
          UpperCAmelCase__							: Tuple 							=						self.get_test_dataloader(__UpperCamelCase  )

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

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

          UpperCAmelCase__							: Any 							=						self.post_process_function(__UpperCamelCase ,			__UpperCamelCase ,			output.predictions ,			"predict"  )
          UpperCAmelCase__							: Union[str, Any] 							=						self.compute_metrics(__UpperCamelCase  )

          # Prefix all keys with metric_key_prefix + '_'
          for key in list(metrics.keys()  ):
               if not key.startswith(F"{metric_key_prefix}_"  ):
                    UpperCAmelCase__							: Optional[int] 							=						metrics.pop(__UpperCamelCase  )

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



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase="./"  )->					Any:
          UpperCAmelCase__							: Optional[Any] 							=						self.eval_dataset
          UpperCAmelCase__							: Optional[Any] 							=						self.get_eval_dataloader(__UpperCamelCase  )

          UpperCAmelCase__							: Dict 							=						next(iter(__UpperCamelCase  )  )

          # saving device - to make it consistent
          UpperCAmelCase__							: Union[str, Any] 							=						torch.device("cuda" if torch.cuda.is_available() else "cpu"  )

          # convert to tuple
          UpperCAmelCase__							: Any 							=						tuple(v.to(__UpperCamelCase  ) for k, v in batch.items()  )

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

          UpperCAmelCase__							: List[Any] 							=						True

          UpperCAmelCase__							: str 							=						self.model.to(__UpperCamelCase  )

          model.eval()
          model.float()

          UpperCAmelCase__							: Any 							=						model.module if hasattr(__UpperCamelCase ,			"module"  ) else model
          quant_trainer.configure_model(__UpperCamelCase ,			self.quant_trainer_args  )

          UpperCAmelCase__							: Dict 							=						os.path.join(__UpperCamelCase ,			"model.onnx"  )
          logger.info(F"exporting model to {output_model_file}"  )

          UpperCAmelCase__							: List[Any] 							=						{0: "batch_size", 1: "seq_len"}

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





"""simple docstring"""




from manim import *







class   _lowercase						(       lowerCAmelCase_				):


     '''simple docstring'''





     def  lowerCAmelCase__       (      self  )->					Tuple:
          UpperCAmelCase__							: str 							=						Rectangle(height=0.5 ,			width=0.5  )
          UpperCAmelCase__							: List[str] 							=						Rectangle(height=0.46 ,			width=0.46  ).set_stroke(width=0  )

          UpperCAmelCase__							: List[Any] 							=						[mem.copy() for i in range(6  )]
          UpperCAmelCase__							: Union[str, Any] 							=						[mem.copy() for i in range(6  )]
          UpperCAmelCase__							: List[Any] 							=						VGroup(*__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0  )
          UpperCAmelCase__							: int 							=						VGroup(*__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0  )
          UpperCAmelCase__							: Tuple 							=						VGroup(__UpperCamelCase ,			__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0  )
          UpperCAmelCase__							: Tuple 							=						Text("CPU" ,			font_size=24  )
          UpperCAmelCase__							: Any 							=						Group(__UpperCamelCase ,			__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0.5 ,			aligned_edge=__UpperCamelCase  )
          cpu.move_to([-2.5, -0.5, 0]  )
          self.add(__UpperCamelCase  )

          UpperCAmelCase__							: List[Any] 							=						[mem.copy() for i in range(4  )]
          UpperCAmelCase__							: List[str] 							=						VGroup(*__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0  )
          UpperCAmelCase__							: Union[str, Any] 							=						Text("GPU" ,			font_size=24  )
          UpperCAmelCase__							: Dict 							=						Group(__UpperCamelCase ,			__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0.5 ,			aligned_edge=__UpperCamelCase  )
          gpu.move_to([-1, -1, 0]  )
          self.add(__UpperCamelCase  )

          UpperCAmelCase__							: Optional[int] 							=						[mem.copy() for i in range(6  )]
          UpperCAmelCase__							: List[str] 							=						VGroup(*__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0  )

          UpperCAmelCase__							: Tuple 							=						Text("Model" ,			font_size=24  )
          UpperCAmelCase__							: Dict 							=						Group(__UpperCamelCase ,			__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0.5 ,			aligned_edge=__UpperCamelCase  )
          model.move_to([3, -1.0, 0]  )
          self.add(__UpperCamelCase  )

          UpperCAmelCase__							: List[str] 							=						[]
          for i, rect in enumerate(__UpperCamelCase  ):
               rect.set_stroke(__UpperCamelCase  )
               # target = fill.copy().set_fill(YELLOW, opacity=0.7)
               # target.move_to(rect)
               # self.add(target)

               UpperCAmelCase__							: int 							=						Rectangle(height=0.46 / 4 ,			width=0.46 / 3  ).set_stroke(width=0.0  ).set_fill(__UpperCamelCase ,			opacity=0.7  )

               if i == 0:
                    cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT  ) ,			buff=0.02 ,			direction=__UpperCamelCase  )
                    cpu_target.set_x(cpu_target.get_x() + 0.1  )
               elif i == 3:
                    cpu_target.next_to(cpu_targs[0] ,			direction=__UpperCamelCase ,			buff=0.0  )
               else:
                    cpu_target.next_to(cpu_targs[i - 1] ,			direction=__UpperCamelCase ,			buff=0.0  )
               self.add(__UpperCamelCase  )
               cpu_targs.append(__UpperCamelCase  )

          UpperCAmelCase__							: List[Any] 							=						[mem.copy() for i in range(6  )]
          UpperCAmelCase__							: Any 							=						VGroup(*__UpperCamelCase  ).arrange(__UpperCamelCase ,			buff=0  )

          UpperCAmelCase__							: Tuple 							=						Text("Loaded Checkpoint" ,			font_size=24  )
          UpperCAmelCase__							: Any 							=						Group(__UpperCamelCase ,			__UpperCamelCase  ).arrange(__UpperCamelCase ,			aligned_edge=__UpperCamelCase ,			buff=0.4  )
          checkpoint.move_to([3, 0.5, 0]  )

          UpperCAmelCase__							: Optional[Any] 							=						Square(side_length=2.2  )
          key.move_to([-5, 2, 0]  )

          UpperCAmelCase__							: Any 							=						MarkupText(
              F"<b>Key:</b>\n\n<span fgcolor='{YELLOW}'>●</span> Empty Model" ,			font_size=18 ,			)

          key_text.move_to([-5, 2.4, 0]  )

          self.add(__UpperCamelCase ,			__UpperCamelCase  )

          UpperCAmelCase__							: str 							=						MarkupText(
              F"<span fgcolor='{BLUE}'>●</span> Checkpoint" ,			font_size=18 ,			)

          blue_text.next_to(__UpperCamelCase ,			DOWN * 2.4 ,			aligned_edge=key_text.get_left()  )

          UpperCAmelCase__							: Optional[Any] 							=						MarkupText(
              F"Next, a <i><span fgcolor=\"{BLUE}\">second</span></i> model is loaded into memory,\nwith the weights of a <span fgcolor=\"{BLUE}\">single shard</span>." ,			font_size=24 ,			)
          step_a.move_to([2, 2, 0]  )
          self.play(Write(__UpperCamelCase  ) ,			Write(__UpperCamelCase  )  )

          self.play(Write(__UpperCamelCase ,			run_time=1  ) ,			Create(__UpperCamelCase ,			run_time=1  )  )

          UpperCAmelCase__							: Union[str, Any] 							=						[]
          UpperCAmelCase__							: List[str] 							=						[]
          for i, rect in enumerate(__UpperCamelCase  ):
               UpperCAmelCase__							: Optional[Any] 							=						fill.copy().set_fill(__UpperCamelCase ,			opacity=0.7  )
               target.move_to(__UpperCamelCase  )
               first_animations.append(GrowFromCenter(__UpperCamelCase ,			run_time=1  )  )

               UpperCAmelCase__							: List[str] 							=						target.copy()
               cpu_target.generate_target()
               if i < 5:
                    cpu_target.target.move_to(cpu_left_col_base[i + 1]  )
               else:
                    cpu_target.target.move_to(cpu_right_col_base[i - 5]  )
               second_animations.append(MoveToTarget(__UpperCamelCase ,			run_time=1.5  )  )

          self.play(*__UpperCamelCase  )
          self.play(*__UpperCamelCase  )
          self.wait()



