



"""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    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       ['''input_features''', '''is_longer''']





		def __init__(					self							,    _lowercase=6_4							,    _lowercase=4_8_0_0_0							,    _lowercase=4_8_0							,    _lowercase=1_0							,    _lowercase=1_0_2_4							,    _lowercase=0.0							,    _lowercase=False							,    _lowercase = 0							,    _lowercase = 1_4_0_0_0							,    _lowercase = None							,    _lowercase = "fusion"							,    _lowercase = "repeatpad"							,    **_lowercase							,    ) ->						Dict:







								'''simple docstring'''

								super().__init__(
								    feature_size=_lowercase							,    sampling_rate=_lowercase							,    padding_value=_lowercase							,    return_attention_mask=_lowercase							,    **_lowercase							,    )
								snake_case_   : Optional[Any]						=			top_db
								snake_case_   : int						=			truncation
								snake_case_   : Tuple						=			padding
								snake_case_   : Any						=			fft_window_size
								snake_case_   : Optional[int]						=			(fft_window_size >> 1) + 1
								snake_case_   : Dict						=			hop_length
								snake_case_   : List[str]						=			max_length_s
								snake_case_   : str						=			max_length_s * sampling_rate
								snake_case_   : str						=			sampling_rate
								snake_case_   : Dict						=			frequency_min
								snake_case_   : Optional[Any]						=			frequency_max
								snake_case_   : Any						=			mel_filter_bank(
								    num_frequency_bins=self.nb_frequency_bins							,    num_mel_filters=_lowercase							,    min_frequency=_lowercase							,    max_frequency=_lowercase							,    sampling_rate=_lowercase							,    norm=_lowercase							,    mel_scale="""htk"""							,    )
								snake_case_   : Union[str, Any]						=			mel_filter_bank(
								    num_frequency_bins=self.nb_frequency_bins							,    num_mel_filters=_lowercase							,    min_frequency=_lowercase							,    max_frequency=_lowercase							,    sampling_rate=_lowercase							,    norm="""slaney"""							,    mel_scale="""slaney"""							,    )





		def    UpperCAmelCase__							(					self   ) ->						Dict[str, Any]:







								'''simple docstring'''

								snake_case_   : Optional[Any]						=			copy.deepcopy(self.__dict__   )
								snake_case_   : Union[str, 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    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase = None   ) ->						np.ndarray:







								'''simple docstring'''

								snake_case_   : str						=			spectrogram(
								    _lowercase							,    window_function(self.fft_window_size							,    """hann"""   )							,    frame_length=self.fft_window_size							,    hop_length=self.hop_length							,    power=2.0							,    mel_filters=_lowercase							,    log_mel="""dB"""							,    )
								return log_mel_spectrogram.T





		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase							,    _lowercase   ) ->						Any:







								'''simple docstring'''

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

								snake_case_   : Optional[Any]						=			mel[idx_front : idx_front + chunk_frames, :]
								snake_case_   : List[str]						=			mel[idx_middle : idx_middle + chunk_frames, :]
								snake_case_   : Optional[Any]						=			mel[idx_back : idx_back + chunk_frames, :]

								snake_case_   : Optional[Any]						=			torch.tensor(mel[None, None, :]   )
								snake_case_   : Tuple						=			torch.nn.functional.interpolate(
								    _lowercase							,    size=[chunk_frames, 6_4]							,    mode="""bilinear"""							,    align_corners=_lowercase   )
								snake_case_   : Optional[Any]						=			mel_shrink[0][0].numpy()
								snake_case_   : Union[str, Any]						=			np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]							,    axis=0   )
								return mel_fusion





		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase							,    _lowercase							,    _lowercase   ) ->						np.array:







								'''simple docstring'''

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

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

														if truncation == "fusion":
																				snake_case_   : Dict						=			self._np_extract_fbank_features(_lowercase							,    self.mel_filters   )
																				snake_case_   : Union[str, Any]						=			np.stack([input_mel, input_mel, input_mel, input_mel]							,    axis=0   )
														else:
																				snake_case_   : int						=			self._np_extract_fbank_features(_lowercase							,    self.mel_filters_slaney   )[None, :]

								return input_mel, longer





		def __call__(					self							,    _lowercase							,    _lowercase = None							,    _lowercase = None							,    _lowercase = None							,    _lowercase = None							,    _lowercase = None							,    **_lowercase							,    ) ->						BatchFeature:







								'''simple docstring'''

								snake_case_   : Union[str, Any]						=			truncation if truncation is not None else self.truncation
								snake_case_   : Optional[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."""   )

								snake_case_   : Optional[Any]						=			isinstance(_lowercase							,    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}'   )
								snake_case_   : Optional[int]						=			is_batched_numpy or (
								    isinstance(_lowercase							,    (list, tuple)   ) and (isinstance(raw_speech[0]							,    (np.ndarray, tuple, list)   ))
								)

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

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

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

								snake_case_   : int						=			[]
								snake_case_   : str						=			[]
								for mel, longer in padded_inputs:
														input_mel.append(_lowercase   )
														is_longer.append(_lowercase   )

								if truncation == "fusion" and sum(_lowercase   ) == 0:
														# if no audio is longer than 10s, then randomly select one audio to be longer
														snake_case_   : Dict						=			np.random.randint(0							,    len(_lowercase   )   )
														snake_case_   : List[str]						=			True

								if isinstance(input_mel[0]							,    _lowercase   ):
														snake_case_   : int						=			[np.asarray(_lowercase							,    dtype=np.floataa   ) for feature in input_mel]

								# is_longer is a list of bool
								snake_case_   : Dict						=			[[longer] for longer in is_longer]

								snake_case_   : str						=			{"""input_features""": input_mel, """is_longer""": is_longer}
								snake_case_   : Tuple						=			BatchFeature(_lowercase   )

								if return_tensors is not None:
														snake_case_   : Any						=			input_features.convert_to_tensors(_lowercase   )

								return input_features






