


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


lowercase						:      Tuple									= logging.getLogger(__name__)

lowercase						:      Optional[int]									= 50  # max width of layer names
lowercase						:      List[Any]									= 70  # max width of quantizer names




def 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    int)      ->					List[Any]:






							'''simple docstring'''





							__UpperCamelCase			:     Optional[int]         =      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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[str])      ->					Dict:






							'''simple docstring'''





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

							__UpperCamelCase			:     str         =      QuantDescriptor(num_bits=args.aprec		,						calib_method=_lowerCamelCase)
							__UpperCamelCase			:     Optional[int]         =      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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    int		,						_lowerCamelCase				:    List[Any]		,						_lowerCamelCase				:    int=False		,						_lowerCamelCase				:    List[Any]=False)      ->					int:






							'''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(_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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    int)      ->					Optional[Any]:






							'''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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    Optional[int]		,						_lowerCamelCase				:    List[Any])      ->					Dict:






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




def 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[str]		,						_lowerCamelCase				:    Optional[int])      ->					str:






							'''simple docstring'''





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

														__UpperCamelCase			:     Tuple         =      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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[Any]		,						_lowerCamelCase				:    Tuple)      ->					Tuple:






							'''simple docstring'''





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




def 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[str])      ->					Dict:






							'''simple docstring'''





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




def 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[Any])      ->					int:






							'''simple docstring'''





							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)
																					__UpperCamelCase			:     List[str]         =      set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis)
																					__UpperCamelCase			:     Optional[Any]         =      set(range(len(mod.weight.size()))) - axis_set
																					__UpperCamelCase			:     str         =      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()}')
																					__UpperCamelCase			:     Dict         =      amax




def 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    Tuple		,						_lowerCamelCase				:    List[Any]=25		,						_lowerCamelCase				:    List[Any]=180		,						_lowerCamelCase				:    str=None)      ->					Any:






							'''simple docstring'''





							if ignore is None:
														__UpperCamelCase			:     List[Any]         =      []
							elif not isinstance(_lowerCamelCase		,						_lowerCamelCase):
														__UpperCamelCase			:     int         =      [ignore]

							__UpperCamelCase			:     Union[str, Any]         =      0
							for name, mod in model.named_modules():
														if not hasattr(_lowerCamelCase		,						"weight"):
																					continue
														__UpperCamelCase			:     int         =      max(_lowerCamelCase		,						len(_lowerCamelCase))

							for name, mod in model.named_modules():
														__UpperCamelCase			:     Tuple         =      getattr(_lowerCamelCase		,						"_input_quantizer"		,						_lowerCamelCase)
														__UpperCamelCase			:     int         =      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
														__UpperCamelCase			:     List[str]         =      F'Act:{input_q.extra_repr()}'
														__UpperCamelCase			:     Union[str, Any]         =      F'Wgt:{weight_q.extra_repr()}'
														__UpperCamelCase			:     Optional[Any]         =      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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    str)      ->					Tuple:






							'''simple docstring'''





							__UpperCamelCase			:     int         =      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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[str]		,						_lowerCamelCase				:    Optional[int]		,						_lowerCamelCase				:    List[Any]		,						_lowerCamelCase				:    Tuple		,						_lowerCamelCase				:    str)      ->					Dict:






							'''simple docstring'''





							__UpperCamelCase			:     Tuple         =      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 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    List[str]		,						_lowerCamelCase				:    int		,						_lowerCamelCase				:    Union[str, Any]="both"		,						**_lowerCamelCase				:    Dict)      ->					Any:






							'''simple docstring'''





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




def 			_SCREAMING_SNAKE_CASE      (	_lowerCamelCase				:    Union[str, Any]		,						_lowerCamelCase				:    int		,						**_lowerCamelCase				:    List[Any])      ->					Optional[Any]:






							'''simple docstring'''





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

import unittest
from dataclasses import dataclass

import pytest

from accelerate.commands.config.config_args import SageMakerConfig
from accelerate.utils import ComputeEnvironment
from accelerate.utils.launch import _convert_nargs_to_dict




@dataclass
class 				lowerCamelCase__							(       __lowercase):




	'''simple docstring'''




	_A			=  ComputeEnvironment.AMAZON_SAGEMAKER
	_A			=  True
	_A			=  'ml.p3.2xlarge'
	_A			=  'accelerate_sagemaker_execution_role'
	_A			=  'hf-sm'
	_A			=  'us-east-1'
	_A			=  1
	_A			=  'accelerate-sagemaker-1'
	_A			=  '1.6'
	_A			=  '4.4'
	_A			=  'train.py'
	_A			=  [
	    '--model_name_or_path',
	    'bert',
	    '--do_train',
	    'False',
	    '--epochs',
	    '3',
	    '--learning_rate',
	    '5e-5',
	    '--max_steps',
	    '50.5',
	]
	_A			=  [
	    '--model_name_or_path',
	    'bert',
	    '--do_train',
	    '--do_test',
	    'False',
	    '--do_predict',
	    '--epochs',
	    '3',
	    '--learning_rate',
	    '5e-5',
	    '--max_steps',
	    '50.5',
	]




class 				lowerCamelCase__							(       unittest.TestCase):




	'''simple docstring'''




	def      _lowerCamelCase       (      self	:Tuple		)	->				str:
								# If no defaults are changed, `to_kwargs` returns an empty dict.
								__UpperCamelCase			:     str         =      _convert_nargs_to_dict(MockLaunchConfig.success_training_script_args		)
								assert isinstance(converted_args["model_name_or_path"]  ,     a		)
								assert isinstance(converted_args["do_train"]  ,     a		)
								assert isinstance(converted_args["epochs"]  ,     a		)
								assert isinstance(converted_args["learning_rate"]  ,     a		)
								assert isinstance(converted_args["max_steps"]  ,     a		)

								with pytest.raises(a		):
															_convert_nargs_to_dict(MockLaunchConfig.fail_training_script_args		)