


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"):
  __UpperCAmelCase : List[str] 			=				True
  from torch.cuda.amp import autocast


__UpperCAmelCase : str 			=				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=_a,     metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"},     )
       __UpperCamelCase      :     Optional[bool]							=					field(
           default=_a,     metadata={"help": "Whether to freeze the feature extractor layers of the model."})
       __UpperCamelCase      :     Optional[bool]							=					field(
           default=_a,     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      a	(							SCREAMING_SNAKE_CASE_      :		ModelArguments							,			SCREAMING_SNAKE_CASE_      :		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       )]							,			)
       UpperCamelCase    :	Optional[Any]   	=		logging.WARNING
       if model_args.verbose_logging:
              UpperCamelCase    :	Dict   	=		logging.DEBUG
       elif trainer_utils.is_main_process(training_args.local_rank       ):
              UpperCamelCase    :	str   	=		logging.INFO
       logger.setLevel(SCREAMING_SNAKE_CASE_       )



@dataclass
class    UpperCAmelCase_	:
       '''simple docstring'''
       __UpperCamelCase      :     str							=					field(
           default=_a,     metadata={"help": "The name of the dataset to use (via the datasets library)."})
       __UpperCamelCase      :     Optional[str]							=					field(
           default=_a,     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=_a,     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=_a,     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 ,					__SCREAMING_SNAKE_CASE						):




              """simple docstring"""




              UpperCamelCase    :	Optional[Any]   	=		self.feature_extractor.pad(
                  __SCREAMING_SNAKE_CASE ,					max_length=self.max_length ,					padding=self.padding ,					pad_to_multiple_of=self.pad_to_multiple_of ,					return_tensors='''pt''' ,					)
              UpperCamelCase    :	Dict   	=		self.model._get_feat_extract_output_lengths(batch['''input_values'''].shape[-1]						)

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

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

              # sample randomly masked indices
              UpperCamelCase    :	str   	=		_compute_mask_indices(
                  (batch_size, mask_indices_seq_length) ,					self.model.config.mask_time_prob ,					self.model.config.mask_time_length ,					attention_mask=__SCREAMING_SNAKE_CASE ,					min_masks=2 ,					)

              return batch




class    UpperCAmelCase_	(							_a):
       '''simple docstring'''






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




              """simple docstring"""




              super().__init__(*__SCREAMING_SNAKE_CASE ,					**__SCREAMING_SNAKE_CASE						)
              UpperCamelCase    :	Tuple   	=		0
              UpperCamelCase    :	List[str]   	=		max_gumbel_temp
              UpperCamelCase    :	Optional[Any]   	=		min_gumbel_temp
              UpperCamelCase    :	Optional[int]   	=		gumbel_temp_decay






       def  _lowercase   (     self ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE						):




              """simple docstring"""




              model.train()
              UpperCamelCase    :	List[str]   	=		self._prepare_inputs(__SCREAMING_SNAKE_CASE						)

              if self.use_amp:
                     with autocast():
                            UpperCamelCase    :	Tuple   	=		self.compute_loss(__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE						)
              else:
                     UpperCamelCase    :	Dict   	=		self.compute_loss(__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE						)

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

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






       """simple docstring"""





       UpperCamelCase    :	List[str]   	=		HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)       )

       UpperCamelCase	, UpperCamelCase	, UpperCamelCase    :	List[str]   	=		parser.parse_args_into_dataclasses()
       configure_logger(SCREAMING_SNAKE_CASE_							,			SCREAMING_SNAKE_CASE_       )

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

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

       # load audio files into numpy arrays
       UpperCamelCase    :	Union[str, Any]   	=		datasets.map(
           SCREAMING_SNAKE_CASE_							,			num_proc=data_args.preprocessing_num_workers							,			remove_columns=datasets['''train'''].column_names       )

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

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

       # normalize and transform to `BatchFeatures`
       UpperCamelCase    :	int   	=		vectorized_datasets.map(
           SCREAMING_SNAKE_CASE_							,			batched=SCREAMING_SNAKE_CASE_							,			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
       UpperCamelCase    :	Union[str, 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\''''       )

       UpperCamelCase    :	Union[str, Any]   	=		WavaVecaForPreTraining(SCREAMING_SNAKE_CASE_       )

       UpperCamelCase    :	Dict   	=		DataCollatorForWavaVecaPretraining(model=SCREAMING_SNAKE_CASE_							,			feature_extractor=SCREAMING_SNAKE_CASE_       )

       UpperCamelCase    :	Tuple   	=		WavaVecaPreTrainer(
           model=SCREAMING_SNAKE_CASE_							,			data_collator=SCREAMING_SNAKE_CASE_							,			args=SCREAMING_SNAKE_CASE_							,			train_dataset=vectorized_datasets['''train''']							,			eval_dataset=vectorized_datasets['''validation''']							,			tokenizer=SCREAMING_SNAKE_CASE_							,			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()





def      a	(							SCREAMING_SNAKE_CASE_      :		int							,			SCREAMING_SNAKE_CASE_      :		int       ):






       """simple docstring"""





       if a < 0 or b < 0:
              raise ValueError('''the value of both inputs must be positive'''       )
       UpperCamelCase    :	int   	=		str(bin(SCREAMING_SNAKE_CASE_       )       )[2:]  # remove the leading "0b"
       UpperCamelCase    :	List[str]   	=		str(bin(SCREAMING_SNAKE_CASE_       )       )[2:]
       UpperCamelCase    :	Tuple   	=		max(len(SCREAMING_SNAKE_CASE_       )							,			len(SCREAMING_SNAKE_CASE_       )       )
       return "0b" + "".join(
           str(int('''1''' in (char_a, char_b)       )       )
           for char_a, char_b in zip(a_binary.zfill(SCREAMING_SNAKE_CASE_       )							,			b_binary.zfill(SCREAMING_SNAKE_CASE_       )       )       )


if __name__ == "__main__":
  import doctest

  doctest.testmod()

