
from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


_SCREAMING_SNAKE_CASE     :   Any							 =      logging.get_logger(__name__)







class 			UpperCamelCase__		(     SCREAMING_SNAKE_CASE_ ):







	'''simple docstring'''
	_lowerCAmelCase											=					["input_values", "attention_mask"]





	def __init__(      self					,							UpperCamelCase__=1					,							UpperCamelCase__=1_6000					,							UpperCamelCase__=128					,							UpperCamelCase__=1024					,							UpperCamelCase__=0.0					,							UpperCamelCase__=True					,							UpperCamelCase__=-4.2_6_7_7_3_9_3					,							UpperCamelCase__=4.5_6_8_9_9_7_4					,							UpperCamelCase__=False					,							**UpperCamelCase__					,							):
					super().__init__(feature_size=UpperCamelCase__					,							sampling_rate=UpperCamelCase__					,							padding_value=UpperCamelCase__					,							**UpperCamelCase__		)
					A__	:						List[Any]								=				num_mel_bins
					A__	:						int								=				max_length
					A__	:						Tuple								=				do_normalize
					A__	:						str								=				mean
					A__	:						Optional[Any]								=				std
					A__	:						Union[str, Any]								=				return_attention_mask





	def 	__snake_case					(      self					,							UpperCamelCase__					,							UpperCamelCase__					,							):
					A__	:						Any								=				torch.from_numpy(UpperCamelCase__		).unsqueeze(0		)
					A__	:						Any								=				ta_kaldi.fbank(
					    UpperCamelCase__					,							htk_compat=UpperCamelCase__					,							sample_frequency=self.sampling_rate					,							use_energy=UpperCamelCase__					,							window_type='''hanning'''					,							num_mel_bins=self.num_mel_bins					,							dither=0.0					,							frame_shift=10					,							)

					A__	:						List[str]								=				fbank.shape[0]
					A__	:						Tuple								=				max_length - n_frames

					# pad or truncate, depending on difference
					if difference > 0:
									A__	:						int								=				torch.nn.ZeroPadad((0, 0, 0, difference)		)
									A__	:						Union[str, Any]								=				pad_module(UpperCamelCase__		)
					elif difference < 0:
									A__	:						List[str]								=				fbank[0:max_length, :]

					A__	:						Any								=				fbank.numpy()

					return fbank





	def 	__snake_case					(      self					,							UpperCamelCase__		):
					return (input_values - (self.mean)) / (self.std * 2)





	def __call__(      self					,							UpperCamelCase__					,							UpperCamelCase__ = None					,							UpperCamelCase__ = None					,							**UpperCamelCase__					,							):

					if sampling_rate is not None:
									if sampling_rate != self.sampling_rate:
													raise ValueError(
													    F"The model corresponding to this feature extractor: {self} was trained using a sampling rate of"
													    F" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"
													    F" {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.'''		)

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

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

					# always return batch
					if not is_batched:
									A__	:						Union[str, Any]								=				[raw_speech]

					# extract fbank features and pad/truncate to max_length
					A__	:						Optional[Any]								=				[self._extract_fbank_features(UpperCamelCase__					,							max_length=self.max_length		) for waveform in raw_speech]

					# convert into BatchFeature
					A__	:						str								=				BatchFeature({'''input_values''': features}		)

					# make sure list is in array format
					A__	:						List[Any]								=				padded_inputs.get('''input_values'''		)
					if isinstance(input_values[0]					,							UpperCamelCase__		):
									A__	:						str								=				[np.asarray(UpperCamelCase__					,							dtype=np.floataa		) for feature in input_values]

					# normalization
					if self.do_normalize:
									A__	:						List[str]								=				[self.normalize(UpperCamelCase__		) for feature in input_values]

					if return_tensors is not None:
									A__	:						Any								=				padded_inputs.convert_to_tensors(UpperCamelCase__		)

					return padded_inputs

import os
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device


if is_torch_available():
					import torch

					from transformers import GPTaLMHeadModel







@require_torch
@require_sentencepiece
@require_tokenizers
class 			UpperCamelCase__		(     unittest.TestCase ):







	'''simple docstring'''





	@slow
	@unittest.skip('''Model is not available.'''		)
	def 	__snake_case					(      self		):
					A__	:						int								=				'''nvidia/megatron-gpt2-345m/'''
					if "MYDIR" in os.environ:
									A__	:						Union[str, Any]								=				os.path.join(os.environ['''MYDIR''']					,							UpperCamelCase__		)
					A__	:						Union[str, Any]								=				GPTaLMHeadModel.from_pretrained(UpperCamelCase__		)
					model.to(UpperCamelCase__		)
					model.half()

					A__	:						Dict								=				torch.tensor(
					    [[101, 7110, 1005, 1056, 2023, 1_1333, 1_7413, 1029, 102]]					,							device=UpperCamelCase__					,							dtype=torch.long					,							)

					with torch.no_grad():
									A__	:						str								=				model(UpperCamelCase__		).logits

					A__	:						int								=				torch.Size((1, 9, 5_0257)		)
					self.assertEqual(output.shape					,							UpperCamelCase__		)

					A__	:						Dict								=				torch.tensor(
					    [
					        4.9_4_1_4,
					        -0.2_9_2_0,
					        -1.2_1_4_8,
					        -4.0_2_7_3,
					        -0.5_1_6_1,
					        -5.2_1_0_9,
					        -1.2_4_1_2,
					        -1.8_3_0_1,
					        -1.7_7_3_4,
					        -4.7_1_4_8,
					        -0.2_3_1_7,
					        -1.0_8_1_1,
					        -2.1_7_7_7,
					        0.4_1_4_1,
					        -3.7_9_6_9,
					        -4.0_5_8_6,
					        -2.5_3_3_2,
					        -3.3_8_0_9,
					        4.3_8_6_7,
					    ]					,							device=UpperCamelCase__					,							dtype=torch.half					,							)

					for i in range(19		):
									A__							,				A__	:						int								=				8 * i // 17, 2792 * i  # along the diagonal
									A__							,				A__	:						Optional[Any]								=				output[0, r, c], expected_diag[i]
									A__	:						Dict								=				F"row={r} col={c} computed={computed} expected={expected}"
									self.assertAlmostEqual(UpperCamelCase__					,							UpperCamelCase__					,							delta=1e-4					,							msg=UpperCamelCase__		)