





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


UpperCamelCase_									=logging.getLogger(__name__)

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





def 	a_    (							_lowercase       ):
						_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_    (							_lowercase       ):

						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      : int         =   '''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=_lowercase       )
						_UpperCamelCase      : 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_    (							_lowercase	,						_lowercase	,						_lowercase=False	,						_lowercase=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_    (							_lowercase       ):
						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_    (							_lowercase	,						_lowercase       ):
						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_    (							_lowercase	,						_lowercase       ):

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

												_UpperCamelCase      : List[str]         =   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_    (							_lowercase	,						_lowercase       ):

						for name, mod in model.named_modules():
												if name.endswith('''.output.dense'''       ) and not name.endswith('''attention.output.dense'''       ):
																		_UpperCamelCase      : List[str]         =   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_    (							_lowercase       ):

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





def 	a_    (							_lowercase       ):

						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      : Tuple         =   set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis       )
																		_UpperCamelCase      : List[Any]         =   set(range(len(mod.weight.size()       )       )       ) - axis_set
																		_UpperCamelCase      : Any         =   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      : Tuple         =   amax





def 	a_    (							_lowercase	,						_lowercase=25	,						_lowercase=180	,						_lowercase=None       ):

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

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

						for name, mod in model.named_modules():
												_UpperCamelCase      : str         =   getattr(_lowercase	,						'''_input_quantizer'''	,						_lowercase       )
												_UpperCamelCase      : List[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      : int         =   F"""Act:{input_q.extra_repr()}"""
												_UpperCamelCase      : Dict         =   F"""Wgt:{weight_q.extra_repr()}"""
												_UpperCamelCase      : int         =   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_    (							_lowercase       ):
						_UpperCamelCase      : Optional[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_    (							_lowercase	,						_lowercase	,						_lowercase	,						_lowercase	,						_lowercase       ):
						_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_    (							_lowercase	,						_lowercase	,						_lowercase="both"	,						**_lowercase       ):
						_UpperCamelCase      : List[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(_lowercase	,						_lowercase	,						'''_input_quantizer'''	,						_lowercase	,						_lowercase       )
												if which in ["weight", "both"]:
																		set_quantizer(_lowercase	,						_lowercase	,						'''_weight_quantizer'''	,						_lowercase	,						_lowercase       )
						logger.info(_lowercase       )






def 	a_    (							_lowercase	,						_lowercase	,						**_lowercase       ):

						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      : Dict         =   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"""




import argparse
import json
from typing import List

from ltp import LTP

from transformers.models.bert.tokenization_bert import BertTokenizer





def 	a_    (							_lowercase       ):
						# This defines a "chinese character" as anything in the CJK Unicode block:
						#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
						#
						# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
						# despite its name. The modern Korean Hangul alphabet is a different block,
						# as is Japanese Hiragana and Katakana. Those alphabets are used to write
						# space-separated words, so they are not treated specially and handled
						# like the all of the other languages.
						if (
						    (cp >= 0X4E00 and cp <= 0X9FFF)
						    or (cp >= 0X3400 and cp <= 0X4DBF)  #
						    or (cp >= 0X20000 and cp <= 0X2A6DF)  #
						    or (cp >= 0X2A700 and cp <= 0X2B73F)  #
						    or (cp >= 0X2B740 and cp <= 0X2B81F)  #
						    or (cp >= 0X2B820 and cp <= 0X2CEAF)  #
						    or (cp >= 0XF900 and cp <= 0XFAFF)
						    or (cp >= 0X2F800 and cp <= 0X2FA1F)  #
						):  #
												return True

						return False





def 	a_    (							_lowercase       ):
						# word like '180' or '身高' or '神'
						for char in word:
												_UpperCamelCase      : Dict         =   ord(_lowercase       )
												if not _is_chinese_char(_lowercase       ):
																		return 0
						return 1





def 	a_    (							_lowercase       ):
						_UpperCamelCase      : List[str]         =   set()

						for token in tokens:
												_UpperCamelCase      : int         =   len(_lowercase       ) > 1 and is_chinese(_lowercase       )
												if chinese_word:
																		word_set.add(_lowercase       )
						_UpperCamelCase      : Optional[int]         =   list(_lowercase       )
						return word_list





def 	a_    (							_lowercase	,						_lowercase       ):
						if not chinese_word_set:
												return bert_tokens
						_UpperCamelCase      : Tuple         =   max([len(_lowercase       ) for w in chinese_word_set]       )

						_UpperCamelCase      : int         =   bert_tokens
						_UpperCamelCase       ,							_UpperCamelCase      : Union[str, Any]         =   0, len(_lowercase       )
						while start < end:
												_UpperCamelCase      : Union[str, Any]         =   True
												if is_chinese(bert_word[start]       ):
																		_UpperCamelCase      : List[Any]         =   min(end - start	,						_lowercase       )
																		for i in range(_lowercase	,						1	,						-1       ):
																								_UpperCamelCase      : str         =   ''''''.join(bert_word[start : start + i]       )
																								if whole_word in chinese_word_set:
																														for j in range(start + 1	,						start + i       ):
																																				_UpperCamelCase      : int         =   '''##''' + bert_word[j]
																														_UpperCamelCase      : int         =   start + i
																														_UpperCamelCase      : Union[str, Any]         =   False
																														break
												if single_word:
																		start += 1
						return bert_word





