


from typing import Dict

from .base import GenericTensor, Pipeline


class     __UpperCAmelCase							(       __A							):


     """simple docstring"""


     def 	snake_case_  (	self  ,       __A=None  ,       __A=None  ,       __A=None  ,       **__A    ):
        if tokenize_kwargs is None:
           __a					=  {}

        if truncation is not None:
           if "truncation" in tokenize_kwargs:
              raise ValueError(
                  """truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)"""    )
           __a					=  truncation

        __a					=  tokenize_kwargs

        __a					=  {}
        if return_tensors is not None:
           __a					=  return_tensors

        return preprocess_params, {}, postprocess_params


     def 	snake_case_  (	self  ,       __A  ,       **__A    ):
        __a					=  self.framework
        __a					=  self.tokenizer(__A  ,       return_tensors=__A  ,       **__A    )
        return model_inputs


     def 	snake_case_  (	self  ,       __A    ):
        __a					=  self.model(**__A    )
        return model_outputs


     def 	snake_case_  (	self  ,       __A  ,       __A=False    ):
        # [0] is the first available tensor, logits or last_hidden_state.
        if return_tensors:
           return model_outputs[0]
        if self.framework == "pt":
           return model_outputs[0].tolist()
        elif self.framework == "tf":
           return model_outputs[0].numpy().tolist()


     def __call__(	self  ,       *__A  ,       **__A    ):
        return super().__call__(*__A  ,       **__A    )







import logging
import re

import pytorch_quantization
import pytorch_quantization.nn as quant_nn
import torch
from pytorch_quantization import calib
from pytorch_quantization.tensor_quant import QuantDescriptor


SCREAMING_SNAKE_CASE							    =							logging.getLogger(__name__)

SCREAMING_SNAKE_CASE							    =							5_0  # max width of layer names
SCREAMING_SNAKE_CASE							    =							7_0  # max width of quantizer names




def  a   (lowerCAmelCase__   ):
   __a					=  parser.add_argument_group("""quant_trainer arguments"""   )
   group.add_argument("""--wprec"""		,				type=lowerCAmelCase__		,				default=8		,				help="""weight precision"""   )
   group.add_argument("""--aprec"""		,				type=lowerCAmelCase__		,				default=8		,				help="""activation precision"""   )
   group.add_argument("""--quant-per-tensor"""		,				action="""store_true"""		,				help="""per tensor weight scaling"""   )
   group.add_argument("""--quant-disable"""		,				action="""store_true"""		,				help="""disable all quantizers"""   )
   group.add_argument("""--quant-disable-embeddings"""		,				action="""store_true"""		,				help="""disable all embeddings quantizers"""   )
   group.add_argument("""--quant-disable-keyword"""		,				type=lowerCAmelCase__		,				nargs="""+"""		,				help="""disable quantizers by keyword"""   )
   group.add_argument("""--quant-disable-layer-module"""		,				type=lowerCAmelCase__		,				help="""disable quantizers by keyword under layer."""   )
   group.add_argument("""--quant-enable-layer-module"""		,				type=lowerCAmelCase__		,				help="""enable quantizers by keyword under layer"""   )
   group.add_argument("""--calibrator"""		,				default="""max"""		,				help="""which quantization range calibrator to use"""   )
   group.add_argument("""--percentile"""		,				default=lowerCAmelCase__		,				type=lowerCAmelCase__		,				help="""percentile for PercentileCalibrator"""   )
   group.add_argument("""--fuse-qkv"""		,				action="""store_true"""		,				help="""use the same scale factor for qkv"""   )
   group.add_argument("""--clip-gelu"""		,				metavar="""N"""		,				type=lowerCAmelCase__		,				help="""clip gelu output maximum value to N"""   )
   group.add_argument(
       """--recalibrate-weights"""		,				action="""store_true"""		,				help=(
           """recalibrate weight amaxes by taking the max of the weights."""
           """ amaxes will be computed with the current quantization granularity (axis)."""
       )		,				)




