




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	:		Optional[Any]							       =   logging.get_logger(__name__)

_lowerCamelCase	:		List[Any]							       =   {"""vocab_file""": """vocab.txt"""}

_lowerCamelCase	:		List[Any]							       =   {
    """vocab_file""": {
        """openbmb/cpm-ant-10b""": """https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt""",
    },
}

_lowerCamelCase	:		List[str]							       =   {
    """openbmb/cpm-ant-10b""": 1024,
}







def SCREAMING_SNAKE_CASE				(       lowercase_						)   ->       Any:



       """simple docstring"""

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


class 				UpperCamelCase_    ( UpperCAmelCase__     ):

    '''simple docstring'''






    def __init__(			self      :	Dict					, UpperCAmelCase__      :	Union[str, Any]					, UpperCAmelCase__      :	List[str]="<unk>"					, UpperCAmelCase__      :	Union[str, Any]=200)						->List[str]:





           '''simple docstring'''

           A__       =    vocab
           A__       =    unk_token
           A__       =    max_input_chars_per_word





    def        SCREAMING_SNAKE_CASE							(			self      :	Optional[Any]					, UpperCAmelCase__      :	Optional[int])						->Optional[int]:





           '''simple docstring'''

           A__       =    list(UpperCAmelCase__)
           if len(UpperCAmelCase__) > self.max_input_chars_per_word:
                  return [self.unk_token]

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

           return sub_tokens


