
"""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     :    str       =			logging.getLogger(__name__)

UpperCamelCase     :    List[str]       =			5_0  # max width of layer names
UpperCamelCase     :    Union[str, Any]       =			7_0  # max width of quantizer names






def 							A						(    snake_case				:Dict		)				->     Optional[Any]:
 __UpperCamelCase      =	parser.add_argument_group('quant_trainer arguments'		)
 group.add_argument('--wprec'    , type=snake_case    , default=8    , help='weight precision'		)
 group.add_argument('--aprec'    , type=snake_case    , 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=snake_case    , nargs='+'    , help='disable quantizers by keyword'		)
 group.add_argument('--quant-disable-layer-module'    , type=snake_case    , help='disable quantizers by keyword under layer.'		)
 group.add_argument('--quant-enable-layer-module'    , type=snake_case    , 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=snake_case    , type=snake_case    , 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=snake_case    , 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						(    snake_case				:Any		)				->     Optional[int]:

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

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






def 							A						(    snake_case				:str    , snake_case				:Tuple    , snake_case				:Optional[Any]=False    , snake_case				:List[str]=False		)				->     Optional[Any]:
 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(snake_case    , ['embeddings']    , which='weight'    , _disabled=snake_case		)

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

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

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

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

  if args.recalibrate_weights:
   recalibrate_weights(snake_case		)

  if args.fuse_qkv:
   fuse_qkv(snake_case    , snake_case		)

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

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






def 							A						(    snake_case				:List[str]		)				->     List[str]:
 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						(    snake_case				:int    , snake_case				:Union[str, Any]		)				->     Optional[int]:
 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(snake_case		)






def 							A						(    snake_case				:Union[str, Any]    , snake_case				:Any		)				->     Any:

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

  __UpperCamelCase      =	max(snake_case    , snake_case    , snake_case		)
  qq._amax.fill_(snake_case		)
  qk._amax.fill_(snake_case		)
  qv._amax.fill_(snake_case		)
  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						(    snake_case				:Optional[int]    , snake_case				:Optional[Any]		)				->     Optional[Any]:

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

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






def 							A						(    snake_case				:List[str]		)				->     int:

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






def 							A						(    snake_case				:Any    , snake_case				:Dict=2_5    , snake_case				:str=1_8_0    , snake_case				:int=None		)				->     str:

 if ignore is None:
  __UpperCamelCase      =	[]
 elif not isinstance(snake_case    , snake_case		):
  __UpperCamelCase      =	[ignore]

 __UpperCamelCase      =	0
 for name, mod in model.named_modules():
  if not hasattr(snake_case    , 'weight'		):
   continue
  __UpperCamelCase      =	max(snake_case    , len(snake_case		)		)

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






def 							A						(    snake_case				:Dict		)				->     Union[str, Any]:
 __UpperCamelCase      =	0
 for name, mod in model.named_modules():
  if isinstance(snake_case    , pytorch_quantization.nn.TensorQuantizer		):
   print(f'{name:80} {mod}'		)
   count += 1
 print(f'{count} TensorQuantizers found in model'		)






def 							A						(    snake_case				:int    , snake_case				:str    , snake_case				:str    , snake_case				:Union[str, Any]    , snake_case				:int		)				->     Union[str, Any]:
 __UpperCamelCase      =	getattr(snake_case    , snake_case    , snake_case		)
 if quantizer_mod is not None:
  assert hasattr(snake_case    , snake_case		)
  setattr(snake_case    , snake_case    , snake_case		)
 else:
  logger.warning(f'{name} has no {quantizer}'		)






def 							A						(    snake_case				:int    , snake_case				:str    , snake_case				:Tuple="both"    , **snake_case				:Optional[Any]		)				->     str:
 __UpperCamelCase      =	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(snake_case    , snake_case    , '_input_quantizer'    , snake_case    , snake_case		)
  if which in ["weight", "both"]:
   set_quantizer(snake_case    , snake_case    , '_weight_quantizer'    , snake_case    , snake_case		)
 logger.info(snake_case		)






def 							A						(    snake_case				:Union[str, Any]    , snake_case				:int    , **snake_case				:Any		)				->     Optional[Any]:

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





"""simple docstring"""

import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCamelCase     :    int       =			logging.get_logger(__name__)

class __lowerCAmelCase		(				__SCREAMING_SNAKE_CASE      ):
     lowercase						       =						["input_values", "attention_mask"]





     def __init__(     self     ,					__UpperCAmelCase = 1     ,					__UpperCAmelCase = 1_6000     ,					__UpperCAmelCase = 0.0     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = 80     ,					__UpperCAmelCase = 16     ,					__UpperCAmelCase = 64     ,					__UpperCAmelCase = "hann_window"     ,					__UpperCAmelCase = 1.0     ,					__UpperCAmelCase = 80     ,					__UpperCAmelCase = 7600     ,					__UpperCAmelCase = 1E-10     ,					__UpperCAmelCase = 2     ,					__UpperCAmelCase = True     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      super().__init__(feature_size=__UpperCAmelCase     ,					sampling_rate=__UpperCAmelCase     ,					padding_value=__UpperCAmelCase     ,					**__UpperCAmelCase				)
      __UpperCamelCase      =	do_normalize
      __UpperCamelCase      =	return_attention_mask

      __UpperCamelCase      =	num_mel_bins
      __UpperCamelCase      =	hop_length
      __UpperCamelCase      =	win_length
      __UpperCamelCase      =	win_function
      __UpperCamelCase      =	frame_signal_scale
      __UpperCamelCase      =	fmin
      __UpperCamelCase      =	fmax
      __UpperCamelCase      =	mel_floor
      __UpperCamelCase      =	reduction_factor

      __UpperCamelCase      =	win_length * sampling_rate // 1000
      __UpperCamelCase      =	hop_length * sampling_rate // 1000
      __UpperCamelCase      =	optimal_fft_length(self.sample_size				)
      __UpperCamelCase      =	(self.n_fft // 2) + 1

      __UpperCamelCase      =	window_function(window_length=self.sample_size     ,					name=self.win_function     ,					periodic=__UpperCAmelCase				)

      __UpperCamelCase      =	mel_filter_bank(
          num_frequency_bins=self.n_freqs     ,					num_mel_filters=self.num_mel_bins     ,					min_frequency=self.fmin     ,					max_frequency=self.fmax     ,					sampling_rate=self.sampling_rate     ,					norm='slaney'     ,					mel_scale='slaney'     ,					)

      if frame_signal_scale != 1.0:
       warnings.warn(
           'The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers'     ,					__UpperCAmelCase     ,					)
      if reduction_factor != 2.0:
       warnings.warn(
           'The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'     ,					__UpperCAmelCase     ,					)





     @staticmethod
     # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
     def 						UpperCAmelCase (     __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase = 0.0				):

      '''simple docstring'''




      if attention_mask is not None:
       __UpperCamelCase      =	np.array(__UpperCAmelCase     ,					np.intaa				)
       __UpperCamelCase      =	[]

       for vector, length in zip(__UpperCAmelCase     ,					attention_mask.sum(-1				)				):
        __UpperCamelCase      =	(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7				)
        if length < normed_slice.shape[0]:
         __UpperCamelCase      =	padding_value

        normed_input_values.append(__UpperCAmelCase				)
      else:
       __UpperCamelCase      =	[(x - x.mean()) / np.sqrt(x.var() + 1E-7				) for x in input_values]

      return normed_input_values





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	spectrogram(
          __UpperCAmelCase     ,					window=self.window     ,					frame_length=self.sample_size     ,					hop_length=self.sample_stride     ,					fft_length=self.n_fft     ,					mel_filters=self.mel_filters     ,					mel_floor=self.mel_floor     ,					log_mel='log10'     ,					)
      return log_mel_spec.T





     def __call__(     self     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      if audio is None and audio_target is None:
       raise ValueError('You must provide either `audio` or `audio_target` values.'				)

      if sampling_rate is not None:
       if sampling_rate != self.sampling_rate:
        raise ValueError(
            F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
            F' {self.sampling_rate}. Please make sure that the provided audio input was sampled with'
            F' {self.sampling_rate} and not {sampling_rate}.'				)
      else:
       logger.warning(
           'It is strongly recommended to pass the ``sampling_rate`` argument to this function. '
           'Failing to do so can result in silent errors that might be hard to debug.'				)

      if audio is not None:
       __UpperCamelCase      =	self._process_audio(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)
      else:
       __UpperCamelCase      =	None

      if audio_target is not None:
       __UpperCamelCase      =	self._process_audio(
           __UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)

       if inputs is None:
        return inputs_target
       else:
        __UpperCamelCase      =	inputs_target['input_values']
        __UpperCamelCase      =	inputs_target.get('attention_mask'				)
        if decoder_attention_mask is not None:
         __UpperCamelCase      =	decoder_attention_mask

      return inputs





     def 						UpperCAmelCase (     self     ,					__UpperCAmelCase     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = False     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					__UpperCAmelCase = None     ,					**__UpperCAmelCase     ,					):

      '''simple docstring'''




      __UpperCamelCase      =	isinstance(__UpperCAmelCase     ,					np.ndarray				) and len(speech.shape				) > 1
      if is_batched_numpy and len(speech.shape				) > 2:
       raise ValueError(F'Only mono-channel audio is supported for input to {self}'				)
      __UpperCamelCase      =	is_batched_numpy or (
          isinstance(__UpperCAmelCase     ,					(list, tuple)				) and (isinstance(speech[0]     ,					(np.ndarray, tuple, list)				))
      )

      if is_batched:
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				) for speech in speech]
      elif not is_batched and not isinstance(__UpperCAmelCase     ,					np.ndarray				):
       __UpperCamelCase      =	np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				)
      elif isinstance(__UpperCAmelCase     ,					np.ndarray				) and speech.dtype is np.dtype(np.floataa				):
       __UpperCamelCase      =	speech.astype(np.floataa				)

      # always return batch
      if not is_batched:
       __UpperCamelCase      =	[speech]

      # needed to make pad() work on spectrogram inputs
      __UpperCamelCase      =	self.feature_size

      # convert into correct format for padding
      if is_target:
       __UpperCamelCase      =	[self._extract_mel_features(__UpperCAmelCase				) for waveform in speech]
       __UpperCamelCase      =	BatchFeature({'input_values': features}				)
       __UpperCamelCase      =	self.num_mel_bins
      else:
       __UpperCamelCase      =	BatchFeature({'input_values': speech}				)

      __UpperCamelCase      =	self.pad(
          __UpperCAmelCase     ,					padding=__UpperCAmelCase     ,					max_length=__UpperCAmelCase     ,					truncation=__UpperCAmelCase     ,					pad_to_multiple_of=__UpperCAmelCase     ,					return_attention_mask=__UpperCAmelCase     ,					**__UpperCAmelCase     ,					)

      __UpperCamelCase      =	feature_size_hack

      # convert input values to correct format
      __UpperCamelCase      =	padded_inputs['input_values']
      if not isinstance(input_values[0]     ,					np.ndarray				):
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.floataa				) for array in input_values]
      elif (
          not isinstance(__UpperCAmelCase     ,					np.ndarray				)
          and isinstance(input_values[0]     ,					np.ndarray				)
          and input_values[0].dtype is np.dtype(np.floataa				)
      ):
       __UpperCamelCase      =	[array.astype(np.floataa				) for array in input_values]
      elif isinstance(__UpperCAmelCase     ,					np.ndarray				) and input_values.dtype is np.dtype(np.floataa				):
       __UpperCamelCase      =	input_values.astype(np.floataa				)

      # convert attention_mask to correct format
      __UpperCamelCase      =	padded_inputs.get('attention_mask'				)
      if attention_mask is not None:
       __UpperCamelCase      =	[np.asarray(__UpperCAmelCase     ,					dtype=np.intaa				) for array in attention_mask]

      # zero-mean and unit-variance normalization
      if not is_target and self.do_normalize:
       __UpperCamelCase      =	(
           attention_mask
           if self._get_padding_strategies(__UpperCAmelCase     ,					max_length=__UpperCAmelCase				) is not PaddingStrategy.DO_NOT_PAD
           else None
       )
       __UpperCamelCase      =	self.zero_mean_unit_var_norm(
           padded_inputs['input_values']     ,					attention_mask=__UpperCAmelCase     ,					padding_value=self.padding_value				)

      if return_tensors is not None:
       __UpperCamelCase      =	padded_inputs.convert_to_tensors(__UpperCAmelCase				)

      return padded_inputs







     def 						UpperCAmelCase (     self				):

      '''simple docstring'''




      __UpperCamelCase      =	super().to_dict()

      # Don't serialize these as they are derived from the other properties.
      __UpperCamelCase      =	['window', 'mel_filters', 'sample_size', 'sample_stride', 'n_fft', 'n_freqs']
      for name in names:
       if name in output:
        del output[name]

      return output