"""simple docstring"""





import unittest

from transformers import RoFormerTokenizer, RoFormerTokenizerFast
from transformers.testing_utils import require_rjieba, require_tokenizers

from ...test_tokenization_common import TokenizerTesterMixin



@require_rjieba
@require_tokenizers
class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__  ,				unittest.TestCase		):





		"""simple docstring"""






		_lowerCamelCase			  =       RoFormerTokenizer
		_lowerCamelCase			  =       RoFormerTokenizerFast
		_lowerCamelCase			  =       True
		_lowerCamelCase			  =       True





		def    UpperCAmelCase__							(					self   ) ->						Any:







								'''simple docstring'''

								super().setUp()





		def    UpperCAmelCase__							(					self							,    **_lowercase   ) ->						str:







								'''simple docstring'''

								return self.tokenizer_class.from_pretrained("""junnyu/roformer_chinese_base"""							,    **_lowercase   )





		def    UpperCAmelCase__							(					self							,    **_lowercase   ) ->						Union[str, Any]:







								'''simple docstring'''

								return self.rust_tokenizer_class.from_pretrained("""junnyu/roformer_chinese_base"""							,    **_lowercase   )





		def    UpperCAmelCase__							(					self   ) ->						Dict:







								'''simple docstring'''

								snake_case_   : Tuple						=			"""永和服装饰品有限公司,今天天气非常好"""
								snake_case_   : int						=			"""永和 服装 饰品 有限公司 , 今 天 天 气 非常 好"""
								return input_text, output_text





		def    UpperCAmelCase__							(					self   ) ->						str:







								'''simple docstring'''

								snake_case_   : List[str]						=			self.get_tokenizer()
								snake_case_			,		snake_case_   : Optional[Any]						=			self.get_chinese_input_output_texts()
								snake_case_   : List[str]						=			tokenizer.tokenize(_lowercase   )

								self.assertListEqual(_lowercase							,    output_text.split()   )

								snake_case_   : str						=			tokens + [tokenizer.unk_token]
								snake_case_   : Tuple						=			[2_2_9_4_3, 2_1_3_3_2, 3_4_4_3_1, 4_5_9_0_4, 1_1_7, 3_0_6, 1_2_3_1, 1_2_3_1, 2_6_5_3, 3_3_9_9_4, 1_2_6_6, 1_0_0]
								self.assertListEqual(tokenizer.convert_tokens_to_ids(_lowercase   )							,    _lowercase   )





		def    UpperCAmelCase__							(					self   ) ->						Any:







								'''simple docstring'''

								snake_case_   : int						=			self.get_rust_tokenizer()
								snake_case_			,		snake_case_   : List[Any]						=			self.get_chinese_input_output_texts()
								snake_case_   : Union[str, Any]						=			tokenizer.tokenize(_lowercase   )
								self.assertListEqual(_lowercase							,    output_text.split()   )
								snake_case_   : Optional[int]						=			tokens + [tokenizer.unk_token]
								snake_case_   : Union[str, Any]						=			[2_2_9_4_3, 2_1_3_3_2, 3_4_4_3_1, 4_5_9_0_4, 1_1_7, 3_0_6, 1_2_3_1, 1_2_3_1, 2_6_5_3, 3_3_9_9_4, 1_2_6_6, 1_0_0]
								self.assertListEqual(tokenizer.convert_tokens_to_ids(_lowercase   )							,    _lowercase   )





		def    UpperCAmelCase__							(					self   ) ->						Tuple:







								'''simple docstring'''

								pass





		def    UpperCAmelCase__							(					self   ) ->						Optional[Any]:







								'''simple docstring'''

								pass





		def    UpperCAmelCase__							(					self   ) ->						Optional[int]:







								'''simple docstring'''

								pass

