




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


__A						:							Optional[Any]							=  logging.get_logger(__name__)




class 			UpperCAmelCase_      (   A       ):




							'''simple docstring'''



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





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





							def    _UpperCAmelCase     (					self       :	Any       ,     a       :	np.ndarray       ,     )			->					np.ndarray:
										SCREAMING_SNAKE_CASE           =					waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
										SCREAMING_SNAKE_CASE           =					torch.from_numpy(a			).unsqueeze(0			)
										SCREAMING_SNAKE_CASE           =					ta_kaldi.fbank(a       ,     num_mel_bins=self.num_mel_bins       ,     sample_frequency=self.sampling_rate			)
										return features.numpy()





							@staticmethod
							def    _UpperCAmelCase     (					a       :	np.ndarray       ,     a       :	int       ,     a       :	Optional[bool] = True       ,     a       :	Optional[bool] = True       ,     a       :	float = 0.0       ,     )			->					np.ndarray:
										# make sure we normalize float32 arrays
										if normalize_means:
													SCREAMING_SNAKE_CASE           =					x[:input_length].mean(axis=0			)
													SCREAMING_SNAKE_CASE           =					np.subtract(a       ,     a			)
										if normalize_vars:
													SCREAMING_SNAKE_CASE           =					x[:input_length].std(axis=0			)
													SCREAMING_SNAKE_CASE           =					np.divide(a       ,     a			)

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

										# make sure array is in float32
										SCREAMING_SNAKE_CASE           =					x.astype(np.floataa			)

										return x





							def    _UpperCAmelCase     (					self       :	Optional[int]       ,     a       :	List[np.ndarray]       ,     a       :	Optional[np.ndarray] = None			)			->					List[np.ndarray]:
										SCREAMING_SNAKE_CASE           =					attention_mask.sum(-1			) if attention_mask is not None else [x.shape[0] for x in input_features]
										return [
										    self.utterance_cmvn(a       ,     a       ,     self.normalize_means       ,     self.normalize_vars       ,     self.padding_value			)
										    for x, n in zip(a       ,     a			)
										]







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

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

										SCREAMING_SNAKE_CASE           =					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}"""			)
										SCREAMING_SNAKE_CASE           =					is_batched_numpy or (
										    isinstance(a       ,     (list, tuple)			) and (isinstance(raw_speech[0]       ,     (np.ndarray, tuple, list)			))
										)

										if is_batched:
													SCREAMING_SNAKE_CASE           =					[np.asarray(a       ,     dtype=np.floataa			) for speech in raw_speech]
										elif not is_batched and not isinstance(a       ,     np.ndarray			):
													SCREAMING_SNAKE_CASE           =					np.asarray(a       ,     dtype=np.floataa			)
										elif isinstance(a       ,     np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
													SCREAMING_SNAKE_CASE           =					raw_speech.astype(np.floataa			)

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

										# extract fbank features
										SCREAMING_SNAKE_CASE           =					[self._extract_fbank_features(a			) for waveform in raw_speech]

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

										SCREAMING_SNAKE_CASE           =					self.pad(
										    a       ,     padding=a       ,     max_length=a       ,     truncation=a       ,     pad_to_multiple_of=a       ,     return_attention_mask=a       ,     **a       ,     )

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

										SCREAMING_SNAKE_CASE           =					padded_inputs.get("""attention_mask"""			)
										if attention_mask is not None:
													SCREAMING_SNAKE_CASE           =					[np.asarray(a       ,     dtype=np.intaa			) for array in attention_mask]

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

										if return_tensors is not None:
													SCREAMING_SNAKE_CASE           =					padded_inputs.convert_to_tensors(a			)

										return padded_inputs












import json
import re
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
		if is_torch_available():
				import torch
		if is_tf_available():
				import tensorflow as tf

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_codegen import CodeGenTokenizer


__A						:							Union[str, Any]							=  logging.get_logger(__name__)

__A						:							str							=  {"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

__A						:							List[Any]							=  {
    """vocab_file""": {
        """Salesforce/codegen-350M-mono""": """https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """Salesforce/codegen-350M-mono""": """https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt""",
    },
    """tokenizer_file""": {
        """Salesforce/codegen-350M-mono""": (
            """https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/tokenizer.json"""
        ),
    },
}

__A						:							Tuple							=  {
    """Salesforce/codegen-350M-mono""": 2_0_4_8,
}




