






import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np


def        _a  (     UpperCamelCase_					:				bytes    ,   UpperCamelCase_					:				int   ) ->   np.array:
						"""simple docstring"""







						lowerCAmelCase__       =						F"{sampling_rate}"
						lowerCAmelCase__       =						"1"
						lowerCAmelCase__       =						"f32le"
						lowerCAmelCase__       =						[
						    "ffmpeg",
						    "-i",
						    "pipe:0",
						    "-ac",
						    ac,
						    "-ar",
						    ar,
						    "-f",
						    format_for_conversion,
						    "-hide_banner",
						    "-loglevel",
						    "quiet",
						    "pipe:1",
						]

						try:
												with subprocess.Popen(UpperCamelCase_    ,   stdin=subprocess.PIPE    ,   stdout=subprocess.PIPE   ) as ffmpeg_process:
																		lowerCAmelCase__       =						ffmpeg_process.communicate(UpperCamelCase_   )
						except FileNotFoundError as error:
												raise ValueError("ffmpeg was not found but is required to load audio files from filename"   ) from error
						lowerCAmelCase__       =						output_stream[0]
						lowerCAmelCase__       =						np.frombuffer(UpperCamelCase_    ,   np.floataa   )
						if audio.shape[0] == 0:
												raise ValueError("Malformed soundfile"   )
						return audio


def        _a  (     UpperCamelCase_					:				int    ,   UpperCamelCase_					:				float    ,   UpperCamelCase_					:				str = "f32le"    ,   ) ->   List[Any]:
						"""simple docstring"""







						lowerCAmelCase__       =						F"{sampling_rate}"
						lowerCAmelCase__       =						"1"
						if format_for_conversion == "s16le":
												lowerCAmelCase__       =						2
						elif format_for_conversion == "f32le":
												lowerCAmelCase__       =						4
						else:
												raise ValueError(F"Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"   )

						lowerCAmelCase__       =						platform.system()
						if system == "Linux":
												lowerCAmelCase__       =						"alsa"
												lowerCAmelCase__       =						"default"
						elif system == "Darwin":
												lowerCAmelCase__       =						"avfoundation"
												lowerCAmelCase__       =						":0"
						elif system == "Windows":
												lowerCAmelCase__       =						"dshow"
												lowerCAmelCase__       =						"default"

						lowerCAmelCase__       =						[
						    "ffmpeg",
						    "-f",
						    format_,
						    "-i",
						    input_,
						    "-ac",
						    ac,
						    "-ar",
						    ar,
						    "-f",
						    format_for_conversion,
						    "-fflags",
						    "nobuffer",
						    "-hide_banner",
						    "-loglevel",
						    "quiet",
						    "pipe:1",
						]
						lowerCAmelCase__       =						int(round(sampling_rate * chunk_length_s   )   ) * size_of_sample
						lowerCAmelCase__       =						_ffmpeg_stream(UpperCamelCase_    ,   UpperCamelCase_   )
						for item in iterator:
												yield item


