




'''simple docstring'''







import math
from typing import Any, Callable, List, Optional, Tuple, Union

import numpy as np
import torch

from ...models import TaFilmDecoder
from ...schedulers import DDPMScheduler
from ...utils import is_onnx_available, logging, randn_tensor


if is_onnx_available():
      from ..onnx_utils import OnnxRuntimeModel

from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline
from .continous_encoder import SpectrogramContEncoder
from .notes_encoder import SpectrogramNotesEncoder


lowercase      :   Optional[int]      =       logging.get_logger(__name__)  # pylint: disable=invalid-name

lowercase      :   str      =       256







class 					_a (__A	):





     '''simple docstring'''



     lowerCAmelCase_  :     str          =					["""melgan"""]







     def __init__(				self							,__a							,__a							,__a							,__a							,__a							,)      ->				Optional[Any]:
          super().__init__()

          # From MELGAN
          snake_case			:      Any        =			math.log(1E-5							)  # Matches MelGAN training.
          snake_case			:      Union[str, Any]        =			4.0  # Largest value for most examples
          snake_case			:      List[Any]        =			128

          self.register_modules(
              notes_encoder=__A							,continuous_encoder=__A							,decoder=__A							,scheduler=__A							,melgan=__A							,)







     def 			snake_case_      (				self							,__a							,__a=(-1.0, 1.0)							,__a=False							)      ->				str:
          snake_case ,		snake_case			:      Union[str, Any]        =			output_range
          if clip:
               snake_case			:      Dict        =			torch.clip(__A							,self.min_value							,self.max_value							)
          # Scale to [0, 1].
          snake_case			:      Optional[Any]        =			(features - self.min_value) / (self.max_value - self.min_value)
          # Scale to [min_out, max_out].
          return zero_one * (max_out - min_out) + min_out







     def 			snake_case_      (				self							,__a							,__a=(-1.0, 1.0)							,__a=False							)      ->				Tuple:
          snake_case ,		snake_case			:      List[Any]        =			input_range
          snake_case			:      Union[str, Any]        =			torch.clip(__A							,__A							,__A							) if clip else outputs
          # Scale to [0, 1].
          snake_case			:      Any        =			(outputs - min_out) / (max_out - min_out)
          # Scale to [self.min_value, self.max_value].
          return zero_one * (self.max_value - self.min_value) + self.min_value







     def 			snake_case_      (				self							,__a							,__a							,__a							)      ->				List[Any]:
          snake_case			:      Tuple        =			input_tokens > 0
          snake_case ,		snake_case			:      Any        =			self.notes_encoder(
              encoder_input_tokens=__A							,encoder_inputs_mask=__A							)

          snake_case ,		snake_case			:      Any        =			self.continuous_encoder(
              encoder_inputs=__A							,encoder_inputs_mask=__A							)

          return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]







     def 			snake_case_      (				self							,__a							,__a							,__a							)      ->				str:
          snake_case			:      Optional[Any]        =			noise_time
          if not torch.is_tensor(__A							):
               snake_case			:      Optional[int]        =			torch.tensor([timesteps]							,dtype=torch.long							,device=input_tokens.device							)
          elif torch.is_tensor(__A							) and len(timesteps.shape							) == 0:
               snake_case			:      str        =			timesteps[None].to(input_tokens.device							)

          # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
          snake_case			:      str        =			timesteps * torch.ones(input_tokens.shape[0]							,dtype=timesteps.dtype							,device=timesteps.device							)

          snake_case			:      Optional[Any]        =			self.decoder(
              encodings_and_masks=__A							,decoder_input_tokens=__A							,decoder_noise_time=__A							)
          return logits







     @torch.no_grad()
     def __call__(				self							,__a							,__a = None							,__a = 100							,__a = True							,__a = "numpy"							,__a = None							,__a = 1							,)      ->				str:
          if (callback_steps is None) or (
              callback_steps is not None and (not isinstance(__A							,__A							) or callback_steps <= 0)
          ):
               raise ValueError(
                   F'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
                   F''' {type(__A							)}.'''							)

          snake_case			:      List[Any]        =			np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]							,dtype=np.floataa							)
          snake_case			:      str        =			np.zeros([1, 0, self.n_dims]							,np.floataa							)
          snake_case			:      Union[str, Any]        =			torch.ones((1, TARGET_FEATURE_LENGTH)							,dtype=__A							,device=self.device							)

          for i, encoder_input_tokens in enumerate(__A							):
               if i == 0:
                    snake_case			:      List[Any]        =			torch.from_numpy(pred_mel[:1].copy()							).to(
                        device=self.device							,dtype=self.decoder.dtype							)
                    # The first chunk has no previous context.
                    snake_case			:      Optional[Any]        =			torch.zeros((1, TARGET_FEATURE_LENGTH)							,dtype=__A							,device=self.device							)
               else:
                    # The full song pipeline does not feed in a context feature, so the mask
                    # will be all 0s after the feature converter. Because we know we're
                    # feeding in a full context chunk from the previous prediction, set it
                    # to all 1s.
                    snake_case			:      Optional[int]        =			ones

               snake_case			:      Tuple        =			self.scale_features(
                   __A							,output_range=[-1.0, 1.0]							,clip=__A							)

               snake_case			:      Union[str, Any]        =			self.encode(
                   input_tokens=torch.IntTensor([encoder_input_tokens]							).to(device=self.device							)							,continuous_inputs=__A							,continuous_mask=__A							,)

               # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
               snake_case			:      Optional[Any]        =			randn_tensor(
                   shape=encoder_continuous_inputs.shape							,generator=__A							,device=self.device							,dtype=self.decoder.dtype							,)

               # set step values
               self.scheduler.set_timesteps(__A							)

               # Denoising diffusion loop
               for j, t in enumerate(self.progress_bar(self.scheduler.timesteps							)							):
                    snake_case			:      str        =			self.decode(
                        encodings_and_masks=__A							,input_tokens=__A							,noise_time=t / self.scheduler.config.num_train_timesteps							,)

                    # Compute previous output: x_t -> x_t-1
                    snake_case			:      Any        =			self.scheduler.step(__A							,__A							,__A							,generator=__A							).prev_sample

               snake_case			:      Optional[Any]        =			self.scale_to_features(__A							,input_range=[-1.0, 1.0]							)
               snake_case			:      Any        =			mel[:1]
               snake_case			:      str        =			mel.cpu().float().numpy()

               snake_case			:      int        =			np.concatenate([full_pred_mel, pred_mel[:1]]							,axis=1							)

               # call the callback, if provided
               if callback is not None and i % callback_steps == 0:
                    callback(__A							,__A							)

               logger.info("""Generated segment"""							,__A							)

          if output_type == "numpy" and not is_onnx_available():
               raise ValueError(
                   """Cannot return output in 'np' format if ONNX is not available. Make sure to have ONNX installed or set 'output_type' to 'mel'."""							)
          elif output_type == "numpy" and self.melgan is None:
               raise ValueError(
                   """Cannot return output in 'np' format if melgan component is not defined. Make sure to define `self.melgan` or set 'output_type' to 'mel'."""							)

          if output_type == "numpy":
               snake_case			:      Optional[int]        =			self.melgan(input_features=full_pred_mel.astype(np.floataa							)							)
          else:
               snake_case			:      List[str]        =			full_pred_mel

          if not return_dict:
               return (output,)

          return AudioPipelineOutput(audios=__A							)







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

