



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__					  =  logging.get_logger(__name__)

class 				snake_case__(__SCREAMING_SNAKE_CASE      ):

					"""simple docstring"""






					lowercase_								=		["input_features", "attention_mask"]







					def __init__(							self :     Any				,       SCREAMING_SNAKE_CASE :     List[str]=80				,       SCREAMING_SNAKE_CASE :     Any=16_000				,       SCREAMING_SNAKE_CASE :     Any=80				,       SCREAMING_SNAKE_CASE :     Dict=0.0				,       SCREAMING_SNAKE_CASE :     List[str]=True				,       SCREAMING_SNAKE_CASE :     int=True				,       SCREAMING_SNAKE_CASE :     Union[str, Any]=True				,       **SCREAMING_SNAKE_CASE :     Any				,       ):
									super().__init__(feature_size=__UpperCAmelCase				,       sampling_rate=__UpperCAmelCase				,       padding_value=__UpperCAmelCase				,       **__UpperCAmelCase					)
									lowercase__					:   List[Any]													=					num_mel_bins
									lowercase__					:   List[str]													=					do_ceptral_normalize
									lowercase__					:   int													=					normalize_means
									lowercase__					:   str													=					normalize_vars
									lowercase__					:   Optional[Any]													=					True







					def 			snake_case					(							self :     Dict				,       SCREAMING_SNAKE_CASE :     int				,       ):
									lowercase__					:   Tuple													=					waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
									lowercase__					:   Dict													=					torch.from_numpy(__UpperCAmelCase					).unsqueeze(0					)
									lowercase__					:   Optional[Any]													=					ta_kaldi.fbank(__UpperCAmelCase				,       num_mel_bins=self.num_mel_bins				,       sample_frequency=self.sampling_rate					)
									return features.numpy()







					@staticmethod
					def 			snake_case					(							SCREAMING_SNAKE_CASE :     Any				,       SCREAMING_SNAKE_CASE :     Optional[int]				,       SCREAMING_SNAKE_CASE :     List[str] = True				,       SCREAMING_SNAKE_CASE :     Optional[Any] = True				,       SCREAMING_SNAKE_CASE :     Any = 0.0				,       ):
									# make sure we normalize float32 arrays
									if normalize_means:
													lowercase__					:   Optional[Any]													=					x[:input_length].mean(axis=0					)
													lowercase__					:   Optional[int]													=					np.subtract(__UpperCAmelCase				,       __UpperCAmelCase					)
									if normalize_vars:
													lowercase__					:   str													=					x[:input_length].std(axis=0					)
													lowercase__					:   Tuple													=					np.divide(__UpperCAmelCase				,       __UpperCAmelCase					)

									if input_length < x.shape[0]:
													lowercase__					:   Any													=					padding_value

									# make sure array is in float32
									lowercase__					:   Dict													=					x.astype(np.floataa					)

									return x







					def 			snake_case					(							self :     Optional[int]				,       SCREAMING_SNAKE_CASE :     Dict				,       SCREAMING_SNAKE_CASE :     List[Any] = None					):
									lowercase__					:   Optional[Any]													=					attention_mask.sum(-1					) if attention_mask is not None else [x.shape[0] for x in input_features]
									return [
									    self.utterance_cmvn(__UpperCAmelCase				,       __UpperCAmelCase				,       self.normalize_means				,       self.normalize_vars				,       self.padding_value					)
									    for x, n in zip(__UpperCAmelCase				,       __UpperCAmelCase					)
									]







					def __call__(							self :     Optional[Any]				,       SCREAMING_SNAKE_CASE :     List[str]				,       SCREAMING_SNAKE_CASE :     Tuple = False				,       SCREAMING_SNAKE_CASE :     str = None				,       SCREAMING_SNAKE_CASE :     List[str] = False				,       SCREAMING_SNAKE_CASE :     Optional[int] = None				,       SCREAMING_SNAKE_CASE :     Union[str, Any] = None				,       SCREAMING_SNAKE_CASE :     Optional[Any] = None				,       SCREAMING_SNAKE_CASE :     Any = None				,       **SCREAMING_SNAKE_CASE :     List[str]				,       ):

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

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

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

									# always return batch
									if not is_batched:
													lowercase__					:   List[Any]													=					[raw_speech]

									# extract fbank features
									lowercase__					:   Dict													=					[self._extract_fbank_features(__UpperCAmelCase					) for waveform in raw_speech]

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

									lowercase__					:   List[str]													=					self.pad(
									    __UpperCAmelCase				,       padding=__UpperCAmelCase				,       max_length=__UpperCAmelCase				,       truncation=__UpperCAmelCase				,       pad_to_multiple_of=__UpperCAmelCase				,       return_attention_mask=__UpperCAmelCase				,       **__UpperCAmelCase				,       )

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

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

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

									if return_tensors is not None:
													lowercase__					:   Dict													=					padded_inputs.convert_to_tensors(__UpperCAmelCase					)

									return padded_inputs









