

"""simple docstring"""





from collections import OrderedDict
from typing import Any, List, Mapping, Optional

from ... import PreTrainedTokenizer, TensorType, is_torch_available
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfigWithPast, PatchingSpec
from ...utils import logging


__UpperCamelCase  :      List[str]							      =		logging.get_logger(__name__)


__UpperCamelCase  :      Dict							      =		{
    '''Salesforce/codegen-350M-nl''': '''https://huggingface.co/Salesforce/codegen-350M-nl/resolve/main/config.json''',
    '''Salesforce/codegen-350M-multi''': '''https://huggingface.co/Salesforce/codegen-350M-multi/resolve/main/config.json''',
    '''Salesforce/codegen-350M-mono''': '''https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/config.json''',
    '''Salesforce/codegen-2B-nl''': '''https://huggingface.co/Salesforce/codegen-2B-nl/resolve/main/config.json''',
    '''Salesforce/codegen-2B-multi''': '''https://huggingface.co/Salesforce/codegen-2B-multi/resolve/main/config.json''',
    '''Salesforce/codegen-2B-mono''': '''https://huggingface.co/Salesforce/codegen-2B-mono/resolve/main/config.json''',
    '''Salesforce/codegen-6B-nl''': '''https://huggingface.co/Salesforce/codegen-6B-nl/resolve/main/config.json''',
    '''Salesforce/codegen-6B-multi''': '''https://huggingface.co/Salesforce/codegen-6B-multi/resolve/main/config.json''',
    '''Salesforce/codegen-6B-mono''': '''https://huggingface.co/Salesforce/codegen-6B-mono/resolve/main/config.json''',
    '''Salesforce/codegen-16B-nl''': '''https://huggingface.co/Salesforce/codegen-16B-nl/resolve/main/config.json''',
    '''Salesforce/codegen-16B-multi''': '''https://huggingface.co/Salesforce/codegen-16B-multi/resolve/main/config.json''',
    '''Salesforce/codegen-16B-mono''': '''https://huggingface.co/Salesforce/codegen-16B-mono/resolve/main/config.json''',
}







class 				SCREAMING_SNAKE_CASE					(		a_       ):






					"""simple docstring"""



					lowercase__				 =       "codegen"
					lowercase__				 =       {
					    "max_position_embeddings": "n_positions",
					    "hidden_size": "n_embd",
					    "num_attention_heads": "n_head",
					    "num_hidden_layers": "n_layer",
					}







					def __init__(       self			:				Tuple   ,lowercase_			:				Tuple=5_0_4_0_0   ,lowercase_			:				Any=2_0_4_8   ,lowercase_			:				Union[str, Any]=2_0_4_8   ,lowercase_			:				List[str]=4_0_9_6   ,lowercase_			:				Dict=2_8   ,lowercase_			:				List[Any]=1_6   ,lowercase_			:				Tuple=6_4   ,lowercase_			:				str=None   ,lowercase_			:				List[str]="gelu_new"   ,lowercase_			:				int=0.0   ,lowercase_			:				Optional[Any]=0.0   ,lowercase_			:				Union[str, Any]=0.0   ,lowercase_			:				Tuple=1E-5   ,lowercase_			:				Union[str, Any]=0.02   ,lowercase_			:				str=True   ,lowercase_			:				Any=5_0_2_5_6   ,lowercase_			:				List[str]=5_0_2_5_6   ,lowercase_			:				List[str]=False   ,**lowercase_			:				Tuple   ,):
						lowerCAmelCase__	:						str  				=	vocab_size
						lowerCAmelCase__	:						Union[str, Any]  				=	n_ctx
						lowerCAmelCase__	:						Optional[int]  				=	n_positions
						lowerCAmelCase__	:						Dict  				=	n_embd
						lowerCAmelCase__	:						Any  				=	n_layer
						lowerCAmelCase__	:						Union[str, Any]  				=	n_head
						lowerCAmelCase__	:						List[str]  				=	n_inner
						lowerCAmelCase__	:						Optional[Any]  				=	rotary_dim
						lowerCAmelCase__	:						Union[str, Any]  				=	activation_function
						lowerCAmelCase__	:						Optional[int]  				=	resid_pdrop
						lowerCAmelCase__	:						Optional[Any]  				=	embd_pdrop
						lowerCAmelCase__	:						List[str]  				=	attn_pdrop
						lowerCAmelCase__	:						Union[str, Any]  				=	layer_norm_epsilon
						lowerCAmelCase__	:						Optional[int]  				=	initializer_range
						lowerCAmelCase__	:						Tuple  				=	use_cache

						lowerCAmelCase__	:						List[str]  				=	bos_token_id
						lowerCAmelCase__	:						Optional[Any]  				=	eos_token_id

						super().__init__(
						    bos_token_id=lowercase_   ,eos_token_id=lowercase_   ,tie_word_embeddings=lowercase_   ,**lowercase_		)