def        _a  (     UpperCamelCase_					:				int    ,   UpperCamelCase_					:				float    ,   UpperCamelCase_					:				Optional[int] = None    ,   UpperCamelCase_					:				Optional[Union[Tuple[float, float], float]] = None    ,   UpperCamelCase_					:				str = "f32le"    ,   ) ->   Tuple:
						"""simple docstring"""







						if stream_chunk_s is not None:
												lowerCAmelCase__       =						stream_chunk_s
						else:
												lowerCAmelCase__       =						chunk_length_s

						lowerCAmelCase__       =						ffmpeg_microphone(UpperCamelCase_    ,   UpperCamelCase_    ,   format_for_conversion=UpperCamelCase_   )
						if format_for_conversion == "s16le":
												lowerCAmelCase__       =						np.intaa
												lowerCAmelCase__       =						2
						elif format_for_conversion == "f32le":
												lowerCAmelCase__       =						np.floataa
												lowerCAmelCase__       =						4
						else:
												raise ValueError(F"Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"   )

						if stride_length_s is None:
												lowerCAmelCase__       =						chunk_length_s / 6
						lowerCAmelCase__       =						int(round(sampling_rate * chunk_length_s   )   ) * size_of_sample
						if isinstance(UpperCamelCase_    ,   (int, float)   ):
												lowerCAmelCase__       =						[stride_length_s, stride_length_s]

						lowerCAmelCase__       =						int(round(sampling_rate * stride_length_s[0]   )   ) * size_of_sample
						lowerCAmelCase__       =						int(round(sampling_rate * stride_length_s[1]   )   ) * size_of_sample
						lowerCAmelCase__       =						datetime.datetime.now()
						lowerCAmelCase__       =						datetime.timedelta(seconds=UpperCamelCase_   )
						for item in chunk_bytes_iter(UpperCamelCase_    ,   UpperCamelCase_    ,   stride=(stride_left, stride_right)    ,   stream=UpperCamelCase_   ):
												# Put everything back in numpy scale
												lowerCAmelCase__       =						np.frombuffer(item["raw"]    ,   dtype=UpperCamelCase_   )
												lowerCAmelCase__       =						(
												    item["stride"][0] // size_of_sample,
												    item["stride"][1] // size_of_sample,
												)
												lowerCAmelCase__       =						sampling_rate
												audio_time += delta
												if datetime.datetime.now() > audio_time + 10 * delta:
																		# We're late !! SKIP
																		continue
												yield item


def        _a  (     UpperCamelCase_					:				Optional[int]    ,   UpperCamelCase_					:				int    ,   UpperCamelCase_					:				Tuple[int, int]    ,   UpperCamelCase_					:				bool = False   ) ->   Optional[int]:
						"""simple docstring"""







						lowerCAmelCase__       =						b""
						lowerCAmelCase__      ,							lowerCAmelCase__       =						stride
						if stride_left + stride_right >= chunk_len:
												raise ValueError(
												    F"Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}"   )
						lowerCAmelCase__       =						0
						for raw in iterator:
												acc += raw
												if stream and len(UpperCamelCase_   ) < chunk_len:
																		lowerCAmelCase__       =						(_stride_left, 0)
																		yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
												else:
																		while len(UpperCamelCase_   ) >= chunk_len:
																								# We are flushing the accumulator
																								lowerCAmelCase__       =						(_stride_left, stride_right)
																								lowerCAmelCase__       =						{"raw": acc[:chunk_len], "stride": stride}
																								if stream:
																														lowerCAmelCase__       =						False
																								yield item
																								lowerCAmelCase__       =						stride_left
																								lowerCAmelCase__       =						acc[chunk_len - stride_left - stride_right :]
    # Last chunk
						if len(UpperCamelCase_   ) > stride_left:
												lowerCAmelCase__       =						{"raw": acc, "stride": (_stride_left, 0)}
												if stream:
																		lowerCAmelCase__       =						False
												yield item





def        _a  (     UpperCamelCase_					:				Optional[Any]    ,   UpperCamelCase_					:				int   ) ->   Dict:
						"""simple docstring"""







						lowerCAmelCase__       =						2**24  # 16Mo
						try:
												with subprocess.Popen(UpperCamelCase_    ,   stdout=subprocess.PIPE    ,   bufsize=UpperCamelCase_   ) as ffmpeg_process:
																		while True:
																								lowerCAmelCase__       =						ffmpeg_process.stdout.read(UpperCamelCase_   )
																								if raw == b"":
																														break
																								yield raw
						except FileNotFoundError as error:
												raise ValueError("ffmpeg was not found but is required to stream audio files from filename"   ) from error














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


a_							=    logging.getLogger(__name__)

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


