



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

		__SCREAMING_SNAKE_CASE	:		Any   	=		parser.parse_args()

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

		return args


def     lowerCAmelCase_       (     ):
		transformers.utils.logging.set_verbosity_error()
		__SCREAMING_SNAKE_CASE	:		Tuple   	=		parse_args()

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

		__SCREAMING_SNAKE_CASE	:		Dict   	=		start_measure()
		__SCREAMING_SNAKE_CASE	:		Optional[int]   	=		model_cls.from_pretrained(args.model_name	,				device_map="""auto"""	,				**_A )
		__SCREAMING_SNAKE_CASE	:		Dict   	=		end_measure(_A )
		log_measures(_A	,				"""Model loading""" )

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

		__SCREAMING_SNAKE_CASE	:		Optional[Any]   	=		AutoTokenizer.from_pretrained(args.tokenizer_name )

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

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

		__SCREAMING_SNAKE_CASE	:		List[str]   	=		end_measure(_A )
		log_measures(_A	,				"""Model generation""" )

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


if __name__ == "__main__":
						main()




'''simple docstring'''




import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin






class 						_UpperCamelCase  (							lowerCamelCase__   ):


			'''simple docstring'''

			_A				:		int         =		['''image_processor''', '''tokenizer''']
			_A				:		Union[str, Any]         =		'''AutoImageProcessor'''
			_A				:		Optional[Any]         =		'''AutoTokenizer'''

			def __init__(							self					:			Tuple							,					lowerCAmelCase__					:			Tuple=None							,					lowerCAmelCase__					:			Union[str, Any]=None							,					**lowerCAmelCase__					:			Any						):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE	:		Tuple   	=		None
					if "feature_extractor" in kwargs:
							warnings.warn(
							    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
							    """ instead."""							,					lowerCAmelCase__							,					)
							__SCREAMING_SNAKE_CASE	:		str   	=		kwargs.pop("""feature_extractor"""						)

					__SCREAMING_SNAKE_CASE	:		Any   	=		image_processor if image_processor is not None else feature_extractor
					if image_processor is None:
							raise ValueError("""You need to specify an `image_processor`."""						)
					if tokenizer is None:
							raise ValueError("""You need to specify a `tokenizer`."""						)

					super().__init__(lowerCAmelCase__							,					lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		List[str]   	=		self.image_processor
					__SCREAMING_SNAKE_CASE	:		Dict   	=		False

			def __call__(							self					:			List[str]							,					*lowerCAmelCase__					:			List[str]							,					**lowerCAmelCase__					:			Dict						):





					"""simple docstring"""






					if self._in_target_context_manager:
							return self.current_processor(*lowerCAmelCase__							,					**lowerCAmelCase__						)

					__SCREAMING_SNAKE_CASE	:		Optional[Any]   	=		kwargs.pop("""images"""							,					lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		Union[str, Any]   	=		kwargs.pop("""text"""							,					lowerCAmelCase__						)
					if len(lowerCAmelCase__						) > 0:
							__SCREAMING_SNAKE_CASE	:		Tuple   	=		args[0]
							__SCREAMING_SNAKE_CASE	:		int   	=		args[1:]

					if images is None and text is None:
							raise ValueError("""You need to specify either an `images` or `text` input to process."""						)

					if images is not None:
							__SCREAMING_SNAKE_CASE	:		List[str]   	=		self.image_processor(lowerCAmelCase__							,					*lowerCAmelCase__							,					**lowerCAmelCase__						)
					if text is not None:
							__SCREAMING_SNAKE_CASE	:		Optional[int]   	=		self.tokenizer(lowerCAmelCase__							,					**lowerCAmelCase__						)

					if text is None:
							return inputs
					elif images is None:
							return encodings
					else:
							__SCREAMING_SNAKE_CASE	:		int   	=		encodings["""input_ids"""]
							return inputs

			def UpperCamelCase__ (							self					:			Any							,					*lowerCAmelCase__					:			Any							,					**lowerCAmelCase__					:			List[str]						):





					"""simple docstring"""






					return self.tokenizer.batch_decode(*lowerCAmelCase__							,					**lowerCAmelCase__						)

			def UpperCamelCase__ (							self					:			int							,					*lowerCAmelCase__					:			str							,					**lowerCAmelCase__					:			Dict						):





					"""simple docstring"""






					return self.tokenizer.decode(*lowerCAmelCase__							,					**lowerCAmelCase__						)

			@contextmanager
			def UpperCamelCase__ (							self					:			Union[str, Any]						):





					"""simple docstring"""






					warnings.warn(
					    """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
					    """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
					    """your images inputs, or in a separate call."""						)
					__SCREAMING_SNAKE_CASE	:		List[Any]   	=		True
					__SCREAMING_SNAKE_CASE	:		Optional[Any]   	=		self.tokenizer
					yield
					__SCREAMING_SNAKE_CASE	:		Union[str, Any]   	=		self.image_processor
					__SCREAMING_SNAKE_CASE	:		List[str]   	=		False

			@property
			def UpperCamelCase__ (							self					:			Optional[Any]						):





					"""simple docstring"""






					warnings.warn(
					    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""							,					lowerCAmelCase__							,					)
					return self.image_processor_class



			@property
			def UpperCamelCase__ (							self					:			List[Any]						):





					"""simple docstring"""






					warnings.warn(
					    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""							,					lowerCAmelCase__							,					)
					return self.image_processor