


"""simple docstring"""





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   _snake_case    (    _snake_case					:      Dict      ):
      if is_torch_available():
            import torch

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

      return isinstance(_snake_case			, np.ndarray      )






class 		snake_case_(							a__      ):






     def   lowerCamelCase__  (						self	:       Optional[Any]					):
           lowerCAmelCase       :						Optional[int]							= fields(self					)

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

           lowerCAmelCase       :						Any							= getattr(self       ,					class_fields[0].name					)
           lowerCAmelCase       :						Union[str, Any]							= 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():
                       lowerCAmelCase       :						Any							= value
           else:
                 for field in class_fields:
                       lowerCAmelCase       :						int							= getattr(self       ,					field.name					)
                       if v is not None:
                             lowerCAmelCase       :						Tuple							= v






     def __delitem__(						self	:       Tuple       ,					*UpperCamelCase_	:       Any       ,					**UpperCamelCase_	:       Union[str, Any]					):
           raise Exception(F'''You cannot use ``__delitem__`` on a {self.__class__.__name__} instance.'''					)






     def   lowerCamelCase__  (						self	:       Tuple       ,					*UpperCamelCase_	:       Dict       ,					**UpperCamelCase_	:       Any					):
           raise Exception(F'''You cannot use ``setdefault`` on a {self.__class__.__name__} instance.'''					)






     def   lowerCamelCase__  (						self	:       Any       ,					*UpperCamelCase_	:       List[str]       ,					**UpperCamelCase_	:       List[Any]					):
           raise Exception(F'''You cannot use ``pop`` on a {self.__class__.__name__} instance.'''					)






     def   lowerCamelCase__  (						self	:       str       ,					*UpperCamelCase_	:       List[str]       ,					**UpperCamelCase_	:       Dict					):
           raise Exception(F'''You cannot use ``update`` on a {self.__class__.__name__} instance.'''					)






     def __getitem__(						self	:       int       ,					UpperCamelCase_	:       Union[str, Any]					):
           if isinstance(UpperCamelCase_       ,					UpperCamelCase_					):
                 lowerCAmelCase       :						Dict							= dict(self.items()					)
                 return inner_dict[k]
           else:
                 return self.to_tuple()[k]






     def __setattr__(						self	:       Dict       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       List[Any]					):
           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	:       Dict       ,					UpperCamelCase_	:       Optional[int]       ,					UpperCamelCase_	:       Any					):
           # Will raise a KeyException if needed
           super().__setitem__(UpperCamelCase_       ,					UpperCamelCase_					)
           # Don't call self.__setattr__ to avoid recursion errors
           super().__setattr__(UpperCamelCase_       ,					UpperCamelCase_					)






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






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


snake_case__		:  Optional[Any]     					=					{
    '''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'''},
}

snake_case__		:  Any     					=					[
    '''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   _snake_case    (    ):
      lowerCAmelCase       :						int							= 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'''      )

      lowerCAmelCase       :						Dict							= parser.parse_args()

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

      return args


def   _snake_case    (    ):
      transformers.utils.logging.set_verbosity_error()
      lowerCAmelCase       :						Union[str, Any]							= parse_args()

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

      lowerCAmelCase       :						str							= start_measure()
      lowerCAmelCase       :						List[str]							= model_cls.from_pretrained(args.model_name			, device_map='''auto'''			, **_snake_case      )
      lowerCAmelCase       :						Tuple							= end_measure(_snake_case      )
      log_measures(_snake_case			, '''Model loading'''      )

      lowerCAmelCase       :						str							= compute_module_sizes(_snake_case      )
      lowerCAmelCase       :						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]
      lowerCAmelCase       :						int							= '''\n'''.join([f'''- {device}: {size // 2**20}MiB''' for device, size in device_size.items()]      )
      print(f'''\nTheoretical use:\n{message}'''      )

      lowerCAmelCase       :						int							= AutoTokenizer.from_pretrained(args.tokenizer_name      )

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

            lowerCAmelCase       :						Tuple							= 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'''      )

      lowerCAmelCase       :						List[Any]							= end_measure(_snake_case      )
      log_measures(_snake_case			, '''Model generation'''      )

      lowerCAmelCase       :						str							= [gen / tok for gen, tok in zip(_snake_case			, _snake_case      )]
      lowerCAmelCase       :						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'''      )
      lowerCAmelCase       :						List[Any]							= 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()