import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase__					  =  logging.get_logger(__name__)

lowerCAmelCase__					  =  '''▁'''

lowerCAmelCase__					  =  {'''vocab_file''': '''sentencepiece.bpe.model''', '''monolingual_vocab_file''': '''dict.txt'''}

lowerCAmelCase__					  =  {
    '''vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/sentencepiece.bpe.model''',
    },
    '''monolingual_vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/dict.txt''',
    },
}

lowerCAmelCase__					  =  {'''vinai/bartpho-syllable''': 1_0_2_4}

class 				snake_case__(_UpperCamelCase      ):

					"""simple docstring"""






					lowercase_								=		VOCAB_FILES_NAMES
					lowercase_								=		PRETRAINED_VOCAB_FILES_MAP
					lowercase_								=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					lowercase_								=		["""input_ids""", """attention_mask"""]







					def __init__(							self :     Optional[int]				,       SCREAMING_SNAKE_CASE :     List[str]				,       SCREAMING_SNAKE_CASE :     Optional[Any]				,       SCREAMING_SNAKE_CASE :     Optional[int]="<s>"				,       SCREAMING_SNAKE_CASE :     Optional[int]="</s>"				,       SCREAMING_SNAKE_CASE :     str="</s>"				,       SCREAMING_SNAKE_CASE :     List[str]="<s>"				,       SCREAMING_SNAKE_CASE :     Union[str, Any]="<unk>"				,       SCREAMING_SNAKE_CASE :     Tuple="<pad>"				,       SCREAMING_SNAKE_CASE :     List[str]="<mask>"				,       SCREAMING_SNAKE_CASE :     Optional[Dict[str, Any]] = None				,       **SCREAMING_SNAKE_CASE :     int				,       ):
									# Mask token behave like a normal word, i.e. include the space before it
									lowercase__					:   Dict													=					AddedToken(SCREAMING_SNAKE_CASE				,       lstrip=SCREAMING_SNAKE_CASE				,       rstrip=SCREAMING_SNAKE_CASE					) if isinstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					) else mask_token

									lowercase__					:   int													=					{} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    bos_token=SCREAMING_SNAKE_CASE				,       eos_token=SCREAMING_SNAKE_CASE				,       unk_token=SCREAMING_SNAKE_CASE				,       sep_token=SCREAMING_SNAKE_CASE				,       cls_token=SCREAMING_SNAKE_CASE				,       pad_token=SCREAMING_SNAKE_CASE				,       mask_token=SCREAMING_SNAKE_CASE				,       sp_model_kwargs=self.sp_model_kwargs				,       **SCREAMING_SNAKE_CASE				,       )

									lowercase__					:   Dict													=					vocab_file
									lowercase__					:   Union[str, Any]													=					monolingual_vocab_file
									lowercase__					:   Union[str, Any]													=					spm.SentencePieceProcessor(**self.sp_model_kwargs					)
									self.sp_model.Load(str(SCREAMING_SNAKE_CASE					)					)

									# Load the reduced vocab

									# Keep order of special tokens for backward compatibility
									lowercase__					:   Any													=					{}
									lowercase__					:   int													=					0
									for token in [bos_token, pad_token, eos_token, unk_token, sep_token, cls_token]:
													if str(SCREAMING_SNAKE_CASE					) not in self.fairseq_tokens_to_ids:
																	lowercase__					:   Dict													=					cnt
																	cnt += 1
									with open(SCREAMING_SNAKE_CASE				,       "r"				,       encoding="utf-8"					) as f:
													for line in f.readlines():
																	lowercase__					:   int													=					line.strip().split()[0]
																	lowercase__					:   List[str]													=					len(self.fairseq_tokens_to_ids					)
									if str(SCREAMING_SNAKE_CASE					) not in self.fairseq_tokens_to_ids:
													lowercase__					:   Optional[Any]													=					len(self.fairseq_tokens_to_ids					)

									lowercase__					:   List[str]													=					{v: k for k, v in self.fairseq_tokens_to_ids.items()}







					def __getstate__(							self :     int					):
									lowercase__					:   Dict													=					self.__dict__.copy()
									lowercase__					:   Union[str, Any]													=					None
									lowercase__					:   Dict													=					self.sp_model.serialized_model_proto()
									return state







					def __setstate__(							self :     Union[str, Any]				,       SCREAMING_SNAKE_CASE :     Optional[Any]					):
									lowercase__					:   Union[str, Any]													=					d

									# for backward compatibility
									if not hasattr(self				,       "sp_model_kwargs"					):
													lowercase__					:   Dict													=					{}

									lowercase__					:   List[str]													=					spm.SentencePieceProcessor(**self.sp_model_kwargs					)
									self.sp_model.LoadFromSerializedProto(self.sp_model_proto					)







					def 			snake_case					(							self :     Optional[int]				,       SCREAMING_SNAKE_CASE :     List[int]				,       SCREAMING_SNAKE_CASE :     Optional[List[int]] = None					):

									if token_ids_a is None:
													return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
									lowercase__					:   List[Any]													=					[self.cls_token_id]
									lowercase__					:   str													=					[self.sep_token_id]
									return cls + token_ids_a + sep + sep + token_ids_a + sep







					def 			snake_case					(							self :     Dict				,       SCREAMING_SNAKE_CASE :     List[int]				,       SCREAMING_SNAKE_CASE :     Optional[List[int]] = None				,       SCREAMING_SNAKE_CASE :     bool = False					):

									if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=SCREAMING_SNAKE_CASE				,       token_ids_a=SCREAMING_SNAKE_CASE				,       already_has_special_tokens=SCREAMING_SNAKE_CASE					)

									if token_ids_a is None:
													return [1] + ([0] * len(SCREAMING_SNAKE_CASE					)) + [1]
									return [1] + ([0] * len(SCREAMING_SNAKE_CASE					)) + [1, 1] + ([0] * len(SCREAMING_SNAKE_CASE					)) + [1]







					def 			snake_case					(							self :     str				,       SCREAMING_SNAKE_CASE :     List[int]				,       SCREAMING_SNAKE_CASE :     Optional[List[int]] = None					):
									lowercase__					:   Tuple													=					[self.sep_token_id]
									lowercase__					:   str													=					[self.cls_token_id]

									if token_ids_a is None:
													return len(cls + token_ids_a + sep					) * [0]
									return len(cls + token_ids_a + sep + sep + token_ids_a + sep					) * [0]







					@property
					def 			snake_case					(							self :     Optional[int]					):
									return len(self.fairseq_ids_to_tokens					)







					def 			snake_case					(							self :     List[Any]					):
									lowercase__					:   Any													=					{self.convert_ids_to_tokens(SCREAMING_SNAKE_CASE					): i for i in range(self.vocab_size					)}
									vocab.update(self.added_tokens_encoder					)
									return vocab







					def 			snake_case					(							self :     Union[str, Any]				,       SCREAMING_SNAKE_CASE :     str					):
									return self.sp_model.encode(SCREAMING_SNAKE_CASE				,       out_type=SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     List[str]				,       SCREAMING_SNAKE_CASE :     int					):
									if token in self.fairseq_tokens_to_ids:
													return self.fairseq_tokens_to_ids[token]
									else:
													return self.unk_token_id







					def 			snake_case					(							self :     List[str]				,       SCREAMING_SNAKE_CASE :     Any					):
									return self.fairseq_ids_to_tokens[index]







					def 			snake_case					(							self :     List[Any]				,       SCREAMING_SNAKE_CASE :     Union[str, Any]					):
									lowercase__					:   str													=					"".join(SCREAMING_SNAKE_CASE					).replace(SCREAMING_SNAKE_CASE				,       " "					).strip()
									return out_string







					def 			snake_case					(							self :     Optional[int]				,       SCREAMING_SNAKE_CASE :     str				,       SCREAMING_SNAKE_CASE :     Optional[str] = None					):
									if not os.path.isdir(SCREAMING_SNAKE_CASE					):
													logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""					)
													return
									lowercase__					:   str													=					os.path.join(
									    SCREAMING_SNAKE_CASE				,       (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]					)
									lowercase__					:   List[str]													=					os.path.join(
									    SCREAMING_SNAKE_CASE				,       (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["monolingual_vocab_file"]				,       )

									if os.path.abspath(self.vocab_file					) != os.path.abspath(SCREAMING_SNAKE_CASE					) and os.path.isfile(self.vocab_file					):
													copyfile(self.vocab_file				,       SCREAMING_SNAKE_CASE					)
									elif not os.path.isfile(self.vocab_file					):
													with open(SCREAMING_SNAKE_CASE				,       "wb"					) as fi:
																	lowercase__					:   str													=					self.sp_model.serialized_model_proto()
																	fi.write(SCREAMING_SNAKE_CASE					)

									if os.path.abspath(self.monolingual_vocab_file					) != os.path.abspath(
									    SCREAMING_SNAKE_CASE					) and os.path.isfile(self.monolingual_vocab_file					):
													copyfile(self.monolingual_vocab_file				,       SCREAMING_SNAKE_CASE					)
									elif not os.path.isfile(self.monolingual_vocab_file					):
													with open(SCREAMING_SNAKE_CASE				,       "w"				,       encoding="utf-8"					) as fp:
																	for token in self.fairseq_tokens_to_ids:
																					if token not in self.all_special_tokens:
																									fp.write(f"""{str(SCREAMING_SNAKE_CASE					)} \n"""					)

									return out_vocab_file, out_monolingual_vocab_file




