


'''simple docstring'''







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


lowerCAmelCase    :     Any					=						logging.getLogger(__name__)

lowerCAmelCase    :     int					=						50  # max width of layer names
lowerCAmelCase    :     Tuple					=						70  # max width of quantizer names






def 			lowercase				(_A					):







       """simple docstring"""







       _lowerCAmelCase    :							Dict   			=       parser.add_argument_group('quant_trainer arguments'					)
       group.add_argument('--wprec'   ,						type=_A   ,						default=8   ,						help='weight precision'					)
       group.add_argument('--aprec'   ,						type=_A   ,						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=_A   ,						nargs='+'   ,						help='disable quantizers by keyword'					)
       group.add_argument('--quant-disable-layer-module'   ,						type=_A   ,						help='disable quantizers by keyword under layer.'					)
       group.add_argument('--quant-enable-layer-module'   ,						type=_A   ,						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=_A   ,						type=_A   ,						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=_A   ,						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 			lowercase				(_A					):







       """simple docstring"""







       if args.calibrator == "max":
              _lowerCAmelCase    :							Dict   			=       'max'
       elif args.calibrator == "percentile":
              if args.percentile is None:
                     raise ValueError('Specify --percentile when using percentile calibrator'					)
              _lowerCAmelCase    :							List[str]   			=       'histogram'
       elif args.calibrator == "mse":
              _lowerCAmelCase    :							Any   			=       'histogram'
       else:
              raise ValueError(f'Invalid calibrator {args.calibrator}'					)

       _lowerCAmelCase    :							Dict   			=       QuantDescriptor(num_bits=args.aprec   ,						calib_method=_A					)
       _lowerCAmelCase    :							Dict   			=       QuantDescriptor(num_bits=args.wprec   ,						axis=(None if args.quant_per_tensor else (0,))					)
       quant_nn.QuantLinear.set_default_quant_desc_input(_A					)
       quant_nn.QuantLinear.set_default_quant_desc_weight(_A					)






def 			lowercase				(_A   ,						_A   ,						_A=False   ,						_A=False					):







       """simple docstring"""







       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(_A   ,						['embeddings']   ,						which='weight'   ,						_disabled=_A					)

              if args.quant_disable:
                     set_quantizer_by_name(_A   ,						['']   ,						_disabled=_A					)

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

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

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

              if args.recalibrate_weights:
                     recalibrate_weights(_A					)

              if args.fuse_qkv:
                     fuse_qkv(_A   ,						_A					)

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

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






def 			lowercase				(_A					):







       """simple docstring"""







       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 			lowercase				(_A   ,						_A					):







       """simple docstring"""







       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(_A					)






def 			lowercase				(_A   ,						_A					):







       """simple docstring"""







       def fusea(_A   ,						_A   ,						_A					):
              for mod in [qq, qk, qv]:
                     if not hasattr(_A   ,						'_amax'					):
                            print('          WARNING: NO AMAX BUFFER'					)
                            return
              _lowerCAmelCase    :							List[str]   			=       qq._amax.detach().item()
              _lowerCAmelCase    :							Union[str, Any]   			=       qk._amax.detach().item()
              _lowerCAmelCase    :							Tuple   			=       qv._amax.detach().item()

              _lowerCAmelCase    :							List[Any]   			=       max(_A   ,						_A   ,						_A					)
              qq._amax.fill_(_A					)
              qk._amax.fill_(_A					)
              qv._amax.fill_(_A					)
              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 			lowercase				(_A   ,						_A					):







       """simple docstring"""







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






def 			lowercase				(_A					):







       """simple docstring"""







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






def 			lowercase				(_A					):







       """simple docstring"""







       for name, mod in model.named_modules():
              if hasattr(_A   ,						'_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)
                     _lowerCAmelCase    :							List[str]   			=       set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis					)
                     _lowerCAmelCase    :							Any   			=       set(range(len(mod.weight.size()					)					)					) - axis_set
                     _lowerCAmelCase    :							str   			=       pytorch_quantization.utils.reduce_amax(mod.weight   ,						axis=_A   ,						keepdims=_A					).detach()
                     logger.info(f'RECALIB: {name:{name_width}} {mod._weight_quantizer._amax.flatten()} -> {amax.flatten()}'					)
                     _lowerCAmelCase    :							int   			=       amax






def 			lowercase				(_A   ,						_A=2_5   ,						_A=1_8_0   ,						_A=None					):







       """simple docstring"""







       if ignore is None:
              _lowerCAmelCase    :							Dict   			=       []
       elif not isinstance(_A   ,						_A					):
              _lowerCAmelCase    :							Tuple   			=       [ignore]

       _lowerCAmelCase    :							Dict   			=       0
       for name, mod in model.named_modules():
              if not hasattr(_A   ,						'weight'					):
                     continue
              _lowerCAmelCase    :							Any   			=       max(_A   ,						len(_A					)					)

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






def 			lowercase				(_A					):







       """simple docstring"""







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






def 			lowercase				(_A   ,						_A   ,						_A   ,						_A   ,						_A					):







       """simple docstring"""







       _lowerCAmelCase    :							Any   			=       getattr(_A   ,						_A   ,						_A					)
       if quantizer_mod is not None:
              assert hasattr(_A   ,						_A					)
              setattr(_A   ,						_A   ,						_A					)
       else:
              logger.warning(f'{name} has no {quantizer}'					)






def 			lowercase				(_A   ,						_A   ,						_A="both"   ,						**_A					):







       """simple docstring"""







       _lowerCAmelCase    :							List[Any]   			=       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(_A   ,						_A   ,						'_input_quantizer'   ,						_A   ,						_A					)
              if which in ["weight", "both"]:
                     set_quantizer(_A   ,						_A   ,						'_weight_quantizer'   ,						_A   ,						_A					)
       logger.info(_A					)






def 			lowercase				(_A   ,						_A   ,						**_A					):







       """simple docstring"""







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





'''simple docstring'''







import unittest

from transformers import AutoTokenizer, is_flax_available
from transformers.testing_utils import require_flax, require_sentencepiece, require_tokenizers, slow


if is_flax_available():
       import jax.numpy as jnp

       from transformers import FlaxXLMRobertaModel
@require_sentencepiece
@require_tokenizers
@require_flax
class 	UpperCamelCase__					( unittest.TestCase	):



     """simple docstring"""






     @slow
     def a	( self   ):

            '''simple docstring'''

            _lowerCAmelCase    :							List[str]   			=       FlaxXLMRobertaModel.from_pretrained('xlm-roberta-base'   )
            _lowerCAmelCase    :							Optional[Any]   			=       AutoTokenizer.from_pretrained('xlm-roberta-base'   )
            _lowerCAmelCase    :							Dict   			=       'The dog is cute and lives in the garden house'
            _lowerCAmelCase    :							List[str]   			=       jnp.array([tokenizer.encode(snake_case__   )]   )

            _lowerCAmelCase    :							Optional[int]   			=       (1, 12, 768)  # batch_size, sequence_length, embedding_vector_dim
            _lowerCAmelCase    :							Tuple   			=       jnp.array(
                [[-0.0101, 0.1218, -0.0803, 0.0801, 0.1327, 0.0776, -0.1215, 0.2383, 0.3338, 0.3106, 0.0300, 0.0252]]   )

            _lowerCAmelCase    :							Union[str, Any]   			=       model(snake_case__   )['last_hidden_state']
            self.assertEqual(output.shape      ,	snake_case__   )
            # compare the actual values for a slice of last dim
            self.assertTrue(jnp.allclose(output[:, :, -1]      ,	snake_case__      ,	atol=1E-3   )   )

