




"""simple docstring"""

import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig


UpperCamelCase__ :    Optional[Any] 					=							logging.get_logger(__name__)

UpperCamelCase__ :    Any 					=							{
    "facebook/musicgen-small": "https://huggingface.co/facebook/musicgen-small/resolve/main/config.json",
    # See all Musicgen models at https://huggingface.co/models?filter=musicgen
}
class     A__      (      lowercase_     ):
	A_							:				Optional[int]						     =      """musicgen_decoder"""
	A_							:				str						     =      ["""past_key_values"""]




	def __init__(   self  :	List[str]      ,    lowercase  :	str=2_048      ,    lowercase  :	List[Any]=2_048      ,    lowercase  :	List[str]=24      ,    lowercase  :	Union[str, Any]=4_096      ,    lowercase  :	Any=16      ,    lowercase  :	Any=0.0      ,    lowercase  :	List[str]=True      ,    lowercase  :	Optional[int]="gelu"      ,    lowercase  :	Dict=1_024      ,    lowercase  :	List[str]=0.1      ,    lowercase  :	Tuple=0.0      ,    lowercase  :	Dict=0.0      ,    lowercase  :	List[str]=0.02      ,    lowercase  :	Union[str, Any]=False      ,    lowercase  :	Optional[Any]=4      ,    lowercase  :	Any=2_048      ,    lowercase  :	int=2_048      ,    lowercase  :	Optional[int]=None      ,    lowercase  :	List[Any]=False      ,    **lowercase  :	Optional[Any]      ,    )					-> Dict:
						__snake_case          =	vocab_size
						__snake_case          =	max_position_embeddings
						__snake_case          =	hidden_size
						__snake_case          =	ffn_dim
						__snake_case          =	num_hidden_layers
						__snake_case          =	num_attention_heads
						__snake_case          =	dropout
						__snake_case          =	attention_dropout
						__snake_case          =	activation_dropout
						__snake_case          =	activation_function
						__snake_case          =	initializer_factor
						__snake_case          =	layerdrop
						__snake_case          =	use_cache
						__snake_case          =	scale_embedding  # scale factor will be sqrt(d_model) if True
						__snake_case          =	num_codebooks
						super().__init__(
						    pad_token_id=lowercase      ,    bos_token_id=lowercase      ,    eos_token_id=lowercase      ,    tie_word_embeddings=lowercase      ,    **lowercase      ,    )

class     A__      (      lowercase_     ):
	A_							:				Optional[Any]						     =      """musicgen"""
	A_							:				Union[str, Any]						     =      True




	def __init__(   self  :	Tuple      ,    **lowercase  :	int	)					-> Optional[Any]:
						super().__init__(**lowercase	)
						if "text_encoder" not in kwargs or "audio_encoder" not in kwargs or "decoder" not in kwargs:
											raise ValueError("""Config has to be initialized with text_encoder, audio_encoder and decoder config"""	)

						__snake_case          =	kwargs.pop("""text_encoder"""	)
						__snake_case          =	text_encoder_config.pop("""model_type"""	)

						__snake_case          =	kwargs.pop("""audio_encoder"""	)
						__snake_case          =	audio_encoder_config.pop("""model_type"""	)

						__snake_case          =	kwargs.pop("""decoder"""	)

						__snake_case          =	AutoConfig.for_model(lowercase      ,    **lowercase	)
						__snake_case          =	AutoConfig.for_model(lowercase      ,    **lowercase	)
						__snake_case          =	MusicgenDecoderConfig(**lowercase	)
						__snake_case          =	True




	@classmethod
	def 							a_		(   cls  :	int      ,    lowercase  :	PretrainedConfig      ,    lowercase  :	PretrainedConfig      ,    lowercase  :	MusicgenDecoderConfig      ,    **lowercase  :	Dict      ,    )					-> Any:
						return cls(
						    text_encoder=text_encoder_config.to_dict()      ,    audio_encoder=audio_encoder_config.to_dict()      ,    decoder=decoder_config.to_dict()      ,    **lowercase      ,    )




	def 							a_		(   self  :	Optional[int]	)					-> List[str]:
						__snake_case          =	copy.deepcopy(self.__dict__	)
						__snake_case          =	self.text_encoder.to_dict()
						__snake_case          =	self.audio_encoder.to_dict()
						__snake_case          =	self.decoder.to_dict()
						__snake_case          =	self.__class__.model_type
						return output








"""simple docstring"""

