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


SCREAMING_SNAKE_CASE__   =					logging.getLogger(__name__)



@dataclass
class 					lowerCAmelCase_      :



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





def 				lowerCAmelCase__      (     _UpperCamelCase    :			ModelArguments	,     _UpperCamelCase    :			TrainingArguments   )   ->			Tuple:





						"""simple docstring"""






						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        =     logging.WARNING
						if model_args.verbose_logging:
												snake_case        =     logging.DEBUG
						elif trainer_utils.is_main_process(training_args.local_rank   ):
												snake_case        =     logging.INFO
						logger.setLevel(lowerCamelCase_   )



@dataclass
class 					lowerCAmelCase_      :



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





@dataclass
class 					lowerCAmelCase_      :



				"""simple docstring"""
				_lowerCAmelCase       :      WavaVecaForPreTraining
				_lowerCAmelCase       :      WavaVecaFeatureExtractor
				_lowerCAmelCase       :      Union[bool, str]     					=  "longest"
				_lowerCAmelCase       :      Optional[int]     					=  None
				_lowerCAmelCase       :      Optional[int]     					=  None

				def __call__(			self						,      lowerCAmelCase      ):






										"""simple docstring"""

										snake_case        =     self.feature_extractor.pad(
										    _UpperCamelCase						,      max_length=self.max_length						,      padding=self.padding						,      pad_to_multiple_of=self.pad_to_multiple_of						,      return_tensors='pt'						,      )
										snake_case        =     self.model._get_feat_extract_output_lengths(batch['input_values'].shape[-1]      )

										snake_case        =     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        =     self.model._get_feat_extract_output_lengths(batch['attention_mask'].sum(-1      )      ).to(
																    torch.long      )

																snake_case        =     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        =     1
																snake_case        =     attention_mask.flip([-1]      ).cumsum(-1      ).flip([-1]      ).bool()

										# sample randomly masked indices
										snake_case        =     _compute_mask_indices(
										    (batch_size, mask_indices_seq_length)						,      self.model.config.mask_time_prob						,      self.model.config.mask_time_length						,      attention_mask=_UpperCamelCase						,      min_masks=2						,      )

										return batch





class 					lowerCAmelCase_      (     lowercase__   ):



				"""simple docstring"""

				def __init__(			self						,      *lowerCAmelCase						,      lowerCAmelCase=1						,      lowerCAmelCase=0						,      lowerCAmelCase=1.0						,      **lowerCAmelCase      ):






										"""simple docstring"""

										super().__init__(*_UpperCamelCase						,      **_UpperCamelCase      )
										snake_case        =     0
										snake_case        =     max_gumbel_temp
										snake_case        =     min_gumbel_temp
										snake_case        =     gumbel_temp_decay

				def 						snake_case	(			self						,      lowerCAmelCase						,      lowerCAmelCase      ):






										"""simple docstring"""

										model.train()
										snake_case        =     self._prepare_inputs(_UpperCamelCase      )

										if self.use_amp:
																with autocast():
																						snake_case        =     self.compute_loss(_UpperCamelCase						,      _UpperCamelCase      )
										else:
																snake_case        =     self.compute_loss(_UpperCamelCase						,      _UpperCamelCase      )

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

										if self.use_amp:
																self.scaler.scale(_UpperCamelCase      ).backward()
										elif self.use_apex:
																with amp.scale_loss(_UpperCamelCase						,      self.optimizer      ) as scaled_loss:
																						scaled_loss.backward()
										elif self.deepspeed:
																self.deepspeed.backward(_UpperCamelCase      )
										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__      (     )   ->			Optional[Any]:





						"""simple docstring"""






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

						snake_case        =     parser.parse_args_into_dataclasses()
						configure_logger(lowerCamelCase_	,     lowerCamelCase_   )

						# Downloading and loading a dataset from the hub.
						snake_case        =     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        =     DatasetDict()
												snake_case        =     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        =     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        =     DatasetDict()
												snake_case        =     load_dataset(
												    data_args.dataset_name	,     data_args.dataset_config_name	,     split='validation'	,     cache_dir=model_args.cache_dir	,     )
												snake_case        =     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        =     WavaVecaFeatureExtractor.from_pretrained(
						    model_args.model_name_or_path	,     cache_dir=model_args.cache_dir	,     do_normalize=lowerCamelCase_   )

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

						# load audio files into numpy arrays
						snake_case        =     datasets.map(
						    lowerCamelCase_	,     num_proc=data_args.preprocessing_num_workers	,     remove_columns=datasets['train'].column_names   )

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

						def normalize(_UpperCamelCase    :			Dict   ):
												return feature_extractor(batch['speech']	,     sampling_rate=feature_extractor.sampling_rate   )

						# normalize and transform to `BatchFeatures`
						snake_case        =     vectorized_datasets.map(
						    lowerCamelCase_	,     batched=lowerCamelCase_	,     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        =     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        =     WavaVecaForPreTraining(lowerCamelCase_   )

						snake_case        =     DataCollatorForWavaVecaPretraining(model=lowerCamelCase_	,     feature_extractor=lowerCamelCase_   )

						snake_case        =     WavaVecaPreTrainer(
						    model=lowerCamelCase_	,     data_collator=lowerCamelCase_	,     args=lowerCamelCase_	,     train_dataset=vectorized_datasets['train']	,     eval_dataset=vectorized_datasets['validation']	,     tokenizer=lowerCamelCase_	,     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"""
SCREAMING_SNAKE_CASE__   =					{str(digit): digit**5 for digit in range(10)}





def 				lowerCAmelCase__      (     _UpperCamelCase    :			int   )   ->			int:





						"""simple docstring"""






						return sum(DIGITS_FIFTH_POWER[digit] for digit in str(_UpperCamelCase   )   )





def 				lowerCAmelCase__      (     )   ->			int:





						"""simple docstring"""






						return sum(
						    number
						    for number in range(1_0_0_0	,     1_0_0_0_0_0_0   )
						    if number == digits_fifth_powers_sum(_UpperCamelCase   )   )


if __name__ == "__main__":
					print(solution())


