


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


__lowerCAmelCase       		=      logging.getLogger(__name__)





@dataclass
class 	SCREAMING_SNAKE_CASE					:
			snake_case__  =				field(
			    metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"}       )
			snake_case__  =				field(
			    default=a__	,   metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}	,   )
			snake_case__  =				field(
			    default=a__	,   metadata={"help": "Whether to freeze the feature extractor layers of the model."}       )
			snake_case__  =				field(
			    default=a__	,   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.999_995	,   metadata={"help": "Decay of gumbel temperature during training."}       )





def 		_UpperCAmelCase   (							__A							:   Dict						,		__A							:   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_		:  Union[str, Any]											=      logging.WARNING
				if model_args.verbose_logging:
								a_		:  str											=      logging.DEBUG
				elif trainer_utils.is_main_process(training_args.local_rank      ):
								a_		:  Any											=      logging.INFO
				logger.setLevel(__A      )





@dataclass
class 	SCREAMING_SNAKE_CASE					:
			snake_case__  =				field(
			    default=a__	,   metadata={"help": "The name of the dataset to use (via the datasets library)."}       )
			snake_case__  =				field(
			    default=a__	,   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=a__	,   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=a__	,   metadata={"help": "The number of processes to use for the preprocessing."}	,   )
			snake_case__  =				field(
			    default=20.0	,   metadata={"help": "Filter audio files that are longer than `max_duration_in_seconds` seconds"}       )





@dataclass
class 	SCREAMING_SNAKE_CASE					:
			snake_case__  =				42
			snake_case__  =				42
			snake_case__  =				"longest"
			snake_case__  =				None
			snake_case__  =				None






			def __call__( self      :			Optional[int]  ,						__SCREAMING_SNAKE_CASE      :			List[str]					)			->			Dict[str, torch.Tensor]:
							# reformat list to dict and set to pytorch format
							a_		:  str											=      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'''  ,						)
							a_		:  int											=      self.model._get_feat_extract_output_lengths(batch['''input_values'''].shape[-1]					)

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

							# sample randomly masked indices
							a_		:  Union[str, 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 	SCREAMING_SNAKE_CASE					(					a__       ):






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







			def 			SCREAMING_SNAKE_CASE      ( self      :			Tuple  ,						__SCREAMING_SNAKE_CASE      :			str  ,						__SCREAMING_SNAKE_CASE      :			Dict					)			->			torch.Tensor:
							model.train()
							a_		:  Dict											=      self._prepare_inputs(lowerCamelCase_					)

							if self.use_amp:
											with autocast():
															a_		:  Dict											=      self.compute_loss(lowerCamelCase_  ,						lowerCamelCase_					)
							else:
											a_		:  List[str]											=      self.compute_loss(lowerCamelCase_  ,						lowerCamelCase_					)

							if self.args.n_gpu > 1 or self.deepspeed:
											if model.module.config.ctc_loss_reduction == "mean":
															a_		:  List[Any]											=      loss.mean()
											elif model.module.config.ctc_loss_reduction == "sum":
															a_		:  List[Any]											=      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_		:  Tuple											=      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 		_UpperCAmelCase   (							):
				# 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_		:  Any											=      HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )

				a_				,			a_				,			a_		:  Dict											=      parser.parse_args_into_dataclasses()
				configure_logger(__A						,		__A      )

				# Downloading and loading a dataset from the hub.
				a_		:  int											=      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_		:  Any											=      DatasetDict()
								a_		:  List[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_		:  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						,		)
				else:
								# make sure only "validation" and "train" keys remain"
								a_		:  str											=      DatasetDict()
								a_		:  Optional[Any]											=      load_dataset(
								    data_args.dataset_name						,		data_args.dataset_config_name						,		split='''validation'''						,		cache_dir=model_args.cache_dir						,		)
								a_		:  Dict											=      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_		:  Optional[int]											=      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_		:  Union[str, Any]											=      librosa.load(batch[data_args.speech_file_column]						,		sr=feature_extractor.sampling_rate      )
								return batch

				# load audio files into numpy arrays
				a_		:  int											=      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[Any]											=      vectorized_datasets.filter(
				    lambda __A      : len(data['''speech''']      ) < int(data_args.max_duration_in_seconds * feature_extractor.sampling_rate      )      )

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

				# normalize and transform to `BatchFeatures`
				a_		:  int											=      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_		:  int											=      DataCollatorForWavaVecaPretraining(model=__A						,		feature_extractor=__A      )

				a_		:  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'''







import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import ClassLabel, Features, Image
from .base import TaskTemplate





@dataclass(frozen=SCREAMING_SNAKE_CASE_       )
class 	SCREAMING_SNAKE_CASE					(					SCREAMING_SNAKE_CASE_       ):
			snake_case__  =				field(default="image-classification"	,   metadata={"include_in_asdict_even_if_is_default": True}       )
			snake_case__  =				Features({"image": Image()}       )
			snake_case__  =				Features({"labels": ClassLabel}       )
			snake_case__  =				"image"
			snake_case__  =				"labels"






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[str]  ,						__SCREAMING_SNAKE_CASE      :			List[Any]					)			->			Any:
							if self.label_column not in features:
											raise ValueError(f'Column {self.label_column} is not present in features.'					)
							if not isinstance(features[self.label_column]  ,						__SCREAMING_SNAKE_CASE					):
											raise ValueError(f'Column {self.label_column} is not a ClassLabel.'					)
							a_		:  Optional[int]											=      copy.deepcopy(self					)
							a_		:  int											=      self.label_schema.copy()
							a_		:  Tuple											=      features[self.label_column]
							a_		:  str											=      label_schema
							return task_template







			@property
			def 			SCREAMING_SNAKE_CASE      ( self      :			Tuple					)			->			Dict[str, str]:
							return {
							    self.image_column: "image",
							    self.label_column: "labels",
							}




