




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


UpperCAmelCase_									=	logging.get_logger(__name__)

UpperCAmelCase_									=	{'vocab_file': 'vocab.json', 'merges_file': 'merges.txt', 'tokenizer_file': 'tokenizer.json'}

UpperCAmelCase_									=	{
    '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'
        ),
    },
}

UpperCAmelCase_									=	{
    'Salesforce/codegen-350M-mono': 2_048,
}





class     lowerCamelCase__(					__lowerCamelCase):
      UpperCAmelCase__      :	Union[str, Any]    		=   VOCAB_FILES_NAMES
      UpperCAmelCase__      :	List[Any]    		=   PRETRAINED_VOCAB_FILES_MAP
      UpperCAmelCase__      :	str    		=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      UpperCAmelCase__      :	List[Any]    		=   ['input_ids', 'attention_mask']
      UpperCAmelCase__      :	Dict    		=   CodeGenTokenizer
      def __init__(				self: Tuple    ,       UpperCamelCase_: Optional[Any]=None    ,       UpperCamelCase_: Tuple=None    ,       UpperCamelCase_: Tuple=None    ,       UpperCamelCase_: Optional[int]="<|endoftext|>"    ,       UpperCamelCase_: Union[str, Any]="<|endoftext|>"    ,       UpperCamelCase_: Union[str, Any]="<|endoftext|>"    ,       UpperCamelCase_: Any=False    ,       **UpperCamelCase_: Tuple    ,       ):
             super().__init__(
                 UpperCamelCase_    ,       UpperCamelCase_    ,       tokenizer_file=UpperCamelCase_    ,       unk_token=UpperCamelCase_    ,       bos_token=UpperCamelCase_    ,       eos_token=UpperCamelCase_    ,       add_prefix_space=UpperCamelCase_    ,       **UpperCamelCase_    ,       )

             if kwargs.pop("""add_bos_token"""    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						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."""   )

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

             __lowerCamelCase   =						add_prefix_space
      def 			lowerCAmelCase__						(				self: Union[str, Any]    ,       *UpperCamelCase_: Union[str, Any]    ,       **UpperCamelCase_: List[str]   ):
             __lowerCamelCase   =						kwargs.get("""is_split_into_words"""    ,       UpperCamelCase_   )
             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(*UpperCamelCase_    ,       **UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: Union[str, Any]    ,       *UpperCamelCase_: Any    ,       **UpperCamelCase_: List[str]   ):
             __lowerCamelCase   =						kwargs.get("""is_split_into_words"""    ,       UpperCamelCase_   )

             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(*UpperCamelCase_    ,       **UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: Optional[Any]    ,       UpperCamelCase_: str    ,       UpperCamelCase_: Optional[str] = None   ):
             __lowerCamelCase   =						self._tokenizer.model.save(UpperCamelCase_    ,       name=UpperCamelCase_   )
             return tuple(UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: List[str]    ,       UpperCamelCase_: Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"]    ,       UpperCamelCase_: bool = False    ,       UpperCamelCase_: bool = None    ,       UpperCamelCase_: Optional[List[str]] = None    ,       **UpperCamelCase_: Union[str, Any]    ,       ):
             __lowerCamelCase   =						super().decode(
                 token_ids=UpperCamelCase_    ,       skip_special_tokens=UpperCamelCase_    ,       clean_up_tokenization_spaces=UpperCamelCase_    ,       **UpperCamelCase_    ,       )

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

             return decoded_text




      def 			lowerCAmelCase__						(				self: int    ,       UpperCamelCase_: Optional[int]    ,       UpperCamelCase_: Any   ):
             def find_re(UpperCamelCase_: int    ,       UpperCamelCase_: Optional[int]    ,       UpperCamelCase_: List[Any]   ):
                    __lowerCamelCase   =						pattern.search(UpperCamelCase_    ,       UpperCamelCase_   )
                    return m.start() if m else -1

             __lowerCamelCase   =						[re.compile(UpperCamelCase_    ,       re.MULTILINE   ) for pattern in truncate_before_pattern]

             __lowerCamelCase   =						list(re.finditer("""^print"""    ,       UpperCamelCase_    ,       re.MULTILINE   )   )

             if len(UpperCamelCase_   ) > 1:
                    __lowerCamelCase   =						completion[: prints[1].start()]

             __lowerCamelCase   =						list(re.finditer("""^def"""    ,       UpperCamelCase_    ,       re.MULTILINE   )   )

             if len(UpperCamelCase_   ) > 1:
                    __lowerCamelCase   =						completion[: defs[1].start()]

             __lowerCamelCase   =						0

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

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



'''simple docstring'''




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


lowercase			:				Optional[int]    		=			logging.get_logger(__name__)





class  A				(							__snake_case						):
     __magic_name__						=		['''input_features''', '''attention_mask''']





     def __init__(							self						,			SCREAMING_SNAKE_CASE=80						,			SCREAMING_SNAKE_CASE=16000						,			SCREAMING_SNAKE_CASE=80						,			SCREAMING_SNAKE_CASE=0.0						,			SCREAMING_SNAKE_CASE=True						,			SCREAMING_SNAKE_CASE=True						,			SCREAMING_SNAKE_CASE=True						,			**SCREAMING_SNAKE_CASE						,			) ->				int:






         """simple docstring"""






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





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			) ->				np.ndarray:






         """simple docstring"""






         A       :   List[Any]					=						waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
         A       :   Any					=						torch.from_numpy(SCREAMING_SNAKE_CASE	).unsqueeze(0	)
         A       :   Any					=						ta_kaldi.fbank(SCREAMING_SNAKE_CASE						,			num_mel_bins=self.num_mel_bins						,			sample_frequency=self.sampling_rate	)
         return features.numpy()





     @staticmethod
     def 				__lowerCAmelCase							(							SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = True						,			SCREAMING_SNAKE_CASE = True						,			SCREAMING_SNAKE_CASE = 0.0						,			) ->				np.ndarray:






         """simple docstring"""






         if normalize_means:
             A       :   Dict					=						x[:input_length].mean(axis=0	)
             A       :   Optional[Any]					=						np.subtract(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)
         if normalize_vars:
             A       :   str					=						x[:input_length].std(axis=0	)
             A       :   int					=						np.divide(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         if input_length < x.shape[0]:
             A       :   List[str]					=						padding_value

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

         return x





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = None	) ->				List[np.ndarray]:






         """simple docstring"""






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





     def __call__(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = False						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = False						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = None						,			**SCREAMING_SNAKE_CASE						,			) ->				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.'''	)

         A       :   List[Any]					=						isinstance(SCREAMING_SNAKE_CASE						,			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       :   Tuple					=						is_batched_numpy or (
             isinstance(SCREAMING_SNAKE_CASE						,			(list, tuple)	) and (isinstance(raw_speech[0]						,			(np.ndarray, tuple, list)	))
         )

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

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

         # extract fbank features
         A       :   List[str]					=						[self._extract_fbank_features(SCREAMING_SNAKE_CASE	) for waveform in raw_speech]

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

         A       :   Union[str, Any]					=						self.pad(
             SCREAMING_SNAKE_CASE						,			padding=SCREAMING_SNAKE_CASE						,			max_length=SCREAMING_SNAKE_CASE						,			truncation=SCREAMING_SNAKE_CASE						,			pad_to_multiple_of=SCREAMING_SNAKE_CASE						,			return_attention_mask=SCREAMING_SNAKE_CASE						,			**SCREAMING_SNAKE_CASE						,			)

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

         A       :   Union[str, Any]					=						padded_inputs.get('''attention_mask'''	)
         if attention_mask is not None:
             A       :   Union[str, Any]					=						[np.asarray(SCREAMING_SNAKE_CASE						,			dtype=np.intaa	) for array in attention_mask]

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

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

         return padded_inputs




