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





  """simple docstring"""

  __UpperCAmelCase				:				str   =					field(
      metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models'''}	)
  __UpperCAmelCase				:				Optional[str]   =					field(
      default=__UpperCAmelCase    ,				metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co'''}    ,				)
  __UpperCAmelCase				:				Optional[bool]   =					field(
      default=__UpperCAmelCase    ,				metadata={'''help''': '''Whether to freeze the feature extractor layers of the model.'''}	)
  __UpperCAmelCase				:				Optional[bool]   =					field(
      default=__UpperCAmelCase    ,				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.9_9_9_9_9_5    ,				metadata={'''help''': '''Decay of gumbel temperature during training.'''}	)





def 					UpperCAmelCase_		(__a      :    Optional[int] ,	__a      :    Any ):






   """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 )] ,	)
   _a				:	Optional[int]			=   logging.WARNING
   if model_args.verbose_logging:
      _a				:	str			=   logging.DEBUG
   elif trainer_utils.is_main_process(training_args.local_rank ):
      _a				:	int			=   logging.INFO
   logger.setLevel(A__ )







@dataclass
class UpperCAmelCase__	:





  """simple docstring"""

  __UpperCAmelCase				:				str   =					field(
      default=__UpperCAmelCase    ,				metadata={'''help''': '''The name of the dataset to use (via the datasets library).'''}	)
  __UpperCAmelCase				:				Optional[str]   =					field(
      default=__UpperCAmelCase    ,				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=__UpperCAmelCase    ,				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=__UpperCAmelCase    ,				metadata={'''help''': '''The number of processes to use for the preprocessing.'''}    ,				)
  __UpperCAmelCase				:				Optional[float]   =					field(
      default=2_0.0    ,				metadata={'''help''': '''Filter audio files that are longer than `max_duration_in_seconds` seconds'''}	)







@dataclass
class UpperCAmelCase__	:





  """simple docstring"""

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


  def __call__(		self      :     str							,_a      :     Optional[int]       ):


     '''simple docstring'''






     _a				:	Optional[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'							,)
     _a				:	str			=   self.model._get_feat_extract_output_lengths(batch['input_values'].shape[-1]       )

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

     # sample randomly masked indices
     _a				:	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 UpperCAmelCase__	( __UpperCAmelCase	):





  """simple docstring"""

  def __init__(		self      :     Tuple							,*_a      :     str							,_a      :     Optional[int]=1							,_a      :     Tuple=0							,_a      :     List[Any]=1.0							,**_a      :     Dict       ):


     '''simple docstring'''






     super().__init__(*_a							,**_a       )
     _a				:	Dict			=   0
     _a				:	int			=   max_gumbel_temp
     _a				:	List[Any]			=   min_gumbel_temp
     _a				:	List[str]			=   gumbel_temp_decay


  def   __lowercase		(		self      :     Optional[int]							,_a      :     Union[str, Any]							,_a      :     List[Any]       ):


     '''simple docstring'''






     model.train()
     _a				:	Any			=   self._prepare_inputs(_a       )

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

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

   _a				:	Tuple			=   HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments) )

   _a				:	List[Any]			=   parser.parse_args_into_dataclasses()
   configure_logger(A__ ,	A__ )

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

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

   # load audio files into numpy arrays
   _a				:	Tuple			=   datasets.map(
       A__ ,	num_proc=data_args.preprocessing_num_workers ,	remove_columns=datasets['train'].column_names )

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

   def normalize(__a      :    Optional[int] ):
      return feature_extractor(batch['speech'] ,	sampling_rate=feature_extractor.sampling_rate )

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

   _a				:	Any			=   WavaVecaForPreTraining(A__ )

   _a				:	int			=   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()










from math import loga



def A_	(		A__				)  ->		int:
       if a < 0:
              raise ValueError('Input value must be a positive integer'				)
       elif isinstance(A__		,      A__				):
              raise TypeError('Input value must be a \'int\' type'				)
       return 0 if (a == 0) else int(loga(a & -a				)				)


if __name__ == "__main__":
    import doctest

    doctest.testmod()