def  a   (lowerCAmelCase__   ):

   if args.calibrator == "max":
      __a					=  """max"""
   elif args.calibrator == "percentile":
      if args.percentile is None:
         raise ValueError("""Specify --percentile when using percentile calibrator"""   )
      __a					=  """histogram"""
   elif args.calibrator == "mse":
      __a					=  """histogram"""
   else:
      raise ValueError(f'''Invalid calibrator {args.calibrator}'''   )

   __a					=  QuantDescriptor(num_bits=args.aprec		,				calib_method=lowerCAmelCase__   )
   __a					=  QuantDescriptor(num_bits=args.wprec		,				axis=(None if args.quant_per_tensor else (0,))   )
   quant_nn.QuantLinear.set_default_quant_desc_input(lowerCAmelCase__   )
   quant_nn.QuantLinear.set_default_quant_desc_weight(lowerCAmelCase__   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__=False		,				lowerCAmelCase__=False   ):
   logger.info("""Configuring Model for Quantization"""   )
   logger.info(f'''using quantization package {pytorch_quantization.__file__}'''   )

   if not calib:
      if args.quant_disable_embeddings:
         set_quantizer_by_name(lowerCAmelCase__		,				["""embeddings"""]		,				which="""weight"""		,				_disabled=lowerCAmelCase__   )

      if args.quant_disable:
         set_quantizer_by_name(lowerCAmelCase__		,				[""""""]		,				_disabled=lowerCAmelCase__   )

      if args.quant_disable_keyword:
         set_quantizer_by_name(lowerCAmelCase__		,				args.quant_disable_keyword		,				_disabled=lowerCAmelCase__   )

      if args.quant_disable_layer_module:
         set_quantizer_by_name(lowerCAmelCase__		,				[r"""layer.\d+.""" + args.quant_disable_layer_module]		,				_disabled=lowerCAmelCase__   )

      if args.quant_enable_layer_module:
         set_quantizer_by_name(lowerCAmelCase__		,				[r"""layer.\d+.""" + args.quant_enable_layer_module]		,				_disabled=lowerCAmelCase__   )

      if args.recalibrate_weights:
         recalibrate_weights(lowerCAmelCase__   )

      if args.fuse_qkv:
         fuse_qkv(lowerCAmelCase__		,				lowerCAmelCase__   )

   if args.clip_gelu:
      clip_gelu(lowerCAmelCase__		,				args.clip_gelu   )

   # if args.local_rank in [-1, 0] and not calib:
   print_quant_summary(lowerCAmelCase__   )




def  a   (lowerCAmelCase__   ):
   logger.info("""Enabling Calibration"""   )
   for name, module in model.named_modules():
      if name.endswith("""_quantizer"""   ):
         if module._calibrator is not None:
            module.disable_quant()
            module.enable_calib()
         else:
            module.disable()
         logger.info(f'''{name:80}: {module}'''   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__   ):
   logger.info("""Loading calibrated amax"""   )
   for name, module in model.named_modules():
      if name.endswith("""_quantizer"""   ):
         if module._calibrator is not None:
            if isinstance(module._calibrator		,				calib.MaxCalibrator   ):
               module.load_calib_amax()
            else:
               module.load_calib_amax("""percentile"""		,				percentile=args.percentile   )
            module.enable_quant()
            module.disable_calib()
         else:
            module.enable()
   model.cuda()
   print_quant_summary(lowerCAmelCase__   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__   ):

   def fusea(lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   ):
      for mod in [qq, qk, qv]:
         if not hasattr(lowerCAmelCase__		,				"""_amax"""   ):
            print("""          WARNING: NO AMAX BUFFER"""   )
            return
      __a					=  qq._amax.detach().item()
      __a					=  qk._amax.detach().item()
      __a					=  qv._amax.detach().item()

      __a					=  max(lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   )
      qq._amax.fill_(lowerCAmelCase__   )
      qk._amax.fill_(lowerCAmelCase__   )
      qv._amax.fill_(lowerCAmelCase__   )
      logger.info(f'''          q={q:5.2f} k={k:5.2f} v={v:5.2f} -> {amax:5.2f}'''   )

   for name, mod in model.named_modules():
      if name.endswith(""".attention.self"""   ):
         logger.info(f'''FUSE_QKV: {name:{name_width}}'''   )
         fusea(mod.matmul_q_input_quantizer		,				mod.matmul_k_input_quantizer		,				mod.matmul_v_input_quantizer   )
         if args.quant_per_tensor:
            fusea(mod.query._weight_quantizer		,				mod.key._weight_quantizer		,				mod.value._weight_quantizer   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__   ):

   for name, mod in model.named_modules():
      if name.endswith(""".output.dense"""   ) and not name.endswith("""attention.output.dense"""   ):
         __a					=  mod._input_quantizer._amax.data.detach().item()
         mod._input_quantizer._amax.data.detach().clamp_(max=lowerCAmelCase__   )
         __a					=  mod._input_quantizer._amax.data.detach().item()
         logger.info(f'''CLIP_GELU: {name:{name_width}} amax: {amax_init:5.2f} -> {amax:5.2f}'''   )




def  a   (lowerCAmelCase__   ):

   for name, mod in model.named_modules():
      if hasattr(lowerCAmelCase__		,				"""_weight_quantizer"""   ) and mod._weight_quantizer.axis is not None:
         __a					=  mod.weight.shape[0]
         __a					=  mod._weight_quantizer._amax.detach()
         __a					=  torch.ones(lowerCAmelCase__		,				dtype=amax.dtype		,				device=amax.device   ) * amax
         print(f'''expanding {name} {amax} -> {mod._weight_quantizer._amax}'''   )




