from typing import List, Optional, Union

import numpy as np
import tensorflow as tf

from .utils import logging


__lowerCamelCase    :				int							     =	logging.get_logger(__name__)



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: Union[tf.Tensor, np.ndarray]			)							->      List[int]:







		"""simple docstring"""





		if isinstance(__UpperCamelCase				,  np.ndarray			):
				return list(tensor.shape			)

		SCREAMING_SNAKE_CASE__    							=   tf.shape(__UpperCamelCase			)

		if tensor.shape == tf.TensorShape(__UpperCamelCase			):
				return dynamic

		SCREAMING_SNAKE_CASE__    							=   tensor.shape.as_list()

		return [dynamic[i] if s is None else s for i, s in enumerate(__UpperCamelCase			)]



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: tf.Tensor				,  __UpperCamelCase					: Optional[int] = None				,  __UpperCamelCase					: Optional[str] = None			)							->      tf.Tensor:







		"""simple docstring"""





		return tf.nn.softmax(logits=logits + 1E-9				,  axis=__UpperCamelCase				,  name=__UpperCamelCase			)



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: Any				,  __UpperCamelCase					: int				,  __UpperCamelCase					: int				,  __UpperCamelCase					: Tuple=1E-5				,  __UpperCamelCase					: Union[str, Any]=-1			)							->      Union[str, Any]:







		"""simple docstring"""





		if weight.shape.rank != 1 or bias.shape.rank != 1 or not isinstance(__UpperCamelCase				,  __UpperCamelCase			):
				raise NotImplementedError("""Only 1D weight and bias tensors are supported for now, with only a single axis."""			)

		# Get mean and variance on the axis to be normalized
		SCREAMING_SNAKE_CASE__							,  SCREAMING_SNAKE_CASE__    							=   tf.nn.moments(__UpperCamelCase				,  axes=[axis]				,  keepdims=__UpperCamelCase			)

		if axis != -1:
				# Reshape scale and weight to have the same rank as inputs, but with 1 dimensions
				# on every dimension except axis
				SCREAMING_SNAKE_CASE__    							=   [1] * inputs.shape.rank
				SCREAMING_SNAKE_CASE__    							=   shape_list(__UpperCamelCase			)[axis]
				SCREAMING_SNAKE_CASE__    							=   tf.reshape(__UpperCamelCase				,  __UpperCamelCase			)
				SCREAMING_SNAKE_CASE__    							=   tf.reshape(__UpperCamelCase				,  __UpperCamelCase			)

		# Compute layer normalization using the batch_normalization
		# function.
		SCREAMING_SNAKE_CASE__    							=   tf.nn.batch_normalization(
		    __UpperCamelCase				,  __UpperCamelCase				,  __UpperCamelCase				,  offset=__UpperCamelCase				,  scale=__UpperCamelCase				,  variance_epsilon=__UpperCamelCase				,  )
		return outputs



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: List[Any]				,  __UpperCamelCase					: Tuple=0				,  __UpperCamelCase					: Any=-1			)							->      Optional[int]:







		"""simple docstring"""





		if end_dim < 0:
				end_dim += input.shape.rank
		if start_dim < 0:
				start_dim += input.shape.rank

		if start_dim == end_dim:
				return input

		SCREAMING_SNAKE_CASE__    							=   tf.shape(__UpperCamelCase			)
		SCREAMING_SNAKE_CASE__    							=   tf.math.reduce_prod(in_shape[start_dim : end_dim + 1]			)
		SCREAMING_SNAKE_CASE__    							=   tf.concat([in_shape[:start_dim], [flattened_dim], in_shape[end_dim + 1 :]]				,  axis=0			)
		return tf.reshape(__UpperCamelCase				,  __UpperCamelCase			)



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: tf.Tensor			)							->      tf.Tensor:







		"""simple docstring"""





		if not isinstance(__UpperCamelCase				,  tf.Tensor			):
				SCREAMING_SNAKE_CASE__    							=   tf.convert_to_tensor(__UpperCamelCase			)  # Catches stray NumPy inputs
		if encoder_attention_mask.shape.rank == 3:
				SCREAMING_SNAKE_CASE__    							=   encoder_attention_mask[:, None, :, :]
		if encoder_attention_mask.shape.rank == 2:
				SCREAMING_SNAKE_CASE__    							=   encoder_attention_mask[:, None, None, :]
		# T5 has a mask that can compare sequence ids, we can simulate this here with this transposition
		# Cf. https://github.com/tensorflow/mesh/blob/8d2465e9bc93129b913b5ccc6a59aa97abd96ec6/mesh_tensorflow
		# /transformer/transformer_layers.py#L270
		# encoder_extended_attention_mask = (encoder_extended_attention_mask ==
		# encoder_extended_attention_mask.transpose(-1, -2))
		SCREAMING_SNAKE_CASE__    							=   (
		    tf.cast(1				,  encoder_attention_mask.dtype			) - encoder_extended_attention_mask
		) * encoder_extended_attention_mask.dtype.min

		return encoder_extended_attention_mask



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: tf.Tensor				,  __UpperCamelCase					: int				,  __UpperCamelCase					: str = "input_ids"			)							->      None:







		"""simple docstring"""





		tf.debugging.assert_less(
		    __UpperCamelCase				,  tf.cast(__UpperCamelCase				,  dtype=tensor.dtype			)				,  message=(
		        f"""The maximum value of {tensor_name} ({tf.math.reduce_max(__UpperCamelCase			)}) must be smaller than the embedding """
		        f"""layer's input dimension ({embed_dim}). The likely cause is some problem at tokenization time."""
		    )				,  )



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: str				,  __UpperCamelCase					: Dict				,  __UpperCamelCase					: Optional[Any]			)							->      Any:







		"""simple docstring"""





		SCREAMING_SNAKE_CASE__    							=   6_45_12
		# Check that no item in `data` is larger than `HDF5_OBJECT_HEADER_LIMIT`
		# because in that case even chunking the array would not make the saving
		# possible.
		SCREAMING_SNAKE_CASE__    							=   [x for x in data if len(__UpperCamelCase			) > HDF5_OBJECT_HEADER_LIMIT]

		# Expecting this to never be true.
		if bad_attributes:
				raise RuntimeError(
				    """The following attributes cannot be saved to HDF5 file because """
				    f"""they are larger than {HDF5_OBJECT_HEADER_LIMIT} """
				    f"""bytes: {bad_attributes}"""			)

		SCREAMING_SNAKE_CASE__    							=   np.asarray(__UpperCamelCase			)

		SCREAMING_SNAKE_CASE__    							=   1
		SCREAMING_SNAKE_CASE__    							=   np.array_split(__UpperCamelCase				,  __UpperCamelCase			)

		# This will never loop forever thanks to the test above.
		while any(x.nbytes > HDF5_OBJECT_HEADER_LIMIT for x in chunked_data			):
				num_chunks += 1
				SCREAMING_SNAKE_CASE__    							=   np.array_split(__UpperCamelCase				,  __UpperCamelCase			)

		if num_chunks > 1:
				for chunk_id, chunk_data in enumerate(__UpperCamelCase			):
						SCREAMING_SNAKE_CASE__    							=   chunk_data
		else:
				SCREAMING_SNAKE_CASE__    							=   data



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: int				,  __UpperCamelCase					: Any			)							->      Union[str, Any]:







		"""simple docstring"""





		if name in group.attrs:
				SCREAMING_SNAKE_CASE__    							=   [n.decode("""utf8"""			) if hasattr(__UpperCamelCase				,  """decode"""			) else n for n in group.attrs[name]]
		else:
				SCREAMING_SNAKE_CASE__    							=   []
				SCREAMING_SNAKE_CASE__    							=   0
				while "%s%d" % (name, chunk_id) in group.attrs:
						data.extend(
						    [n.decode("""utf8"""			) if hasattr(__UpperCamelCase				,  """decode"""			) else n for n in group.attrs["""%s%d""" % (name, chunk_id)]]			)
						chunk_id += 1
		return data