SCREAMING_SNAKE_CASE							    =							5_0  # max width of layer names
SCREAMING_SNAKE_CASE							    =							7_0  # max width of quantizer names




def  a   (lowerCAmelCase__   ):
   __a					=  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					=  """max"""
   elif args.calibrator == "percentile":
      if args.percentile is None:
         raise ValueError("""Specify --percentile when using percentile calibrator"""   )
      __a					=  """histogram"""
   elif args.calibrator == "mse":
      __a					=  """histogram"""
   else:
      raise ValueError(f'''Invalid calibrator {args.calibrator}'''   )

   __a					=  QuantDescriptor(num_bits=args.aprec		,				calib_method=lowerCAmelCase__   )
   __a					=  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					=  qq._amax.detach().item()
      __a					=  qk._amax.detach().item()
      __a					=  qv._amax.detach().item()

      __a					=  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					=  mod._input_quantizer._amax.data.detach().item()
         mod._input_quantizer._amax.data.detach().clamp_(max=lowerCAmelCase__   )
         __a					=  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					=  mod.weight.shape[0]
         __a					=  mod._weight_quantizer._amax.detach()
         __a					=  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					=  set() if mod._weight_quantizer.axis is None else set(mod._weight_quantizer.axis   )
         __a					=  set(range(len(mod.weight.size()   )   )   ) - axis_set
         __a					=  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					=  amax




def  a   (lowerCAmelCase__		,				lowerCAmelCase__=25		,				lowerCAmelCase__=180		,				lowerCAmelCase__=None   ):

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

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

   for name, mod in model.named_modules():
      __a					=  getattr(lowerCAmelCase__		,				"""_input_quantizer"""		,				lowerCAmelCase__   )
      __a					=  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					=  f'''Act:{input_q.extra_repr()}'''
      __a					=  f'''Wgt:{weight_q.extra_repr()}'''
      __a					=  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					=  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					=  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					=  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					=  f'''Warning: changing {name:{name_width}}'''
               for k, v in kwargs.items():
                  s += f''' {k}={v}'''
                  setattr(lowerCAmelCase__		,				lowerCAmelCase__		,				lowerCAmelCase__   )
               logger.info(lowerCAmelCase__   )



