






"""simple docstring"""



import argparse
import time

import torch
import transformers
from measures_util import end_measure, log_measures, start_measure
from transformers import AutoConfig, AutoModelForCausalLM, AutoModelForSeqaSeqLM, AutoTokenizer

from accelerate.utils import compute_module_sizes


A_			=							{
    '''gpt-j-6b''': {'''is_causal''': True, '''model''': '''sgugger/sharded-gpt-j-6B''', '''tokenizer''': '''EleutherAI/gpt-j-6B'''},
    '''gpt-neox''': {'''is_causal''': True, '''model''': '''EleutherAI/gpt-neox-20b'''},
    '''opt''': {'''is_causal''': True, '''model''': '''facebook/opt-30b'''},
    '''T0pp''': {'''is_causal''': False, '''model''': '''bigscience/T0pp''', '''model_revision''': '''sharded'''},
}

A_			=							[
    '''Hello, my name is''',
    '''Are unicorns real? Unicorns are''',
    '''For the first time in several years,''',
    '''My name is Julien and I am''',
    '''The goal of life is''',
    '''Whenever I\'m sad, I like to''',
]




def    _lowerCAmelCase						(							)						->List[str]:
						A__  :     List[Any]						     =     argparse.ArgumentParser(description="""Run and time generations on a big model using Accelerate."""					)
						parser.add_argument("""model_name""",  type=UpperCAmelCase__,  default=UpperCAmelCase__,  help="""The name of the model to try."""					)
						parser.add_argument(
						    """--tokenizer_name""",  type=UpperCAmelCase__,  default=UpperCAmelCase__,  help="""The name of the tokenizer (if different from the model."""					)
						parser.add_argument("""--is_causal""",  type=UpperCAmelCase__,  default=UpperCAmelCase__,  help="""Whether or not the model is causal."""					)
						parser.add_argument(
						    """--model_revision""",  type=UpperCAmelCase__,  default=UpperCAmelCase__,  help="""The revision to use for the model checkpoint."""					)
						parser.add_argument("""--torch_dtype""",  type=UpperCAmelCase__,  default=UpperCAmelCase__,  help="""The dtype for the model."""					)
						parser.add_argument("""--disk_offload""",  action="""store_true"""					)

						A__  :     str						     =     parser.parse_args()

						# Sanitize args
						if args.model_name in DEFAULT_MODELS:
												A__  :     Optional[Any]						     =     DEFAULT_MODELS[args.model_name]
												A__  :     List[str]						     =     defaults["model"]
												if args.tokenizer_name is None:
																		A__  :     str						     =     defaults.get("""tokenizer""",  args.model_name					)
												if args.is_causal is None:
																		A__  :     List[str]						     =     defaults["is_causal"]
												if args.model_revision is None:
																		A__  :     List[str]						     =     defaults.get("""model_revision""",  """main"""					)

						if args.is_causal is None:
												raise ValueError("""Could not infer the default for `--is_causal`, pass either True or False for it."""					)
						if args.tokenizer_name is None:
												A__  :     Any						     =     args.model_name
						if args.model_revision is None:
												A__  :     Optional[int]						     =     "main"

						return args




