


import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE							    =							logging.get_logger(__name__)
SCREAMING_SNAKE_CASE							    =							{'vocab_file': 'spiece.model'}

SCREAMING_SNAKE_CASE							    =							{
    'vocab_file': {
        'albert-base-v1': 'https://huggingface.co/albert-base-v1/resolve/main/spiece.model',
        'albert-large-v1': 'https://huggingface.co/albert-large-v1/resolve/main/spiece.model',
        'albert-xlarge-v1': 'https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model',
        'albert-xxlarge-v1': 'https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model',
        'albert-base-v2': 'https://huggingface.co/albert-base-v2/resolve/main/spiece.model',
        'albert-large-v2': 'https://huggingface.co/albert-large-v2/resolve/main/spiece.model',
        'albert-xlarge-v2': 'https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model',
        'albert-xxlarge-v2': 'https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model',
    }
}

SCREAMING_SNAKE_CASE							    =							{
    'albert-base-v1': 5_1_2,
    'albert-large-v1': 5_1_2,
    'albert-xlarge-v1': 5_1_2,
    'albert-xxlarge-v1': 5_1_2,
    'albert-base-v2': 5_1_2,
    'albert-large-v2': 5_1_2,
    'albert-xlarge-v2': 5_1_2,
    'albert-xxlarge-v2': 5_1_2,
}

SCREAMING_SNAKE_CASE							    =							'▁'


class     __UpperCAmelCase							(       __A							):


     """simple docstring"""
     _lowerCamelCase   =      VOCAB_FILES_NAMES
     _lowerCamelCase   =      PRETRAINED_VOCAB_FILES_MAP
     _lowerCamelCase   =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES


     def __init__(	self  ,       __A  ,       __A=True  ,       __A=True  ,       __A=False  ,       __A="[CLS]"  ,       __A="[SEP]"  ,       __A="<unk>"  ,       __A="[SEP]"  ,       __A="<pad>"  ,       __A="[CLS]"  ,       __A="[MASK]"  ,       __A = None  ,       **__A  ,       ):
        # Mask token behave like a normal word, i.e. include the space before it and
        # is included in the raw text, there should be a match in a non-normalized sentence.
        __a					=  (
            AddedToken(__A  ,       lstrip=__A  ,       rstrip=__A  ,       normalized=__A    )
            if isinstance(__A  ,       __A    )
            else mask_token
        )

        __a					=  {} if sp_model_kwargs is None else sp_model_kwargs

        super().__init__(
            do_lower_case=__A  ,       remove_space=__A  ,       keep_accents=__A  ,       bos_token=__A  ,       eos_token=__A  ,       unk_token=__A  ,       sep_token=__A  ,       pad_token=__A  ,       cls_token=__A  ,       mask_token=__A  ,       sp_model_kwargs=self.sp_model_kwargs  ,       **__A  ,       )

        __a					=  do_lower_case
        __a					=  remove_space
        __a					=  keep_accents
        __a					=  vocab_file

        __a					=  spm.SentencePieceProcessor(**self.sp_model_kwargs    )
        self.sp_model.Load(__A    )


     @property
     def 	snake_case_  (	self    ):
        return len(self.sp_model    )


     def 	snake_case_  (	self    ):
        __a					=  {self.convert_ids_to_tokens(__A    ): i for i in range(self.vocab_size    )}
        vocab.update(self.added_tokens_encoder    )
        return vocab


     def __getstate__(	self    ):
        __a					=  self.__dict__.copy()
        __a					=  None
        return state


     def __setstate__(	self  ,       __A    ):
        __a					=  d

        # for backward compatibility
        if not hasattr(self  ,       """sp_model_kwargs"""    ):
           __a					=  {}

        __a					=  spm.SentencePieceProcessor(**self.sp_model_kwargs    )
        self.sp_model.Load(self.vocab_file    )


     def 	snake_case_  (	self  ,       __A    ):
        if self.remove_space:
           __a					=  """ """.join(inputs.strip().split()    )
        else:
           __a					=  inputs
        __a					=  outputs.replace("""``"""  ,       """\""""    ).replace("""''"""  ,       """\""""    )

        if not self.keep_accents:
           __a					=  unicodedata.normalize("""NFKD"""  ,       __A    )
           __a					=  """""".join([c for c in outputs if not unicodedata.combining(__A    )]    )
        if self.do_lower_case:
           __a					=  outputs.lower()

        return outputs


     def 	snake_case_  (	self  ,       __A    ):
        __a					=  self.preprocess_text(__A    )
        __a					=  self.sp_model.encode(__A  ,       out_type=__A    )
        __a					=  []
        for piece in pieces:
           if len(__A    ) > 1 and piece[-1] == str(""","""    ) and piece[-2].isdigit():
              __a					=  self.sp_model.EncodeAsPieces(piece[:-1].replace(__A  ,       """"""    )    )
              if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
                 if len(cur_pieces[0]    ) == 1:
                    __a					=  cur_pieces[1:]
                 else:
                    __a					=  cur_pieces[0][1:]
              cur_pieces.append(piece[-1]    )
              new_pieces.extend(__A    )
           else:
              new_pieces.append(__A    )

        return new_pieces


     def 	snake_case_  (	self  ,       __A    ):
        return self.sp_model.PieceToId(__A    )


     def 	snake_case_  (	self  ,       __A    ):
        return self.sp_model.IdToPiece(__A    )


     def 	snake_case_  (	self  ,       __A    ):
        __a					=  []
        __a					=  """"""
        __a					=  False
        for token in tokens:
           # make sure that special tokens are not decoded using sentencepiece model
           if token in self.all_special_tokens:
              if not prev_is_special:
                 out_string += " "
              out_string += self.sp_model.decode(__A    ) + token
              __a					=  True
              __a					=  []
           else:
              current_sub_tokens.append(__A    )
              __a					=  False
        out_string += self.sp_model.decode(__A    )
        return out_string.strip()


     def 	snake_case_  (	self  ,       __A  ,       __A = None    ):
        __a					=  [self.sep_token_id]
        __a					=  [self.cls_token_id]
        if token_ids_a is None:
           return cls + token_ids_a + sep
        return cls + token_ids_a + sep + token_ids_a + sep


     def 	snake_case_  (	self  ,       __A  ,       __A = None  ,       __A = False    ):

        if already_has_special_tokens:
           return super().get_special_tokens_mask(
               token_ids_a=__A  ,       token_ids_a=__A  ,       already_has_special_tokens=__A    )

        if token_ids_a is not None:
           return [1] + ([0] * len(__A    )) + [1] + ([0] * len(__A    )) + [1]
        return [1] + ([0] * len(__A    )) + [1]


     def 	snake_case_  (	self  ,       __A  ,       __A = None    ):
        __a					=  [self.sep_token_id]
        __a					=  [self.cls_token_id]

        if token_ids_a is None:
           return len(cls + token_ids_a + sep    ) * [0]
        return len(cls + token_ids_a + sep    ) * [0] + len(token_ids_a + sep    ) * [1]


     def 	snake_case_  (	self  ,       __A  ,       __A = None    ):
        if not os.path.isdir(__A    ):
           logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''    )
           return
        __a					=  os.path.join(
            __A  ,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]    )

        if os.path.abspath(self.vocab_file    ) != os.path.abspath(__A    ) and os.path.isfile(self.vocab_file    ):
           copyfile(self.vocab_file  ,       __A    )
        elif not os.path.isfile(self.vocab_file    ):
           with open(__A  ,       """wb"""    ) as fi:
              __a					=  self.sp_model.serialized_model_proto()
              fi.write(__A    )

        return (out_vocab_file,)







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__   )



