
"""simple docstring"""






from math import ceil
from typing import List, Optional, Union

import numpy as np

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


__UpperCAmelCase       =  logging.get_logger(__name__)




class 			_SCREAMING_SNAKE_CASE					(			A__							):
			UpperCAmelCase_  :Tuple				 =						["audio_values", "audio_mask"]






			def __init__(					self       ,   __A=2048       ,   __A=1       ,   __A=[16, 16]       ,   __A=128       ,   __A=4_4100       ,   __A=86       ,   __A=2048       ,   __A=0.0       ,   **__A       ,   )			->     int:
						super().__init__(
						    feature_size=__A       ,   sampling_rate=__A       ,   padding_value=__A       ,   **__A       ,   )

						lowerCAmelCase_    :str			     =  spectrogram_length
						lowerCAmelCase_    :Union[str, Any]			     =  num_channels
						lowerCAmelCase_    :List[Any]			     =  patch_size
						lowerCAmelCase_    :Dict			     =  feature_size // self.patch_size[1]
						lowerCAmelCase_    :int			     =  n_fft
						lowerCAmelCase_    :Optional[Any]			     =  sampling_rate // hop_length_to_sampling_rate
						lowerCAmelCase_    :List[Any]			     =  sampling_rate
						lowerCAmelCase_    :List[Any]			     =  padding_value
						lowerCAmelCase_    :Dict			     =  mel_filter_bank(
						    num_frequency_bins=1 + n_fft // 2       ,   num_mel_filters=__A       ,   min_frequency=0.0       ,   max_frequency=2_2_0_5_0.0       ,   sampling_rate=__A       ,   norm="""slaney"""       ,   mel_scale="""slaney"""       ,   ).T






			def  __lowerCAmelCase (					self       ,   __A       )			->     np.ndarray:
						lowerCAmelCase_    :Dict			     =  spectrogram(
						    __A       ,   window_function(self.n_fft       ,   """hann"""       )       ,   frame_length=self.n_fft       ,   hop_length=self.hop_length       ,   power=2.0       ,   mel_filters=self.mel_filters.T       ,   log_mel="""dB"""       ,   db_range=8_0.0       ,   )
						lowerCAmelCase_    :Dict			     =  log_spec[:, :-1]
						lowerCAmelCase_    :Union[str, Any]			     =  log_spec - 2_0.0
						lowerCAmelCase_    :Union[str, Any]			     =  np.clip(log_spec / 4_0.0       ,   -2.0       ,   0.0       ) + 1.0
						return log_spec






			def __call__(					self       ,   __A       ,   __A = None       ,   __A = True       ,   __A = None       ,   __A = False       ,   __A = False       ,   **__A       ,   )			->     BatchFeature:

						if sampling_rate is not None:
									if sampling_rate != self.sampling_rate:
												raise ValueError(
												    """This feature extractor is set to support sampling rate"""
												    f""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
												    f""" 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."""       )

						lowerCAmelCase_    :Tuple			     =  isinstance(__A       ,   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}"""       )
						lowerCAmelCase_    :Tuple			     =  is_batched_numpy or (
						    isinstance(__A       ,   (list, tuple)       ) and (isinstance(raw_speech[0]       ,   (np.ndarray, tuple, list)       ))
						)
						if is_batched:
									lowerCAmelCase_    :Tuple			     =  [np.asarray([speech]       ,   dtype=np.floataa       ).T for speech in raw_speech]
						elif not is_batched and not isinstance(__A       ,   np.ndarray       ):
									lowerCAmelCase_    :List[Any]			     =  np.asarray(__A       ,   dtype=np.floataa       )
						elif isinstance(__A       ,   np.ndarray       ) and raw_speech.dtype is np.dtype(np.floataa       ):
									lowerCAmelCase_    :Union[str, Any]			     =  raw_speech.astype(np.floataa       )
						# always return batch
						if not is_batched:
									lowerCAmelCase_    :Union[str, Any]			     =  [np.asarray([raw_speech]       ).T]

						# Convert audio signals to log mel spectrograms, truncate by time axis
						lowerCAmelCase_    :Optional[int]			     =  [
						    self._np_extract_fbank_features(waveform.squeeze()       ).T[: self.spectrogram_length] for waveform in raw_speech
						]
						if isinstance(audio_features[0]       ,   __A       ):
									lowerCAmelCase_    :Union[str, Any]			     =  [np.asarray(__A       ,   dtype=np.floataa       ) for feature in audio_features]

						# Create audio attention mask
						lowerCAmelCase_    :Dict			     =  max(
						    [ceil(feature.shape[0] / self.patch_size[0]       ) * self.freq_len for feature in audio_features]       )  # The maximum number of audio patches in a batch
						if return_attention_mask:
									lowerCAmelCase_    :Dict			     =  [
									    (ceil(feature.shape[0] / self.patch_size[0]       ) * self.freq_len) * [1]
									    + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]       ) * self.freq_len) * [0]
									    for feature in audio_features
									]
									lowerCAmelCase_    :str			     =  np.array(__A       ).astype(np.floataa       )

						# convert into correct format for padding
						lowerCAmelCase_    :Optional[Any]			     =  max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
						lowerCAmelCase_    :int			     =  np.ones([len(__A       ), 1, max_time_len, self.feature_size]       ).astype(np.floataa       )
						lowerCAmelCase_    :List[Any]			     =  padded_audio_features * self.padding_value
						for i in range(len(__A       )       ):
									lowerCAmelCase_    :Optional[Any]			     =  audio_features[i]
									lowerCAmelCase_    :str			     =  feature

						# return as BatchFeature
						if return_attention_mask:
									lowerCAmelCase_    :Tuple			     =  {"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
						else:
									lowerCAmelCase_    :Optional[int]			     =  {"""audio_values""": padded_audio_features}

						lowerCAmelCase_    :Union[str, Any]			     =  BatchFeature(data=__A       ,   tensor_type=__A       )
						return encoded_inputs







import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFGPTaLMHeadModel, is_keras_nlp_available, is_tf_available
from transformers.models.gpta.tokenization_gpta import GPTaTokenizer
from transformers.testing_utils import require_keras_nlp, require_tf, slow


if is_tf_available():
					import tensorflow as tf

if is_keras_nlp_available():
					from transformers.models.gpta import TFGPTaTokenizer


__UpperCamelCase		:       Optional[Any]									=     ['gpt2']
__UpperCamelCase		:       str									=     'gpt2'

if is_tf_available():





					class 					lowercase__						(   tf.Module):


											def __init__(       self	:				Optional[Any]    ,		UpperCamelCase__	:				Union[str, Any]						):





																	'''simple docstring'''





																	super().__init__()
																	SCREAMING_SNAKE_CASE							:    Dict							=							tokenizer
																	SCREAMING_SNAKE_CASE							:    int							=							AutoConfig.from_pretrained(UpperCamelCase__						)
																	SCREAMING_SNAKE_CASE							:    Dict							=							TFGPTaLMHeadModel.from_config(UpperCamelCase__						)



											@tf.function(input_signature=(tf.TensorSpec((None,)    ,		tf.string    ,		name='''text'''						),)						)
											def        __A       (       self	:				str    ,		UpperCamelCase__	:				int						):





																	'''simple docstring'''





																	SCREAMING_SNAKE_CASE							:    Optional[Any]							=							self.tokenizer(UpperCamelCase__						)
																	SCREAMING_SNAKE_CASE							:    Optional[Any]							=							tokenized['''input_ids'''].to_tensor()

																	SCREAMING_SNAKE_CASE							:    Any							=							tf.cast(input_ids_dense > 0    ,		tf.intaa						)
																	# input_mask = tf.reshape(input_mask, [-1, MAX_SEQ_LEN])

																	SCREAMING_SNAKE_CASE							:    List[Any]							=							self.model(input_ids=UpperCamelCase__    ,		attention_mask=UpperCamelCase__						)['''logits''']

																	return outputs





@require_tf
@require_keras_nlp
class 					lowercase__						(   unittest.TestCase):


						def        __A       (       self	:				int						):





												'''simple docstring'''





												super().setUp()

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							[GPTaTokenizer.from_pretrained(UpperCamelCase__						) for checkpoint in (TOKENIZER_CHECKPOINTS)]
												SCREAMING_SNAKE_CASE							:    List[str]							=							[TFGPTaTokenizer.from_pretrained(UpperCamelCase__						) for checkpoint in TOKENIZER_CHECKPOINTS]
												assert len(self.tokenizers						) == len(self.tf_tokenizers						)

												SCREAMING_SNAKE_CASE							:    Tuple							=							[
												    '''This is a straightforward English test sentence.''',
												    '''This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.''',
												    '''Now we\'re going to add some Chinese: 一 二 三 一二三''',
												    '''And some much more rare Chinese: 齉 堃 齉堃''',
												    '''Je vais aussi écrire en français pour tester les accents''',
												    '''Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ''',
												]
												SCREAMING_SNAKE_CASE							:    Dict							=							list(zip(self.test_sentences    ,		self.test_sentences[::-1]						)						)


						def        __A       (       self	:				str						):





												'''simple docstring'''





												for tokenizer, tf_tokenizer in zip(self.tokenizers    ,		self.tf_tokenizers						):
																		for test_inputs in self.test_sentences:
																								SCREAMING_SNAKE_CASE							:    Dict							=							tokenizer([test_inputs]    ,		return_tensors='''tf'''						)
																								SCREAMING_SNAKE_CASE							:    Any							=							tf_tokenizer([test_inputs]						)

																								for key in python_outputs.keys():
																														# convert them to numpy to avoid messing with ragged tensors
																														SCREAMING_SNAKE_CASE							:    int							=							python_outputs[key].numpy()
																														SCREAMING_SNAKE_CASE							:    int							=							tf_outputs[key].numpy()

																														self.assertTrue(tf.reduce_all(python_outputs_values.shape == tf_outputs_values.shape						)						)
																														self.assertTrue(tf.reduce_all(tf.cast(UpperCamelCase__    ,		tf.intaa						) == tf_outputs_values						)						)


						@slow
						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												for tf_tokenizer in self.tf_tokenizers:
																		SCREAMING_SNAKE_CASE							:    Optional[int]							=							tf.function(UpperCamelCase__						)
																		for test_inputs in self.test_sentences:
																								SCREAMING_SNAKE_CASE							:    Optional[int]							=							tf.constant(UpperCamelCase__						)
																								SCREAMING_SNAKE_CASE							:    List[str]							=							compiled_tokenizer(UpperCamelCase__						)
																								SCREAMING_SNAKE_CASE							:    List[Any]							=							tf_tokenizer(UpperCamelCase__						)

																								for key in eager_outputs.keys():
																														self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]						)						)


						@slow
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												for tf_tokenizer in self.tf_tokenizers:
																		SCREAMING_SNAKE_CASE							:    str							=							ModelToSave(tokenizer=UpperCamelCase__						)
																		SCREAMING_SNAKE_CASE							:    List[Any]							=							tf.convert_to_tensor([self.test_sentences[0]]						)
																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							model.serving(UpperCamelCase__						)  # Build model with some sample inputs
																		with TemporaryDirectory() as tempdir:
																								SCREAMING_SNAKE_CASE							:    List[str]							=							Path(UpperCamelCase__						) / '''saved.model'''
																								tf.saved_model.save(UpperCamelCase__    ,		UpperCamelCase__    ,		signatures={'''serving_default''': model.serving}						)
																								SCREAMING_SNAKE_CASE							:    str							=							tf.saved_model.load(UpperCamelCase__						)
																		SCREAMING_SNAKE_CASE							:    int							=							loaded_model.signatures['''serving_default'''](UpperCamelCase__						)['''output_0''']
																		# We may see small differences because the loaded model is compiled, so we need an epsilon for the test
																		self.assertTrue(tf.reduce_all(out == loaded_output						)						)


						@slow
						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												for tf_tokenizer in self.tf_tokenizers:
																		SCREAMING_SNAKE_CASE							:    List[Any]							=							tf.convert_to_tensor([self.test_sentences[0]]						)
																		SCREAMING_SNAKE_CASE							:    Tuple							=							tf_tokenizer(UpperCamelCase__						)  # Build model with some sample inputs

																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							tf_tokenizer.get_config()
																		SCREAMING_SNAKE_CASE							:    Optional[Any]							=							TFGPTaTokenizer.from_config(UpperCamelCase__						)
																		SCREAMING_SNAKE_CASE							:    str							=							model_from_config(UpperCamelCase__						)

																		for key in from_config_output.keys():
																								self.assertTrue(tf.reduce_all(from_config_output[key] == out[key]						)						)



						@slow
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												for tf_tokenizer in self.tf_tokenizers:
																		# for the test to run
																		SCREAMING_SNAKE_CASE							:    Tuple							=							12_3123

																		for max_length in [3, 5, 1024]:
																								SCREAMING_SNAKE_CASE							:    Dict							=							tf.convert_to_tensor([self.test_sentences[0]]						)
																								SCREAMING_SNAKE_CASE							:    Tuple							=							tf_tokenizer(UpperCamelCase__    ,		max_length=UpperCamelCase__						)

																								SCREAMING_SNAKE_CASE							:    Tuple							=							out['''input_ids'''].numpy().shape[1]

																								assert out_length == max_length





