



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


snake_case_			:     Optional[int]  	=  logging.getLogger(__name__)

snake_case_			:     List[Any]  	=  50  # max width of layer names
snake_case_			:     Dict  	=  70  # max width of quantizer names






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






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

						_UpperCamelCase :		Optional[int]			    =       QuantDescriptor(num_bits=args.aprec  ,  calib_method=_lowercase							)
						_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(_lowercase							)
						quant_nn.QuantLinear.set_default_quant_desc_weight(_lowercase							)






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_=False  ,  UpperCAmelCase_=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(_lowercase  ,  ['embeddings']  ,  which='weight'  ,  _disabled=_lowercase							)

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

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

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

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

												if args.recalibrate_weights:
																		recalibrate_weights(_lowercase							)

												if args.fuse_qkv:
																		fuse_qkv(_lowercase  ,  _lowercase							)

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

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






def A__   (					UpperCAmelCase_							):
						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__   (					UpperCAmelCase_  ,  UpperCAmelCase_							):
						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(_lowercase							)






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_							):






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

												_UpperCamelCase :		Any			    =       max(_lowercase  ,  _lowercase  ,  _lowercase							)
												qq._amax.fill_(_lowercase							)
												qk._amax.fill_(_lowercase							)
												qv._amax.fill_(_lowercase							)
												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__   (					UpperCAmelCase_  ,  UpperCAmelCase_							):






						for name, mod in model.named_modules():
												if name.endswith('.output.dense'							) and not name.endswith('attention.output.dense'							):
																		_UpperCamelCase :		Dict			    =       mod._input_quantizer._amax.data.detach().item()
																		mod._input_quantizer._amax.data.detach().clamp_(max=_lowercase							)
																		_UpperCamelCase :		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 A__   (					UpperCAmelCase_							):






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






def A__   (					UpperCAmelCase_							):






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






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_=2_5  ,  UpperCAmelCase_=1_8_0  ,  UpperCAmelCase_=None							):






						if ignore is None:
												_UpperCamelCase :		Optional[Any]			    =       []
						elif not isinstance(_lowercase  ,  _lowercase							):
												_UpperCamelCase :		int			    =       [ignore]

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

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






def A__   (					UpperCAmelCase_							):
						_UpperCamelCase :		int			    =       0
						for name, mod in model.named_modules():
												if isinstance(_lowercase  ,  pytorch_quantization.nn.TensorQuantizer							):
																		print(f'{name:80} {mod}'							)
																		count += 1
						print(f'{count} TensorQuantizers found in model'							)






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_							):
						_UpperCamelCase :		Optional[int]			    =       getattr(_lowercase  ,  _lowercase  ,  _lowercase							)
						if quantizer_mod is not None:
												assert hasattr(_lowercase  ,  _lowercase							)
												setattr(_lowercase  ,  _lowercase  ,  _lowercase							)
						else:
												logger.warning(f'{name} has no {quantizer}'							)






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_="both"  ,  **UpperCAmelCase_							):
						_UpperCamelCase :		int			    =       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(_lowercase  ,  _lowercase  ,  '_input_quantizer'  ,  _lowercase  ,  _lowercase							)
												if which in ["weight", "both"]:
																		set_quantizer(_lowercase  ,  _lowercase  ,  '_weight_quantizer'  ,  _lowercase  ,  _lowercase							)
						logger.info(_lowercase							)






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_  ,  **UpperCAmelCase_							):






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







'''simple docstring'''
from __future__ import annotations

import time
from collections.abc import Sequence
from random import randint

from matplotlib import pyplot as plt






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_							):
						if not arr:
												return None, None, 0
						if low == high:
												return low, high, arr[low]

						_UpperCamelCase :		str			    =       (low + high) // 2
						_UpperCamelCase    ,      _UpperCamelCase    ,      _UpperCamelCase :		Dict			    =       max_subarray(UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_							)
						_UpperCamelCase    ,      _UpperCamelCase    ,      _UpperCamelCase :		List[Any]			    =       max_subarray(UpperCAmelCase_  ,  mid + 1  ,  UpperCAmelCase_							)
						_UpperCamelCase    ,      _UpperCamelCase    ,      _UpperCamelCase :		Tuple			    =       max_cross_sum(UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_							)
						if left_sum >= right_sum and left_sum >= cross_sum:
												return left_low, left_high, left_sum
						elif right_sum >= left_sum and right_sum >= cross_sum:
												return right_low, right_high, right_sum
						return cross_left, cross_right, cross_sum






def A__   (					UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_  ,  UpperCAmelCase_							):
						_UpperCamelCase    ,      _UpperCamelCase :		Optional[Any]			    =       float('-inf'							), -1
						_UpperCamelCase    ,      _UpperCamelCase :		int			    =       float('-inf'							), -1

						_UpperCamelCase :		int | float			    =       0
						for i in range(UpperCAmelCase_  ,  low - 1  ,  -1							):
												summ += arr[i]
												if summ > left_sum:
																		_UpperCamelCase :		Optional[int]			    =       summ
																		_UpperCamelCase :		Union[str, Any]			    =       i

						_UpperCamelCase :		List[Any]			    =       0
						for i in range(mid + 1  ,  high + 1							):
												summ += arr[i]
												if summ > right_sum:
																		_UpperCamelCase :		List[Any]			    =       summ
																		_UpperCamelCase :		List[str]			    =       i

						return max_left, max_right, (left_sum + right_sum)






def A__   (					UpperCAmelCase_							):
						_UpperCamelCase :		List[str]			    =       [randint(1  ,  UpperCAmelCase_							) for _ in range(UpperCAmelCase_							)]
						_UpperCamelCase :		Optional[Any]			    =       time.time()
						max_subarray(UpperCAmelCase_  ,  0  ,  input_size - 1							)
						_UpperCamelCase :		str			    =       time.time()
						return end - start






def A__   (					):
						_UpperCamelCase :		Any			    =       [1_0, 1_0_0, 1_0_0_0, 1_0_0_0_0, 5_0_0_0_0, 1_0_0_0_0_0, 2_0_0_0_0_0, 3_0_0_0_0_0, 4_0_0_0_0_0, 5_0_0_0_0_0]
						_UpperCamelCase :		Dict			    =       [time_max_subarray(UpperCAmelCase_							) for input_size in input_sizes]
						print('No of Inputs\t\tTime Taken'							)
						for input_size, runtime in zip(UpperCAmelCase_  ,  UpperCAmelCase_							):
												print(UpperCAmelCase_  ,  '\t\t'  ,  UpperCAmelCase_							)
						plt.plot(UpperCAmelCase_  ,  UpperCAmelCase_							)
						plt.xlabel('Number of Inputs'							)
						plt.ylabel('Time taken in seconds'							)
						plt.show()


if __name__ == "__main__":
					from doctest import testmod

					testmod()


