





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__   : Any         =       True
  from torch.cuda.amp import autocast


UpperCAmelCase__   : Optional[Any]         =       logging.getLogger(__name__)






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






def  A							(			snake_case__   :						Any ,				snake_case__   :						int				)					->				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(snake_case__				)






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






@dataclass
class 							__lowercase			:
      __UpperCAmelCase		      =						42
      __UpperCAmelCase		      =						42
      __UpperCAmelCase		      =						"longest"
      __UpperCAmelCase		      =						None
      __UpperCAmelCase		      =						None






      def __call__(		self			,							lowercase_)     ->       Dict[str, torch.Tensor]:
       # reformat list to dict and set to pytorch format
       __snake_case         =			self.feature_extractor.pad(
           __snake_case			,							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=__snake_case			,							min_masks=2			,							)

       return batch






class 							__lowercase			(     snake_case_       ):






      def __init__(		self			,							*lowercase_			,							lowercase_=1			,							lowercase_=0			,							lowercase_=1.0			,							**lowercase_)     ->       List[str]:
       super().__init__(*__snake_case			,							**__snake_case)
       __snake_case         =			0
       __snake_case         =			max_gumbel_temp
       __snake_case         =			min_gumbel_temp
       __snake_case         =			gumbel_temp_decay







      def 			_a						(		self			,							lowercase_			,							lowercase_)     ->       torch.Tensor:
       model.train()
       __snake_case         =			self._prepare_inputs(__snake_case)

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

       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(__snake_case).backward()
       elif self.use_apex:
        with amp.scale_loss(__snake_case			,							self.optimizer) as scaled_loss:
         scaled_loss.backward()
       elif self.deepspeed:
        self.deepspeed.backward(__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							(			)					->				Dict:
 '''simple docstring'''
 __snake_case         =			HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)				)

 __snake_case   ,    __snake_case   ,    __snake_case         =			parser.parse_args_into_dataclasses()
 configure_logger(snake_case__ ,				snake_case__				)

 # 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=snake_case__				)

 def prepare_dataset(snake_case__   :						str				):
  # check that all files have the correct sampling rate
  __snake_case   ,    __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(
     snake_case__ ,				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 snake_case__				: len(data['speech']				) < int(data_args.max_duration_in_seconds * feature_extractor.sampling_rate				)				)

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

 # normalize and transform to `BatchFeatures`
 __snake_case         =			vectorized_datasets.map(
     snake_case__ ,				batched=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
 __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(snake_case__				)

 __snake_case         =			DataCollatorForWavaVecaPretraining(model=snake_case__ ,				feature_extractor=snake_case__				)

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








'''simple docstring'''






import os
import unittest

from transformers import BatchEncoding
from transformers.models.bert.tokenization_bert import (
    BasicTokenizer,
    WordpieceTokenizer,
    _is_control,
    _is_punctuation,
    _is_whitespace,
)
from transformers.models.prophetnet.tokenization_prophetnet import VOCAB_FILES_NAMES, ProphetNetTokenizer
from transformers.testing_utils import require_torch, slow

from ...test_tokenization_common import TokenizerTesterMixin




class 					UpperCAmelCase	(     snake_case_				,						unittest.TestCase  ):
      _lowercase:     Optional[int]			=  ProphetNetTokenizer
      _lowercase:     Union[str, Any]			=  False




      def 							lowercase__       (     self     :					Dict    )							-> Optional[int]:
            super().setUp()

            _lowerCAmelCase					=				[
                """[UNK]""",
                """[CLS]""",
                """[SEP]""",
                """[PAD]""",
                """[MASK]""",
                """want""",
                """##want""",
                """##ed""",
                """wa""",
                """un""",
                """runn""",
                """##ing""",
                """,""",
                """low""",
                """lowest""",
            ]
            _lowerCAmelCase					=				os.path.join(self.tmpdirname     ,     VOCAB_FILES_NAMES["""vocab_file"""]    )
            with open(self.vocab_file     ,     """w"""     ,     encoding="""utf-8"""    ) as vocab_writer:
                  vocab_writer.write("""""".join([x + """\n""" for x in vocab_tokens]    )    )




      def 							lowercase__       (     self     :					Tuple     ,     __snake_case     :					str    )							-> List[str]:
            _lowerCAmelCase					=				"""UNwant\u00E9d,running"""
            _lowerCAmelCase					=				"""unwanted, running"""
            return input_text, output_text




      def 							lowercase__       (     self     :					Union[str, Any]    )							-> Optional[int]:
            _lowerCAmelCase					=				self.tokenizer_class(self.vocab_file    )

            _lowerCAmelCase					=				tokenizer.tokenize("""UNwant\u00E9d,running"""    )
            self.assertListEqual(__snake_case     ,     ["""un""", """##want""", """##ed""", """,""", """runn""", """##ing"""]    )
            self.assertListEqual(tokenizer.convert_tokens_to_ids(__snake_case    )     ,     [9, 6, 7, 12, 10, 11]    )




      def 							lowercase__       (     self     :					List[Any]    )							-> List[str]:
            _lowerCAmelCase					=				BasicTokenizer()

            self.assertListEqual(tokenizer.tokenize("""ah\u535A\u63A8zz"""    )     ,     ["""ah""", """\u535A""", """\u63A8""", """zz"""]    )




      def 							lowercase__       (     self     :					Optional[Any]    )							-> List[Any]:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHeLLo!how  \n Are yoU?  """    )     ,     ["""hello""", """!""", """how""", """are""", """you""", """?"""]    )
            self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""    )     ,     ["""hello"""]    )




      def 							lowercase__       (     self     :					Dict    )							-> int:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case     ,     strip_accents=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """    )     ,     ["""hällo""", """!""", """how""", """are""", """you""", """?"""]    )
            self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""    )     ,     ["""h\u00E9llo"""]    )




      def 							lowercase__       (     self     :					Dict    )							-> int:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case     ,     strip_accents=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """    )     ,     ["""hallo""", """!""", """how""", """are""", """you""", """?"""]    )
            self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""    )     ,     ["""hello"""]    )




      def 							lowercase__       (     self     :					Dict    )							-> Optional[Any]:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """    )     ,     ["""hallo""", """!""", """how""", """are""", """you""", """?"""]    )
            self.assertListEqual(tokenizer.tokenize("""H\u00E9llo"""    )     ,     ["""hello"""]    )




      def 							lowercase__       (     self     :					str    )							-> List[str]:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHeLLo!how  \n Are yoU?  """    )     ,     ["""HeLLo""", """!""", """how""", """Are""", """yoU""", """?"""]    )




      def 							lowercase__       (     self     :					Union[str, Any]    )							-> str:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case     ,     strip_accents=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """    )     ,     ["""HäLLo""", """!""", """how""", """Are""", """yoU""", """?"""]    )




      def 							lowercase__       (     self     :					Tuple    )							-> str:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case     ,     strip_accents=__snake_case    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHäLLo!how  \n Are yoU?  """    )     ,     ["""HaLLo""", """!""", """how""", """Are""", """yoU""", """?"""]    )




      def 							lowercase__       (     self     :					Dict    )							-> Optional[int]:
            _lowerCAmelCase					=				BasicTokenizer(do_lower_case=__snake_case     ,     never_split=["""[UNK]"""]    )

            self.assertListEqual(
                tokenizer.tokenize(""" \tHeLLo!how  \n Are yoU? [UNK]"""    )     ,     ["""HeLLo""", """!""", """how""", """Are""", """yoU""", """?""", """[UNK]"""]    )




      def 							lowercase__       (     self     :					Any    )							-> Dict:
            _lowerCAmelCase					=				["""[UNK]""", """[CLS]""", """[SEP]""", """want""", """##want""", """##ed""", """wa""", """un""", """runn""", """##ing"""]

            _lowerCAmelCase					=				{}
            for i, token in enumerate(__snake_case    ):
                  _lowerCAmelCase					=				i
            _lowerCAmelCase					=				WordpieceTokenizer(vocab=__snake_case     ,     unk_token="""[UNK]"""    )

            self.assertListEqual(tokenizer.tokenize(""""""    )     ,     []    )

            self.assertListEqual(tokenizer.tokenize("""unwanted running"""    )     ,     ["""un""", """##want""", """##ed""", """runn""", """##ing"""]    )

            self.assertListEqual(tokenizer.tokenize("""unwantedX running"""    )     ,     ["""[UNK]""", """runn""", """##ing"""]    )




      @require_torch
      def 							lowercase__       (     self     :					Optional[int]    )							-> Optional[Any]:
            _lowerCAmelCase					=				self.tokenizer_class.from_pretrained("""microsoft/prophetnet-large-uncased"""    )

            _lowerCAmelCase					=				["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
            _lowerCAmelCase					=				[10_37, 21_46, 2_04_23, 20_05, 76_80, 78_49, 39_89, 10_12, 1_02]
            _lowerCAmelCase					=				tokenizer(__snake_case     ,     padding=__snake_case     ,     return_tensors="""pt"""    )
            self.assertIsInstance(__snake_case     ,     __snake_case    )
            _lowerCAmelCase					=				list(batch.input_ids.numpy()[0]    )
            self.assertListEqual(__snake_case     ,     __snake_case    )

            self.assertEqual((2, 9)     ,     batch.input_ids.shape    )
            self.assertEqual((2, 9)     ,     batch.attention_mask.shape    )




      def 							lowercase__       (     self     :					List[Any]    )							-> Any:
            self.assertTrue(_is_whitespace(""" """    )    )
            self.assertTrue(_is_whitespace("""\t"""    )    )
            self.assertTrue(_is_whitespace("""\r"""    )    )
            self.assertTrue(_is_whitespace("""\n"""    )    )
            self.assertTrue(_is_whitespace("""\u00A0"""    )    )

            self.assertFalse(_is_whitespace("""A"""    )    )
            self.assertFalse(_is_whitespace("""-"""    )    )




      def 							lowercase__       (     self     :					str    )							-> Union[str, Any]:
            self.assertTrue(_is_control("""\u0005"""    )    )

            self.assertFalse(_is_control("""A"""    )    )
            self.assertFalse(_is_control(""" """    )    )
            self.assertFalse(_is_control("""\t"""    )    )
            self.assertFalse(_is_control("""\r"""    )    )




      def 							lowercase__       (     self     :					int    )							-> int:
            self.assertTrue(_is_punctuation("""-"""    )    )
            self.assertTrue(_is_punctuation("""$"""    )    )
            self.assertTrue(_is_punctuation("""`"""    )    )
            self.assertTrue(_is_punctuation("""."""    )    )

            self.assertFalse(_is_punctuation("""A"""    )    )
            self.assertFalse(_is_punctuation(""" """    )    )




      @slow
      def 							lowercase__       (     self     :					int    )							-> Optional[int]:
            _lowerCAmelCase					=				self.tokenizer_class.from_pretrained("""microsoft/prophetnet-large-uncased"""    )

            _lowerCAmelCase					=				tokenizer.encode("""sequence builders"""     ,     add_special_tokens=__snake_case    )
            _lowerCAmelCase					=				tokenizer.encode("""multi-sequence build"""     ,     add_special_tokens=__snake_case    )

            _lowerCAmelCase					=				tokenizer.build_inputs_with_special_tokens(__snake_case    )
            _lowerCAmelCase					=				tokenizer.build_inputs_with_special_tokens(__snake_case     ,     __snake_case    )

            assert encoded_sentence == text + [1_02]
            assert encoded_pair == text + [1_02] + text_a + [1_02]






