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


UpperCAmelCase							   =      {
    """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"""},
}

UpperCAmelCase							   =      [
    """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 	UpperCAmelCase       (		)	->	int:
     a  :    Optional[int]				=						argparse.ArgumentParser(description='''Run and time generations on a big model using Accelerate.'''       )
     parser.add_argument('''model_name'''						,      type=lowerCAmelCase_						,      default=lowerCAmelCase_						,      help='''The name of the model to try.'''       )
     parser.add_argument(
         '''--tokenizer_name'''						,      type=lowerCAmelCase_						,      default=lowerCAmelCase_						,      help='''The name of the tokenizer (if different from the model.'''       )
     parser.add_argument('''--is_causal'''						,      type=lowerCAmelCase_						,      default=lowerCAmelCase_						,      help='''Whether or not the model is causal.'''       )
     parser.add_argument(
         '''--model_revision'''						,      type=lowerCAmelCase_						,      default=lowerCAmelCase_						,      help='''The revision to use for the model checkpoint.'''       )
     parser.add_argument('''--torch_dtype'''						,      type=lowerCAmelCase_						,      default=lowerCAmelCase_						,      help='''The dtype for the model.'''       )
     parser.add_argument('''--disk_offload'''						,      action='''store_true'''       )

     a  :    Any				=						parser.parse_args()

     # Sanitize args
     if args.model_name in DEFAULT_MODELS:
          a  :    int				=						DEFAULT_MODELS[args.model_name]
          a  :    Optional[int]				=						defaults["model"]
          if args.tokenizer_name is None:
               a  :    Union[str, Any]				=						defaults.get('''tokenizer'''						,      args.model_name       )
          if args.is_causal is None:
               a  :    Tuple				=						defaults["is_causal"]
          if args.model_revision is None:
               a  :    Dict				=						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  :    str				=						args.model_name
     if args.model_revision is None:
          a  :    Any				=						"main"

     return args







def 	UpperCAmelCase       (		)	->	Union[str, Any]:
     transformers.utils.logging.set_verbosity_error()
     a  :    str				=						parse_args()

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

     a  :    Tuple				=						start_measure()
     a  :    Any				=						model_cls.from_pretrained(args.model_name						,      device_map='''auto'''						,      **lowerCAmelCase_       )
     a  :    Optional[int]				=						end_measure(lowerCAmelCase_       )
     log_measures(lowerCAmelCase_						,      '''Model loading'''       )

     a  :    List[Any]				=						compute_module_sizes(lowerCAmelCase_       )
     a  :    Tuple				=						{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  :    int				=						"\n".join([f"""- {device}: {size // 2**2_0}MiB""" for device, size in device_size.items()]       )
     print(f"""\nTheoretical use:\n{message}"""       )

     a  :    str				=						AutoTokenizer.from_pretrained(args.tokenizer_name       )

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

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

     a  :    Dict				=						end_measure(lowerCAmelCase_       )
     log_measures(lowerCAmelCase_						,      '''Model generation'''       )

     a  :    Any				=						[gen / tok for gen, tok in zip(lowerCAmelCase_						,      lowerCAmelCase_       )]
     a  :    Any				=						sum(lowerCAmelCase_       ) / len(lowerCAmelCase_       )
     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  :    int				=						sum(generation_times_per_token[1:]       ) / (len(lowerCAmelCase_       ) - 1)
     print(f"""Average time of generation per token (excluding the first): {avg_gen:.2f}s"""       )


if __name__ == "__main__":
  main()
# this script builds a small sample spm file tests/fixtures/test_sentencepiece_no_bos.model, with features needed by pegasus

# 1. pip install sentencepiece
#
# 2. wget https://raw.githubusercontent.com/google/sentencepiece/master/data/botchan.txt

# 3. build
import sentencepiece as spm

# pegasus:
# 1. no bos
# 2. eos_id is 1
# 3. unk_id is 2
# build a sample spm file accordingly
spm.SentencePieceTrainer.train(
    """--input=botchan.txt --model_prefix=test_sentencepiece_no_bos --bos_id=-1 --unk_id=2  --eos_id=1  --vocab_size=1000"""
)

# 4. now update the fixture
# mv test_sentencepiece_no_bos.model ../../tests/fixtures/