

"""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						:   Union[str, 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'''},
}

__UpperCamelCase						:   int							   =	[
    '''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 						__SCREAMING_SNAKE_CASE						(   ):
	lowerCAmelCase__			:		int						       =			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'''				)

	lowerCAmelCase__			:		int						       =			parser.parse_args()

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

	return args

def 						__SCREAMING_SNAKE_CASE						(   ):
	transformers.utils.logging.set_verbosity_error()
	lowerCAmelCase__			:		Dict						       =			parse_args()

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

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

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

	lowerCAmelCase__			:		Tuple						       =			AutoTokenizer.from_pretrained(args.tokenizer_name				)

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

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

	lowerCAmelCase__			:		Optional[int]						       =			end_measure(A_				)
	log_measures(A_		,  '''Model generation'''				)

	lowerCAmelCase__			:		Union[str, Any]						       =			[gen / tok for gen, tok in zip(A_		,  A_				)]
	lowerCAmelCase__			:		Tuple						       =			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'				)
	lowerCAmelCase__			:		str						       =			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 unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
							import torch

if is_vision_available():
							from PIL import Image

							from transformers import SwinaSRImageProcessor
							from transformers.image_transforms import get_image_size







class        SCREAMING_SNAKE_CASE      (	unittest.TestCase ):






					"""simple docstring"""



					def __init__( self    :       Dict     ,lowercase_    :       List[str]     ,lowercase_    :       List[str]=7     ,lowercase_    :       List[Any]=3     ,lowercase_    :       str=1_8     ,lowercase_    :       Any=3_0     ,lowercase_    :       int=4_0_0     ,lowercase_    :       Tuple=True     ,lowercase_    :       Optional[int]=1 / 2_5_5     ,lowercase_    :       int=True     ,lowercase_    :       Union[str, Any]=8     ,):
						lowerCAmelCase__			:		Tuple						       =			parent
						lowerCAmelCase__			:		Optional[Any]						       =			batch_size
						lowerCAmelCase__			:		int						       =			num_channels
						lowerCAmelCase__			:		Union[str, Any]						       =			image_size
						lowerCAmelCase__			:		Optional[Any]						       =			min_resolution
						lowerCAmelCase__			:		Union[str, Any]						       =			max_resolution
						lowerCAmelCase__			:		int						       =			do_rescale
						lowerCAmelCase__			:		Tuple						       =			rescale_factor
						lowerCAmelCase__			:		Dict						       =			do_pad
						lowerCAmelCase__			:		str						       =			pad_size







					def  __lowerCAmelCase				( self    :       int	):
						return {
						    "do_rescale": self.do_rescale,
						    "rescale_factor": self.rescale_factor,
						    "do_pad": self.do_pad,
						    "pad_size": self.pad_size,
						}







					def  __lowerCAmelCase				( self    :       Union[str, Any]     ,lowercase_    :       List[str]=False     ,lowercase_    :       Any=False     ,lowercase_    :       Dict=False	):
						assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

						if equal_resolution:
							lowerCAmelCase__			:		int						       =			[]
							for i in range(self.batch_size	):
								image_inputs.append(
								    np.random.randint(
								        2_5_5     ,size=(self.num_channels, self.max_resolution, self.max_resolution)     ,dtype=np.uinta	)	)
						else:
							lowerCAmelCase__			:		str						       =			[]
							for i in range(self.batch_size	):
								lowerCAmelCase__       ,lowerCAmelCase__			:		Any						       =			np.random.choice(np.arange(self.min_resolution     ,self.max_resolution	)     ,2	)
								image_inputs.append(np.random.randint(2_5_5     ,size=(self.num_channels, width, height)     ,dtype=np.uinta	)	)

						if not numpify and not torchify:
							# PIL expects the channel dimension as last dimension
							lowerCAmelCase__			:		Optional[int]						       =			[Image.fromarray(np.moveaxis(lowercase_     ,0     ,-1	)	) for x in image_inputs]

						if torchify:
							lowerCAmelCase__			:		int						       =			[torch.from_numpy(lowercase_	) for x in image_inputs]

						return image_inputs







@require_torch
@require_vision
class        SCREAMING_SNAKE_CASE      (	a_					,							unittest.TestCase ):






					"""simple docstring"""



					lowercase__       				= SwinaSRImageProcessor if is_vision_available() else None







					def  __lowerCAmelCase				( self    :       Dict	):
						lowerCAmelCase__			:		Dict						       =			SwinaSRImageProcessingTester(self	)







					@property
					def  __lowerCAmelCase				( self    :       str	):
						return self.image_processor_tester.prepare_image_processor_dict()







					def  __lowerCAmelCase				( self    :       List[str]	):
						lowerCAmelCase__			:		Any						       =			self.image_processing_class(**self.image_processor_dict	)
						self.assertTrue(hasattr(lowercase_     ,'''do_rescale'''	)	)
						self.assertTrue(hasattr(lowercase_     ,'''rescale_factor'''	)	)
						self.assertTrue(hasattr(lowercase_     ,'''do_pad'''	)	)
						self.assertTrue(hasattr(lowercase_     ,'''pad_size'''	)	)







					def  __lowerCAmelCase				( self    :       Dict	):
						pass







					def  __lowerCAmelCase				( self    :       List[str]     ,lowercase_    :       Optional[Any]	):
						lowerCAmelCase__       ,lowerCAmelCase__			:		Tuple						       =			get_image_size(lowercase_	)
						lowerCAmelCase__			:		Dict						       =			self.image_processor_tester.pad_size

						lowerCAmelCase__			:		List[Any]						       =			(old_height // size + 1) * size - old_height
						lowerCAmelCase__			:		Union[str, Any]						       =			(old_width // size + 1) * size - old_width
						return old_height + pad_height, old_width + pad_width







					def  __lowerCAmelCase				( self    :       List[Any]	):
						# Initialize image_processor
						lowerCAmelCase__			:		str						       =			self.image_processing_class(**self.image_processor_dict	)
						# create random PIL images
						lowerCAmelCase__			:		int						       =			self.image_processor_tester.prepare_inputs(equal_resolution=lowercase_	)
						for image in image_inputs:
							self.assertIsInstance(lowercase_     ,Image.Image	)

						# Test not batched input
						lowerCAmelCase__			:		List[str]						       =			image_processor(image_inputs[0]     ,return_tensors='''pt'''	).pixel_values
						lowerCAmelCase__       ,lowerCAmelCase__			:		List[Any]						       =			self.calculate_expected_size(np.array(image_inputs[0]	)	)
						self.assertEqual(
						    encoded_images.shape     ,(
						        1,
						        self.image_processor_tester.num_channels,
						        expected_height,
						        expected_width,
						    )     ,)







					def  __lowerCAmelCase				( self    :       int	):
						# Initialize image_processor
						lowerCAmelCase__			:		str						       =			self.image_processing_class(**self.image_processor_dict	)
						# create random numpy tensors
						lowerCAmelCase__			:		Tuple						       =			self.image_processor_tester.prepare_inputs(equal_resolution=lowercase_     ,numpify=lowercase_	)
						for image in image_inputs:
							self.assertIsInstance(lowercase_     ,np.ndarray	)

						# Test not batched input
						lowerCAmelCase__			:		Optional[Any]						       =			image_processor(image_inputs[0]     ,return_tensors='''pt'''	).pixel_values
						lowerCAmelCase__       ,lowerCAmelCase__			:		List[Any]						       =			self.calculate_expected_size(image_inputs[0]	)
						self.assertEqual(
						    encoded_images.shape     ,(
						        1,
						        self.image_processor_tester.num_channels,
						        expected_height,
						        expected_width,
						    )     ,)







					def  __lowerCAmelCase				( self    :       Dict	):
						# Initialize image_processor
						lowerCAmelCase__			:		Tuple						       =			self.image_processing_class(**self.image_processor_dict	)
						# create random PyTorch tensors
						lowerCAmelCase__			:		Union[str, Any]						       =			self.image_processor_tester.prepare_inputs(equal_resolution=lowercase_     ,torchify=lowercase_	)
						for image in image_inputs:
							self.assertIsInstance(lowercase_     ,torch.Tensor	)

						# Test not batched input
						lowerCAmelCase__			:		Tuple						       =			image_processor(image_inputs[0]     ,return_tensors='''pt'''	).pixel_values
						lowerCAmelCase__       ,lowerCAmelCase__			:		int						       =			self.calculate_expected_size(image_inputs[0]	)
						self.assertEqual(
						    encoded_images.shape     ,(
						        1,
						        self.image_processor_tester.num_channels,
						        expected_height,
						        expected_width,
						    )     ,)