class 				SCREAMING_SNAKE_CASE					(		a_       ):






					"""simple docstring"""



					def __init__(       self			:				Optional[int]   ,lowercase_			:				PretrainedConfig   ,lowercase_			:				str = "default"   ,lowercase_			:				List[PatchingSpec] = None   ,lowercase_			:				bool = False   ,):
						super().__init__(lowercase_   ,task=lowercase_   ,patching_specs=lowercase_   ,use_past=lowercase_		)
						if not getattr(self._config   ,'''pad_token_id'''   ,lowercase_		):
							# TODO: how to do that better?
							lowerCAmelCase__	:						List[Any]  				=	0







					@property
					def        __lowerCAmelCase (       self			:				Tuple		):
						lowerCAmelCase__	:						Union[str, Any]  				=	OrderedDict({'''input_ids''': {0: '''batch''', 1: '''sequence'''}}		)
						if self.use_past:
							self.fill_with_past_key_values_(lowercase_   ,direction='''inputs'''		)
							lowerCAmelCase__	:						Union[str, Any]  				=	{0: '''batch''', 1: '''past_sequence + sequence'''}
						else:
							lowerCAmelCase__	:						List[Any]  				=	{0: '''batch''', 1: '''sequence'''}

						return common_inputs







					@property
					def        __lowerCAmelCase (       self			:				int		):
						return self._config.n_layer







					@property
					def        __lowerCAmelCase (       self			:				List[Any]		):
						return self._config.n_head







					def        __lowerCAmelCase (       self			:				Optional[Any]   ,lowercase_			:				PreTrainedTokenizer   ,lowercase_			:				int = -1   ,lowercase_			:				int = -1   ,lowercase_			:				bool = False   ,lowercase_			:				Optional[TensorType] = None   ,):
						lowerCAmelCase__	:						str  				=	super(lowercase_   ,self		).generate_dummy_inputs(
						    lowercase_   ,batch_size=lowercase_   ,seq_length=lowercase_   ,is_pair=lowercase_   ,framework=lowercase_		)

						# We need to order the input in the way they appears in the forward()
						lowerCAmelCase__	:						Any  				=	OrderedDict({'''input_ids''': common_inputs['''input_ids''']}		)

						# Need to add the past_keys
						if self.use_past:
							if not is_torch_available():
								raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''		)
							else:
								import torch

								lowerCAmelCase__				,lowerCAmelCase__	:						Union[str, Any]  				=	common_inputs['''input_ids'''].shape
								# Not using the same length for past_key_values
								lowerCAmelCase__	:						Optional[Any]  				=	seqlen + 2
								lowerCAmelCase__	:						Union[str, Any]  				=	(
								    batch,
								    self.num_attention_heads,
								    past_key_values_length,
								    self._config.hidden_size // self.num_attention_heads,
								)
								lowerCAmelCase__	:						Dict  				=	[
								    (torch.zeros(lowercase_		), torch.zeros(lowercase_		)) for _ in range(self.num_layers		)
								]

						lowerCAmelCase__	:						Any  				=	common_inputs['''attention_mask''']
						if self.use_past:
							lowerCAmelCase__	:						List[str]  				=	ordered_inputs['''attention_mask'''].dtype
							lowerCAmelCase__	:						Dict  				=	torch.cat(
							    [ordered_inputs['''attention_mask'''], torch.ones(lowercase_   ,lowercase_   ,dtype=lowercase_		)]   ,dim=1		)

						return ordered_inputs







					@property
					def        __lowerCAmelCase (       self			:				Any		):
						return 1_3





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


UpperCamelCase  							=						logging.get_logger(__name__)  # pylint: disable=invalid-name

UpperCamelCase  							=						256