def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: List[str]			)							->      Optional[int]:







		"""simple docstring"""





		def _expand_single_ad_tensor(__UpperCamelCase					: str			):
				if isinstance(__UpperCamelCase				,  tf.Tensor			) and t.shape.rank == 1:
						return tf.expand_dims(__UpperCamelCase				,  axis=-1			)
				return t

		return tf.nest.map_structure(_expand_single_ad_tensor				,  __UpperCamelCase			)

import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
   import jieba

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__lowerCamelCase    :				Any							     =	logging.get_logger(__name__)

__lowerCamelCase    :				Optional[Any]							     =	{'''vocab_file''': '''vocab.txt'''}

__lowerCamelCase    :				Union[str, Any]							     =	{
    '''vocab_file''': {
        '''openbmb/cpm-ant-10b''': '''https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt''',
    },
}

__lowerCamelCase    :				Optional[Any]							     =	{
    '''openbmb/cpm-ant-10b''': 1024,
}



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: str			)							->      Optional[Any]:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__    							=   collections.OrderedDict()
  with open(__UpperCamelCase				,  """r"""				,  encoding="""utf-8"""			) as reader:
    SCREAMING_SNAKE_CASE__    							=   reader.readlines()
  for index, token in enumerate(__UpperCamelCase			):
    SCREAMING_SNAKE_CASE__    							=   token.rstrip("""\n"""			)
    SCREAMING_SNAKE_CASE__    							=   index
  return vocab