class 				UpperCamelCase_    ( UpperCAmelCase__     ):

    '''simple docstring'''






    UpperCAmelCase__							  =      VOCAB_FILES_NAMES
    UpperCAmelCase__							  =      PRETRAINED_VOCAB_FILES_MAP
    UpperCAmelCase__							  =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    UpperCAmelCase__							  =      ['''input_ids''', '''attention_mask''']
    UpperCAmelCase__							  =      False


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





           '''simple docstring'''

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

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

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

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


    @property
    def        SCREAMING_SNAKE_CASE							(			self      :	List[str])						->Dict:





           '''simple docstring'''

           return self.encoder[self.bod_token]


    @property
    def        SCREAMING_SNAKE_CASE							(			self      :	Optional[Any])						->List[Any]:





           '''simple docstring'''

           return self.encoder[self.eod_token]


    @property
    def        SCREAMING_SNAKE_CASE							(			self      :	int)						->List[Any]:





           '''simple docstring'''

           return self.encoder["\n"]


    @property
    def        SCREAMING_SNAKE_CASE							(			self      :	int)						->int:





           '''simple docstring'''

           return len(self.encoder)


    def        SCREAMING_SNAKE_CASE							(			self      :	Dict)						->str:





           '''simple docstring'''

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


    def        SCREAMING_SNAKE_CASE							(			self      :	Any					, UpperCAmelCase__      :	Any)						->Union[str, Any]:





           '''simple docstring'''

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


    def        SCREAMING_SNAKE_CASE							(			self      :	Optional[Any]					, UpperCAmelCase__      :	Optional[Any]					, **UpperCAmelCase__      :	Dict)						->List[str]:





           '''simple docstring'''

           A__       =    [i for i in token_ids if i >= 0]
           A__       =    [
               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(UpperCAmelCase__					, **UpperCAmelCase__)


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





           '''simple docstring'''

           return token in self.encoder


    def        SCREAMING_SNAKE_CASE							(			self      :	Any					, UpperCAmelCase__      :	List[str])						->str:





           '''simple docstring'''

           return "".join(UpperCAmelCase__)


    def        SCREAMING_SNAKE_CASE							(			self      :	int					, UpperCAmelCase__      :	int)						->Union[str, Any]:





           '''simple docstring'''

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


    def        SCREAMING_SNAKE_CASE							(			self      :	int					, UpperCAmelCase__      :	List[str])						->Any:





           '''simple docstring'''

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


    def        SCREAMING_SNAKE_CASE							(			self      :	List[str]					, UpperCAmelCase__      :	str					, UpperCAmelCase__      :	Optional[str] = None)						->Tuple[str]:





           '''simple docstring'''

           if os.path.isdir(UpperCAmelCase__):
                  A__       =    os.path.join(
                      UpperCAmelCase__					, (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file'''])
           else:
                  A__       =    (filename_prefix + '''-''' if filename_prefix else '''''') + save_directory
           A__       =    0
           if " " in self.encoder:
                  A__       =    self.encoder[''' ''']
                  del self.encoder[" "]
           if "\n" in self.encoder:
                  A__       =    self.encoder['''\n''']
                  del self.encoder["\n"]
           A__       =    collections.OrderedDict(sorted(self.encoder.items()					, key=lambda UpperCAmelCase__: x[1]))
           with open(UpperCAmelCase__					, '''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!''')
                                A__       =    token_index
                         writer.write(token + '''\n''')
                         index += 1
           return (vocab_file,)


    def        SCREAMING_SNAKE_CASE							(			self      :	int					, UpperCAmelCase__      :	List[int]					, UpperCAmelCase__      :	List[int] = None)						->List[int]:





           '''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        SCREAMING_SNAKE_CASE							(			self      :	Optional[int]					, UpperCAmelCase__      :	List[int]					, UpperCAmelCase__      :	Optional[List[int]] = None					, UpperCAmelCase__      :	bool = False)						->List[int]:





           '''simple docstring'''

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

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








'''simple docstring'''

import unittest

from transformers import MPNetConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch

							from transformers import (
							    MPNetForMaskedLM,
							    MPNetForMultipleChoice,
							    MPNetForQuestionAnswering,
							    MPNetForSequenceClassification,
							    MPNetForTokenClassification,
							    MPNetModel,
							)






class 						a				:


				def __init__(		self    :					List[Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Any=13       ,				lowercase_    :					Optional[Any]=7       ,				lowercase_    :					Optional[Any]=True       ,				lowercase_    :					Dict=True       ,				lowercase_    :					Tuple=False       ,				lowercase_    :					Optional[Any]=True       ,				lowercase_    :					Any=99       ,				lowercase_    :					Union[str, Any]=64       ,				lowercase_    :					str=5       ,				lowercase_    :					int=4       ,				lowercase_    :					List[Any]=64       ,				lowercase_    :					Dict="gelu"       ,				lowercase_    :					Optional[int]=0.1       ,				lowercase_    :					Optional[Any]=0.1       ,				lowercase_    :					Tuple=512       ,				lowercase_    :					List[Any]=16       ,				lowercase_    :					str=2       ,				lowercase_    :					List[str]=0.02       ,				lowercase_    :					Optional[Any]=3       ,				lowercase_    :					Optional[Any]=4       ,				lowercase_    :					List[Any]=None       ,				):
									snake_case_									=				parent
									snake_case_									=				batch_size
									snake_case_									=				seq_length
									snake_case_									=				is_training
									snake_case_									=				use_input_mask
									snake_case_									=				use_token_type_ids
									snake_case_									=				use_labels
									snake_case_									=				vocab_size
									snake_case_									=				hidden_size
									snake_case_									=				num_hidden_layers
									snake_case_									=				num_attention_heads
									snake_case_									=				intermediate_size
									snake_case_									=				hidden_act
									snake_case_									=				hidden_dropout_prob
									snake_case_									=				attention_probs_dropout_prob
									snake_case_									=				max_position_embeddings
									snake_case_									=				type_vocab_size
									snake_case_									=				type_sequence_label_size
									snake_case_									=				initializer_range
									snake_case_									=				num_labels
									snake_case_									=				num_choices
									snake_case_									=				scope


				def   A_ (		self    :					List[str]					):
									return MPNetConfig.from_pretrained('''microsoft/mpnet-base'''					)


				def   A_ (		self    :					str					):
									snake_case_									=				ids_tensor([self.batch_size, self.seq_length]       ,				self.vocab_size					)

									snake_case_									=				None
									if self.use_input_mask:
														snake_case_									=				random_attention_mask([self.batch_size, self.seq_length]					)

									snake_case_									=				None
									snake_case_									=				None
									snake_case_									=				None
									if self.use_labels:
														snake_case_									=				ids_tensor([self.batch_size]       ,				self.type_sequence_label_size					)
														snake_case_									=				ids_tensor([self.batch_size, self.seq_length]       ,				self.num_labels					)
														snake_case_									=				ids_tensor([self.batch_size]       ,				self.num_choices					)

									snake_case_									=				self.get_config()
									return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels


				def   A_ (		self    :					Tuple					):
									return MPNetConfig(
									    vocab_size=self.vocab_size       ,				hidden_size=self.hidden_size       ,				num_hidden_layers=self.num_hidden_layers       ,				num_attention_heads=self.num_attention_heads       ,				intermediate_size=self.intermediate_size       ,				hidden_act=self.hidden_act       ,				hidden_dropout_prob=self.hidden_dropout_prob       ,				attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,				max_position_embeddings=self.max_position_embeddings       ,				initializer_range=self.initializer_range       ,				)


				def   A_ (		self    :					Any       ,				lowercase_    :					Dict       ,				lowercase_    :					str       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				MPNetModel(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(lowercase_       ,				lowercase_					)
									snake_case_									=				model(lowercase_					)
									self.parent.assertEqual(result.last_hidden_state.shape       ,				(self.batch_size, self.seq_length, self.hidden_size)					)
									self.parent.assertEqual(result.pooler_output.shape       ,				(self.batch_size, self.hidden_size)					)


				def   A_ (		self    :					str       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Any       ,				lowercase_    :					str       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				MPNetForQuestionAnswering(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(
									    lowercase_       ,				attention_mask=lowercase_       ,				start_positions=lowercase_       ,				end_positions=lowercase_       ,				)
									self.parent.assertEqual(result.start_logits.shape       ,				(self.batch_size, self.seq_length)					)
									self.parent.assertEqual(result.end_logits.shape       ,				(self.batch_size, self.seq_length)					)


				def   A_ (		self    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					List[str]       ,				lowercase_    :					str       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Any					):
									snake_case_									=				self.num_labels
									snake_case_									=				MPNetForSequenceClassification(lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(lowercase_       ,				attention_mask=lowercase_       ,				labels=lowercase_					)
									self.parent.assertEqual(result.logits.shape       ,				(self.batch_size, self.num_labels)					)


				def   A_ (		self    :					Any       ,				lowercase_    :					Any       ,				lowercase_    :					str       ,				lowercase_    :					Tuple       ,				lowercase_    :					List[str]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict					):
									snake_case_									=				self.num_choices
									snake_case_									=				MPNetForMultipleChoice(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				input_ids.unsqueeze(1					).expand(-1       ,				self.num_choices       ,				-1					).contiguous()
									snake_case_									=				input_mask.unsqueeze(1					).expand(-1       ,				self.num_choices       ,				-1					).contiguous()
									snake_case_									=				model(
									    lowercase_       ,				attention_mask=lowercase_       ,				labels=lowercase_       ,				)
									self.parent.assertEqual(result.logits.shape       ,				(self.batch_size, self.num_choices)					)


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					List[Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					str       ,				lowercase_    :					int       ,				lowercase_    :					List[str]					):
									snake_case_									=				self.num_labels
									snake_case_									=				MPNetForTokenClassification(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(lowercase_       ,				attention_mask=lowercase_       ,				labels=lowercase_					)
									self.parent.assertEqual(result.logits.shape       ,				(self.batch_size, self.seq_length, self.num_labels)					)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				self.prepare_config_and_inputs()
									((snake_case_)			,(snake_case_)			,(snake_case_)			,(snake_case_)			,(snake_case_)			,(snake_case_))									=				config_and_inputs
									snake_case_									=				{'''input_ids''': input_ids, '''attention_mask''': input_mask}
									return config, inputs_dict







@require_torch
class 						a				(   _lowerCamelCase   ,     _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		(
				    (
				        MPNetForMaskedLM,
				        MPNetForMultipleChoice,
				        MPNetForQuestionAnswering,
				        MPNetForSequenceClassification,
				        MPNetForTokenClassification,
				        MPNetModel,
				    )
				    if is_torch_available()
				    else ()
				)
				snake_case_      =		(
				    {
				        "feature-extraction": MPNetModel,
				        "fill-mask": MPNetForMaskedLM,
				        "question-answering": MPNetForQuestionAnswering,
				        "text-classification": MPNetForSequenceClassification,
				        "token-classification": MPNetForTokenClassification,
				        "zero-shot": MPNetForSequenceClassification,
				    }
				    if is_torch_available()
				    else {}
				)
				snake_case_      =		False
				snake_case_      =		True


				def   A_ (		self    :					Tuple					):
									snake_case_									=				MPNetModelTester(self					)
									snake_case_									=				ConfigTester(self       ,				config_class=lowercase_       ,				hidden_size=37					)


				def   A_ (		self    :					Union[str, Any]					):
									self.config_tester.run_common_tests()


				def   A_ (		self    :					Tuple					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_model(*lowercase_					)


				def   A_ (		self    :					List[Any]					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_sequence_classification(*lowercase_					)


				def   A_ (		self    :					List[Any]					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_multiple_choice(*lowercase_					)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_token_classification(*lowercase_					)


				def   A_ (		self    :					Tuple					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_question_answering(*lowercase_					)







@require_torch
class 						a				(   unittest.TestCase     ):


				@slow
				def   A_ (		self    :					List[Any]					):
									snake_case_									=				MPNetModel.from_pretrained('''microsoft/mpnet-base'''					)
									snake_case_									=				torch.tensor([[0, 345, 232, 328, 740, 140, 1695, 69, 6078, 1588, 2]]					)
									snake_case_									=				model(lowercase_					)[0]
									snake_case_									=				torch.Size((1, 11, 768)					)
									self.assertEqual(output.shape       ,				lowercase_					)
									snake_case_									=				torch.tensor(
									    [[[-0.0550, 0.1943, -0.0740], [-0.0562, 0.2211, -0.0579], [-0.0437, 0.3337, -0.0641]]]					)
									# compare the actual values for a slice.
									self.assertTrue(torch.allclose(output[:, :3, :3]       ,				lowercase_       ,				atol=1e-4					)					)