def  a   (lowerCAmelCase__   ):

   for name, mod in model.named_modules():
      if hasattr(lowerCAmelCase__		,				"""_weight_quantizer"""   ):
         if not hasattr(mod.weight_quantizer		,				"""_amax"""   ):
            print("""RECALIB: {name:{name_width}} WARNING: NO AMAX BUFFER"""   )
            continue

         # determine which axes to reduce across
         # e.g. a 4D tensor quantized per axis 0 should reduce over (1,2,3)
         __a					=  set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis   )
         __a					=  set(range(len(mod.weight.size()   )   )   ) - axis_set
         __a					=  pytorch_quantization.utils.reduce_amax(mod.weight		,				axis=lowerCAmelCase__		,				keepdims=lowerCAmelCase__   ).detach()
         logger.info(f'''RECALIB: {name:{name_width}} {mod._weight_quantizer._amax.flatten()} -> {amax.flatten()}'''   )
         __a					=  amax




def  a   (lowerCAmelCase__		,				lowerCAmelCase__=25		,				lowerCAmelCase__=180		,				lowerCAmelCase__=None   ):

   if ignore is None:
      __a					=  []
   elif not isinstance(lowerCAmelCase__		,				lowerCAmelCase__   ):
      __a					=  [ignore]

   __a					=  0
   for name, mod in model.named_modules():
      if not hasattr(lowerCAmelCase__		,				"""weight"""   ):
         continue
      __a					=  max(lowerCAmelCase__		,				len(lowerCAmelCase__   )   )

   for name, mod in model.named_modules():
      __a					=  getattr(lowerCAmelCase__		,				"""_input_quantizer"""		,				lowerCAmelCase__   )
      __a					=  getattr(lowerCAmelCase__		,				"""_weight_quantizer"""		,				lowerCAmelCase__   )
      if not hasattr(lowerCAmelCase__		,				"""weight"""   ):
         continue
      if type(lowerCAmelCase__   ) in ignore:
         continue
      if [True for s in ignore if type(lowerCAmelCase__   ) is str and s in name]:
         continue
      __a					=  f'''Act:{input_q.extra_repr()}'''
      __a					=  f'''Wgt:{weight_q.extra_repr()}'''
      __a					=  f'''{name:{name_width}} {act_str} {wgt_str}'''
      if len(lowerCAmelCase__   ) <= line_width:
         logger.info(lowerCAmelCase__   )
      else:
         logger.info(f'''{name:{name_width}} {act_str}'''   )
         logger.info(f'''{'  ':{name_width}} {wgt_str}'''   )




def  a   (lowerCAmelCase__   ):
   __a					=  0
   for name, mod in model.named_modules():
      if isinstance(lowerCAmelCase__		,				pytorch_quantization.nn.TensorQuantizer   ):
         print(f'''{name:80} {mod}'''   )
         count += 1
   print(f'''{count} TensorQuantizers found in model'''   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   ):
   __a					=  getattr(lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   )
   if quantizer_mod is not None:
      assert hasattr(lowerCAmelCase__		,				lowerCAmelCase__   )
      setattr(lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   )
   else:
      logger.warning(f'''{name} has no {quantizer}'''   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__="both"		,				**lowerCAmelCase__   ):
   __a					=  f'''Warning: changing {which} quantizers of {name:{qname_width}}'''
   for k, v in kwargs.items():
      s += f''' {k}={v}'''
      if which in ["input", "both"]:
         set_quantizer(lowerCAmelCase__		,				lowerCAmelCase__		,				"""_input_quantizer"""		,				lowerCAmelCase__		,				lowerCAmelCase__   )
      if which in ["weight", "both"]:
         set_quantizer(lowerCAmelCase__		,				lowerCAmelCase__		,				"""_weight_quantizer"""		,				lowerCAmelCase__		,				lowerCAmelCase__   )
   logger.info(lowerCAmelCase__   )




def  a   (lowerCAmelCase__		,				lowerCAmelCase__		,				**lowerCAmelCase__   ):

   for name, mod in model.named_modules():
      if hasattr(lowerCAmelCase__		,				"""_input_quantizer"""   ) or hasattr(lowerCAmelCase__		,				"""_weight_quantizer"""   ):
         for n in names:
            if re.search(lowerCAmelCase__		,				lowerCAmelCase__   ):
               set_quantizers(lowerCAmelCase__		,				lowerCAmelCase__		,				**lowerCAmelCase__   )
      elif name.endswith("""_quantizer"""   ):
         for n in names:
            if re.search(lowerCAmelCase__		,				lowerCAmelCase__   ):
               __a					=  f'''Warning: changing {name:{name_width}}'''
               for k, v in kwargs.items():
                  s += f''' {k}={v}'''
                  setattr(lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   )
               logger.info(lowerCAmelCase__   )