def    _lowerCAmelCase						(							)						->Optional[int]:
						transformers.utils.logging.set_verbosity_error()
						A__  :     Any						     =     parse_args()

						if args.torch_dtype is None:
												A__  :     Dict						     =     AutoConfig.from_pretrained(args.model_name					)
												A__  :     int						     =     getattr(UpperCAmelCase__,  """torch_dtype""",  torch.floataa					)
						else:
												A__  :     List[Any]						     =     getattr(UpperCAmelCase__,  args.torch_dtype					)
						A__  :     int						     =     AutoModelForCausalLM if args.is_causal else AutoModelForSeqaSeqLM
						A__  :     int						     =     {
						    "torch_dtype": torch_dtype,
						    "revision": args.model_revision,
						}
						if args.disk_offload:
												A__  :     Any						     =     "tmp_offload"
												A__  :     Tuple						     =     True

						A__  :     Any						     =     start_measure()
						A__  :     int						     =     model_cls.from_pretrained(args.model_name,  device_map="""auto""",  **UpperCAmelCase__					)
						A__  :     List[Any]						     =     end_measure(UpperCAmelCase__					)
						log_measures(UpperCAmelCase__,  """Model loading"""					)

						A__  :     Optional[Any]						     =     compute_module_sizes(UpperCAmelCase__					)
						A__  :     str						     =     {v: 0 for v in model.hf_device_map.values()}
						for module, device in model.hf_device_map.items():
												device_size[device] += module_sizes[module]
						A__  :     str						     =     "\n".join([f'- {device}: {size // 2**2_0}MiB' for device, size in device_size.items()]					)
						print(f'\nTheoretical use:\n{message}'					)

						A__  :     List[str]						     =     AutoTokenizer.from_pretrained(args.tokenizer_name					)

						A__  :     Tuple						     =     start_measure()
						A__  :     str						     =     []
						A__  :     Optional[Any]						     =     []
						A__  :     List[str]						     =     []
						for prompt in PROMPTS:
												A__  :     str						     =     tokenizer(UpperCAmelCase__,  return_tensors="""pt"""					).to(0					)
												A__  :     Any						     =     inputs["input_ids"][0].tolist()
												A__  :     Dict						     =     time.time()
												A__  :     Any						     =     model.generate(inputs["""input_ids"""]					)
												A__  :     Dict						     =     time.time()
												A__  :     Dict						     =     outputs[0].tolist()
												A__  :     Any						     =     len(UpperCAmelCase__					) if outputs[: len(UpperCAmelCase__					)] != tokens else len(UpperCAmelCase__					) - len(UpperCAmelCase__					)
												A__  :     Union[str, Any]						     =     after_generate - before_generate

												A__  :     Union[str, Any]						     =     tokenizer.decode(UpperCAmelCase__,  skip_special_tokens=UpperCAmelCase__					)
												texts_outs.append(UpperCAmelCase__					)
												generation_times.append(UpperCAmelCase__					)
												gen_tokens.append(UpperCAmelCase__					)
												print(f'Prompt: {prompt}\nGeneration {text_out}\nIn {generation_time:.2f}s for {num_gen_tokens} tokens\n'					)

						A__  :     Any						     =     end_measure(UpperCAmelCase__					)
						log_measures(UpperCAmelCase__,  """Model generation"""					)

						A__  :     str						     =     [gen / tok for gen, tok in zip(UpperCAmelCase__,  UpperCAmelCase__					)]
						A__  :     int						     =     sum(UpperCAmelCase__					) / len(UpperCAmelCase__					)
						print(f'Average time of generation per token: {avg_gen:.2f}s'					)
						print(f'First generation (avg time per token): {generation_times_per_token[0]:.2f}s'					)
						A__  :     Optional[Any]						     =     sum(generation_times_per_token[1:]					) / (len(UpperCAmelCase__					) - 1)
						print(f'Average time of generation per token (excluding the first): {avg_gen:.2f}s'					)


if __name__ == "__main__":
						main()














import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch

   from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
       SpeechaTextaDecoder,
       SpeechaTextaForCausalLM,
   )


