




'''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   __A (							_UpperCamelCase  ):



   """simple docstring"""







   def __init__(      self				, *_lowerCamelCase				, _lowerCamelCase=None				, _lowerCamelCase=None				, _lowerCamelCase=None				, **_lowerCamelCase						)->       int:
      super().__init__(*_UpperCAmelCase				, **_UpperCAmelCase						)
      lowercase__       							=							eval_examples
      lowercase__       							=							post_process_function
      lowercase__       							=							quant_trainer_args
      lowercase__       							=							1_2_8  # default number of calibration samples

   def snake_case_(      self				, _lowerCamelCase=None						)->       Tuple:
      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(_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 snake_case_(      self				, _lowerCamelCase=None						)->       Tuple:
      lowercase__       							=							self.train_dataset if calib_dataset is None else calib_dataset
      lowercase__       							=							self.get_calib_dataloader(_UpperCAmelCase						)

      lowercase__       							=							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
         lowercase__       							=							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						)
      lowercase__       							=							model

   def snake_case_(      self				, _lowerCamelCase=None				, _lowerCamelCase=None				, _lowerCamelCase=None				, _lowerCamelCase = "eval"						)->       Tuple:
      lowercase__       							=							self.eval_dataset if eval_dataset is None else eval_dataset
      lowercase__       							=							self.get_eval_dataloader(_UpperCAmelCase						)
      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(
             _UpperCAmelCase				, description='''Evaluation'''				, prediction_loss_only=True if compute_metrics is None else None				, ignore_keys=_UpperCAmelCase				, )
      finally:
         lowercase__       							=							compute_metrics

      if self.post_process_function is not None and self.compute_metrics is not None:
         lowercase__       							=							self.post_process_function(_UpperCAmelCase				, _UpperCAmelCase				, output.predictions						)
         lowercase__       							=							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}_'''						):
               lowercase__       							=							metrics.pop(_UpperCAmelCase						)

         self.log(_UpperCAmelCase						)
      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				, _UpperCAmelCase						)
      return metrics

   def snake_case_(      self				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase=None				, _lowerCamelCase = "test"						)->       str:
      lowercase__       							=							self.get_test_dataloader(_UpperCAmelCase						)

      # 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(
             _UpperCAmelCase				, description='''Prediction'''				, prediction_loss_only=True if compute_metrics is None else None				, ignore_keys=_UpperCAmelCase				, )
      finally:
         lowercase__       							=							compute_metrics

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

      lowercase__       							=							self.post_process_function(_UpperCAmelCase				, _UpperCAmelCase				, output.predictions				, '''predict'''						)
      lowercase__       							=							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}_'''						):
            lowercase__       							=							metrics.pop(_UpperCAmelCase						)

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

   def snake_case_(      self				, _lowerCamelCase="./"						)->       List[str]:
      lowercase__       							=							self.eval_dataset
      lowercase__       							=							self.get_eval_dataloader(_UpperCAmelCase						)

      lowercase__       							=							next(iter(_UpperCAmelCase						)						)

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

      # convert to tuple
      lowercase__       							=							tuple(v.to(_UpperCAmelCase						) 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(_UpperCAmelCase						)

      model.eval()
      model.float()

      lowercase__       							=							model.module if hasattr(_UpperCAmelCase				, '''module'''						) else model
      quant_trainer.configure_model(_UpperCAmelCase				, self.quant_trainer_args						)

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

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

      torch.onnx.export(
          _UpperCAmelCase				, _UpperCAmelCase				, _UpperCAmelCase				, export_params=_UpperCAmelCase				, opset_version=1_3				, 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 __future__ import annotations

from math import gcd



def 	__lowerCamelCase	( lowerCAmelCase_		,     lowerCAmelCase_ = 2		,     lowerCAmelCase_ = 1		,     lowerCAmelCase_ = 3		,     )     -> int | None:
		# A value less than 2 can cause an infinite loop in the algorithm.
		if num < 2:
				raise ValueError('The input value cannot be less than 2'    )

		# Because of the relationship between ``f(f(x))`` and ``f(x)``, this
		# algorithm struggles to find factors that are divisible by two.
		# As a workaround, we specifically check for two and even inputs.
		#   See: https://math.stackexchange.com/a/2856214/165820
		if num > 2 and num % 2 == 0:
				return 2

		# Pollard's Rho algorithm requires a function that returns pseudorandom
		# values between 0 <= X < ``num``.  It doesn't need to be random in the
		# sense that the output value is cryptographically secure or difficult
		# to calculate, it only needs to be random in the sense that all output
		# values should be equally likely to appear.
		# For this reason, Pollard suggested using ``f(x) = (x**2 - 1) % num``
		# However, the success of Pollard's algorithm isn't guaranteed and is
		# determined in part by the initial seed and the chosen random function.
		# To make retries easier, we will instead use ``f(x) = (x**2 + C) % num``
		# where ``C`` is a value that we can modify between each attempt.
		def rand_fn(lowerCAmelCase_		,     lowerCAmelCase_		,     lowerCAmelCase_    ) -> int:
				return (pow(lowerCAmelCase_		,     2    ) + step) % modulus

		for _ in range(lowerCAmelCase_    ):
				# These track the position within the cycle detection logic.
				_a	: Optional[int]								=				seed
				_a	: str								=				seed

				while True:
						# At each iteration, the tortoise moves one step and the hare moves two.
						_a	: str								=				rand_fn(lowerCAmelCase_		,     lowerCAmelCase_		,     lowerCAmelCase_    )
						_a	: Dict								=				rand_fn(lowerCAmelCase_		,     lowerCAmelCase_		,     lowerCAmelCase_    )
						_a	: Tuple								=				rand_fn(lowerCAmelCase_		,     lowerCAmelCase_		,     lowerCAmelCase_    )

						# At some point both the tortoise and the hare will enter a cycle whose
						# length ``p`` is a divisor of ``num``.  Once in that cycle, at some point
						# the tortoise and hare will end up on the same value modulo ``p``.
						# We can detect when this happens because the position difference between
						# the tortoise and the hare will share a common divisor with ``num``.
						_a	: Optional[Any]								=				gcd(hare - tortoise		,     lowerCAmelCase_    )

						if divisor == 1:
								# No common divisor yet, just keep searching.
								continue
						else:
								# We found a common divisor!
								if divisor == num:
										# Unfortunately, the divisor is ``num`` itself and is useless.
										break
								else:
										# The divisor is a nontrivial factor of ``num``!
										return divisor

        # If we made it here, then this attempt failed.
        # We need to pick a new starting seed for the tortoise and hare
        # in addition to a new step value for the random function.
        # To keep this example implementation deterministic, the
        # new values will be generated based on currently available
        # values instead of using something like ``random.randint``.

        # We can use the hare's position as the new seed.
        # This is actually what Richard Brent's the "optimized" variant does.
				_a	: Union[str, Any]								=				hare

				# The new step value for the random function can just be incremented.
				# At first the results will be similar to what the old function would
				# have produced, but the value will quickly diverge after a bit.
				step += 1

		# We haven't found a divisor within the requested number of attempts.
		# We were unlucky or ``num`` itself is actually prime.
		return None


if __name__ == "__main__":
							import argparse

							__lowerCAmelCase		     =  argparse.ArgumentParser()
							parser.add_argument(
							    '''num''',
							    type=int,
							    help='''The value to find a divisor of''',
							)
							parser.add_argument(
							    '''--attempts''',
							    type=int,
							    default=3,
							    help='''The number of attempts before giving up''',
							)
							__lowerCAmelCase		     =  parser.parse_args()

							__lowerCAmelCase		     =  pollard_rho(args.num, attempts=args.attempts)
							if divisor is None:
														print(f"""{args.num} is probably prime""")
							else:
														__lowerCAmelCase		     =  args.num // divisor
														print(f"""{args.num} = {divisor} * {quotient}""")




