





'''simple docstring'''




from collections import Counter
from pathlib import Path
from typing import Optional, Tuple

import yaml






class 				__snake_case     (					yaml.SafeLoader							):





	def   _snake_case   (		self      ,       UpperCamelCase_		)      ->		List[Any]:
		snake_case__	       =   [self.constructed_objects[key_node] for key_node, _ in node.value]
		snake_case__	       =   [tuple(UpperCamelCase_		) if isinstance(UpperCamelCase_      ,       UpperCamelCase_		) else key for key in keys]
		snake_case__	       =   Counter(UpperCamelCase_		)
		snake_case__	       =   [key for key in counter if counter[key] > 1]
		if duplicate_keys:
			raise TypeError(F'''Got duplicate yaml keys: {duplicate_keys}'''		)





	def   _snake_case   (		self      ,       UpperCamelCase_      ,       UpperCamelCase_=False		)      ->		int:
		snake_case__	       =   super().construct_mapping(UpperCamelCase_      ,       deep=UpperCamelCase_		)
		self._check_no_duplicates_on_constructed_node(UpperCamelCase_		)
		return mapping





def 						__lowerCamelCase			(	UpperCAmelCase_							)   ->Tuple[Optional[str], str]:
	snake_case__	       =   list(readme_content.splitlines()							)
	if full_content and full_content[0] == "---" and "---" in full_content[1:]:
		snake_case__	       =   full_content[1:].index('---'							) + 1
		snake_case__	       =   '\n'.join(full_content[1:sep_idx]							)
		return yamlblock, "\n".join(full_content[sep_idx + 1 :]							)

	return None, "\n".join(__UpperCAmelCase							)






class 				__snake_case     (					_snake_case							):
	# class attributes
	__lowerCAmelCase       						=  {'train_eval_index'}  # train-eval-index in the YAML metadata





	@classmethod
	def   _snake_case   (		cls      ,       UpperCamelCase_		)      ->		Tuple:

		with open(UpperCamelCase_      ,       encoding='utf-8'		) as readme_file:
			snake_case__       ,  snake_case__	       =   _split_yaml_from_readme(readme_file.read()		)
		if yaml_string is not None:
			return cls.from_yaml_string(UpperCamelCase_		)
		else:
			return cls()





	def   _snake_case   (		self      ,       UpperCamelCase_		)      ->		int:

		if path.exists():
			with open(UpperCamelCase_      ,       encoding='utf-8'		) as readme_file:
				snake_case__	       =   readme_file.read()
		else:
			snake_case__	       =   None
		snake_case__	       =   self._to_readme(UpperCamelCase_		)
		with open(UpperCamelCase_      ,       'w'      ,       encoding='utf-8'		) as readme_file:
			readme_file.write(UpperCamelCase_		)





	def   _snake_case   (		self      ,       UpperCamelCase_ = None		)      ->		Optional[int]:

		if readme_content is not None:
			snake_case__       ,  snake_case__	       =   _split_yaml_from_readme(UpperCamelCase_		)
			snake_case__	       =   '---\n' + self.to_yaml_string() + '---\n' + content
		else:
			snake_case__	       =   '---\n' + self.to_yaml_string() + '---\n'
		return full_content





	@classmethod
	def   _snake_case   (		cls      ,       UpperCamelCase_		)      ->		Optional[int]:
		snake_case__	       =   yaml.load(UpperCamelCase_      ,       Loader=_NoDuplicateSafeLoader		) or {}

		# Convert the YAML keys to DatasetMetadata fields
		snake_case__	       =   {
		    (key.replace('-'      ,       '_'		) if key.replace('-'      ,       '_'		) in cls._FIELDS_WITH_DASHES else key): value
		    for key, value in metadata_dict.items()
		}
		return cls(**UpperCamelCase_		)





	def   _snake_case   (		self		)      ->		Optional[Any]:
		return yaml.safe_dump(
		    {
		        (key.replace('_'      ,       '-'		) if key in self._FIELDS_WITH_DASHES else key): value
		        for key, value in self.items()
		    }      ,       sort_keys=UpperCamelCase_      ,       allow_unicode=UpperCamelCase_      ,       encoding='utf-8'      ,       ).decode('utf-8'		)