@require_torch
class        SCREAMING_SNAKE_CASE	:
     '''simple docstring'''


     def __init__( self				:		List[str]    ,       UpperCAmelCase_				:		Any    ,       UpperCAmelCase_				:		List[Any]=99    ,       UpperCAmelCase_				:		Tuple=13    ,       UpperCAmelCase_				:		Dict=16    ,       UpperCAmelCase_				:		int=7    ,       UpperCAmelCase_				:		Optional[Any]=True    ,       UpperCAmelCase_				:		List[Any]=True    ,       UpperCAmelCase_				:		Any=True    ,       UpperCAmelCase_				:		List[str]=False    ,       UpperCAmelCase_				:		Tuple=True    ,       UpperCAmelCase_				:		Tuple=2    ,       UpperCAmelCase_				:		List[str]=32    ,       UpperCAmelCase_				:		Union[str, Any]=4    ,       UpperCAmelCase_				:		Any=4    ,       UpperCAmelCase_				:		Optional[int]=30    ,       UpperCAmelCase_				:		Union[str, Any]=0    ,       UpperCAmelCase_				:		Union[str, Any]=1    ,       UpperCAmelCase_				:		Any=2    ,       UpperCAmelCase_				:		Dict=None    ,       ):
       SCREAMING_SNAKE_CASE				:					Any		  =			parent
       SCREAMING_SNAKE_CASE				:					Tuple		  =			batch_size
       SCREAMING_SNAKE_CASE				:					str		  =			decoder_seq_length
       # For common tests
       SCREAMING_SNAKE_CASE				:					Tuple		  =			self.decoder_seq_length
       SCREAMING_SNAKE_CASE				:					str		  =			is_training
       SCREAMING_SNAKE_CASE				:					Dict		  =			use_attention_mask
       SCREAMING_SNAKE_CASE				:					int		  =			use_labels

       SCREAMING_SNAKE_CASE				:					Tuple		  =			vocab_size
       SCREAMING_SNAKE_CASE				:					int		  =			d_model
       SCREAMING_SNAKE_CASE				:					List[str]		  =			d_model
       SCREAMING_SNAKE_CASE				:					int		  =			decoder_layers
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			decoder_layers
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			decoder_ffn_dim
       SCREAMING_SNAKE_CASE				:					Tuple		  =			decoder_attention_heads
       SCREAMING_SNAKE_CASE				:					Dict		  =			decoder_attention_heads
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			eos_token_id
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			bos_token_id
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			pad_token_id
       SCREAMING_SNAKE_CASE				:					List[str]		  =			decoder_start_token_id
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			use_cache
       SCREAMING_SNAKE_CASE				:					Tuple		  =			max_position_embeddings

       SCREAMING_SNAKE_CASE				:					Any		  =			None
       SCREAMING_SNAKE_CASE				:					Tuple		  =			decoder_seq_length
       SCREAMING_SNAKE_CASE				:					int		  =			2
       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			1


     def  _A				( self				:		Optional[int]					):
       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			ids_tensor([self.batch_size, self.decoder_seq_length]    ,       self.vocab_size					)

       SCREAMING_SNAKE_CASE				:					str		  =			None
       if self.use_attention_mask:
         SCREAMING_SNAKE_CASE				:					Tuple		  =			ids_tensor([self.batch_size, self.decoder_seq_length]    ,       vocab_size=2					)

       SCREAMING_SNAKE_CASE				:					str		  =			None
       if self.use_labels:
         SCREAMING_SNAKE_CASE				:					List[str]		  =			ids_tensor([self.batch_size, self.decoder_seq_length]    ,       self.vocab_size					)

       SCREAMING_SNAKE_CASE				:					str		  =			SpeechaTextaConfig(
           vocab_size=self.vocab_size    ,       d_model=self.d_model    ,       decoder_layers=self.decoder_layers    ,       decoder_ffn_dim=self.decoder_ffn_dim    ,       decoder_attention_heads=self.decoder_attention_heads    ,       eos_token_id=self.eos_token_id    ,       bos_token_id=self.bos_token_id    ,       use_cache=self.use_cache    ,       pad_token_id=self.pad_token_id    ,       decoder_start_token_id=self.decoder_start_token_id    ,       max_position_embeddings=self.max_position_embeddings    ,       )

       return (
           config,
           input_ids,
           attention_mask,
           lm_labels,
       )


     def  _A				( self				:		Optional[Any]    ,       UpperCAmelCase_				:		int    ,       UpperCAmelCase_				:		Dict    ,       UpperCAmelCase_				:		int    ,       UpperCAmelCase_				:		Any    ,       ):
       SCREAMING_SNAKE_CASE				:					str		  =			True
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			SpeechaTextaDecoder(config=UpperCAmelCase_					).to(UpperCAmelCase_					).eval()
       SCREAMING_SNAKE_CASE				:					Tuple		  =			input_ids[:2]

       input_ids[input_ids == 0] += 1
       # first forward pass
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			model(UpperCAmelCase_    ,       use_cache=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					List[str]		  =			model(UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			model(UpperCAmelCase_    ,       use_cache=UpperCAmelCase_					)

       self.parent.assertTrue(len(UpperCAmelCase_					) == len(UpperCAmelCase_					)					)
       self.parent.assertTrue(len(UpperCAmelCase_					) == len(UpperCAmelCase_					) + 1					)

       SCREAMING_SNAKE_CASE				:					List[Any]		  =			outputs["past_key_values"]

       # create hypothetical next token and extent to next_input_ids
       SCREAMING_SNAKE_CASE				:					str		  =			ids_tensor((2, 1)    ,       config.vocab_size - 1					) + 1

       # append to next input_ids and
       SCREAMING_SNAKE_CASE				:					str		  =			torch.cat([input_ids, next_tokens]    ,       dim=-1					)

       print(UpperCAmelCase_					)

       SCREAMING_SNAKE_CASE				:					List[str]		  =			model(UpperCAmelCase_					)["last_hidden_state"]
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			model(UpperCAmelCase_    ,       past_key_values=UpperCAmelCase_					)["last_hidden_state"]

       # select random slice
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			ids_tensor((1,)    ,       output_from_past.shape[-1]					).item()
       SCREAMING_SNAKE_CASE				:					str		  =			output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
       SCREAMING_SNAKE_CASE				:					int		  =			output_from_past[:, 0, random_slice_idx].detach()

       # test that outputs are equal for slice
       assert torch.allclose(UpperCAmelCase_    ,       UpperCAmelCase_    ,       atol=1E-3					)






     def  _A				( self				:		str					):
       SCREAMING_SNAKE_CASE				:					int		  =			self.prepare_config_and_inputs()
       (
           (
           SCREAMING_SNAKE_CASE
       )  ,     (
           SCREAMING_SNAKE_CASE
       )  ,     (
           SCREAMING_SNAKE_CASE
       )  ,     (
           SCREAMING_SNAKE_CASE
       )  ,     
       )				:					Optional[Any]		  =			config_and_inputs

       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			{
           "input_ids": input_ids,
           "attention_mask": attention_mask,
       }
       return config, inputs_dict





@require_torch
class        SCREAMING_SNAKE_CASE	(	lowerCAmelCase					,						lowerCAmelCase					,						lowerCAmelCase					,						unittest.TestCase    ):
     '''simple docstring'''


     UpperCamelCase_							:						int								=		(SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
     UpperCamelCase_							:						Tuple								=		(SpeechaTextaForCausalLM,) if is_torch_available() else ()
     UpperCamelCase_							:						int								=		{'''text-generation''': SpeechaTextaForCausalLM} if is_torch_available() else {}
     UpperCamelCase_							:						Tuple								=		True
     UpperCamelCase_							:						Union[str, Any]								=		False


     def  _A				( self				:		Tuple    ,       ):
       SCREAMING_SNAKE_CASE				:					Any		  =			SpeechaTextaStandaloneDecoderModelTester(self    ,       is_training=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			ConfigTester(self    ,       config_class=UpperCAmelCase_					)


     def  _A				( self				:		Optional[Any]					):
       pass


     def  _A				( self				:		str					):
       pass


     def  _A				( self				:		Dict					):
       pass


     def  _A				( self				:		Optional[Any]					):
       self.config_tester.run_common_tests()


     def  _A				( self				:		Optional[int]					):
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_decoder_model_past(*UpperCAmelCase_					)






     def  _A				( self				:		Optional[int]					):
       return