def 	a_    (							_lowercase	,						_lowercase	,						_lowercase       ):
						_UpperCamelCase      : List[Any]         =   []

						for i in range(0	,						len(_lowercase       )	,						100       ):
												_UpperCamelCase      : Optional[int]         =   ltp_tokenizer.pipeline(lines[i : i + 100]	,						tasks=['''cws''']       ).cws
												_UpperCamelCase      : Optional[int]         =   [get_chinese_word(_lowercase       ) for r in res]
												ltp_res.extend(_lowercase       )
						assert len(_lowercase       ) == len(_lowercase       )

						_UpperCamelCase      : Dict         =   []
						for i in range(0	,						len(_lowercase       )	,						100       ):
												_UpperCamelCase      : Optional[int]         =   bert_tokenizer(lines[i : i + 100]	,						add_special_tokens=_lowercase	,						truncation=_lowercase	,						max_length=512       )
												bert_res.extend(res['''input_ids''']       )
						assert len(_lowercase       ) == len(_lowercase       )

						_UpperCamelCase      : Optional[Any]         =   []
						for input_ids, chinese_word in zip(_lowercase	,						_lowercase       ):
												_UpperCamelCase      : str         =   []
												for id in input_ids:
																		_UpperCamelCase      : Dict         =   bert_tokenizer._convert_id_to_token(_lowercase       )
																		input_tokens.append(_lowercase       )
												_UpperCamelCase      : str         =   add_sub_symbol(_lowercase	,						_lowercase       )
												_UpperCamelCase      : List[str]         =   []
												# We only save pos of chinese subwords start with ##, which mean is part of a whole word.
												for i, token in enumerate(_lowercase       ):
																		if token[:2] == "##":
																								_UpperCamelCase      : int         =   token[2:]
																								# save chinese tokens' pos
																								if len(_lowercase       ) == 1 and _is_chinese_char(ord(_lowercase       )       ):
																														ref_id.append(_lowercase       )
												ref_ids.append(_lowercase       )

						assert len(_lowercase       ) == len(_lowercase       )

						return ref_ids





def 	a_    (							_lowercase       ):
						# For Chinese (Ro)Bert, the best result is from : RoBERTa-wwm-ext (https://github.com/ymcui/Chinese-BERT-wwm)
						# If we want to fine-tune these model, we have to use same tokenizer : LTP (https://github.com/HIT-SCIR/ltp)
						with open(args.file_name	,						'''r'''	,						encoding='''utf-8'''       ) as f:
												_UpperCamelCase      : Union[str, Any]         =   f.readlines()
						_UpperCamelCase      : Tuple         =   [line.strip() for line in data if len(_lowercase       ) > 0 and not line.isspace()]  # avoid delimiter like '\u2029'
						_UpperCamelCase      : List[Any]         =   LTP(args.ltp       )  # faster in GPU device
						_UpperCamelCase      : int         =   BertTokenizer.from_pretrained(args.bert       )

						_UpperCamelCase      : List[str]         =   prepare_ref(_lowercase	,						_lowercase	,						_lowercase       )

						with open(args.save_path	,						'''w'''	,						encoding='''utf-8'''       ) as f:
												_UpperCamelCase      : List[Any]         =   [json.dumps(_lowercase       ) + '''\n''' for ref in ref_ids]
												f.writelines(_lowercase       )


if __name__ == "__main__":
						UpperCamelCase_									=argparse.ArgumentParser(description="""prepare_chinese_ref""")
						parser.add_argument(
						    """--file_name""",
						    required=False,
						    type=str,
						    default="""./resources/chinese-demo.txt""",
						    help="""file need process, same as training data in lm""",
						)
						parser.add_argument(
						    """--ltp""",
						    required=False,
						    type=str,
						    default="""./resources/ltp""",
						    help="""resources for LTP tokenizer, usually a path""",
						)
						parser.add_argument(
						    """--bert""",
						    required=False,
						    type=str,
						    default="""./resources/robert""",
						    help="""resources for Bert tokenizer""",
						)
						parser.add_argument(
						    """--save_path""",
						    required=False,
						    type=str,
						    default="""./resources/ref.txt""",
						    help="""path to save res""",
						)

						UpperCamelCase_									=parser.parse_args()
						main(args)



