






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 PaddingStrategy, TensorType, logging


__lowerCamelCase   :					str   							=					logging.get_logger(__name__)







class    A_	(a_							):
							"""simple docstring"""




							a__   		=	['''input_features''', '''attention_mask''']

							def __init__(		self			:Optional[Any]       ,							lowerCAmelCase__			:str=80       ,							lowerCAmelCase__			:Optional[Any]=16_000       ,							lowerCAmelCase__			:List[str]=80       ,							lowerCAmelCase__			:str=0.0       ,							lowerCAmelCase__			:int=True       ,							lowerCAmelCase__			:int=True       ,							lowerCAmelCase__			:Union[str, Any]=True       ,							**lowerCAmelCase__			:Tuple       ,							)				->  List[str]:


										'''simple docstring'''




										super().__init__(feature_size=lowerCAmelCase__       ,							sampling_rate=lowerCAmelCase__       ,							padding_value=lowerCAmelCase__       ,							**lowerCAmelCase__				)
										snake_case_				:					List[str]   =       num_mel_bins
										snake_case_				:					Any   =       do_ceptral_normalize
										snake_case_				:					Tuple   =       normalize_means
										snake_case_				:					Optional[int]   =       normalize_vars
										snake_case_				:					Dict   =       True

							def 							_A	(		self			:List[Any]       ,							lowerCAmelCase__			:np.ndarray       ,							)				->  np.ndarray:


										'''simple docstring'''




										snake_case_				:					Optional[int]   =       waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
										snake_case_				:					Dict   =       torch.from_numpy(lowerCAmelCase__				).unsqueeze(0				)
										snake_case_				:					Union[str, Any]   =       ta_kaldi.fbank(lowerCAmelCase__       ,							num_mel_bins=self.num_mel_bins       ,							sample_frequency=self.sampling_rate				)
										return features.numpy()

							@staticmethod
							def 							_A	(		lowerCAmelCase__			:np.ndarray       ,							lowerCAmelCase__			:int       ,							lowerCAmelCase__			:Optional[bool] = True       ,							lowerCAmelCase__			:Optional[bool] = True       ,							lowerCAmelCase__			:float = 0.0       ,							)				->  np.ndarray:


										'''simple docstring'''




										if normalize_means:
													snake_case_				:					int   =       x[:input_length].mean(axis=0				)
													snake_case_				:					List[str]   =       np.subtract(lowerCAmelCase__       ,							lowerCAmelCase__				)
										if normalize_vars:
													snake_case_				:					Union[str, Any]   =       x[:input_length].std(axis=0				)
													snake_case_				:					int   =       np.divide(lowerCAmelCase__       ,							lowerCAmelCase__				)

										if input_length < x.shape[0]:
													snake_case_				:					Union[str, Any]   =       padding_value

										# make sure array is in float32
										snake_case_				:					Union[str, Any]   =       x.astype(np.floataa				)

										return x

							def 							_A	(		self			:Union[str, Any]       ,							lowerCAmelCase__			:List[np.ndarray]       ,							lowerCAmelCase__			:Optional[np.ndarray] = None				)				->  List[np.ndarray]:


										'''simple docstring'''




										snake_case_				:					int   =       attention_mask.sum(-1				) if attention_mask is not None else [x.shape[0] for x in input_features]
										return [
										    self.utterance_cmvn(lowerCAmelCase__       ,							lowerCAmelCase__       ,							self.normalize_means       ,							self.normalize_vars       ,							self.padding_value				)
										    for x, n in zip(lowerCAmelCase__       ,							lowerCAmelCase__				)
										]






							def __call__(		self			:Union[str, Any]       ,							lowerCAmelCase__			:Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]       ,							lowerCAmelCase__			:Union[bool, str, PaddingStrategy] = False       ,							lowerCAmelCase__			:Optional[int] = None       ,							lowerCAmelCase__			:bool = False       ,							lowerCAmelCase__			:Optional[int] = None       ,							lowerCAmelCase__			:Optional[Union[str, TensorType]] = None       ,							lowerCAmelCase__			:Optional[int] = None       ,							lowerCAmelCase__			:Optional[bool] = None       ,							**lowerCAmelCase__			:Optional[int]       ,							)				->  BatchFeature:


										'''simple docstring'''




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

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

										if is_batched:
													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				):
													snake_case_				:					Optional[int]   =       np.asarray(lowerCAmelCase__       ,							dtype=np.floataa				)
										elif isinstance(lowerCAmelCase__       ,							np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
													snake_case_				:					Optional[int]   =       raw_speech.astype(np.floataa				)

										# always return batch
										if not is_batched:
													snake_case_				:					Tuple   =       [raw_speech]

										# extract fbank features
										snake_case_				:					int   =       [self._extract_fbank_features(lowerCAmelCase__				) for waveform in raw_speech]

										# convert into correct format for padding
										snake_case_				:					int   =       BatchFeature({"input_features": features}				)

										snake_case_				:					Optional[int]   =       self.pad(
										    lowerCAmelCase__       ,							padding=lowerCAmelCase__       ,							max_length=lowerCAmelCase__       ,							truncation=lowerCAmelCase__       ,							pad_to_multiple_of=lowerCAmelCase__       ,							return_attention_mask=lowerCAmelCase__       ,							**lowerCAmelCase__       ,							)

										# make sure list is in array format
										snake_case_				:					str   =       padded_inputs.get("input_features"				)
										if isinstance(input_features[0]       ,							lowerCAmelCase__				):
													snake_case_				:					Tuple   =       [np.asarray(lowerCAmelCase__       ,							dtype=np.floataa				) for feature in input_features]

										snake_case_				:					Any   =       padded_inputs.get("attention_mask"				)
										if attention_mask is not None:
													snake_case_				:					str   =       [np.asarray(lowerCAmelCase__       ,							dtype=np.intaa				) for array in attention_mask]

										# Utterance-level cepstral mean and variance normalization
										if self.do_ceptral_normalize:
													snake_case_				:					int   =       (
													    np.array(lowerCAmelCase__       ,							dtype=np.intaa				)
													    if self._get_padding_strategies(lowerCAmelCase__       ,							max_length=lowerCAmelCase__				) is not PaddingStrategy.DO_NOT_PAD
													    else None
													)
													snake_case_				:					Dict   =       self.normalize(
													    padded_inputs["input_features"]       ,							attention_mask=lowerCAmelCase__				)

										if return_tensors is not None:
													snake_case_				:					List[str]   =       padded_inputs.convert_to_tensors(lowerCAmelCase__				)

										return padded_inputs












'''simple docstring'''




import json
import os
import unittest

from transformers import MgpstrTokenizer
from transformers.models.mgp_str.tokenization_mgp_str import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers

from ...test_tokenization_common import TokenizerTesterMixin







@require_tokenizers
class    A_	(a_						,					unittest.TestCase							):
							"""simple docstring"""




							a__   		=	MgpstrTokenizer
							a__   		=	False
							a__   		=	{}
							a__   		=	False

							def 							_A	(		self			:List[str]				)				->  List[str]:


										'''simple docstring'''




										super().setUp()

										# fmt: off
										snake_case_				:					Dict   =       ["[GO]", "[s]", "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x", "y", "z"]
										# fmt: on
										snake_case_				:					List[str]   =       dict(zip(lowerCAmelCase__       ,							range(len(lowerCAmelCase__				)				)				)				)

										snake_case_				:					Tuple   =       os.path.join(self.tmpdirname       ,							VOCAB_FILES_NAMES["vocab_file"]				)
										with open(self.vocab_file       ,							"w"       ,							encoding="utf-8"				) as fp:
													fp.write(json.dumps(lowerCAmelCase__				) + "\n"				)

							def 							_A	(		self			:Optional[Any]       ,							**lowerCAmelCase__			:Optional[Any]				)				->  Dict:


										'''simple docstring'''




										return MgpstrTokenizer.from_pretrained(self.tmpdirname       ,							**lowerCAmelCase__				)

							def 							_A	(		self			:Dict       ,							lowerCAmelCase__			:Any				)				->  str:


										'''simple docstring'''




										snake_case_				:					Dict   =       "tester"
										snake_case_				:					Tuple   =       "tester"
										return input_text, output_text

							@unittest.skip("MGP-STR always lower cases letters."				)
							def 							_A	(		self			:Dict				)				->  str:


										'''simple docstring'''




										pass

							def 							_A	(		self			:Tuple				)				->  Union[str, Any]:


										'''simple docstring'''




										snake_case_				:					List[str]   =       self.get_tokenizers(do_lower_case=lowerCAmelCase__				)
										for tokenizer in tokenizers:
													with self.subTest(F'''{tokenizer.__class__.__name__}'''				):
																snake_case_				:					Tuple   =       "[SPECIAL_TOKEN]"

																tokenizer.add_special_tokens({"cls_token": special_token}				)
																snake_case_				:					str   =       tokenizer.encode([special_token]       ,							add_special_tokens=lowerCAmelCase__				)
																self.assertEqual(len(lowerCAmelCase__				)       ,							1				)

																snake_case_				:					Tuple   =       tokenizer.decode(lowerCAmelCase__       ,							skip_special_tokens=lowerCAmelCase__				)
																self.assertTrue(special_token not in decoded				)

							def 							_A	(		self			:int				)				->  List[str]:


										'''simple docstring'''




										snake_case_				:					Dict   =       self.get_tokenizers()
										for tokenizer in tokenizers:
													with self.subTest(F'''{tokenizer.__class__.__name__}'''				):
																snake_case_,	snake_case_				:					str   =       self.get_input_output_texts(lowerCAmelCase__				)

																snake_case_				:					Union[str, Any]   =       tokenizer.tokenize(lowerCAmelCase__				)
																snake_case_				:					List[Any]   =       tokenizer.convert_tokens_to_ids(lowerCAmelCase__				)
																snake_case_				:					Dict   =       tokenizer.encode(lowerCAmelCase__       ,							add_special_tokens=lowerCAmelCase__				)
																self.assertListEqual(lowerCAmelCase__       ,							lowerCAmelCase__				)

																snake_case_				:					List[str]   =       tokenizer.convert_ids_to_tokens(lowerCAmelCase__				)
																self.assertNotEqual(len(lowerCAmelCase__				)       ,							0				)
																snake_case_				:					List[str]   =       tokenizer.decode(lowerCAmelCase__				)
																self.assertIsInstance(lowerCAmelCase__       ,							lowerCAmelCase__				)

																self.assertEqual(text_a.replace(" "       ,							""				)       ,							lowerCAmelCase__				)

							@unittest.skip("MGP-STR tokenizer only handles one sequence."				)
							def 							_A	(		self			:Union[str, Any]				)				->  Any:


										'''simple docstring'''




										pass






							@unittest.skip("inputs cannot be pretokenized in MgpstrTokenizer"				)
							def 							_A	(		self			:int				)				->  Dict:


										'''simple docstring'''




										pass




