





"""simple docstring"""






import logging
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union

import librosa
import torch
from datasets import DatasetDict, load_dataset
from packaging import version
from torch import nn

from transformers import (
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaForPreTraining,
    is_apex_available,
    trainer_utils,
)
from transformers.models.wavaveca.modeling_wavaveca import _compute_mask_indices


if is_apex_available():
		from apex import amp

if version.parse(version.parse(torch.__version__).base_version) >= version.parse('''1.6'''):
		A	     =					True
		from torch.cuda.amp import autocast


A	     =					logging.getLogger(__name__)




@dataclass
class 			__lowercase     :
					__lowerCAmelCase					=					field(
					    metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models'''}							)
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co'''}				, )
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''Whether to freeze the feature extractor layers of the model.'''}							)
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''Whether to log verbose messages or not.'''}				, )
					__lowerCAmelCase					=					field(
					    default=2.0				, metadata={'''help''': '''Maximum temperature for gumbel softmax.'''}							)
					__lowerCAmelCase					=					field(
					    default=0.5				, metadata={'''help''': '''Minimum temperature for gumbel softmax.'''}							)
					__lowerCAmelCase					=					field(
					    default=0.99_9995				, metadata={'''help''': '''Decay of gumbel temperature during training.'''}							)






def       __A (     a_  :Any				,    a_  :Any)						->						Union[str, Any]:
				logging.basicConfig(
				    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s'''				,    datefmt='''%m/%d/%Y %H:%M:%S'''				,    handlers=[logging.StreamHandler(sys.stdout)]				,    )
				__a	:  Any         =					logging.WARNING
				if model_args.verbose_logging:
								__a	:  Optional[Any]         =					logging.DEBUG
				elif trainer_utils.is_main_process(training_args.local_rank):
								__a	:  List[Any]         =					logging.INFO
				logger.setLevel(a__)




@dataclass
class 			__lowercase     :
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''The name of the dataset to use (via the datasets library).'''}							)
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''The configuration name of the dataset to use (via the datasets library).'''}							)
					__lowerCAmelCase					=					field(
					    default='''train'''				, metadata={
					        '''help''': '''The name of the training data set split to use (via the datasets library). Defaults to \'train\''''
					    }				, )
					__lowerCAmelCase					=					field(
					    default='''validation'''				, metadata={
					        '''help''': (
					            '''The name of the validation data set split to use (via the datasets library). Defaults to \'validation\''''
					        )
					    }				, )
					__lowerCAmelCase					=					field(
					    default='''file'''				, metadata={'''help''': '''Column in the dataset that contains speech file path. Defaults to \'file\''''}				, )
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''Overwrite the cached preprocessed datasets or not.'''}							)
					__lowerCAmelCase					=					field(
					    default=1				, metadata={
					        '''help''': '''The percentage of the train set used as validation set in case there\'s no validation split'''
					    }				, )
					__lowerCAmelCase					=					field(
					    default=_UpperCamelCase				, metadata={'''help''': '''The number of processes to use for the preprocessing.'''}				, )
					__lowerCAmelCase					=					field(
					    default=20.0				, metadata={'''help''': '''Filter audio files that are longer than `max_duration_in_seconds` seconds'''}							)






@dataclass
class 			__lowercase     :
					__lowerCAmelCase					=					42
					__lowerCAmelCase					=					42
					__lowerCAmelCase					=					'''longest'''
					__lowerCAmelCase					=					None
					__lowerCAmelCase					=					None







					def __call__(  self		,     _UpperCAmelCase       ):
									__a	:  List[Any]         =					self.feature_extractor.pad(
									    _snake_case		,     max_length=self.max_length		,     padding=self.padding		,     pad_to_multiple_of=self.pad_to_multiple_of		,     return_tensors='''pt'''		,     )
									__a	:  List[Any]         =					self.model._get_feat_extract_output_lengths(batch['''input_values'''].shape[-1]       )

									__a	:  Dict         =					batch['''input_values'''].shape[0]

									# make sure that no loss is computed on padded inputs
									if batch["attention_mask"] is not None:
													# compute real output lengths according to convolution formula
													__a	:  int         =					self.model._get_feat_extract_output_lengths(batch['''attention_mask'''].sum(-1       )       ).to(
													    torch.long       )

													__a	:  Dict         =					torch.zeros(
													    (batch_size, mask_indices_seq_length)		,     dtype=torch.long		,     device=batch['''input_values'''].device       )

													# these two operations makes sure that all values
													# before the output lengths indices are attended to
													__a	:  Dict         =					1
													__a	:  List[str]         =					attention_mask.flip([-1]       ).cumsum(-1       ).flip([-1]       ).bool()

									# sample randomly masked indices
									__a	:  List[str]         =					_compute_mask_indices(
									    (batch_size, mask_indices_seq_length)		,     self.model.config.mask_time_prob		,     self.model.config.mask_time_length		,     attention_mask=_snake_case		,     min_masks=2		,     )

									return batch