class __snake_case       (		lowerCamelCase_	):




   def __init__(					self    :       Optional[int] ,			_lowercase    :       Optional[Any] ,			_lowercase    :       int="<unk>" ,			_lowercase    :       int=2_00   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   vocab
     SCREAMING_SNAKE_CASE__    							=   unk_token
     SCREAMING_SNAKE_CASE__    							=   max_input_chars_per_word




   def 			__a		(					self    :       Optional[int] ,			_lowercase    :       List[Any]   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   list(_lowercase   )
     if len(_lowercase   ) > self.max_input_chars_per_word:
       return [self.unk_token]

     SCREAMING_SNAKE_CASE__    							=   0
     SCREAMING_SNAKE_CASE__    							=   []
     while start < len(_lowercase   ):
       SCREAMING_SNAKE_CASE__    							=   len(_lowercase   )
       SCREAMING_SNAKE_CASE__    							=   None
       while start < end:
         SCREAMING_SNAKE_CASE__    							=   """""".join(chars[start:end]   )
         if substr in self.vocab:
           SCREAMING_SNAKE_CASE__    							=   substr
           break
         end -= 1
       if cur_substr is None:
         sub_tokens.append(self.unk_token   )
         start += 1
       else:
         sub_tokens.append(_lowercase   )
         SCREAMING_SNAKE_CASE__    							=   end

     return sub_tokens







class __snake_case       (		lowerCamelCase_	):
   lowerCAmelCase_														=							VOCAB_FILES_NAMES
   lowerCAmelCase_														=							PRETRAINED_VOCAB_FILES_MAP
   lowerCAmelCase_														=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   lowerCAmelCase_														=							["input_ids", "attention_mask"]
   lowerCAmelCase_														=							False




   def __init__(					self    :       int ,			_lowercase    :       str ,			_lowercase    :       List[Any]="<d>" ,			_lowercase    :       List[Any]="</d>" ,			_lowercase    :       Union[str, Any]="<s>" ,			_lowercase    :       List[str]="</s>" ,			_lowercase    :       str="<pad>" ,			_lowercase    :       int="<unk>" ,			_lowercase    :       List[str]="</n>" ,			_lowercase    :       Tuple="</_>" ,			_lowercase    :       Any="left" ,			**_lowercase    :       Any ,			):



     """simple docstring"""






     requires_backends(self ,			["""jieba"""]   )
     super().__init__(
         bod_token=_lowercase ,			eod_token=_lowercase ,			bos_token=_lowercase ,			eos_token=_lowercase ,			pad_token=_lowercase ,			unk_token=_lowercase ,			line_token=_lowercase ,			space_token=_lowercase ,			padding_side=_lowercase ,			**_lowercase ,			)
     SCREAMING_SNAKE_CASE__    							=   bod_token
     SCREAMING_SNAKE_CASE__    							=   eod_token
     SCREAMING_SNAKE_CASE__    							=   load_vocab(_lowercase   )
     SCREAMING_SNAKE_CASE__    							=   self.encoder[space_token]
     SCREAMING_SNAKE_CASE__    							=   self.encoder[line_token]

     del self.encoder[space_token]
     del self.encoder[line_token]

     SCREAMING_SNAKE_CASE__    							=   collections.OrderedDict(sorted(self.encoder.items() ,			key=lambda _lowercase   : x[1]   )   )
     SCREAMING_SNAKE_CASE__    							=   {v: k for k, v in self.encoder.items()}

     SCREAMING_SNAKE_CASE__    							=   WordpieceTokenizer(vocab=self.encoder ,			unk_token=self.unk_token   )




   @property
   def 			__a		(					self    :       Optional[Any]   ):



     """simple docstring"""






     return self.encoder[self.bod_token]




   @property
   def 			__a		(					self    :       List[Any]   ):



     """simple docstring"""






     return self.encoder[self.eod_token]




   @property
   def 			__a		(					self    :       Any   ):



     """simple docstring"""






     return self.encoder["\n"]




   @property
   def 			__a		(					self    :       Union[str, Any]   ):



     """simple docstring"""






     return len(self.encoder   )




   def 			__a		(					self    :       int   ):



     """simple docstring"""






     return dict(self.encoder ,			**self.added_tokens_encoder   )




   def 			__a		(					self    :       Union[str, Any] ,			_lowercase    :       str   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   []
     for x in jieba.cut(_lowercase ,			cut_all=_lowercase   ):
       output_tokens.extend(self.wordpiece_tokenizer.tokenize(_lowercase   )   )
     return output_tokens




   def 			__a		(					self    :       int ,			_lowercase    :       Any ,			**_lowercase    :       Any   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   [i for i in token_ids if i >= 0]
     SCREAMING_SNAKE_CASE__    							=   [
         x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
     ]
     return super()._decode(_lowercase ,			**_lowercase   )




   def 			__a		(					self    :       Optional[int] ,			_lowercase    :       List[Any]   ):



     """simple docstring"""






     return token in self.encoder




   def 			__a		(					self    :       List[str] ,			_lowercase    :       List[str]   ):



     """simple docstring"""






     return "".join(_lowercase   )




   def 			__a		(					self    :       Optional[int] ,			_lowercase    :       Any   ):



     """simple docstring"""






     return self.encoder.get(_lowercase ,			self.encoder.get(self.unk_token   )   )




   def 			__a		(					self    :       Tuple ,			_lowercase    :       List[Any]   ):



     """simple docstring"""






     return self.decoder.get(_lowercase ,			self.unk_token   )




   def 			__a		(					self    :       Optional[Any] ,			_lowercase    :       str ,			_lowercase    :       Optional[str] = None   ):



     """simple docstring"""






     if os.path.isdir(_lowercase   ):
       SCREAMING_SNAKE_CASE__    							=   os.path.join(
           _lowercase ,			(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]   )
     else:
       SCREAMING_SNAKE_CASE__    							=   (filename_prefix + """-""" if filename_prefix else """""") + save_directory
     SCREAMING_SNAKE_CASE__    							=   0
     if " " in self.encoder:
       SCREAMING_SNAKE_CASE__    							=   self.encoder[""" """]
       del self.encoder[" "]
     if "\n" in self.encoder:
       SCREAMING_SNAKE_CASE__    							=   self.encoder["""\n"""]
       del self.encoder["\n"]
     SCREAMING_SNAKE_CASE__    							=   collections.OrderedDict(sorted(self.encoder.items() ,			key=lambda _lowercase   : x[1]   )   )
     with open(_lowercase ,			"""w""" ,			encoding="""utf-8"""   ) as writer:
       for token, token_index in self.encoder.items():
         if index != token_index:
           logger.warning(
               f"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
               """ Please check that the vocabulary is not corrupted!"""   )
           SCREAMING_SNAKE_CASE__    							=   token_index
         writer.write(token + """\n"""   )
         index += 1
     return (vocab_file,)




   def 			__a		(					self    :       int ,			_lowercase    :       List[int] ,			_lowercase    :       List[int] = None   ):



     """simple docstring"""






     if token_ids_a is None:
       return [self.bos_token_id] + token_ids_a
     return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a




   def 			__a		(					self    :       Union[str, Any] ,			_lowercase    :       List[int] ,			_lowercase    :       Optional[List[int]] = None ,			_lowercase    :       bool = False   ):



     """simple docstring"""






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

     if token_ids_a is not None:
       return [1] + ([0] * len(_lowercase   )) + [1] + ([0] * len(_lowercase   ))
     return [1] + ([0] * len(_lowercase   ))
