






'''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


SCREAMING_SNAKE_CASE							=						logging.getLogger(__name__)

SCREAMING_SNAKE_CASE							=						50  # max width of layer names
SCREAMING_SNAKE_CASE							=						70  # max width of quantizer names


def 				lowercase_					(     __A			:							List[str]      )							->   Optional[Any]:



 """simple docstring"""




 lowercase       : str						=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			:							Union[str, Any]      )							->   Optional[int]:



 """simple docstring"""




 if args.calibrator == "max":
  lowercase       : Union[str, Any]						='''max'''
 elif args.calibrator == "percentile":
  if args.percentile is None:
   raise ValueError('''Specify --percentile when using percentile calibrator'''      )
  lowercase       : int						='''histogram'''
 elif args.calibrator == "mse":
  lowercase       : int						='''histogram'''
 else:
  raise ValueError(F'Invalid calibrator {args.calibrator}'      )

 lowercase       : Tuple						=QuantDescriptor(num_bits=args.aprec       ,   calib_method=__A      )
 lowercase       : List[Any]						=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			:							int       ,   __A			:							int       ,   __A			:							List[Any]=False       ,   __A			:							int=False      )							->   List[Any]:



 """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			:							Union[str, Any]      )							->   Tuple:



 """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			:							Any       ,   __A			:							Optional[Any]      )							->   Optional[int]:



 """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			:							str       ,   __A			:							Optional[Any]      )							->   Union[str, Any]:



 """simple docstring"""




 def fusea(__A			:							str       ,   __A			:							Optional[int]       ,   __A			:							Tuple      ):
  for mod in [qq, qk, qv]:
   if not hasattr(__A       ,   '''_amax'''      ):
    print('''          WARNING: NO AMAX BUFFER'''      )
    return
  lowercase       : Optional[int]						=qq._amax.detach().item()
  lowercase       : Dict						=qk._amax.detach().item()
  lowercase       : str						=qv._amax.detach().item()

  lowercase       : str						=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			:							int       ,   __A			:							Optional[Any]      )							->   Optional[int]:



 """simple docstring"""




 for name, mod in model.named_modules():
  if name.endswith('''.output.dense'''      ) and not name.endswith('''attention.output.dense'''      ):
   lowercase       : Union[str, Any]						=mod._input_quantizer._amax.data.detach().item()
   mod._input_quantizer._amax.data.detach().clamp_(max=__A      )
   lowercase       : List[Any]						=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			:							Optional[int]      )							->   List[Any]:



 """simple docstring"""




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


def 				lowercase_					(     __A			:							str      )							->   Any:



 """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)
   lowercase       : Union[str, Any]						=set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis      )
   lowercase       : Tuple						=set(range(len(mod.weight.size()      )      )      ) - axis_set
   lowercase       : Tuple						=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()}'      )
   lowercase       : Tuple						=amax


def 				lowercase_					(     __A			:							str       ,   __A			:							Tuple=2_5       ,   __A			:							int=1_8_0       ,   __A			:							Optional[Any]=None      )							->   Optional[Any]:



 """simple docstring"""




 if ignore is None:
  lowercase       : Tuple						=[]
 elif not isinstance(__A       ,   __A      ):
  lowercase       : int						=[ignore]

 lowercase       : Any						=0
 for name, mod in model.named_modules():
  if not hasattr(__A       ,   '''weight'''      ):
   continue
  lowercase       : Optional[int]						=max(__A       ,   len(__A      )      )

 for name, mod in model.named_modules():
  lowercase       : Any						=getattr(__A       ,   '''_input_quantizer'''       ,   __A      )
  lowercase       : 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
  lowercase       : Optional[Any]						=F'Act:{input_q.extra_repr()}'
  lowercase       : Dict						=F'Wgt:{weight_q.extra_repr()}'
  lowercase       : str						=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			:							int      )							->   Dict:



 """simple docstring"""




 lowercase       : Any						=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			:							Dict       ,   __A			:							int       ,   __A			:							Dict       ,   __A			:							Tuple       ,   __A			:							Tuple      )							->   List[Any]:



 """simple docstring"""




 lowercase       : int						=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			:							Dict       ,   __A			:							int       ,   __A			:							Union[str, Any]="both"       ,   **__A			:							Tuple      )							->   Dict:



 """simple docstring"""




 lowercase       : str						=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			:							Union[str, Any]       ,   __A			:							Union[str, Any]       ,   **__A			:							Optional[int]      )							->   List[Any]:



 """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      ):
     lowercase       : Optional[int]						=F'Warning: changing {name:{name_width}}'
     for k, v in kwargs.items():
      s += F' {k}={v}'
      setattr(__A       ,   __A       ,   __A      )
     logger.info(__A      )





# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os

from accelerate.utils import ComputeEnvironment

from .cluster import get_cluster_input
from .config_args import cache_dir, default_config_file, default_yaml_config_file, load_config_from_file  # noqa: F401
from .config_utils import _ask_field, _ask_options, _convert_compute_environment  # noqa: F401
from .sagemaker import get_sagemaker_input


__lowerCAmelCase     :	Optional[Any]       			=  'Launches a series of prompts to create and save a `default_config.yaml` configuration file for your training system. Should always be ran first on your machine'

def a__       (			):






  '''simple docstring'''





  __magic_name__								=			_ask_options(
      """In which compute environment are you running?""",	["""This machine""", """AWS (Amazon SageMaker)"""],	_convert_compute_environment,	)
  if compute_environment == ComputeEnvironment.AMAZON_SAGEMAKER:
    __magic_name__								=			get_sagemaker_input()
  else:
    __magic_name__								=			get_cluster_input()
  return config

def a__       (			A_=None   ):






  '''simple docstring'''





  if subparsers is not None:
    __magic_name__								=			subparsers.add_parser("""config""",	description=A_   )
  else:
    __magic_name__								=			argparse.ArgumentParser("""Accelerate config command""",	description=A_   )

  parser.add_argument(
      """--config_file""",	default=A_,	help=(
          """The path to use to store the config file. Will default to a file named default_config.yaml in the cache """
          """location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have """
          """such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed """
          """with 'huggingface'."""
      ),	)

  if subparsers is not None:
    parser.set_defaults(func=A_   )
  return parser

def a__       (			A_   ):






  '''simple docstring'''





  __magic_name__								=			get_user_input()
  if args.config_file is not None:
    __magic_name__								=			args.config_file
  else:
    if not os.path.isdir(A_   ):
      os.makedirs(A_   )
    __magic_name__								=			default_yaml_config_file

  if config_file.endswith(""".json"""   ):
    config.to_json_file(A_   )
  else:
    config.to_yaml_file(A_   )
  print(f'''accelerate configuration saved at {config_file}'''   )

def a__       (			):






  '''simple docstring'''





  __magic_name__								=			config_command_parser()
  __magic_name__								=			parser.parse_args()
  config_command(A_   )


if __name__ == "__main__":
    main()