class 			__lowercase     (     _UpperCamelCase							):







					def __init__(  self		,     *_UpperCAmelCase		,     _UpperCAmelCase=1		,     _UpperCAmelCase=0		,     _UpperCAmelCase=1.0		,     **_UpperCAmelCase       ):
									super().__init__(*_snake_case		,     **_snake_case       )
									__a	:  Union[str, Any]         =					0
									__a	:  int         =					max_gumbel_temp
									__a	:  Optional[Any]         =					min_gumbel_temp
									__a	:  Tuple         =					gumbel_temp_decay







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase		,     _UpperCAmelCase       ):
									model.train()
									__a	:  int         =					self._prepare_inputs(_snake_case       )

									if self.use_amp:
													with autocast():
																	__a	:  Dict         =					self.compute_loss(_snake_case		,     _snake_case       )
									else:
													__a	:  Any         =					self.compute_loss(_snake_case		,     _snake_case       )

									if self.args.n_gpu > 1 or self.deepspeed:
													if model.module.config.ctc_loss_reduction == "mean":
																	__a	:  Tuple         =					loss.mean()
													elif model.module.config.ctc_loss_reduction == "sum":
																	__a	:  Dict         =					loss.sum() / (inputs['''mask_time_indices''']).sum()
													else:
																	raise ValueError(f"""{model.config.ctc_loss_reduction} is not valid. Choose one of ['mean', 'sum']"""       )

									if self.args.gradient_accumulation_steps > 1:
													__a	:  List[str]         =					loss / self.args.gradient_accumulation_steps

									if self.use_amp:
													self.scaler.scale(_snake_case       ).backward()
									elif self.use_apex:
													with amp.scale_loss(_snake_case		,     self.optimizer       ) as scaled_loss:
																	scaled_loss.backward()
									elif self.deepspeed:
													self.deepspeed.backward(_snake_case       )
									else:
													loss.backward()

									self.num_update_step += 1
									# make sure gumbel softmax temperature is decayed
									if self.args.n_gpu > 1 or self.deepspeed:
													model.module.set_gumbel_temperature(
													    max(self.max_gumbel_temp * self.gumbel_temp_decay**self.num_update_step		,     self.min_gumbel_temp       )       )
									else:
													model.set_gumbel_temperature(
													    max(self.max_gumbel_temp * self.gumbel_temp_decay**self.num_update_step		,     self.min_gumbel_temp       )       )

									return loss.detach()






