





"""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


lowerCamelCase__ :       Union[str, Any]						     =			logging.get_logger(__name__)




class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	UpperCamelCase				=  ["""input_features""", """is_longer"""]





	def __init__(						self    :Any		, lowerCamelCase_    :Union[str, Any]=64		, lowerCamelCase_    :int=4_80_00		, lowerCamelCase_    :int=4_80		, lowerCamelCase_    :Any=10		, lowerCamelCase_    :Union[str, Any]=10_24		, lowerCamelCase_    :List[Any]=0.0		, lowerCamelCase_    :Union[str, Any]=False		, lowerCamelCase_    :float = 0		, lowerCamelCase_    :float = 1_40_00		, lowerCamelCase_    :int = None		, lowerCamelCase_    :str = "fusion"		, lowerCamelCase_    :str = "repeatpad"		, **lowerCamelCase_    :Any		, )							->			int:







			'''simple docstring'''




			super().__init__(
			    feature_size=lowerCamelCase_		, sampling_rate=lowerCamelCase_		, padding_value=lowerCamelCase_		, return_attention_mask=lowerCamelCase_		, **lowerCamelCase_		, )
			SCREAMING_SNAKE_CASE       :     List[str]				= top_db
			SCREAMING_SNAKE_CASE       :     Dict				= truncation
			SCREAMING_SNAKE_CASE       :     Dict				= padding
			SCREAMING_SNAKE_CASE       :     List[str]				= fft_window_size
			SCREAMING_SNAKE_CASE       :     Any				= (fft_window_size >> 1) + 1
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= hop_length
			SCREAMING_SNAKE_CASE       :     Optional[int]				= max_length_s
			SCREAMING_SNAKE_CASE       :     Any				= max_length_s * sampling_rate
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= sampling_rate
			SCREAMING_SNAKE_CASE       :     Any				= frequency_min
			SCREAMING_SNAKE_CASE       :     Dict				= frequency_max
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= 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       :     Tuple				= 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   __lowerCAmelCase (						self    :Dict					)							->			Tuple:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Any				= copy.deepcopy(self.__dict__					)
			SCREAMING_SNAKE_CASE       :     Any				= 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   __lowerCAmelCase (						self    :Any		, lowerCamelCase_    :np.array		, lowerCamelCase_    :Optional[np.array] = None					)							->			int:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= 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   __lowerCAmelCase (						self    :Any		, lowerCamelCase_    :Dict		, lowerCamelCase_    :Dict		, lowerCamelCase_    :List[str]					)							->			Any:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= 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       :     List[Any]				= [0]
			if len(ranges[2]					) == 0:
					# if the audio is too short, we just use the first chunk
					SCREAMING_SNAKE_CASE       :     Optional[int]				= [0]
			# randomly choose index for each part
			SCREAMING_SNAKE_CASE       :     Dict				= np.random.choice(ranges[0]					)
			SCREAMING_SNAKE_CASE       :     Tuple				= np.random.choice(ranges[1]					)
			SCREAMING_SNAKE_CASE       :     Optional[int]				= np.random.choice(ranges[2]					)

			SCREAMING_SNAKE_CASE       :     Optional[int]				= mel[idx_front : idx_front + chunk_frames, :]
			SCREAMING_SNAKE_CASE       :     Optional[int]				= mel[idx_middle : idx_middle + chunk_frames, :]
			SCREAMING_SNAKE_CASE       :     Optional[int]				= mel[idx_back : idx_back + chunk_frames, :]

			SCREAMING_SNAKE_CASE       :     List[Any]				= torch.tensor(mel[None, None, :]					)
			SCREAMING_SNAKE_CASE       :     List[str]				= torch.nn.functional.interpolate(
			    lowerCamelCase_		, size=[chunk_frames, 64]		, mode='''bilinear'''		, align_corners=lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     List[str]				= mel_shrink[0][0].numpy()
			SCREAMING_SNAKE_CASE       :     Any				= np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]		, axis=0					)
			return mel_fusion





	def   __lowerCAmelCase (						self    :List[Any]		, lowerCamelCase_    :np.array		, lowerCamelCase_    :List[str]		, lowerCamelCase_    :int		, lowerCamelCase_    :Optional[int]					)							->			Dict:







			'''simple docstring'''




			if waveform.shape[0] > max_length:
					if truncation == "rand_trunc":
							SCREAMING_SNAKE_CASE       :     Optional[Any]				= True
							# random crop to max_length (for compatibility) -> this should be handled by self.pad
							SCREAMING_SNAKE_CASE       :     Optional[int]				= len(lowerCamelCase_					) - max_length
							SCREAMING_SNAKE_CASE       :     int				= np.random.randint(0		, overflow + 1					)
							SCREAMING_SNAKE_CASE       :     Union[str, Any]				= waveform[idx : idx + max_length]
							SCREAMING_SNAKE_CASE       :     Union[str, Any]				= self._np_extract_fbank_features(lowerCamelCase_		, self.mel_filters_slaney					)[None, :]
					elif truncation == "fusion":
							SCREAMING_SNAKE_CASE       :     Dict				= self._np_extract_fbank_features(lowerCamelCase_		, self.mel_filters					)
							SCREAMING_SNAKE_CASE       :     List[str]				= max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
							SCREAMING_SNAKE_CASE       :     Union[str, Any]				= 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       :     List[Any]				= np.stack([mel, mel, mel, mel]		, axis=0					)
									SCREAMING_SNAKE_CASE       :     Any				= False
							else:
									SCREAMING_SNAKE_CASE       :     Union[str, Any]				= self._random_mel_fusion(lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_					)
									SCREAMING_SNAKE_CASE       :     Tuple				= True
					else:
							raise NotImplementedError(f"data_truncating {truncation} not implemented"					)

			else:
					SCREAMING_SNAKE_CASE       :     Dict				= 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       :     Optional[Any]				= int(max_length / len(lowerCamelCase_					)					)
									SCREAMING_SNAKE_CASE       :     Any				= np.stack(np.tile(lowerCamelCase_		, n_repeat + 1					)					)[:max_length]
							if padding == "repeatpad":
									SCREAMING_SNAKE_CASE       :     str				= int(max_length / len(lowerCamelCase_					)					)
									SCREAMING_SNAKE_CASE       :     Tuple				= np.stack(np.tile(lowerCamelCase_		, lowerCamelCase_					)					)
							SCREAMING_SNAKE_CASE       :     Optional[Any]				= np.pad(lowerCamelCase_		, (0, max_length - waveform.shape[0])		, mode='''constant'''		, constant_values=0					)

					if truncation == "fusion":
							SCREAMING_SNAKE_CASE       :     Optional[int]				= self._np_extract_fbank_features(lowerCamelCase_		, self.mel_filters					)
							SCREAMING_SNAKE_CASE       :     Tuple				= np.stack([input_mel, input_mel, input_mel, input_mel]		, axis=0					)
					else:
							SCREAMING_SNAKE_CASE       :     str				= self._np_extract_fbank_features(lowerCamelCase_		, self.mel_filters_slaney					)[None, :]

			return input_mel, longer







	def __call__(						self    :Union[str, 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_    :Optional[int]		, )							->			Optional[Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Any				= truncation if truncation is not None else self.truncation
			SCREAMING_SNAKE_CASE       :     Any				= 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       :     str				= 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       :     Optional[Any]				= is_batched_numpy or (
			    isinstance(lowerCamelCase_		, (list, tuple)					) and (isinstance(raw_speech[0]		, (np.ndarray, tuple, list)					))
			)

			if is_batched:
					SCREAMING_SNAKE_CASE       :     List[Any]				= [np.asarray(lowerCamelCase_		, dtype=np.floataa					) for speech in raw_speech]
			elif not is_batched and not isinstance(lowerCamelCase_		, np.ndarray					):
					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= np.asarray(lowerCamelCase_		, dtype=np.floataa					)
			elif isinstance(lowerCamelCase_		, np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
					SCREAMING_SNAKE_CASE       :     Dict				= raw_speech.astype(np.floataa					)

			# always return batch
			if not is_batched:
					SCREAMING_SNAKE_CASE       :     Optional[int]				= [np.asarray(lowerCamelCase_					)]

			# convert to mel spectrogram, truncate and pad if needed.
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= [
			    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       :     Any				= []
			SCREAMING_SNAKE_CASE       :     Any				= []
			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       :     str				= np.random.randint(0		, len(lowerCamelCase_					)					)
					SCREAMING_SNAKE_CASE       :     Any				= True

			if isinstance(input_mel[0]		, lowerCamelCase_					):
					SCREAMING_SNAKE_CASE       :     List[str]				= [np.asarray(lowerCamelCase_		, dtype=np.floataa					) for feature in input_mel]

			# is_longer is a list of bool
			SCREAMING_SNAKE_CASE       :     str				= [[longer] for longer in is_longer]

			SCREAMING_SNAKE_CASE       :     str				= {'input_features': input_mel, 'is_longer': is_longer}
			SCREAMING_SNAKE_CASE       :     Dict				= BatchFeature(lowerCamelCase_					)

			if return_tensors is not None:
					SCREAMING_SNAKE_CASE       :     Tuple				= input_features.convert_to_tensors(lowerCamelCase_					)

			return input_features







"""simple docstring"""






import json
import os
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCamelCase__ :       List[str]						     =			logging.get_logger(__name__)

lowerCamelCase__ :       Optional[int]						     =			{"vocab_file": "vocab.json"}

lowerCamelCase__ :       Dict						     =			{
    "vocab_file": {
        "mgp-str": "https://huggingface.co/alibaba-damo/mgp-str-base/blob/main/vocab.json",
    }
}

lowerCamelCase__ :       Optional[Any]						     =			{"mgp-str": 27}




class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	UpperCamelCase				=  VOCAB_FILES_NAMES
	UpperCamelCase				=  PRETRAINED_VOCAB_FILES_MAP
	UpperCamelCase				=  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES





	def __init__(						self    :int		, lowerCamelCase_    :Optional[int]		, lowerCamelCase_    :Optional[int]="[GO]"		, lowerCamelCase_    :int="[GO]"		, lowerCamelCase_    :str="[s]"		, lowerCamelCase_    :Dict="[GO]"		, **lowerCamelCase_    :List[str]					)							->			Tuple:







			'''simple docstring'''




			super().__init__(
			    unk_token=lowerCamelCase_		, bos_token=lowerCamelCase_		, eos_token=lowerCamelCase_		, pad_token=lowerCamelCase_		, **lowerCamelCase_		, )

			with open(lowerCamelCase_		, encoding='''utf-8'''					) as vocab_handle:
					SCREAMING_SNAKE_CASE       :     int				= json.load(lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= {v: k for k, v in self.vocab.items()}





	@property
	def   __lowerCAmelCase (						self    :int					)							->			Dict:







			'''simple docstring'''




			return len(self.vocab					)





	def   __lowerCAmelCase (						self    :List[str]					)							->			Dict:







			'''simple docstring'''




			return dict(self.vocab		, **self.added_tokens_encoder					)





	def   __lowerCAmelCase (						self    :List[str]		, lowerCamelCase_    :Optional[Any]					)							->			List[str]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= []
			for s in text:
					char_tokens.extend(lowerCamelCase_					)
			return char_tokens





	def   __lowerCAmelCase (						self    :Dict		, lowerCamelCase_    :Tuple					)							->			Optional[int]:







			'''simple docstring'''




			return self.vocab.get(lowerCamelCase_		, self.vocab.get(self.unk_token					)					)





	def   __lowerCAmelCase (						self    :List[str]		, lowerCamelCase_    :Dict					)							->			Optional[int]:







			'''simple docstring'''




			return self.decoder.get(lowerCamelCase_					)







	def   __lowerCAmelCase (						self    :Optional[int]		, lowerCamelCase_    :str		, lowerCamelCase_    :Optional[str] = None					)							->			Tuple[str]:







			'''simple docstring'''




			if not os.path.isdir(lowerCamelCase_					):
					logger.error('''Vocabulary path ({}) should be a directory'''.format(lowerCamelCase_					)					)
					return
			SCREAMING_SNAKE_CASE       :     List[Any]				= os.path.join(
			    lowerCamelCase_		, (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']					)

			with open(lowerCamelCase_		, '''w'''		, encoding='''utf-8'''					) as f:
					f.write(json.dumps(self.vocab		, indent=2		, sort_keys=lowerCamelCase_		, ensure_ascii=lowerCamelCase_					) + '''\n'''					)

			return (vocab_file,)





