






"""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


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 							A(    ):
       """simple docstring"""





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

       lowercase__:							Optional[Any]         =     parser.parse_args()

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

       return args






def 							A(    ):
       """simple docstring"""





       transformers.utils.logging.set_verbosity_error()
       lowercase__:							Optional[Any]         =     parse_args()

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

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

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

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

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

              lowercase__:							str         =     tokenizer.decode(snake_case_		,  skip_special_tokens=snake_case_						)
              texts_outs.append(snake_case_						)
              generation_times.append(snake_case_						)
              gen_tokens.append(snake_case_						)
              print(F"""Prompt: {prompt}\nGeneration {text_out}\nIn {generation_time:.2f}s for {num_gen_tokens} tokens\n"""						)

       lowercase__:							Dict         =     end_measure(snake_case_						)
       log_measures(snake_case_		,  "Model generation"						)

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


if __name__ == "__main__":
  main()










from collections import OrderedDict
from dataclasses import fields
from typing import Any, Tuple

import numpy as np

from .import_utils import is_torch_available






def 			__lowerCAmelCase       (SCREAMING_SNAKE_CASE   )->    Union[str, Any]:




      """simple docstring"""

      if is_torch_available():
            import torch

            if isinstance(SCREAMING_SNAKE_CASE   ,   torch.Tensor   ):
                  return True

      return isinstance(SCREAMING_SNAKE_CASE   ,   np.ndarray   )







class lowerCAmelCase_		(					lowerCamelCase__			):

 '''simple docstring'''

 def 		UpperCamelCase__	(      self   ):
       snake_case_			= fields(self   )

       # Safety and consistency checks
       if not len(_UpperCAmelCase   ):
             raise ValueError(F'''{self.__class__.__name__} has no fields.'''   )

       snake_case_			= getattr(self						, class_fields[0].name   )
       snake_case_			= all(getattr(self						, field.name   ) is None for field in class_fields[1:]   )

       if other_fields_are_none and isinstance(_UpperCAmelCase						, _UpperCAmelCase   ):
             for key, value in first_field.items():
                   snake_case_			= value
       else:
             for field in class_fields:
                   snake_case_			= getattr(self						, field.name   )
                   if v is not None:
                         snake_case_			= v



 def __delitem__(      self						, *_UpperCAmelCase						, **_UpperCAmelCase   ):
       raise Exception(F'''You cannot use ``__delitem__`` on a {self.__class__.__name__} instance.'''   )



 def 		UpperCamelCase__	(      self						, *_UpperCAmelCase						, **_UpperCAmelCase   ):
       raise Exception(F'''You cannot use ``setdefault`` on a {self.__class__.__name__} instance.'''   )



 def 		UpperCamelCase__	(      self						, *_UpperCAmelCase						, **_UpperCAmelCase   ):
       raise Exception(F'''You cannot use ``pop`` on a {self.__class__.__name__} instance.'''   )



 def 		UpperCamelCase__	(      self						, *_UpperCAmelCase						, **_UpperCAmelCase   ):
       raise Exception(F'''You cannot use ``update`` on a {self.__class__.__name__} instance.'''   )



 def __getitem__(      self						, _UpperCAmelCase   ):
       if isinstance(_UpperCAmelCase						, _UpperCAmelCase   ):
             snake_case_			= dict(self.items()   )
             return inner_dict[k]
       else:
             return self.to_tuple()[k]



 def __setattr__(      self						, _UpperCAmelCase						, _UpperCAmelCase   ):
       if name in self.keys() and value is not None:
             # Don't call self.__setitem__ to avoid recursion errors
             super().__setitem__(_UpperCAmelCase						, _UpperCAmelCase   )
       super().__setattr__(_UpperCAmelCase						, _UpperCAmelCase   )



 def __setitem__(      self						, _UpperCAmelCase						, _UpperCAmelCase   ):
       # Will raise a KeyException if needed
       super().__setitem__(_UpperCAmelCase						, _UpperCAmelCase   )
       # Don't call self.__setattr__ to avoid recursion errors
       super().__setattr__(_UpperCAmelCase						, _UpperCAmelCase   )





 def 		UpperCamelCase__	(      self   ):
       return tuple(self[k] for k in self.keys()   )