


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"):
			_snake_case     :		Tuple						=				True
			from torch.cuda.amp import autocast


_snake_case     :		int						=				logging.getLogger(__name__)

@dataclass
class 				a		:




							"""simple docstring"""


							__UpperCAmelCase    :							str       =  field(
							    metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"}      )
							__UpperCAmelCase    :							Optional[str]       =  field(
							    default=lowercase_     ,				metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}     ,				)
							__UpperCAmelCase    :							Optional[bool]       =  field(
							    default=lowercase_     ,				metadata={"help": "Whether to freeze the feature extractor layers of the model."}      )
							__UpperCAmelCase    :							Optional[bool]       =  field(
							    default=lowercase_     ,				metadata={"help": "Whether to log verbose messages or not."}     ,				)
							__UpperCAmelCase    :							Optional[float]       =  field(
							    default=2.0     ,				metadata={"help": "Maximum temperature for gumbel softmax."}      )
							__UpperCAmelCase    :							Optional[float]       =  field(
							    default=0.5     ,				metadata={"help": "Minimum temperature for gumbel softmax."}      )
							__UpperCAmelCase    :							Optional[float]       =  field(
							    default=0.999995     ,				metadata={"help": "Decay of gumbel temperature during training."}      )







def 				lowerCAmelCase_     (					__lowerCamelCase							,	__lowerCamelCase	):
							logging.basicConfig(
							    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"							,	datefmt="%m/%d/%Y %H:%M:%S"							,	handlers=[logging.StreamHandler(sys.stdout	)]							,	)
							__snake_case		:			Optional[Any]							 =   logging.WARNING
							if model_args.verbose_logging:
														__snake_case		:			List[Any]							 =   logging.DEBUG
							elif trainer_utils.is_main_process(training_args.local_rank	):
														__snake_case		:			Optional[Any]							 =   logging.INFO
							logger.setLevel(UpperCamelCase__	)

@dataclass
class 				a		:




							"""simple docstring"""


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

@dataclass
class 				a		:




							"""simple docstring"""


							__UpperCAmelCase    :							WavaVecaForPreTraining
							__UpperCAmelCase    :							WavaVecaFeatureExtractor
							__UpperCAmelCase    :							Union[bool, str]       =  "longest"
							__UpperCAmelCase    :							Optional[int]       =  None
							__UpperCAmelCase    :							Optional[int]       =  None

							def __call__(							self       :  str    ,	lowerCamelCase       :  str				)	->    Union[str, Any]:
														# reformat list to dict and set to pytorch format
														__snake_case		:			Union[str, Any]							 =   self.feature_extractor.pad(
														    lowerCamelCase    ,	max_length=self.max_length    ,	padding=self.padding    ,	pad_to_multiple_of=self.pad_to_multiple_of    ,	return_tensors="pt"    ,	)
														__snake_case		:			List[str]							 =   self.model._get_feat_extract_output_lengths(batch["input_values"].shape[-1]				)

														__snake_case		:			Union[str, Any]							 =   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
																					__snake_case		:			int							 =   self.model._get_feat_extract_output_lengths(batch["attention_mask"].sum(-1				)				).to(
																					    torch.long				)

																					__snake_case		:			List[Any]							 =   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
																					__snake_case		:			List[Any]							 =   1
																					__snake_case		:			Optional[Any]							 =   attention_mask.flip([-1]				).cumsum(-1				).flip([-1]				).bool()

														# sample randomly masked indices
														__snake_case		:			Any							 =   _compute_mask_indices(
														    (batch_size, mask_indices_seq_length)    ,	self.model.config.mask_time_prob    ,	self.model.config.mask_time_length    ,	attention_mask=lowerCamelCase    ,	min_masks=2    ,	)

														return batch

class 				a		(lowercase_      ):




							"""simple docstring"""


							def __init__(							self       :  str    ,	*lowerCamelCase       :  Union[str, Any]    ,	lowerCamelCase       :  Dict=1    ,	lowerCamelCase       :  List[str]=0    ,	lowerCamelCase       :  int=1.0    ,	**lowerCamelCase       :  Tuple				)	->    int:
														super().__init__(*lowerCamelCase    ,	**lowerCamelCase				)
														__snake_case		:			List[str]							 =   0
														__snake_case		:			Optional[int]							 =   max_gumbel_temp
														__snake_case		:			int							 =   min_gumbel_temp
														__snake_case		:			List[Any]							 =   gumbel_temp_decay






							def     __snake_case					(							self       :  List[Any]    ,	lowerCamelCase       :  Optional[Any]    ,	lowerCamelCase       :  List[str]				)	->    Optional[int]:
														model.train()
														__snake_case		:			List[Any]							 =   self._prepare_inputs(lowerCamelCase				)

														if self.use_amp:
																					with autocast():
																												__snake_case		:			int							 =   self.compute_loss(lowerCamelCase    ,	lowerCamelCase				)
														else:
																					__snake_case		:			int							 =   self.compute_loss(lowerCamelCase    ,	lowerCamelCase				)

														if self.args.n_gpu > 1 or self.deepspeed:
																					if model.module.config.ctc_loss_reduction == "mean":
																												__snake_case		:			Any							 =   loss.mean()
																					elif model.module.config.ctc_loss_reduction == "sum":
																												__snake_case		:			List[str]							 =   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:
																					__snake_case		:			List[str]							 =   loss / self.args.gradient_accumulation_steps

														if self.use_amp:
																					self.scaler.scale(lowerCamelCase				).backward()
														elif self.use_apex:
																					with amp.scale_loss(lowerCamelCase    ,	self.optimizer				) as scaled_loss:
																												scaled_loss.backward()
														elif self.deepspeed:
																					self.deepspeed.backward(lowerCamelCase				)
														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 				lowerCAmelCase_     (					):
							__snake_case		:			Dict							 =   HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)	)

							__snake_case ,			__snake_case ,			__snake_case		:			Dict							 =   parser.parse_args_into_dataclasses()
							configure_logger(UpperCamelCase__							,	UpperCamelCase__	)

							# Downloading and loading a dataset from the hub.
							__snake_case		:			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"
														__snake_case		:			Dict							 =   DatasetDict()
														__snake_case		:			List[str]							 =   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							,	)
														__snake_case		:			Tuple							 =   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"
														__snake_case		:			str							 =   DatasetDict()
														__snake_case		:			List[str]							 =   load_dataset(
														    data_args.dataset_name							,	data_args.dataset_config_name							,	split="validation"							,	cache_dir=model_args.cache_dir							,	)
														__snake_case		:			Optional[Any]							 =   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
							__snake_case		:			List[Any]							 =   WavaVecaFeatureExtractor.from_pretrained(
							    model_args.model_name_or_path							,	cache_dir=model_args.cache_dir							,	do_normalize=UpperCamelCase__	)

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

							# load audio files into numpy arrays
							__snake_case		:			Optional[Any]							 =   datasets.map(
							    UpperCamelCase__							,	num_proc=data_args.preprocessing_num_workers							,	remove_columns=datasets["train"].column_names	)

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

							def normalize(__lowerCamelCase	):
														return feature_extractor(batch["speech"]							,	sampling_rate=feature_extractor.sampling_rate	)

							# normalize and transform to `BatchFeatures`
							__snake_case		:			Tuple							 =   vectorized_datasets.map(
							    UpperCamelCase__							,	batched=UpperCamelCase__							,	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
							__snake_case		:			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\'"	)

							__snake_case		:			Any							 =   WavaVecaForPreTraining(UpperCamelCase__	)

							__snake_case		:			str							 =   DataCollatorForWavaVecaPretraining(model=UpperCamelCase__							,	feature_extractor=UpperCamelCase__	)

							__snake_case		:			Any							 =   WavaVecaPreTrainer(
							    model=UpperCamelCase__							,	data_collator=UpperCamelCase__							,	args=UpperCamelCase__							,	train_dataset=vectorized_datasets["train"]							,	eval_dataset=vectorized_datasets["validation"]							,	tokenizer=UpperCamelCase__							,	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()




from abc import ABC, abstractmethod
from argparse import ArgumentParser







class  lowercase				(       lowercase_			):




				@staticmethod
				@abstractmethod
				def     a							(					snake_case       ):
											raise NotImplementedError()




				@abstractmethod
				def     a							(					self       ):
											raise NotImplementedError()




