





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


_A :   Any  			=  logging.getLogger(__name__)


@dataclass
class        __SCREAMING_SNAKE_CASE      :






       """simple docstring"""




       _UpperCAmelCase				:  int			   =   field(
           metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"} )
       _UpperCAmelCase				:  Any			   =   field(
           default=_lowerCamelCase					,metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}					,)
       _UpperCAmelCase				:  Dict			   =   field(
           default=_lowerCamelCase					,metadata={"help": "Whether to freeze the feature extractor layers of the model."} )
       _UpperCAmelCase				:  int			   =   field(
           default=_lowerCamelCase					,metadata={"help": "Whether to log verbose messages or not."}					,)
       _UpperCAmelCase				:  Optional[Any]			   =   field(
           default=2.0					,metadata={"help": "Maximum temperature for gumbel softmax."} )
       _UpperCAmelCase				:  Optional[int]			   =   field(
           default=0.5					,metadata={"help": "Minimum temperature for gumbel softmax."} )
       _UpperCAmelCase				:  Dict			   =   field(
           default=0.99_9995					,metadata={"help": "Decay of gumbel temperature during training."} )



def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase					)    ->    str:







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


@dataclass
class        __SCREAMING_SNAKE_CASE      :






       """simple docstring"""




       _UpperCAmelCase				:  Optional[int]			   =   field(
           default=_lowerCamelCase					,metadata={"help": "The name of the dataset to use (via the datasets library)."} )
       _UpperCAmelCase				:  str			   =   field(
           default=_lowerCamelCase					,metadata={"help": "The configuration name of the dataset to use (via the datasets library)."} )
       _UpperCAmelCase				:  Any			   =   field(
           default="train"					,metadata={
               "help": "The name of the training data set split to use (via the datasets library). Defaults to 'train'"
           }					,)
       _UpperCAmelCase				:  Dict			   =   field(
           default="validation"					,metadata={
               "help": (
                   "The name of the validation data set split to use (via the datasets library). Defaults to 'validation'"
               )
           }					,)
       _UpperCAmelCase				:  Tuple			   =   field(
           default="file"					,metadata={"help": "Column in the dataset that contains speech file path. Defaults to 'file'"}					,)
       _UpperCAmelCase				:  str			   =   field(
           default=_lowerCamelCase					,metadata={"help": "Overwrite the cached preprocessed datasets or not."} )
       _UpperCAmelCase				:  List[str]			   =   field(
           default=1					,metadata={
               "help": "The percentage of the train set used as validation set in case there's no validation split"
           }					,)
       _UpperCAmelCase				:  Optional[Any]			   =   field(
           default=_lowerCamelCase					,metadata={"help": "The number of processes to use for the preprocessing."}					,)
       _UpperCAmelCase				:  Optional[Any]			   =   field(
           default=20.0					,metadata={"help": "Filter audio files that are longer than `max_duration_in_seconds` seconds"} )


@dataclass
class        __SCREAMING_SNAKE_CASE      :






       """simple docstring"""




       _UpperCAmelCase				:  List[Any]			   =   4_2
       _UpperCAmelCase				:  int			   =   4_2
       _UpperCAmelCase				:  List[str]			   =   "longest"
       _UpperCAmelCase				:  Dict			   =   None
       _UpperCAmelCase				:  Union[str, Any]			   =   None







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

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

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

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

          return batch


class        __SCREAMING_SNAKE_CASE      (			_lowerCamelCase ):






       """simple docstring"""




       def __init__(				self					:     Any		, *A					:     Any		, A					:     Tuple=1		, A					:     Optional[Any]=0		, A					:     List[Any]=1.0		, **A					:     Any      )       ->int:
          super().__init__(*lowercase_		, **lowercase_      )
          lowerCamelCase__       :						Optional[Any]        =							0
          lowerCamelCase__       :						List[str]        =							max_gumbel_temp
          lowerCamelCase__       :						Optional[Any]        =							min_gumbel_temp
          lowerCamelCase__       :						List[Any]        =							gumbel_temp_decay







       def 						__lowerCamelCase   (				self					:     Any		, A					:     nn.Module		, A					:     Dict[str, Union[torch.Tensor, Any]]      )       ->Dict:
          model.train()
          lowerCamelCase__       :						Union[str, Any]        =							self._prepare_inputs(lowercase_      )

          if self.use_amp:
             with autocast():
                lowerCamelCase__       :						Union[str, Any]        =							self.compute_loss(lowercase_		, lowercase_      )
          else:
             lowerCamelCase__       :						Union[str, Any]        =							self.compute_loss(lowercase_		, lowercase_      )

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

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







   """simple docstring"""
   lowerCamelCase__       :						int        =							HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)					)

   lowerCamelCase__       ,							lowerCamelCase__       ,							lowerCamelCase__       :						List[str]        =							parser.parse_args_into_dataclasses()
   configure_logger(__UpperCAmelCase      ,  __UpperCAmelCase					)

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

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

   # load audio files into numpy arrays
   lowerCamelCase__       :						Optional[int]        =							datasets.map(
       __UpperCAmelCase      ,  num_proc=data_args.preprocessing_num_workers      ,  remove_columns=datasets['''train'''].column_names					)

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

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

   # normalize and transform to `BatchFeatures`
   lowerCamelCase__       :						str        =							vectorized_datasets.map(
       __UpperCAmelCase      ,  batched=__UpperCAmelCase      ,  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
   lowerCamelCase__       :						str        =							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\''''					)

   lowerCamelCase__       :						List[str]        =							WavaVecaForPreTraining(__UpperCAmelCase					)

   lowerCamelCase__       :						Optional[Any]        =							DataCollatorForWavaVecaPretraining(model=__UpperCAmelCase      ,  feature_extractor=__UpperCAmelCase					)

   lowerCamelCase__       :						List[Any]        =							WavaVecaPreTrainer(
       model=__UpperCAmelCase      ,  data_collator=__UpperCAmelCase      ,  args=__UpperCAmelCase      ,  train_dataset=vectorized_datasets['''train''']      ,  eval_dataset=vectorized_datasets['''validation''']      ,  tokenizer=__UpperCAmelCase      ,  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 ceil, sqrt



def 	_a						(   UpperCAmelCase = 1000000					)    ->    int:







   """simple docstring"""
   lowerCamelCase__       :						Any        =							0

   for outer_width in range(3      ,  (limit // 4) + 2					):
      if outer_width**2 > limit:
         lowerCamelCase__       :						List[Any]        =							max(ceil(sqrt(outer_width**2 - limit					)					)      ,  1					)
      else:
         lowerCamelCase__       :						Union[str, Any]        =							1
      if (outer_width - hole_width_lower_bound) % 2:
         hole_width_lower_bound += 1

      answer += (outer_width - hole_width_lower_bound - 2) // 2 + 1

   return answer


if __name__ == "__main__":
       print(F'''{solution() = }''')



