





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


SCREAMING_SNAKE_CASE__       :			Optional[Any]			      =						logging.getLogger(__name__)

SCREAMING_SNAKE_CASE__       :			str			      =						5_0  # max width of layer names
SCREAMING_SNAKE_CASE__       :			int			      =						7_0  # max width of quantizer names






def       _A  (      lowerCamelCase   ):
   a__	:						Any      		=  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       _A  (      lowerCamelCase   ):

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

   a__	:						Union[str, Any]      		=  QuantDescriptor(num_bits=args.aprec							, calib_method=lowerCamelCase   )
   a__	:						Any      		=  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       _A  (      lowerCamelCase							, lowerCamelCase							, lowerCamelCase=False							, lowerCamelCase=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(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       _A  (      lowerCamelCase   ):
   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  (      lowerCamelCase							, lowerCamelCase   ):
   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       _A  (      lowerCamelCase							, lowerCamelCase   ):

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

      a__	:						str      		=  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       _A  (      lowerCamelCase							, lowerCamelCase   ):

   for name, mod in model.named_modules():
      if name.endswith(".output.dense"   ) and not name.endswith("attention.output.dense"   ):
         a__	:						Any      		=  mod._input_quantizer._amax.data.detach().item()
         mod._input_quantizer._amax.data.detach().clamp_(max=lowerCamelCase   )
         a__	:						Dict      		=  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  (      lowerCamelCase   ):

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






def       _A  (      lowerCamelCase   ):

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






def       _A  (      lowerCamelCase							, lowerCamelCase=25							, lowerCamelCase=180							, lowerCamelCase=None   ):

   if ignore is None:
      a__	:						Union[str, Any]      		=  []
   elif not isinstance(lowerCamelCase							, lowerCamelCase   ):
      a__	:						Optional[int]      		=  [ignore]

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

   for name, mod in model.named_modules():
      a__	:						int      		=  getattr(lowerCamelCase							, "_input_quantizer"							, lowerCamelCase   )
      a__	:						List[Any]      		=  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
      a__	:						Dict      		=  F"""Act:{input_q.extra_repr()}"""
      a__	:						Union[str, Any]      		=  F"""Wgt:{weight_q.extra_repr()}"""
      a__	:						str      		=  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       _A  (      lowerCamelCase   ):
   a__	:						Any      		=  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       _A  (      lowerCamelCase							, lowerCamelCase							, lowerCamelCase							, lowerCamelCase							, lowerCamelCase   ):
   a__	:						int      		=  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       _A  (      lowerCamelCase							, lowerCamelCase							, lowerCamelCase="both"							, **lowerCamelCase   ):
   a__	:						Dict      		=  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       _A  (      lowerCamelCase							, lowerCamelCase							, **lowerCamelCase   ):

   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   ):
               a__	:						List[Any]      		=  F"""Warning: changing {name:{name_width}}"""
               for k, v in kwargs.items():
                  s += F""" {k}={v}"""
                  setattr(lowerCamelCase							, lowerCamelCase							, lowerCamelCase   )
               logger.info(lowerCamelCase   )








from typing import List, Optional, Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


SCREAMING_SNAKE_CASE__       :			Any			      =						logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__       :			str			      =						{
    """huggingface/informer-tourism-monthly""": (
        """https://huggingface.co/huggingface/informer-tourism-monthly/resolve/main/config.json"""
    ),
    # See all Informer models at https://huggingface.co/models?filter=informer
}







class   __lowerCAmelCase			(					_UpperCamelCase  ):
   _UpperCamelCase :						Optional[Any]     		=				"""informer"""
   _UpperCamelCase :						Any     		=				{
       """hidden_size""": """d_model""",
       """num_attention_heads""": """encoder_attention_heads""",
       """num_hidden_layers""": """encoder_layers""",
   }





   def __init__( self					,			snake_case = None					,			snake_case = None					,			snake_case = "student_t"					,			snake_case = "nll"					,			snake_case = 1					,			snake_case = None					,			snake_case = "mean"					,			snake_case = 0					,			snake_case = 0					,			snake_case = 0					,			snake_case = 0					,			snake_case = None					,			snake_case = None					,			snake_case = 64					,			snake_case = 32					,			snake_case = 32					,			snake_case = 2					,			snake_case = 2					,			snake_case = 2					,			snake_case = 2					,			snake_case = True					,			snake_case = "gelu"					,			snake_case = 0.05					,			snake_case = 0.1					,			snake_case = 0.1					,			snake_case = 0.1					,			snake_case = 0.1					,			snake_case = 100					,			snake_case = 0.02					,			snake_case=True					,			snake_case = "prob"					,			snake_case = 5					,			snake_case = True					,			**snake_case					,			)				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Optional[Any]      		=  prediction_length
      a__	:						Optional[int]      		=  context_length or prediction_length
      a__	:						Optional[int]      		=  distribution_output
      a__	:						str      		=  loss
      a__	:						Optional[Any]      		=  input_size
      a__	:						int      		=  num_time_features
      a__	:						Optional[int]      		=  lags_sequence if lags_sequence is not None else [1, 2, 3, 4, 5, 6, 7]
      a__	:						Optional[int]      		=  scaling
      a__	:						List[str]      		=  num_dynamic_real_features
      a__	:						Optional[int]      		=  num_static_real_features
      a__	:						Optional[int]      		=  num_static_categorical_features

      # set cardinality
      if cardinality and num_static_categorical_features > 0:
         if len(snake_case      ) != num_static_categorical_features:
            raise ValueError(
                "The cardinality should be a list of the same length as `num_static_categorical_features`"      )
         a__	:						List[Any]      		=  cardinality
      else:
         a__	:						Tuple      		=  [0]

      # set embedding_dimension
      if embedding_dimension and num_static_categorical_features > 0:
         if len(snake_case      ) != num_static_categorical_features:
            raise ValueError(
                "The embedding dimension should be a list of the same length as `num_static_categorical_features`"      )
         a__	:						Tuple      		=  embedding_dimension
      else:
         a__	:						int      		=  [min(50					,			(cat + 1) // 2      ) for cat in self.cardinality]

      a__	:						Optional[Any]      		=  num_parallel_samples

      # Transformer architecture configuration
      a__	:						int      		=  input_size * len(self.lags_sequence      ) + self._number_of_features
      a__	:						Union[str, Any]      		=  d_model
      a__	:						Any      		=  encoder_attention_heads
      a__	:						Optional[Any]      		=  decoder_attention_heads
      a__	:						int      		=  encoder_ffn_dim
      a__	:						List[Any]      		=  decoder_ffn_dim
      a__	:						List[str]      		=  encoder_layers
      a__	:						Any      		=  decoder_layers

      a__	:						List[str]      		=  dropout
      a__	:						int      		=  attention_dropout
      a__	:						List[Any]      		=  activation_dropout
      a__	:						Optional[int]      		=  encoder_layerdrop
      a__	:						Tuple      		=  decoder_layerdrop

      a__	:						Any      		=  activation_function
      a__	:						Tuple      		=  init_std

      a__	:						Optional[int]      		=  use_cache

      # Informer
      a__	:						Union[str, Any]      		=  attention_type
      a__	:						List[str]      		=  sampling_factor
      a__	:						Optional[int]      		=  distil

      super().__init__(is_encoder_decoder=snake_case					,			**snake_case      )





   @property
   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      return (
          sum(self.embedding_dimension      )
          + self.num_dynamic_real_features
          + self.num_time_features
          + self.num_static_real_features
          + self.input_size * 2  # the log1p(abs(loc)) and log(scale) features
      )