import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
		import torch

		from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
		import pyanvml.pyanvml as nvml


UpperCamelCase__ :    List[str] 					=							logging.get_logger(__name__)
class     A__      (      lowercase_     ):
	A_							:				PyTorchBenchmarkArguments
	A_							:				PretrainedConfig
	A_							:				str						     =      "PyTorch"




	@property
	def 							a_		(   self  :	Optional[Any]	)					-> Tuple:
						return torch.__version__




	def 							a_		(   self  :	int      ,    lowercase  :	str      ,    lowercase  :	int      ,    lowercase  :	int	)					-> float:
						__snake_case          =	self._prepare_inference_func(lowercase      ,    lowercase      ,    lowercase	)
						return self._measure_speed(_inference	)




	def 							a_		(   self  :	Dict      ,    lowercase  :	str      ,    lowercase  :	int      ,    lowercase  :	int	)					-> [Memory, Optional[MemorySummary]]:
						__snake_case          =	self._prepare_inference_func(lowercase      ,    lowercase      ,    lowercase	)
						return self._measure_memory(_inference	)




	def 							a_		(   self  :	str      ,    lowercase  :	str      ,    lowercase  :	int      ,    lowercase  :	int	)					-> float:
						__snake_case          =	self._prepare_train_func(lowercase      ,    lowercase      ,    lowercase	)
						return self._measure_speed(_train	)




	def 							a_		(   self  :	List[Any]      ,    lowercase  :	str      ,    lowercase  :	int      ,    lowercase  :	int	)					-> [Memory, Optional[MemorySummary]]:
						__snake_case          =	self._prepare_train_func(lowercase      ,    lowercase      ,    lowercase	)
						return self._measure_memory(_train	)




	def 							a_		(   self  :	Union[str, Any]      ,    lowercase  :	str      ,    lowercase  :	int      ,    lowercase  :	int	)					-> Callable[[], None]:
						__snake_case          =	self.config_dict[model_name]

						if self.args.torchscript:
											__snake_case          =	True

						__snake_case          =	(
						    hasattr(lowercase      ,    """architectures"""	)
						    and isinstance(config.architectures      ,    lowercase	)
						    and len(config.architectures	) > 0
						)
						if not self.args.only_pretrain_model and has_model_class_in_config:
											try:
																__snake_case          =	config.architectures[0]
																__snake_case          =	__import__("""transformers"""      ,    fromlist=[model_class]	)
																__snake_case          =	getattr(lowercase      ,    lowercase	)
																__snake_case          =	model_cls(lowercase	)
											except ImportError:
																raise ImportError(
																    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
																    """ set `--only_pretrain_model` or `args.only_pretrain_model=True`."""	)
						else:
											__snake_case          =	MODEL_MAPPING[config.__class__](lowercase	)

						model.eval()
						model.to(self.args.device	)

						# encoder-decoder has vocab size saved differently
						__snake_case          =	config.vocab_size if hasattr(lowercase      ,    """vocab_size"""	) else config.encoder.vocab_size
						__snake_case          =	torch.randint(lowercase      ,    (batch_size, sequence_length)      ,    dtype=torch.long      ,    device=self.args.device	)

						if self.args.fpaa:
											logger.info("""Running training in Mixed Precision..."""	)
											if not self.args.is_gpu:
																raise ValueError("""Mixed precision is possible only for GPU."""	)
											# amp seems to have memory leaks so that memory usage
											# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
											model.half()

						if self.args.torchscript:
											with torch.no_grad():
																__snake_case          =	torch.jit.trace(lowercase      ,    lowercase	)
						else:
											__snake_case          =	model

						def encoder_decoder_forward():
											with torch.no_grad():
																__snake_case          =	inference_model(lowercase      ,    decoder_input_ids=lowercase	)
											return outputs

						def encoder_forward():
											with torch.no_grad():
																__snake_case          =	inference_model(lowercase	)
											return outputs

						__snake_case          =	encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
						return _forward




	def 							a_		(   self  :	Optional[int]      ,    lowercase  :	str      ,    lowercase  :	int      ,    lowercase  :	int	)					-> Callable[[], None]:
						__snake_case          =	self.config_dict[model_name]

						__snake_case          =	(
						    hasattr(lowercase      ,    """architectures"""	)
						    and isinstance(config.architectures      ,    lowercase	)
						    and len(config.architectures	) > 0
						)
						if not self.args.only_pretrain_model and has_model_class_in_config:
											try:
																__snake_case          =	config.architectures[0]
																__snake_case          =	__import__("""transformers"""      ,    fromlist=[model_class]	)
																__snake_case          =	getattr(lowercase      ,    lowercase	)
																__snake_case          =	model_cls(lowercase	)
											except ImportError:
																raise ImportError(
																    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
																    """ set `--only_pretrain_model` or `args.only_pretrain_model=True`."""	)
						else:
											__snake_case          =	MODEL_WITH_LM_HEAD_MAPPING[config.__class__](lowercase	)

						if self.args.torchscript:
											raise NotImplementedError("""Training for torchscript is currently not implemented"""	)
						else:
											__snake_case          =	model

						model.train()
						model.to(self.args.device	)

						# encoder-decoder has vocab size saved differently
						__snake_case          =	config.vocab_size if hasattr(lowercase      ,    """vocab_size"""	) else config.encoder.vocab_size
						__snake_case          =	torch.randint(lowercase      ,    (batch_size, sequence_length)      ,    dtype=torch.long      ,    device=self.args.device	)

						if self.args.fpaa:
											logger.info("""Running training in Mixed Precision..."""	)
											if not self.args.is_gpu:
																raise ValueError("""Mixed precision is possible only for GPU."""	)

											# amp seems to have memory leaks so that memory usage
											# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
											model.half()

						def compute_loss_and_backprob_encoder():
											__snake_case          =	train_model(lowercase      ,    labels=lowercase	)[0]
											loss.backward()
											return loss

						def compute_loss_and_backprob_encoder_decoder():
											__snake_case          =	train_model(lowercase      ,    decoder_input_ids=lowercase      ,    labels=lowercase	)[0]
											loss.backward()
											return loss

						__snake_case          =	(
						    compute_loss_and_backprob_encoder_decoder
						    if config.is_encoder_decoder
						    else compute_loss_and_backprob_encoder
						)
						return _train




	def 							a_		(   self  :	Tuple      ,    lowercase  :	List[Any]	)					-> float:
						try:
											if self.args.is_tpu or self.args.torchscript:
																# run additional 10 times to stabilize compilation for tpu and torchscript
																logger.info("""Do inference on TPU or torchscript. Running model 5 times to stabilize compilation"""	)
																timeit.repeat(
																    lowercase      ,    repeat=1      ,    number=5      ,    )

											# as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
											__snake_case          =	timeit.repeat(
											    lowercase      ,    repeat=self.args.repeat      ,    number=10      ,    )

											if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
																import torch_xla.debug.metrics as met

																self.print_fn(met.metrics_report()	)

											return min(lowercase	) / 10.0
						except RuntimeError as e:
											self.print_fn(f'''Doesn\'t fit on GPU. {e}'''	)
											return "N/A"




	def 							a_		(   self  :	Union[str, Any]      ,    lowercase  :	Callable[[], None]	)					-> [Memory, MemorySummary]:
						try:
											if self.args.trace_memory_line_by_line:
																__snake_case          =	start_memory_tracing("""transformers"""	)

											if self.args.is_tpu:
																# tpu
																raise NotImplementedError(
																    """Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with"""
																    """ `--no-memory` or `args.memory=False`"""	)
											elif self.args.is_gpu:
																if not is_pyanvml_available():
																					logger.warning(
																					    """py3nvml not installed, we won't log GPU memory usage. """
																					    """Install py3nvml (pip install py3nvml) to log information about GPU."""	)
																					__snake_case          =	"""N/A"""
																else:
																					logger.info(
																					    """Measuring total GPU usage on GPU device. Make sure to not have additional processes running"""
																					    """ on the same GPU."""	)
																					# init nvml
																					nvml.nvmlInit()
																					func()
																					__snake_case          =	nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx	)
																					__snake_case          =	nvml.nvmlDeviceGetMemoryInfo(lowercase	)
																					__snake_case          =	meminfo.used
																					__snake_case          =	Memory(lowercase	)
																					# shutdown nvml
																					nvml.nvmlShutdown()
											else:
																# cpu
																__snake_case          =	measure_peak_memory_cpu(lowercase	)
																__snake_case          =	Memory(lowercase	) if isinstance(lowercase      ,    lowercase	) else memory_bytes

											if self.args.trace_memory_line_by_line:
																__snake_case          =	stop_memory_tracing(lowercase	)
											else:
																__snake_case          =	None

											return memory, summary
						except RuntimeError as e:
											self.print_fn(f'''Doesn\'t fit on GPU. {e}'''	)
											return "N/A", None


