



'''simple docstring'''
from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


snake_case_				:			List[str]						 =						logging.get_logger(__name__)






class 							lowercase__			(							lowercase     ):
	lowercase__						       =    ["""input_values""", """attention_mask"""]





	def __init__(				self      :       Optional[int]				,lowerCamelCase__      :       int=1				,lowerCamelCase__      :       Optional[int]=16000				,lowerCamelCase__      :       Dict=128				,lowerCamelCase__      :       Dict=1024				,lowerCamelCase__      :       List[str]=0.0				,lowerCamelCase__      :       Optional[Any]=True				,lowerCamelCase__      :       Optional[int]=-4.2_6_7_7_3_9_3				,lowerCamelCase__      :       Union[str, Any]=4.5_6_8_9_9_7_4				,lowerCamelCase__      :       List[str]=False				,**lowerCamelCase__      :       Optional[int]				,):







							'''simple docstring'''
							super().__init__(feature_size=lowerCamelCase__				,sampling_rate=lowerCamelCase__				,padding_value=lowerCamelCase__				,**lowerCamelCase__				)
							_UpperCamelCase      :	Optional[int]				       =     num_mel_bins
							_UpperCamelCase      :	Any				       =     max_length
							_UpperCamelCase      :	Tuple				       =     do_normalize
							_UpperCamelCase      :	Tuple				       =     mean
							_UpperCamelCase      :	str				       =     std
							_UpperCamelCase      :	List[Any]				       =     return_attention_mask





	def 			UpperCamelCase_						(				self      :       Tuple				,lowerCamelCase__      :       np.ndarray				,lowerCamelCase__      :       int				,):







							'''simple docstring'''
							_UpperCamelCase      :	Union[str, Any]				       =     torch.from_numpy(lowerCamelCase__				).unsqueeze(0				)
							_UpperCamelCase      :	Dict				       =     ta_kaldi.fbank(
							    lowerCamelCase__				,htk_compat=lowerCamelCase__				,sample_frequency=self.sampling_rate				,use_energy=lowerCamelCase__				,window_type='hanning'				,num_mel_bins=self.num_mel_bins				,dither=0.0				,frame_shift=10				,)

							_UpperCamelCase      :	List[Any]				       =     fbank.shape[0]
							_UpperCamelCase      :	Optional[int]				       =     max_length - n_frames

							# pad or truncate, depending on difference
							if difference > 0:
													_UpperCamelCase      :	Tuple				       =     torch.nn.ZeroPadad((0, 0, 0, difference)				)
													_UpperCamelCase      :	Any				       =     pad_module(lowerCamelCase__				)
							elif difference < 0:
													_UpperCamelCase      :	List[str]				       =     fbank[0:max_length, :]

							_UpperCamelCase      :	Optional[int]				       =     fbank.numpy()

							return fbank





	def 			UpperCamelCase_						(				self      :       Tuple				,lowerCamelCase__      :       np.ndarray				):







							'''simple docstring'''
							return (input_values - (self.mean)) / (self.std * 2)





	def __call__(				self      :       List[str]				,lowerCamelCase__      :       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]				,lowerCamelCase__      :       Optional[int] = None				,lowerCamelCase__      :       Optional[Union[str, TensorType]] = None				,**lowerCamelCase__      :       Dict				,):







							'''simple docstring'''

							if sampling_rate is not None:
													if sampling_rate != self.sampling_rate:
																			raise ValueError(
																			    F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
																			    F' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
																			    F' {self.sampling_rate} and not {sampling_rate}.'				)
							else:
													logger.warning(
													    'It is strongly recommended to pass the `sampling_rate` argument to this function. '
													    'Failing to do so can result in silent errors that might be hard to debug.'				)

							_UpperCamelCase      :	Dict				       =     isinstance(lowerCamelCase__				,np.ndarray				) and len(raw_speech.shape				) > 1
							if is_batched_numpy and len(raw_speech.shape				) > 2:
													raise ValueError(F'Only mono-channel audio is supported for input to {self}'				)
							_UpperCamelCase      :	int				       =     is_batched_numpy or (
							    isinstance(lowerCamelCase__				,(list, tuple)				) and (isinstance(raw_speech[0]				,(np.ndarray, tuple, list)				))
							)

							if is_batched:
													_UpperCamelCase      :	List[Any]				       =     [np.asarray(lowerCamelCase__				,dtype=np.floataa				) for speech in raw_speech]
							elif not is_batched and not isinstance(lowerCamelCase__				,np.ndarray				):
													_UpperCamelCase      :	Tuple				       =     np.asarray(lowerCamelCase__				,dtype=np.floataa				)
							elif isinstance(lowerCamelCase__				,np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
													_UpperCamelCase      :	List[str]				       =     raw_speech.astype(np.floataa				)

							# always return batch
							if not is_batched:
													_UpperCamelCase      :	Dict				       =     [raw_speech]

							# extract fbank features and pad/truncate to max_length
							_UpperCamelCase      :	Optional[Any]				       =     [self._extract_fbank_features(lowerCamelCase__				,max_length=self.max_length				) for waveform in raw_speech]

							# convert into BatchFeature
							_UpperCamelCase      :	Dict				       =     BatchFeature({'input_values': features}				)

							# make sure list is in array format
							_UpperCamelCase      :	Any				       =     padded_inputs.get('input_values'				)
							if isinstance(input_values[0]				,lowerCamelCase__				):
													_UpperCamelCase      :	List[Any]				       =     [np.asarray(lowerCamelCase__				,dtype=np.floataa				) for feature in input_values]

							# normalization
							if self.do_normalize:
													_UpperCamelCase      :	List[Any]				       =     [self.normalize(lowerCamelCase__				) for feature in input_values]

							if return_tensors is not None:
													_UpperCamelCase      :	Union[str, Any]				       =     padded_inputs.convert_to_tensors(lowerCamelCase__				)

							return padded_inputs







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

						_UpperCamelCase      :	Dict				       =     parser.parse_args()

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

						return args






def A__					(  ):
						transformers.utils.logging.set_verbosity_error()
						_UpperCamelCase      :	Optional[Any]				       =     parse_args()

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

						_UpperCamelCase      :	Dict				       =     start_measure()
						_UpperCamelCase      :	Optional[int]				       =     model_cls.from_pretrained(args.model_name						,						device_map='auto'						,						**UpperCAmelCase_							)
						_UpperCamelCase      :	List[str]				       =     end_measure(UpperCAmelCase_							)
						log_measures(UpperCAmelCase_						,						'Model loading'							)

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

						_UpperCamelCase      :	List[str]				       =     AutoTokenizer.from_pretrained(args.tokenizer_name							)

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

												_UpperCamelCase      :	Optional[int]				       =     tokenizer.decode(UpperCAmelCase_						,						skip_special_tokens=UpperCAmelCase_							)
												texts_outs.append(UpperCAmelCase_							)
												generation_times.append(UpperCAmelCase_							)
												gen_tokens.append(UpperCAmelCase_							)
												print(f'Prompt: {prompt}\nGeneration {text_out}\nIn {generation_time:.2f}s for {num_gen_tokens} tokens\n'							)

						_UpperCamelCase      :	Union[str, Any]				       =     end_measure(UpperCAmelCase_							)
						log_measures(UpperCAmelCase_						,						'Model generation'							)

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


if __name__ == "__main__":
					main()