a__					:			Union[str, Any]          =     {
    '''image-classification''': [],
    '''translation''': [],
    '''image-segmentation''': [],
    '''fill-mask''': [],
    '''automatic-speech-recognition''': [],
    '''token-classification''': [],
    '''sentence-similarity''': [],
    '''audio-classification''': [],
    '''question-answering''': [],
    '''summarization''': [],
    '''zero-shot-classification''': [],
    '''table-to-text''': [],
    '''feature-extraction''': [],
    '''other''': [],
    '''multiple-choice''': [],
    '''text-classification''': [],
    '''text-to-image''': [],
    '''text2text-generation''': [],
    '''zero-shot-image-classification''': [],
    '''tabular-classification''': [],
    '''tabular-regression''': [],
    '''image-to-image''': [],
    '''tabular-to-text''': [],
    '''unconditional-image-generation''': [],
    '''text-retrieval''': [],
    '''text-to-speech''': [],
    '''object-detection''': [],
    '''audio-to-audio''': [],
    '''text-generation''': [],
    '''conversational''': [],
    '''table-question-answering''': [],
    '''visual-question-answering''': [],
    '''image-to-text''': [],
    '''reinforcement-learning''': [],
    '''voice-activity-detection''': [],
    '''time-series-forecasting''': [],
    '''document-question-answering''': [],
}


if __name__ == "__main__":
			from argparse import ArgumentParser

			a__					:			int          =     ArgumentParser(usage='''Validate the yaml metadata block of a README.md file.''')
			ap.add_argument('''readme_filepath''')
			a__					:			Union[str, Any]          =     ap.parse_args()

			a__					:			Dict          =     Path(args.readme_filepath)
			a__					:			int          =     DatasetMetadata.from_readme(readme_filepath)
			print(dataset_metadata)
			dataset_metadata.to_readme(readme_filepath)










'''simple docstring'''

import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


a   	=    logging.get_logger(__name__)