class 			UpperCAmelCase_      (   A       ):




							'''simple docstring'''



							a__						  =     VOCAB_FILES_NAMES
							a__						  =     PRETRAINED_VOCAB_FILES_MAP
							a__						  =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							a__						  =     ['''input_ids''', '''attention_mask''']
							a__						  =     CodeGenTokenizer





							def __init__(					self       :	Tuple       ,     a       :	Dict=None       ,     a       :	Tuple=None       ,     a       :	Tuple=None       ,     a       :	Optional[Any]="<|endoftext|>"       ,     a       :	Any="<|endoftext|>"       ,     a       :	Dict="<|endoftext|>"       ,     a       :	Optional[int]=False       ,     **a       :	Tuple       ,     )			->					str:
										super().__init__(
										    a       ,     a       ,     tokenizer_file=a       ,     unk_token=a       ,     bos_token=a       ,     eos_token=a       ,     add_prefix_space=a       ,     **a       ,     )

										if kwargs.pop("""add_bos_token"""       ,     a			):
													SCREAMING_SNAKE_CASE           =					kwargs.pop("""name_or_path"""       ,     """"""			)
													raise ValueError(
													    """Currenty GPT2's fast tokenizer does NOT support adding a BOS token."""
													    """Instead you should use GPT2's slow tokenizer class `CodeGenTokenizer` as follows: \n"""
													    f"""`CodeGenTokenizer.from_pretrained('{model_id}')`\nor\n"""
													    f"""`AutoTokenizer.from_pretrained('{model_id}', use_fast=False)`\n"""
													    """This issue will be fixed soon, see: https://github.com/huggingface/tokenizers/pull/1005."""
													    """ so that the fast tokenizer works correctly."""			)

										SCREAMING_SNAKE_CASE           =					json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()			)
										if pre_tok_state.get("""add_prefix_space"""       ,     a			) != add_prefix_space:
													SCREAMING_SNAKE_CASE           =					getattr(a       ,     pre_tok_state.pop("""type"""			)			)
													SCREAMING_SNAKE_CASE           =					add_prefix_space
													SCREAMING_SNAKE_CASE           =					pre_tok_class(**a			)

										SCREAMING_SNAKE_CASE           =					add_prefix_space





							def    _UpperCAmelCase     (					self       :	Optional[Any]       ,     *a       :	Optional[Any]       ,     **a       :	List[Any]			)			->					BatchEncoding:
										SCREAMING_SNAKE_CASE           =					kwargs.get("""is_split_into_words"""       ,     a			)
										assert self.add_prefix_space or not is_split_into_words, (
										    f"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._batch_encode_plus(*a       ,     **a			)





							def    _UpperCAmelCase     (					self       :	Union[str, Any]       ,     *a       :	Tuple       ,     **a       :	List[Any]			)			->					BatchEncoding:
										SCREAMING_SNAKE_CASE           =					kwargs.get("""is_split_into_words"""       ,     a			)

										assert self.add_prefix_space or not is_split_into_words, (
										    f"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._encode_plus(*a       ,     **a			)





							def    _UpperCAmelCase     (					self       :	Union[str, Any]       ,     a       :	str       ,     a       :	Optional[str] = None			)			->					Tuple[str]:
										SCREAMING_SNAKE_CASE           =					self._tokenizer.model.save(a       ,     name=a			)
										return tuple(a			)





							def    _UpperCAmelCase     (					self       :	Union[str, Any]       ,     a       :	Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"]       ,     a       :	bool = False       ,     a       :	bool = None       ,     a       :	Optional[List[str]] = None       ,     **a       :	List[Any]       ,     )			->					str:
										SCREAMING_SNAKE_CASE           =					super().decode(
										    token_ids=a       ,     skip_special_tokens=a       ,     clean_up_tokenization_spaces=a       ,     **a       ,     )

										if truncate_before_pattern is not None and len(a			) > 0:
													SCREAMING_SNAKE_CASE           =					self.truncate(a       ,     a			)

										return decoded_text







							def    _UpperCAmelCase     (					self       :	Any       ,     a       :	Optional[Any]       ,     a       :	List[str]			)			->					Any:
										def find_re(a       :	Tuple       ,     a       :	Any       ,     a       :	List[str]			):
													SCREAMING_SNAKE_CASE           =					pattern.search(a       ,     a			)
													return m.start() if m else -1

										SCREAMING_SNAKE_CASE           =					[re.compile(a       ,     re.MULTILINE			) for pattern in truncate_before_pattern]

										SCREAMING_SNAKE_CASE           =					list(re.finditer("""^print"""       ,     a       ,     re.MULTILINE			)			)

										if len(a			) > 1:
													SCREAMING_SNAKE_CASE           =					completion[: prints[1].start()]

										SCREAMING_SNAKE_CASE           =					list(re.finditer("""^def"""       ,     a       ,     re.MULTILINE			)			)

										if len(a			) > 1:
													SCREAMING_SNAKE_CASE           =					completion[: defs[1].start()]

										SCREAMING_SNAKE_CASE           =					0

										SCREAMING_SNAKE_CASE           =					[
										    pos for pos in [find_re(a       ,     a       ,     a			) for terminal in terminals] if pos != -1
										]

										if len(a			) > 0:
													return completion[: min(a			)]
										else:
													return completion






