





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


__lowercase       :str       =  logging.getLogger(__name__)



@dataclass
class    _a     :






							"""simple docstring"""

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







def       UpperCAmelCase							(     _lowerCamelCase      :  ModelArguments			,   _lowerCamelCase      :  TrainingArguments     ):





						'''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     )]			,   )
						SCREAMING_SNAKE_CASE__     :				str   =					logging.WARNING
						if model_args.verbose_logging:
												SCREAMING_SNAKE_CASE__     :				str   =					logging.DEBUG
						elif trainer_utils.is_main_process(training_args.local_rank     ):
												SCREAMING_SNAKE_CASE__     :				List[str]   =					logging.INFO
						logger.setLevel(_lowerCamelCase     )



@dataclass
class    _a     :






							"""simple docstring"""

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







@dataclass
class    _a     :






							"""simple docstring"""

							snake_case_							 =						42
							snake_case_							 =						42
							snake_case_							 =						"longest"
							snake_case_							 =						None
							snake_case_							 =						None

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

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

																			SCREAMING_SNAKE_CASE__     :				int   =					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
																			SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					1
																			SCREAMING_SNAKE_CASE__     :				Optional[int]   =					attention_mask.flip([-1]    ).cumsum(-1    ).flip([-1]    ).bool()

													# sample randomly masked indices
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					_compute_mask_indices(
													    (batch_size, mask_indices_seq_length)							,	self.model.config.mask_time_prob							,	self.model.config.mask_time_length							,	attention_mask=a							,	min_masks=2							,	)

													return batch







class    _a     (				lowercase__							):






							"""simple docstring"""

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






							def      A_  (      self				:     Any							,	a				:     nn.Module							,	a				:     Dict[str, Union[torch.Tensor, Any]]    )	->torch.Tensor:
													model.train()
													SCREAMING_SNAKE_CASE__     :				int   =					self._prepare_inputs(a    )

													if self.use_amp:
																			with autocast():
																									SCREAMING_SNAKE_CASE__     :				Optional[int]   =					self.compute_loss(a							,	a    )
													else:
																			SCREAMING_SNAKE_CASE__     :				List[str]   =					self.compute_loss(a							,	a    )

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

													if self.use_amp:
																			self.scaler.scale(a    ).backward()
													elif self.use_apex:
																			with amp.scale_loss(a							,	self.optimizer    ) as scaled_loss:
																									scaled_loss.backward()
													elif self.deepspeed:
																			self.deepspeed.backward(a    )
													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       UpperCAmelCase							(     ):





						'''simple docstring'''


						SCREAMING_SNAKE_CASE__     :				List[Any]   =					HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)     )

						SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				int   =					parser.parse_args_into_dataclasses()
						configure_logger(_lowerCamelCase			,   _lowerCamelCase     )

						# Downloading and loading a dataset from the hub.
						SCREAMING_SNAKE_CASE__     :				Tuple   =					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"
												SCREAMING_SNAKE_CASE__     :				Optional[int]   =					DatasetDict()
												SCREAMING_SNAKE_CASE__     :				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			,   )
												SCREAMING_SNAKE_CASE__     :				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			,   )
						else:
												# make sure only "validation" and "train" keys remain"
												SCREAMING_SNAKE_CASE__     :				str   =					DatasetDict()
												SCREAMING_SNAKE_CASE__     :				int   =					load_dataset(
												    data_args.dataset_name			,   data_args.dataset_config_name			,   split="validation"			,   cache_dir=model_args.cache_dir			,   )
												SCREAMING_SNAKE_CASE__     :				Optional[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
						SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					WavaVecaFeatureExtractor.from_pretrained(
						    model_args.model_name_or_path			,   cache_dir=model_args.cache_dir			,   do_normalize=_lowerCamelCase     )

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

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

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

						def normalize(_lowerCamelCase      :  Union[str, Any]     ):
												return feature_extractor(batch["speech"]			,   sampling_rate=feature_extractor.sampling_rate     )

						# normalize and transform to `BatchFeatures`
						SCREAMING_SNAKE_CASE__     :				int   =					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
						SCREAMING_SNAKE_CASE__     :				Any   =					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'"     )

						SCREAMING_SNAKE_CASE__     :				Dict   =					WavaVecaForPreTraining(_lowerCamelCase     )

						SCREAMING_SNAKE_CASE__     :				Optional[int]   =					DataCollatorForWavaVecaPretraining(model=_lowerCamelCase			,   feature_extractor=_lowerCamelCase     )

						SCREAMING_SNAKE_CASE__     :				Any   =					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()






import numpy



class    _a     :






							"""simple docstring"""

							def __init__(      self				:     Optional[int]							,	a				:     numpy.ndarray							,	a				:     numpy.ndarray    )	->None:
													SCREAMING_SNAKE_CASE__     :				Any   =					input_array

													# Random initial weights are assigned where first argument is the
													# number of nodes in previous layer and second argument is the
													# number of nodes in the next layer.

													# Random initial weights are assigned.
													# self.input_array.shape[1] is used to represent number of nodes in input layer.
													# First hidden layer consists of 4 nodes.
													SCREAMING_SNAKE_CASE__     :				int   =					numpy.random.rand(
													    self.input_array.shape[1]							,	4    )

													# Random initial values for the first hidden layer.
													# First hidden layer has 4 nodes.
													# Second hidden layer has 3 nodes.
													SCREAMING_SNAKE_CASE__     :				Dict   =					numpy.random.rand(
													    4							,	3    )

													# Random initial values for the second hidden layer.
													# Second hidden layer has 3 nodes.
													# Output layer has 1 node.
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					numpy.random.rand(3							,	1    )

													# Real output values provided.
													SCREAMING_SNAKE_CASE__     :				str   =					output_array

													# Predicted output values by the neural network.
													# Predicted_output array initially consists of zeroes.
													SCREAMING_SNAKE_CASE__     :				Tuple   =					numpy.zeros(output_array.shape    )

							def      A_  (      self				:     Union[str, Any]    )	->numpy.ndarray:
													SCREAMING_SNAKE_CASE__     :				List[Any]   =					sigmoid(
													    numpy.dot(self.input_array							,	self.input_layer_and_first_hidden_layer_weights    )    )

													# layer_between_first_hidden_layer_and_second_hidden_layer is the layer
													# connecting the first hidden set of nodes with the second hidden set of nodes.
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					sigmoid(
													    numpy.dot(
													        self.layer_between_input_and_first_hidden_layer							,	self.first_hidden_layer_and_second_hidden_layer_weights							,	)    )

													# layer_between_second_hidden_layer_and_output is the layer connecting
													# second hidden layer with the output node.
													SCREAMING_SNAKE_CASE__     :				int   =					sigmoid(
													    numpy.dot(
													        self.layer_between_first_hidden_layer_and_second_hidden_layer							,	self.second_hidden_layer_and_output_layer_weights							,	)    )

													return self.layer_between_second_hidden_layer_and_output

							def      A_  (      self				:     int    )	->None:
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					numpy.dot(
													    self.layer_between_first_hidden_layer_and_second_hidden_layer.T							,	2
													    * (self.output_array - self.predicted_output)
													    * sigmoid_derivative(self.predicted_output    )							,	)
													SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					numpy.dot(
													    self.layer_between_input_and_first_hidden_layer.T							,	numpy.dot(
													        2
													        * (self.output_array - self.predicted_output)
													        * sigmoid_derivative(self.predicted_output    )							,	self.second_hidden_layer_and_output_layer_weights.T							,	)
													    * sigmoid_derivative(
													        self.layer_between_first_hidden_layer_and_second_hidden_layer    )							,	)
													SCREAMING_SNAKE_CASE__     :				int   =					numpy.dot(
													    self.input_array.T							,	numpy.dot(
													        numpy.dot(
													            2
													            * (self.output_array - self.predicted_output)
													            * sigmoid_derivative(self.predicted_output    )							,	self.second_hidden_layer_and_output_layer_weights.T							,	)
													        * sigmoid_derivative(
													            self.layer_between_first_hidden_layer_and_second_hidden_layer    )							,	self.first_hidden_layer_and_second_hidden_layer_weights.T							,	)
													    * sigmoid_derivative(self.layer_between_input_and_first_hidden_layer    )							,	)

													self.input_layer_and_first_hidden_layer_weights += (
													    updated_input_layer_and_first_hidden_layer_weights
													)
													self.first_hidden_layer_and_second_hidden_layer_weights += (
													    updated_first_hidden_layer_and_second_hidden_layer_weights
													)
													self.second_hidden_layer_and_output_layer_weights += (
													    updated_second_hidden_layer_and_output_layer_weights
													)

							def      A_  (      self				:     int							,	a				:     numpy.ndarray							,	a				:     int							,	a				:     bool    )	->None:
													for iteration in range(1							,	iterations + 1    ):
																			SCREAMING_SNAKE_CASE__     :				Dict   =					self.feedforward()
																			self.back_propagation()
																			if give_loss:
																									SCREAMING_SNAKE_CASE__     :				int   =					numpy.mean(numpy.square(output - self.feedforward()    )    )
																									print(f"""Iteration {iteration} Loss: {loss}"""    )






							def      A_  (      self				:     Tuple							,	a				:     numpy.ndarray    )	->int:
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					input_arr

													SCREAMING_SNAKE_CASE__     :				Dict   =					sigmoid(
													    numpy.dot(self.array							,	self.input_layer_and_first_hidden_layer_weights    )    )

													SCREAMING_SNAKE_CASE__     :				Any   =					sigmoid(
													    numpy.dot(
													        self.layer_between_input_and_first_hidden_layer							,	self.first_hidden_layer_and_second_hidden_layer_weights							,	)    )

													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					sigmoid(
													    numpy.dot(
													        self.layer_between_first_hidden_layer_and_second_hidden_layer							,	self.second_hidden_layer_and_output_layer_weights							,	)    )

													return int(self.layer_between_second_hidden_layer_and_output > 0.6    )







def       UpperCAmelCase							(     _lowerCamelCase      :  numpy.ndarray     ):





						'''simple docstring'''


						return 1 / (1 + numpy.exp(-value     ))







def       UpperCAmelCase							(     _lowerCamelCase      :  numpy.ndarray     ):





						'''simple docstring'''


						return (value) * (1 - (value))







def       UpperCAmelCase							(     ):





						'''simple docstring'''


						SCREAMING_SNAKE_CASE__     :				str   =					numpy.array(
						    (
						        [0, 0, 0],
						        [0, 0, 1],
						        [0, 1, 0],
						        [0, 1, 1],
						        [1, 0, 0],
						        [1, 0, 1],
						        [1, 1, 0],
						        [1, 1, 1],
						    )			,   dtype=numpy.floataa			,   )

						# True output values for the given input values.
						SCREAMING_SNAKE_CASE__     :				Any   =					numpy.array(([0], [1], [1], [0], [1], [0], [0], [1])			,   dtype=numpy.floataa     )

						# Calling neural network class.
						SCREAMING_SNAKE_CASE__     :				List[Any]   =					TwoHiddenLayerNeuralNetwork(
						    input_array=_lowerCamelCase			,   output_array=_lowerCamelCase     )

						# Calling training function.
						# Set give_loss to True if you want to see loss in every iteration.
						neural_network.train(output=_lowerCamelCase			,   iterations=10			,   give_loss=_lowerCamelCase     )

						return neural_network.predict(numpy.array(([1, 1, 1])			,   dtype=numpy.floataa     )     )


if __name__ == "__main__":
				example()