class __a			( _snake_case      ):
   __UpperCamelCase :			List[str]						   =						['input_features', 'is_longer']





   def __init__(       self					:      Union[str, Any]					,lowerCamelCase					:      Union[str, Any]=64					,lowerCamelCase					:      Any=4_8000					,lowerCamelCase					:      int=480					,lowerCamelCase					:      Any=10					,lowerCamelCase					:      Dict=1024					,lowerCamelCase					:      Union[str, Any]=0.0					,lowerCamelCase					:      int=False					,lowerCamelCase					:      float = 0					,lowerCamelCase					:      float = 1_4000					,lowerCamelCase					:      int = None					,lowerCamelCase					:      str = "fusion"					,lowerCamelCase					:      str = "repeatpad"					,**lowerCamelCase					:      Dict					,):

      '''simple docstring'''

      super().__init__(
          feature_size=lowerCamelCase					,sampling_rate=lowerCamelCase					,padding_value=lowerCamelCase					,return_attention_mask=lowerCamelCase					,**lowerCamelCase					,)
      __SCREAMING_SNAKE_CASE   				=							top_db
      __SCREAMING_SNAKE_CASE   				=							truncation
      __SCREAMING_SNAKE_CASE   				=							padding
      __SCREAMING_SNAKE_CASE   				=							fft_window_size
      __SCREAMING_SNAKE_CASE   				=							(fft_window_size >> 1) + 1
      __SCREAMING_SNAKE_CASE   				=							hop_length
      __SCREAMING_SNAKE_CASE   				=							max_length_s
      __SCREAMING_SNAKE_CASE   				=							max_length_s * sampling_rate
      __SCREAMING_SNAKE_CASE   				=							sampling_rate
      __SCREAMING_SNAKE_CASE   				=							frequency_min
      __SCREAMING_SNAKE_CASE   				=							frequency_max
      __SCREAMING_SNAKE_CASE   				=							mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins					,num_mel_filters=lowerCamelCase					,min_frequency=lowerCamelCase					,max_frequency=lowerCamelCase					,sampling_rate=lowerCamelCase					,norm=lowerCamelCase					,mel_scale="""htk"""					,)
      __SCREAMING_SNAKE_CASE   				=							mel_filter_bank(
          num_frequency_bins=self.nb_frequency_bins					,num_mel_filters=lowerCamelCase					,min_frequency=lowerCamelCase					,max_frequency=lowerCamelCase					,sampling_rate=lowerCamelCase					,norm="""slaney"""					,mel_scale="""slaney"""					,)





   def   UpperCAmelCase__  (       self					:      int     ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE   				=							copy.deepcopy(self.__dict__     )
      __SCREAMING_SNAKE_CASE   				=							self.__class__.__name__
      if "mel_filters" in output:
         del output["mel_filters"]
      if "mel_filters_slaney" in output:
         del output["mel_filters_slaney"]
      return output





   def   UpperCAmelCase__  (       self					:      str					,lowerCamelCase					:      np.array					,lowerCamelCase					:      Optional[np.array] = None     ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE   				=							spectrogram(
          lowerCamelCase					,window_function(self.fft_window_size					,"""hann"""     )					,frame_length=self.fft_window_size					,hop_length=self.hop_length					,power=2.0					,mel_filters=lowerCamelCase					,log_mel="""dB"""					,)
      return log_mel_spectrogram.T





   def   UpperCAmelCase__  (       self					:      int					,lowerCamelCase					:      int					,lowerCamelCase					:      Union[str, Any]					,lowerCamelCase					:      int     ):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE   				=							np.array_split(list(range(0					,total_frames - chunk_frames + 1     )     )					,3     )
      if len(ranges[1]     ) == 0:
         # if the audio is too short, we just use the first chunk
         __SCREAMING_SNAKE_CASE   				=							[0]
      if len(ranges[2]     ) == 0:
         # if the audio is too short, we just use the first chunk
         __SCREAMING_SNAKE_CASE   				=							[0]
      # randomly choose index for each part
      __SCREAMING_SNAKE_CASE   				=							np.random.choice(ranges[0]     )
      __SCREAMING_SNAKE_CASE   				=							np.random.choice(ranges[1]     )
      __SCREAMING_SNAKE_CASE   				=							np.random.choice(ranges[2]     )

      __SCREAMING_SNAKE_CASE   				=							mel[idx_front : idx_front + chunk_frames, :]
      __SCREAMING_SNAKE_CASE   				=							mel[idx_middle : idx_middle + chunk_frames, :]
      __SCREAMING_SNAKE_CASE   				=							mel[idx_back : idx_back + chunk_frames, :]

      __SCREAMING_SNAKE_CASE   				=							torch.tensor(mel[None, None, :]     )
      __SCREAMING_SNAKE_CASE   				=							torch.nn.functional.interpolate(
          lowerCamelCase					,size=[chunk_frames, 64]					,mode="""bilinear"""					,align_corners=lowerCamelCase     )
      __SCREAMING_SNAKE_CASE   				=							mel_shrink[0][0].numpy()
      __SCREAMING_SNAKE_CASE   				=							np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]					,axis=0     )
      return mel_fusion





   def   UpperCAmelCase__  (       self					:      Optional[Any]					,lowerCamelCase					:      np.array					,lowerCamelCase					:      str					,lowerCamelCase					:      Tuple					,lowerCamelCase					:      Optional[Any]     ):

      '''simple docstring'''

      if waveform.shape[0] > max_length:
         if truncation == "rand_trunc":
            __SCREAMING_SNAKE_CASE   				=							True
            # random crop to max_length (for compatibility) -> this should be handled by self.pad
            __SCREAMING_SNAKE_CASE   				=							len(lowerCamelCase     ) - max_length
            __SCREAMING_SNAKE_CASE   				=							np.random.randint(0					,overflow + 1     )
            __SCREAMING_SNAKE_CASE   				=							waveform[idx : idx + max_length]
            __SCREAMING_SNAKE_CASE   				=							self._np_extract_fbank_features(lowerCamelCase					,self.mel_filters_slaney     )[None, :]
         elif truncation == "fusion":
            __SCREAMING_SNAKE_CASE   				=							self._np_extract_fbank_features(lowerCamelCase					,self.mel_filters     )
            __SCREAMING_SNAKE_CASE   				=							max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
            __SCREAMING_SNAKE_CASE   				=							mel.shape[0]
            if chunk_frames == total_frames:
               # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
               # In this case, we just use the whole audio.
               __SCREAMING_SNAKE_CASE   				=							np.stack([mel, mel, mel, mel]					,axis=0     )
               __SCREAMING_SNAKE_CASE   				=							False
            else:
               __SCREAMING_SNAKE_CASE   				=							self._random_mel_fusion(lowerCamelCase					,lowerCamelCase					,lowerCamelCase     )
               __SCREAMING_SNAKE_CASE   				=							True
         else:
            raise NotImplementedError(f"""data_truncating {truncation} not implemented"""     )

      else:
         __SCREAMING_SNAKE_CASE   				=							False
         # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
         if waveform.shape[0] < max_length:
            if padding == "repeat":
               __SCREAMING_SNAKE_CASE   				=							int(max_length / len(lowerCamelCase     )     )
               __SCREAMING_SNAKE_CASE   				=							np.stack(np.tile(lowerCamelCase					,n_repeat + 1     )     )[:max_length]
            if padding == "repeatpad":
               __SCREAMING_SNAKE_CASE   				=							int(max_length / len(lowerCamelCase     )     )
               __SCREAMING_SNAKE_CASE   				=							np.stack(np.tile(lowerCamelCase					,lowerCamelCase     )     )
            __SCREAMING_SNAKE_CASE   				=							np.pad(lowerCamelCase					,(0, max_length - waveform.shape[0])					,mode="""constant"""					,constant_values=0     )

         if truncation == "fusion":
            __SCREAMING_SNAKE_CASE   				=							self._np_extract_fbank_features(lowerCamelCase					,self.mel_filters     )
            __SCREAMING_SNAKE_CASE   				=							np.stack([input_mel, input_mel, input_mel, input_mel]					,axis=0     )
         else:
            __SCREAMING_SNAKE_CASE   				=							self._np_extract_fbank_features(lowerCamelCase					,self.mel_filters_slaney     )[None, :]

      return input_mel, longer





   def __call__(       self					:      Optional[Any]					,lowerCamelCase					:      Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]					,lowerCamelCase					:      str = None					,lowerCamelCase					:      Optional[str] = None					,lowerCamelCase					:      Optional[int] = None					,lowerCamelCase					:      Optional[int] = None					,lowerCamelCase					:      Optional[Union[str, TensorType]] = None					,**lowerCamelCase					:      str					,):

      '''simple docstring'''

      __SCREAMING_SNAKE_CASE   				=							truncation if truncation is not None else self.truncation
      __SCREAMING_SNAKE_CASE   				=							padding if padding else self.padding

      if sampling_rate is not None:
         if sampling_rate != self.sampling_rate:
            raise ValueError(
                f"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
                f""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
                f""" was sampled with {self.sampling_rate} and not {sampling_rate}."""     )
      else:
         logger.warning(
             """It is strongly recommended to pass the `sampling_rate` argument to this function. """
             """Failing to do so can result in silent errors that might be hard to debug."""     )

      __SCREAMING_SNAKE_CASE   				=							isinstance(lowerCamelCase					,np.ndarray     ) and len(raw_speech.shape     ) > 1
      if is_batched_numpy and len(raw_speech.shape     ) > 2:
         raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""     )
      __SCREAMING_SNAKE_CASE   				=							is_batched_numpy or (
          isinstance(lowerCamelCase					,(list, tuple)     ) and (isinstance(raw_speech[0]					,(np.ndarray, tuple, list)     ))
      )

      if is_batched:
         __SCREAMING_SNAKE_CASE   				=							[np.asarray(lowerCamelCase					,dtype=np.floataa     ) for speech in raw_speech]
      elif not is_batched and not isinstance(lowerCamelCase					,np.ndarray     ):
         __SCREAMING_SNAKE_CASE   				=							np.asarray(lowerCamelCase					,dtype=np.floataa     )
      elif isinstance(lowerCamelCase					,np.ndarray     ) and raw_speech.dtype is np.dtype(np.floataa     ):
         __SCREAMING_SNAKE_CASE   				=							raw_speech.astype(np.floataa     )

      # always return batch
      if not is_batched:
         __SCREAMING_SNAKE_CASE   				=							[np.asarray(lowerCamelCase     )]

      # convert to mel spectrogram, truncate and pad if needed.
      __SCREAMING_SNAKE_CASE   				=							[
          self._get_input_mel(lowerCamelCase					,max_length if max_length else self.nb_max_samples					,lowerCamelCase					,lowerCamelCase     )
          for waveform in raw_speech
      ]

      __SCREAMING_SNAKE_CASE   				=							[]
      __SCREAMING_SNAKE_CASE   				=							[]
      for mel, longer in padded_inputs:
         input_mel.append(lowerCamelCase     )
         is_longer.append(lowerCamelCase     )

      if truncation == "fusion" and sum(lowerCamelCase     ) == 0:
         # if no audio is longer than 10s, then randomly select one audio to be longer
         __SCREAMING_SNAKE_CASE   				=							np.random.randint(0					,len(lowerCamelCase     )     )
         __SCREAMING_SNAKE_CASE   				=							True

      if isinstance(input_mel[0]					,lowerCamelCase     ):
         __SCREAMING_SNAKE_CASE   				=							[np.asarray(lowerCamelCase					,dtype=np.floataa     ) for feature in input_mel]

      # is_longer is a list of bool
      __SCREAMING_SNAKE_CASE   				=							[[longer] for longer in is_longer]

      __SCREAMING_SNAKE_CASE   				=							{"""input_features""": input_mel, """is_longer""": is_longer}
      __SCREAMING_SNAKE_CASE   				=							BatchFeature(lowerCamelCase     )

      if return_tensors is not None:
         __SCREAMING_SNAKE_CASE   				=							input_features.convert_to_tensors(lowerCamelCase     )

      return input_features