class   snake_case_       (		__A       ):
   __A	:		str         =				["melgan"]






   def __init__(      self				:     str	,						lowercase_				:     SpectrogramNotesEncoder	,						lowercase_				:     SpectrogramContEncoder	,						lowercase_				:     TaFilmDecoder	,						lowercase_				:     DDPMScheduler	,						lowercase_				:     OnnxRuntimeModel if is_onnx_available() else Any	,						)					->						None:
       super().__init__()

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

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






   def       __UpperCamelCase     (      self				:     Union[str, Any]	,						lowercase_				:     Union[str, Any]	,						lowercase_				:     List[str]=(-1.0, 1.0)	,						lowercase_				:     Dict=False  )					->						Optional[Any]:
       lowercase__					,   lowercase__ :   int    					=				output_range
       if clip:
           lowercase__ :   Optional[Any]    					=				torch.clip(lowercase_	,						self.min_value	,						self.max_value  )
       # Scale to [0, 1].
       lowercase__ :   List[str]    					=				(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       __UpperCamelCase     (      self				:     Optional[int]	,						lowercase_				:     List[str]	,						lowercase_				:     List[str]=(-1.0, 1.0)	,						lowercase_				:     List[Any]=False  )					->						Union[str, Any]:
       lowercase__					,   lowercase__ :   Tuple    					=				input_range
       lowercase__ :   Optional[Any]    					=				torch.clip(lowercase_	,						lowercase_	,						lowercase_  ) if clip else outputs
       # Scale to [0, 1].
       lowercase__ :   Union[str, 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       __UpperCamelCase     (      self				:     List[str]	,						lowercase_				:     Any	,						lowercase_				:     Optional[Any]	,						lowercase_				:     Tuple  )					->						List[str]:
       lowercase__ :   Optional[Any]    					=				input_tokens > 0
       lowercase__					,   lowercase__ :   int    					=				self.notes_encoder(
           encoder_input_tokens=lowercase_	,						encoder_inputs_mask=lowercase_  )

       lowercase__					,   lowercase__ :   List[Any]    					=				self.continuous_encoder(
           encoder_inputs=lowercase_	,						encoder_inputs_mask=lowercase_  )

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






   def       __UpperCamelCase     (      self				:     Optional[Any]	,						lowercase_				:     Optional[int]	,						lowercase_				:     Optional[int]	,						lowercase_				:     str  )					->						Tuple:
       lowercase__ :   Union[str, Any]    					=				noise_time
       if not torch.is_tensor(lowercase_  ):
           lowercase__ :   Optional[Any]    					=				torch.tensor([timesteps]	,						dtype=torch.long	,						device=input_tokens.device  )
       elif torch.is_tensor(lowercase_  ) and len(timesteps.shape  ) == 0:
           lowercase__ :   Optional[Any]    					=				timesteps[None].to(input_tokens.device  )

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

       lowercase__ :   str    					=				self.decoder(
           encodings_and_masks=lowercase_	,						decoder_input_tokens=lowercase_	,						decoder_noise_time=lowercase_  )
       return logits







   @torch.no_grad()
   def __call__(      self				:     List[str]	,						lowercase_				:     List[List[int]]	,						lowercase_				:     Optional[torch.Generator] = None	,						lowercase_				:     int = 1_00	,						lowercase_				:     bool = True	,						lowercase_				:     str = "numpy"	,						lowercase_				:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,						lowercase_				:     int = 1	,						)					->						Union[AudioPipelineOutput, Tuple]:
       if (callback_steps is None) or (
           callback_steps is not None and (not isinstance(lowercase_	,						lowercase_  ) or callback_steps <= 0)
       ):
           raise ValueError(
               F'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
               F''' {type(lowercase_  )}.'''  )

       lowercase__ :   str    					=				np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]	,						dtype=np.floataa  )
       lowercase__ :   Optional[int]    					=				np.zeros([1, 0, self.n_dims]	,						np.floataa  )
       lowercase__ :   str    					=				torch.ones((1, TARGET_FEATURE_LENGTH)	,						dtype=lowercase_	,						device=self.device  )

       for i, encoder_input_tokens in enumerate(lowercase_  ):
           if i == 0:
               lowercase__ :   Union[str, Any]    					=				torch.from_numpy(pred_mel[:1].copy()  ).to(
                   device=self.device	,						dtype=self.decoder.dtype  )
               # The first chunk has no previous context.
               lowercase__ :   List[str]    					=				torch.zeros((1, TARGET_FEATURE_LENGTH)	,						dtype=lowercase_	,						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.
               lowercase__ :   str    					=				ones

           lowercase__ :   str    					=				self.scale_features(
               lowercase_	,						output_range=[-1.0, 1.0]	,						clip=lowercase_  )

           lowercase__ :   str    					=				self.encode(
               input_tokens=torch.IntTensor([encoder_input_tokens]  ).to(device=self.device  )	,						continuous_inputs=lowercase_	,						continuous_mask=lowercase_	,						)

           # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
           lowercase__ :   List[str]    					=				randn_tensor(
               shape=encoder_continuous_inputs.shape	,						generator=lowercase_	,						device=self.device	,						dtype=self.decoder.dtype	,						)

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

           # Denoising diffusion loop
           for j, t in enumerate(self.progress_bar(self.scheduler.timesteps  )  ):
               lowercase__ :   Optional[int]    					=				self.decode(
                   encodings_and_masks=lowercase_	,						input_tokens=lowercase_	,						noise_time=t / self.scheduler.config.num_train_timesteps	,						)

               # Compute previous output: x_t -> x_t-1
               lowercase__ :   Optional[Any]    					=				self.scheduler.step(lowercase_	,						lowercase_	,						lowercase_	,						generator=lowercase_  ).prev_sample

           lowercase__ :   Tuple    					=				self.scale_to_features(lowercase_	,						input_range=[-1.0, 1.0]  )
           lowercase__ :   List[str]    					=				mel[:1]
           lowercase__ :   Optional[int]    					=				mel.cpu().float().numpy()

           lowercase__ :   str    					=				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(lowercase_	,						lowercase_  )

           logger.info("Generated segment"	,						lowercase_  )

       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":
           lowercase__ :   Union[str, Any]    					=				self.melgan(input_features=full_pred_mel.astype(np.floataa  )  )
       else:
           lowercase__ :   Dict    					=				full_pred_mel

       if not return_dict:
           return (output,)

       return AudioPipelineOutput(audios=lowercase_  )



