






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


__UpperCAmelCase								=						logging.getLogger(__name__)



@dataclass
class   lowerCamelCase      :







		'''simple docstring'''



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





def      lowercase__      (     __snake_case    :							ModelArguments     ,       __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_    :      List[str]            =						logging.WARNING
					if model_args.verbose_logging:
										UpperCAmelCase_    :      Any            =						logging.DEBUG
					elif trainer_utils.is_main_process(training_args.local_rank  ):
										UpperCAmelCase_    :      Any            =						logging.INFO
					logger.setLevel(__snake_case  )



@dataclass
class   lowerCamelCase      :







		'''simple docstring'''



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





@dataclass
class   lowerCamelCase      :







		'''simple docstring'''



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



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

							UpperCAmelCase_    :      Union[str, Any]            =						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_    :      str            =						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_    :      Union[str, Any]            =						1
												UpperCAmelCase_    :      Dict            =						attention_mask.flip([-1]   ).cumsum(-1   ).flip([-1]   ).bool()

							# sample randomly masked indices
							UpperCAmelCase_    :      int            =						_compute_mask_indices(
							    (batch_size, mask_indices_seq_length)				,		self.model.config.mask_time_prob				,		self.model.config.mask_time_length				,		attention_mask=_UpperCamelCase				,		min_masks=2				,		)

							return batch





class   lowerCamelCase      (_snake_case			):







		'''simple docstring'''



		def __init__(							self				,		*_UpperCamelCase				,		_UpperCamelCase=1				,		_UpperCamelCase=0				,		_UpperCamelCase=1.0				,		**_UpperCamelCase   )	->							Dict:
							super().__init__(*_UpperCamelCase				,		**_UpperCamelCase   )
							UpperCAmelCase_    :      List[Any]            =						0
							UpperCAmelCase_    :      Any            =						max_gumbel_temp
							UpperCAmelCase_    :      int            =						min_gumbel_temp
							UpperCAmelCase_    :      Dict            =						gumbel_temp_decay



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase   )	->							torch.Tensor:
							model.train()
							UpperCAmelCase_    :      int            =						self._prepare_inputs(_UpperCamelCase   )

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

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

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


					'''simple docstring'''


					UpperCAmelCase_    :      Union[str, Any]            =						HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)  )

					UpperCAmelCase_       ,  UpperCAmelCase_       ,  UpperCAmelCase_    :      Union[str, Any]            =						parser.parse_args_into_dataclasses()
					configure_logger(__snake_case     ,       __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_    :      Tuple            =						DatasetDict()
										UpperCAmelCase_    :      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     ,       )
										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[int]            =						DatasetDict()
										UpperCAmelCase_    :      List[Any]            =						load_dataset(
										    data_args.dataset_name     ,       data_args.dataset_config_name     ,       split='validation'     ,       cache_dir=model_args.cache_dir     ,       )
										UpperCAmelCase_    :      Union[str, 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_    :      int            =						WavaVecaFeatureExtractor.from_pretrained(
					    model_args.model_name_or_path     ,       cache_dir=model_args.cache_dir     ,       do_normalize=__snake_case  )

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

					# load audio files into numpy arrays
					UpperCAmelCase_    :      List[Any]            =						datasets.map(
					    __snake_case     ,       num_proc=data_args.preprocessing_num_workers     ,       remove_columns=datasets['train'].column_names  )

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

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

					# normalize and transform to `BatchFeatures`
					UpperCAmelCase_    :      Tuple            =						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
					UpperCAmelCase_    :      Tuple            =						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_    :      Optional[Any]            =						WavaVecaForPreTraining(__snake_case  )

					UpperCAmelCase_    :      Optional[int]            =						DataCollatorForWavaVecaPretraining(model=__snake_case     ,       feature_extractor=__snake_case  )

					UpperCAmelCase_    :      Optional[int]            =						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()














from __future__ import annotations




def      lowercase__      (     __snake_case    :							tuple[int, int]     ,       __snake_case    :							int  ):


					'''simple docstring'''


					UpperCAmelCase_       ,  UpperCAmelCase_    :      Tuple            =						position
					UpperCAmelCase_    :      str            =						[
					    (y + 1, x + 2),
					    (y - 1, x + 2),
					    (y + 1, x - 2),
					    (y - 1, x - 2),
					    (y + 2, x + 1),
					    (y + 2, x - 1),
					    (y - 2, x + 1),
					    (y - 2, x - 1),
					]
					UpperCAmelCase_    :      Optional[Any]            =						[]

					for position in positions:
										UpperCAmelCase_       ,  UpperCAmelCase_    :      Tuple            =						position
										if 0 <= y_test < n and 0 <= x_test < n:
															permissible_positions.append(__snake_case  )

					return permissible_positions




def      lowercase__      (     __snake_case    :							list[list[int]]  ):


					'''simple docstring'''


					return not any(elem == 0 for row in board for elem in row  )




def      lowercase__      (     __snake_case    :							list[list[int]]     ,       __snake_case    :							tuple[int, int]     ,       __snake_case    :							int  ):


					'''simple docstring'''


					if is_complete(__snake_case  ):
										return True

					for position in get_valid_pos(__snake_case     ,       len(__snake_case  )  ):
										UpperCAmelCase_       ,  UpperCAmelCase_    :      Any            =						position

										if board[y][x] == 0:
															UpperCAmelCase_    :      Optional[Any]            =						curr + 1
															if open_knight_tour_helper(__snake_case     ,       __snake_case     ,       curr + 1  ):
																				return True
															UpperCAmelCase_    :      List[Any]            =						0

					return False




def      lowercase__      (     __snake_case    :							int  ):


					'''simple docstring'''


					UpperCAmelCase_    :      str            =						[[0 for i in range(__snake_case  )] for j in range(__snake_case  )]

					for i in range(__snake_case  ):
										for j in range(__snake_case  ):
															UpperCAmelCase_    :      Optional[Any]            =						1
															if open_knight_tour_helper(__snake_case     ,       (i, j)     ,       1  ):
																				return board
															UpperCAmelCase_    :      List[Any]            =						0

					UpperCAmelCase_    :      List[str]            =						F"Open Kight Tour cannot be performed on a board of size {n}"
					raise ValueError(__snake_case  )


if __name__ == "__main__":
		import doctest

		doctest.testmod()