def       __A (     )						->						Union[str, Any]:
				# See all possible arguments in src/transformers/training_args.py
				# or by passing the --help flag to this script.
				# We now keep distinct sets of args, for a cleaner separation of concerns.

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

				__a    ,  __a    ,  __a	:  List[Any]         =					parser.parse_args_into_dataclasses()
				configure_logger(a__				,    a__)

				# Downloading and loading a dataset from the hub.
				__a	:  List[str]         =					load_dataset(data_args.dataset_name				,    data_args.dataset_config_name				,    cache_dir=model_args.cache_dir)

				if "validation" not in datasets.keys():
								# make sure only "validation" and "train" keys remain"
								__a	:  Tuple         =					DatasetDict()
								__a	:  Optional[Any]         =					load_dataset(
								    data_args.dataset_name				,    data_args.dataset_config_name				,    split=F"""{data_args.train_split_name}[:{data_args.validation_split_percentage}%]"""				,    cache_dir=model_args.cache_dir				,    )
								__a	:  Optional[int]         =					load_dataset(
								    data_args.dataset_name				,    data_args.dataset_config_name				,    split=F"""{data_args.train_split_name}[{data_args.validation_split_percentage}%:]"""				,    cache_dir=model_args.cache_dir				,    )
				else:
								# make sure only "validation" and "train" keys remain"
								__a	:  Optional[int]         =					DatasetDict()
								__a	:  Optional[Any]         =					load_dataset(
								    data_args.dataset_name				,    data_args.dataset_config_name				,    split='''validation'''				,    cache_dir=model_args.cache_dir				,    )
								__a	:  int         =					load_dataset(
								    data_args.dataset_name				,    data_args.dataset_config_name				,    split=F"""{data_args.train_split_name}"""				,    cache_dir=model_args.cache_dir				,    )

				# only normalized-inputs-training is supported
				__a	:  Dict         =					WavaVecaFeatureExtractor.from_pretrained(
				    model_args.model_name_or_path				,    cache_dir=model_args.cache_dir				,    do_normalize=a__)

				def prepare_dataset(a_  :Tuple):
								# check that all files have the correct sampling rate
								__a    ,  __a	:  List[str]         =					librosa.load(batch[data_args.speech_file_column]				,    sr=feature_extractor.sampling_rate)
								return batch

				# load audio files into numpy arrays
				__a	:  List[Any]         =					datasets.map(
				    a__				,    num_proc=data_args.preprocessing_num_workers				,    remove_columns=datasets['''train'''].column_names)

				# filter audio files that are too long
				__a	:  Optional[int]         =					vectorized_datasets.filter(
				    lambda a_: len(data['''speech''']) < int(data_args.max_duration_in_seconds * feature_extractor.sampling_rate))

				def normalize(a_  :List[Any]):
								return feature_extractor(batch['''speech''']				,    sampling_rate=feature_extractor.sampling_rate)

				# normalize and transform to `BatchFeatures`
				__a	:  Tuple         =					vectorized_datasets.map(
				    a__				,    batched=a__				,    num_proc=data_args.preprocessing_num_workers				,    load_from_cache_file=not data_args.overwrite_cache				,    remove_columns=vectorized_datasets['''train'''].column_names				,    )

				# pretraining is only supported for "newer" stable layer norm architecture
				# apply_spec_augment has to be True, mask_feature_prob has to be 0.0
				__a	:  int         =					WavaVecaConfig.from_pretrained(
				    model_args.model_name_or_path				,    cache_dir=model_args.cache_dir				,    gradient_checkpointing=training_args.gradient_checkpointing				,    )

				if not config.do_stable_layer_norm or config.feat_extract_norm != "layer":
								raise ValueError(
								    '''PreTraining is only supported for ``config.do_stable_layer_norm=True`` and'''
								    ''' ``config.feat_extract_norm=\'layer\'''')

				__a	:  Any         =					WavaVecaForPreTraining(a__)

				__a	:  Dict         =					DataCollatorForWavaVecaPretraining(model=a__				,    feature_extractor=a__)

				__a	:  Optional[int]         =					WavaVecaPreTrainer(
				    model=a__				,    data_collator=a__				,    args=a__				,    train_dataset=vectorized_datasets['''train''']				,    eval_dataset=vectorized_datasets['''validation''']				,    tokenizer=a__				,    max_gumbel_temp=model_args.max_gumbel_temperature				,    min_gumbel_temp=model_args.min_gumbel_temperature				,    gumbel_temp_decay=model_args.gumbel_temperature_decay				,    )
				trainer.train()


if __name__ == "__main__":
		main()






"""simple docstring"""






def       __A (     a_  :float)						->						float:

				if edge <= 0 or not isinstance(a_				,    a_):
								raise ValueError('''Length must be a positive.''')
				return 3 * ((25 + 10 * (5 ** (1 / 2))) ** (1 / 2)) * (edge**2)

def       __A (     a_  :float)						->						float:

				if edge <= 0 or not isinstance(a_				,    a_):
								raise ValueError('''Length must be a positive.''')
				return ((15 + (7 * (5 ** (1 / 2)))) / 4) * (edge**3)


if __name__ == "__main__":
		import doctest

		doctest.testmod()