def        _a  (     UpperCamelCase_					:				str   ) ->   List[str]:
						"""simple docstring"""







						lowerCAmelCase__       =						parser.add_argument_group("quant_trainer arguments"   )
						group.add_argument("--wprec"    ,   type=UpperCamelCase_    ,   default=8    ,   help="weight precision"   )
						group.add_argument("--aprec"    ,   type=UpperCamelCase_    ,   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=UpperCamelCase_    ,   nargs="+"    ,   help="disable quantizers by keyword"   )
						group.add_argument("--quant-disable-layer-module"    ,   type=UpperCamelCase_    ,   help="disable quantizers by keyword under layer."   )
						group.add_argument("--quant-enable-layer-module"    ,   type=UpperCamelCase_    ,   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=UpperCamelCase_    ,   type=UpperCamelCase_    ,   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=UpperCamelCase_    ,   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_					:				int   ) ->   List[Any]:
						"""simple docstring"""







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

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


def        _a  (     UpperCamelCase_					:				Union[str, Any]    ,   UpperCamelCase_					:				List[Any]    ,   UpperCamelCase_					:				Any=False    ,   UpperCamelCase_					:				List[Any]=False   ) ->   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(UpperCamelCase_    ,   ["embeddings"]    ,   which="weight"    ,   _disabled=UpperCamelCase_   )

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

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

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

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

												if args.recalibrate_weights:
																		recalibrate_weights(UpperCamelCase_   )

												if args.fuse_qkv:
																		fuse_qkv(UpperCamelCase_    ,   UpperCamelCase_   )

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

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


def        _a  (     UpperCamelCase_					:				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        _a  (     UpperCamelCase_					:				Dict    ,   UpperCamelCase_					:				Tuple   ) ->   str:
						"""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(UpperCamelCase_   )


def        _a  (     UpperCamelCase_					:				Optional[Any]    ,   UpperCamelCase_					:				int   ) ->   int:
						"""simple docstring"""







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

												lowerCAmelCase__       =						max(UpperCamelCase_    ,   UpperCamelCase_    ,   UpperCamelCase_   )
												qq._amax.fill_(UpperCamelCase_   )
												qk._amax.fill_(UpperCamelCase_   )
												qv._amax.fill_(UpperCamelCase_   )
												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_					:				Any    ,   UpperCamelCase_					:				List[Any]   ) ->   Optional[int]:
						"""simple docstring"""







						for name, mod in model.named_modules():
												if name.endswith(".output.dense"   ) and not name.endswith("attention.output.dense"   ):
																		lowerCAmelCase__       =						mod._input_quantizer._amax.data.detach().item()
																		mod._input_quantizer._amax.data.detach().clamp_(max=UpperCamelCase_   )
																		lowerCAmelCase__       =						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_					:				str   ) ->   int:
						"""simple docstring"""







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


def        _a  (     UpperCamelCase_					:				Any   ) ->   Optional[int]:
						"""simple docstring"""







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


def        _a  (     UpperCamelCase_					:				str    ,   UpperCamelCase_					:				Optional[Any]=25    ,   UpperCamelCase_					:				str=180    ,   UpperCamelCase_					:				Optional[int]=None   ) ->   Tuple:
						"""simple docstring"""







						if ignore is None:
												lowerCAmelCase__       =						[]
						elif not isinstance(UpperCamelCase_    ,   UpperCamelCase_   ):
												lowerCAmelCase__       =						[ignore]

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

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


def        _a  (     UpperCamelCase_					:				str   ) ->   Dict:
						"""simple docstring"""







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


def        _a  (     UpperCamelCase_					:				List[str]    ,   UpperCamelCase_					:				Union[str, Any]    ,   UpperCamelCase_					:				Union[str, Any]    ,   UpperCamelCase_					:				Optional[Any]    ,   UpperCamelCase_					:				int   ) ->   Tuple:
						"""simple docstring"""







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


def        _a  (     UpperCamelCase_					:				List[Any]    ,   UpperCamelCase_					:				int    ,   UpperCamelCase_					:				Union[str, Any]="both"    ,   **UpperCamelCase_					:				Dict   ) ->   Optional[int]:
						"""simple docstring"""







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





def        _a  (     UpperCamelCase_					:				str    ,   UpperCamelCase_					:				List[Any]    ,   **UpperCamelCase_					:				Any   ) ->   Any:
						"""simple docstring"""







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






