



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


lowerCamelCase__     =	logging.get_logger(__name__)






class 				SCREAMING_SNAKE_CASE     (							lowerCamelCase__   ):
							__lowerCamelCase		:		Any      =['input_values', 'attention_mask']





							def __init__(				self						:						Dict ,			__lowercase						:						List[Any]=1 ,			__lowercase						:						int=16000 ,			__lowercase						:						Optional[Any]=128 ,			__lowercase						:						List[Any]=1024 ,			__lowercase						:						List[Any]=0.0 ,			__lowercase						:						Optional[Any]=True ,			__lowercase						:						Optional[Any]=-4.2677393 ,			__lowercase						:						List[str]=4.5689974 ,			__lowercase						:						Union[str, Any]=False ,			**__lowercase						:						Tuple ,			):

											'''simple docstring'''
											super().__init__(feature_size=__lowercase ,			sampling_rate=__lowercase ,			padding_value=__lowercase ,			**__lowercase		)
											__a					       =						num_mel_bins
											__a					       =						max_length
											__a					       =						do_normalize
											__a					       =						mean
											__a					       =						std
											__a					       =						return_attention_mask





							def   UpperCamelCase_   (				self						:						int ,			__lowercase						:						np.ndarray ,			__lowercase						:						int ,			):

											'''simple docstring'''
											__a					       =						torch.from_numpy(__lowercase		).unsqueeze(0		)
											__a					       =						ta_kaldi.fbank(
											    __lowercase ,			htk_compat=__lowercase ,			sample_frequency=self.sampling_rate ,			use_energy=__lowercase ,			window_type="""hanning""" ,			num_mel_bins=self.num_mel_bins ,			dither=0.0 ,			frame_shift=10 ,			)

											__a					       =						fbank.shape[0]
											__a					       =						max_length - n_frames

											# pad or truncate, depending on difference
											if difference > 0:
															__a					       =						torch.nn.ZeroPadad((0, 0, 0, difference)		)
															__a					       =						pad_module(__lowercase		)
											elif difference < 0:
															__a					       =						fbank[0:max_length, :]

											__a					       =						fbank.numpy()

											return fbank





							def   UpperCamelCase_   (				self						:						Dict ,			__lowercase						:						np.ndarray		):

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







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

											'''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."""		)

											__a					       =						isinstance(__lowercase ,			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}"		)
											__a					       =						is_batched_numpy or (
											    isinstance(__lowercase ,			(list, tuple)		) and (isinstance(raw_speech[0] ,			(np.ndarray, tuple, list)		))
											)

											if is_batched:
															__a					       =						[np.asarray(__lowercase ,			dtype=np.floataa		) for speech in raw_speech]
											elif not is_batched and not isinstance(__lowercase ,			np.ndarray		):
															__a					       =						np.asarray(__lowercase ,			dtype=np.floataa		)
											elif isinstance(__lowercase ,			np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
															__a					       =						raw_speech.astype(np.floataa		)

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

											# extract fbank features and pad/truncate to max_length
											__a					       =						[self._extract_fbank_features(__lowercase ,			max_length=self.max_length		) for waveform in raw_speech]

											# convert into BatchFeature
											__a					       =						BatchFeature({"""input_values""": features}		)

											# make sure list is in array format
											__a					       =						padded_inputs.get("""input_values"""		)
											if isinstance(input_values[0] ,			__lowercase		):
															__a					       =						[np.asarray(__lowercase ,			dtype=np.floataa		) for feature in input_values]

											# normalization
											if self.do_normalize:
															__a					       =						[self.normalize(__lowercase		) for feature in input_values]

											if return_tensors is not None:
															__a					       =						padded_inputs.convert_to_tensors(__lowercase		)

											return padded_inputs





import logging
import os
import sys
from dataclasses import dataclass, field
from typing import Optional

import transformers
from transformers import (
    AutoConfig,
    AutoModelForQuestionAnswering,
    AutoTokenizer,
    DataCollatorWithPadding,
    HfArgumentParser,
    SquadDataset,
    Trainer,
    TrainingArguments,
)
from transformers import SquadDataTrainingArguments as DataTrainingArguments
from transformers.trainer_utils import is_main_process


lowerCamelCase__     =	logging.getLogger(__name__)






@dataclass
class 				SCREAMING_SNAKE_CASE     :
							__lowerCamelCase		:		str      =field(
							    metadata={'help': 'Path to pretrained model or model identifier from huggingface.co/models'}   )
							__lowerCamelCase		:		Optional[str]      =field(
							    default=lowerCamelCase__     , metadata={'help': 'Pretrained config name or path if not the same as model_name'}   )
							__lowerCamelCase		:		Optional[str]      =field(
							    default=lowerCamelCase__     , metadata={'help': 'Pretrained tokenizer name or path if not the same as model_name'}   )
							__lowerCamelCase		:		bool      =field(default=lowerCamelCase__     , metadata={'help': 'Set this flag to use fast tokenization.'}   )
							# If you want to tweak more attributes on your tokenizer, you should do it in a distinct script,
							# or just modify its tokenizer_config.json.
							__lowerCamelCase		:		Optional[str]      =field(
							    default=lowerCamelCase__     , metadata={'help': 'Where do you want to store the pretrained models downloaded from huggingface.co'}     , )



def 					lowerCAmelCase__  (							):
				"""simple docstring"""







				__a					       =						HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )

				if len(sys.argv      ) == 2 and sys.argv[1].endswith(""".json"""      ):
								# If we pass only one argument to the script and it's the path to a json file,
								# let's parse it to get our arguments.
								__a       ,   __a       ,   __a					       =						parser.parse_json_file(json_file=os.path.abspath(sys.argv[1]      )      )
				else:
								__a       ,   __a       ,   __a					       =						parser.parse_args_into_dataclasses()

				if (
				    os.path.exists(training_args.output_dir      )
				    and os.listdir(training_args.output_dir      )
				    and training_args.do_train
				    and not training_args.overwrite_output_dir
				):
								raise ValueError(
								    f"Output directory ({training_args.output_dir}) already exists and is not empty. Use"
								    """ --overwrite_output_dir to overcome."""      )

				# Setup logging
				logging.basicConfig(
				    format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s"""			,       datefmt="""%m/%d/%Y %H:%M:%S"""			,       level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN			,       )
				logger.warning(
				    """Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"""			,       training_args.local_rank			,       training_args.device			,       training_args.n_gpu			,       bool(training_args.local_rank != -1      )			,       training_args.fpaa			,       )
				# Set the verbosity to info of the Transformers logger (on main process only):
				if is_main_process(training_args.local_rank      ):
								transformers.utils.logging.set_verbosity_info()
								transformers.utils.logging.enable_default_handler()
								transformers.utils.logging.enable_explicit_format()
				logger.info("""Training/evaluation parameters %s"""			,       _SCREAMING_SNAKE_CASE      )

				# Prepare Question-Answering task
				# Load pretrained model and tokenizer
				#
				# Distributed training:
				# The .from_pretrained methods guarantee that only one local process can concurrently
				# download model & vocab.

				__a					       =						AutoConfig.from_pretrained(
				    model_args.config_name if model_args.config_name else model_args.model_name_or_path			,       cache_dir=model_args.cache_dir			,       )
				__a					       =						AutoTokenizer.from_pretrained(
				    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path			,       cache_dir=model_args.cache_dir			,       use_fast=_SCREAMING_SNAKE_CASE			,       )
				__a					       =						AutoModelForQuestionAnswering.from_pretrained(
				    model_args.model_name_or_path			,       from_tf=bool(""".ckpt""" in model_args.model_name_or_path      )			,       config=_SCREAMING_SNAKE_CASE			,       cache_dir=model_args.cache_dir			,       )

				# Get datasets
				__a					       =						hasattr(model.config			,       """lang2id"""      )
				__a					       =						(
				    SquadDataset(
				        _SCREAMING_SNAKE_CASE			,       tokenizer=_SCREAMING_SNAKE_CASE			,       is_language_sensitive=_SCREAMING_SNAKE_CASE			,       cache_dir=model_args.cache_dir      )
				    if training_args.do_train
				    else None
				)
				__a					       =						(
				    SquadDataset(
				        _SCREAMING_SNAKE_CASE			,       tokenizer=_SCREAMING_SNAKE_CASE			,       mode="""dev"""			,       is_language_sensitive=_SCREAMING_SNAKE_CASE			,       cache_dir=model_args.cache_dir			,       )
				    if training_args.do_eval
				    else None
				)

				# Data collator
				__a					       =						DataCollatorWithPadding(_SCREAMING_SNAKE_CASE			,       pad_to_multiple_of=8      ) if training_args.fpaa else None

				# Initialize our Trainer
				__a					       =						Trainer(
				    model=_SCREAMING_SNAKE_CASE			,       args=_SCREAMING_SNAKE_CASE			,       train_dataset=_SCREAMING_SNAKE_CASE			,       eval_dataset=_SCREAMING_SNAKE_CASE			,       data_collator=_SCREAMING_SNAKE_CASE			,       )

				# Training
				if training_args.do_train:
								trainer.train(
								    model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path      ) else None      )
								trainer.save_model()
								# For convenience, we also re-save the tokenizer to the same directory,
								# so that you can share your model easily on huggingface.co/models =)
								if trainer.is_world_master():
												tokenizer.save_pretrained(training_args.output_dir      )

def 					lowerCAmelCase__  (							_SCREAMING_SNAKE_CASE    :     Optional[Any]      ):
				"""simple docstring"""







				main()


if __name__ == "__main__":
						main()
