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_:					List[Any]  =logging.getLogger(__name__)

SCREAMING_SNAKE_CASE_:					List[Any]  =50  # max width of layer names
SCREAMING_SNAKE_CASE_:					Tuple  =70  # max width of quantizer names





def      lowerCAmelCase_    ( snake_case_	:			List[Any]      )   ->				int:


     '''simple docstring'''




     UpperCAmelCase_	   =       parser.add_argument_group("quant_trainer arguments"      )
     group.add_argument("--wprec" ,   type=snake_case_ ,   default=8 ,   help="weight precision"      )
     group.add_argument("--aprec" ,   type=snake_case_ ,   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=snake_case_ ,   nargs="+" ,   help="disable quantizers by keyword"      )
     group.add_argument("--quant-disable-layer-module" ,   type=snake_case_ ,   help="disable quantizers by keyword under layer."      )
     group.add_argument("--quant-enable-layer-module" ,   type=snake_case_ ,   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=snake_case_ ,   type=snake_case_ ,   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=snake_case_ ,   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      lowerCAmelCase_    ( snake_case_	:			int      )   ->				str:


     '''simple docstring'''




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

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





def      lowerCAmelCase_    ( snake_case_	:			List[Any] ,   snake_case_	:			Any ,   snake_case_	:			Union[str, Any]=False ,   snake_case_	:			int=False      )   ->				Dict:


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

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

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

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

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

          if args.recalibrate_weights:
               recalibrate_weights(snake_case_      )

          if args.fuse_qkv:
               fuse_qkv(snake_case_ ,   snake_case_      )

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

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





def      lowerCAmelCase_    ( snake_case_	:			Dict      )   ->				str:


     '''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      lowerCAmelCase_    ( snake_case_	:			Tuple ,   snake_case_	:			str      )   ->				List[Any]:


     '''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(snake_case_      )





def      lowerCAmelCase_    ( snake_case_	:			int ,   snake_case_	:			Optional[Any]      )   ->				Tuple:


     '''simple docstring'''




     def fusea(snake_case_	:			Union[str, Any] ,   snake_case_	:			Optional[Any] ,   snake_case_	:			str      ):
          for mod in [qq, qk, qv]:
               if not hasattr(snake_case_ ,   "_amax"      ):
                    print("          WARNING: NO AMAX BUFFER"      )
                    return
          UpperCAmelCase_	   =       qq._amax.detach().item()
          UpperCAmelCase_	   =       qk._amax.detach().item()
          UpperCAmelCase_	   =       qv._amax.detach().item()

          UpperCAmelCase_	   =       max(snake_case_ ,   snake_case_ ,   snake_case_      )
          qq._amax.fill_(snake_case_      )
          qk._amax.fill_(snake_case_      )
          qv._amax.fill_(snake_case_      )
          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      lowerCAmelCase_    ( snake_case_	:			int ,   snake_case_	:			Tuple      )   ->				Any:


     '''simple docstring'''




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





def      lowerCAmelCase_    ( snake_case_	:			Optional[Any]      )   ->				Optional[int]:


     '''simple docstring'''




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





def      lowerCAmelCase_    ( snake_case_	:			Union[str, Any]      )   ->				Dict:


     '''simple docstring'''




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





def      lowerCAmelCase_    ( snake_case_	:			List[str] ,   snake_case_	:			Union[str, Any]=25 ,   snake_case_	:			Tuple=1_80 ,   snake_case_	:			List[Any]=None      )   ->				int:


     '''simple docstring'''




     if ignore is None:
          UpperCAmelCase_	   =       []
     elif not isinstance(snake_case_ ,   snake_case_      ):
          UpperCAmelCase_	   =       [ignore]

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

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





def      lowerCAmelCase_    ( snake_case_	:			Optional[Any]      )   ->				List[Any]:


     '''simple docstring'''




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





def      lowerCAmelCase_    ( snake_case_	:			Any ,   snake_case_	:			Any ,   snake_case_	:			Dict ,   snake_case_	:			List[str] ,   snake_case_	:			Any      )   ->				Any:


     '''simple docstring'''




     UpperCAmelCase_	   =       getattr(snake_case_ ,   snake_case_ ,   snake_case_      )
     if quantizer_mod is not None:
          assert hasattr(snake_case_ ,   snake_case_      )
          setattr(snake_case_ ,   snake_case_ ,   snake_case_      )
     else:
          logger.warning(f"""{name} has no {quantizer}"""      )





def      lowerCAmelCase_    ( snake_case_	:			Dict ,   snake_case_	:			List[str] ,   snake_case_	:			Optional[Any]="both" ,   **snake_case_	:			Optional[int]      )   ->				List[Any]:


     '''simple docstring'''




     UpperCAmelCase_	   =       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(snake_case_ ,   snake_case_ ,   "_input_quantizer" ,   snake_case_ ,   snake_case_      )
          if which in ["weight", "both"]:
               set_quantizer(snake_case_ ,   snake_case_ ,   "_weight_quantizer" ,   snake_case_ ,   snake_case_      )
     logger.info(snake_case_      )






def      lowerCAmelCase_    ( snake_case_	:			Any ,   snake_case_	:			Tuple ,   **snake_case_	:			Optional[int]      )   ->				Any:


     '''simple docstring'''




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





'''simple docstring'''





from unittest import TestCase

from datasets import Dataset
from minhash_deduplication import deduplicate_dataset, make_duplicate_clusters





def      lowerCAmelCase_    ( )   ->				str:


     '''simple docstring'''




     UpperCAmelCase_	   =       {
         "repo_name": ["test_repo1", "test_repo2", "test_repo3"],
         "path": ["test_1.py", "test_2.py", "unit_test.py"],
         "content": ["a " * 20, "a " * 30, "b " * 7],
     }
     UpperCAmelCase_	   =       Dataset.from_dict(snake_case_      )
     return dataset






class 	__A				(			UpperCamelCase__     ):



    def  _lowercase						(self   :   str			):
         UpperCAmelCase_	   =       get_dataset()
         UpperCAmelCase_	   =       make_duplicate_clusters(__a       ,      0.85			)
         self.assertEqual(len(duplicate_clusters[0]			)       ,      2			)






    def  _lowercase						(self   :   Optional[Any]			):
         UpperCAmelCase_	   =       get_dataset()
         UpperCAmelCase_   ,			UpperCAmelCase_	   =       deduplicate_dataset(__a			)
         self.assertEqual(len(__a			)       ,      2			)
         print(__a			)
         self.assertEqual(duplicate_clusters[0][0]["copies"]       ,      2			)
         self.assertEqual(duplicate_clusters[0][0]["is_extreme"]       ,      __a			)




