"""simple docstring"""







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


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

_a					:					int=			{'vocab_file': 'vocab.txt'}

_a					:					Any=			{
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

_a					:					Optional[Any]=			{
    'openbmb/cpm-ant-10b': 1_024,
}
def   __UpperCAmelCase    (   UpperCAmelCase_  )							->      Optional[int]:

	'''simple docstring'''





	__snake_case      :							Optional[int]						=			collections.OrderedDict()
	with open(snake_case_      ,							'r'      ,							encoding='utf-8'  ) as reader:
		__snake_case      :							int						=			reader.readlines()
	for index, token in enumerate(snake_case_  ):
		__snake_case      :							Any						=			token.rstrip('\n'  )
		__snake_case      :							List[Any]						=			index
	return vocab

class 		UpperCamelCase      (   _lowerCAmelCase						):



				def __init__(self    :				Optional[int]     ,  _A    :				List[str]     ,  _A    :				str="<unk>"     ,  _A    :				Optional[Any]=2_00)   ->     int:
					__snake_case      :							str						=			vocab
					__snake_case      :							Tuple						=			unk_token
					__snake_case      :							Optional[Any]						=			max_input_chars_per_word






				def  _lowercase					(self    :				int     ,  _A    :				List[Any])   ->     str:
					__snake_case      :							Dict						=			list(_lowercase)
					if len(_lowercase) > self.max_input_chars_per_word:
						return [self.unk_token]

					__snake_case      :							List[str]						=			0
					__snake_case      :							List[Any]						=			[]
					while start < len(_lowercase):
						__snake_case      :							Union[str, Any]						=			len(_lowercase)
						__snake_case      :							Optional[Any]						=			None
						while start < end:
							__snake_case      :							Any						=			''.join(chars[start:end])
							if substr in self.vocab:
								__snake_case      :							Tuple						=			substr
								break
							end -= 1
						if cur_substr is None:
							sub_tokens.append(self.unk_token)
							start += 1
						else:
							sub_tokens.append(_lowercase)
							__snake_case      :							List[str]						=			end

					return sub_tokens







class 		UpperCamelCase      (   _lowerCAmelCase						):
				UpperCAmelCase    :				List[str]   =		VOCAB_FILES_NAMES
				UpperCAmelCase    :				Union[str, Any]   =		PRETRAINED_VOCAB_FILES_MAP
				UpperCAmelCase    :				List[Any]   =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				UpperCAmelCase    :				Optional[Any]   =		["input_ids", "attention_mask"]
				UpperCAmelCase    :				List[Any]   =		False



				def __init__(self    :				List[Any]     ,  _A    :				Union[str, Any]     ,  _A    :				Optional[Any]="<d>"     ,  _A    :				int="</d>"     ,  _A    :				Tuple="<s>"     ,  _A    :				List[str]="</s>"     ,  _A    :				Optional[int]="<pad>"     ,  _A    :				str="<unk>"     ,  _A    :				int="</n>"     ,  _A    :				Optional[int]="</_>"     ,  _A    :				List[str]="left"     ,  **_A    :				Optional[Any]     ,  )   ->     str:
					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     ,  )
					__snake_case      :							int						=			bod_token
					__snake_case      :							Optional[Any]						=			eod_token
					__snake_case      :							Dict						=			load_vocab(_lowercase)
					__snake_case      :							List[str]						=			self.encoder[space_token]
					__snake_case      :							Optional[Any]						=			self.encoder[line_token]

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

					__snake_case      :							Union[str, Any]						=			collections.OrderedDict(sorted(self.encoder.items()     ,  key=lambda _A: x[1]))
					__snake_case      :							Any						=			{v: k for k, v in self.encoder.items()}

					__snake_case      :							Any						=			WordpieceTokenizer(vocab=self.encoder     ,  unk_token=self.unk_token)



				@property
				def  _lowercase					(self    :				int)   ->     List[Any]:
					return self.encoder[self.bod_token]



				@property
				def  _lowercase					(self    :				Optional[int])   ->     Tuple:
					return self.encoder[self.eod_token]



				@property
				def  _lowercase					(self    :				Optional[Any])   ->     Optional[Any]:
					return self.encoder["\n"]



				@property
				def  _lowercase					(self    :				Dict)   ->     List[Any]:
					return len(self.encoder)



				def  _lowercase					(self    :				Any)   ->     str:
					return dict(self.encoder     ,  **self.added_tokens_encoder)



				def  _lowercase					(self    :				str     ,  _A    :				Dict)   ->     str:
					__snake_case      :							Optional[int]						=			[]
					for x in jieba.cut(_lowercase     ,  cut_all=_lowercase):
						output_tokens.extend(self.wordpiece_tokenizer.tokenize(_lowercase))
					return output_tokens



				def  _lowercase					(self    :				Optional[Any]     ,  _A    :				Optional[int]     ,  **_A    :				Dict)   ->     str:
					__snake_case      :							List[str]						=			[i for i in token_ids if i >= 0]
					__snake_case      :							List[str]						=			[
					    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  _lowercase					(self    :				Any     ,  _A    :				int)   ->     int:
					return token in self.encoder



				def  _lowercase					(self    :				int     ,  _A    :				List[str])   ->     Dict:
					return "".join(_lowercase)



				def  _lowercase					(self    :				int     ,  _A    :				int)   ->     List[Any]:
					return self.encoder.get(_lowercase     ,  self.encoder.get(self.unk_token))



				def  _lowercase					(self    :				List[Any]     ,  _A    :				Any)   ->     List[Any]:
					return self.decoder.get(_lowercase     ,  self.unk_token)



				def  _lowercase					(self    :				Any     ,  _A    :				str     ,  _A    :				Optional[str] = None)   ->     Optional[Any]:
					if os.path.isdir(_lowercase):
						__snake_case      :							int						=			os.path.join(
						    _lowercase     ,  (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'])
					else:
						__snake_case      :							Dict						=			(filename_prefix + '-' if filename_prefix else '') + save_directory
					__snake_case      :							Tuple						=			0
					if " " in self.encoder:
						__snake_case      :							Optional[Any]						=			self.encoder[' ']
						del self.encoder[" "]
					if "\n" in self.encoder:
						__snake_case      :							Any						=			self.encoder['\n']
						del self.encoder["\n"]
					__snake_case      :							List[Any]						=			collections.OrderedDict(sorted(self.encoder.items()     ,  key=lambda _A: 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!')
								__snake_case      :							str						=			token_index
							writer.write(token + '\n')
							index += 1
					return (vocab_file,)



				def  _lowercase					(self    :				Any     ,  _A    :				List[int]     ,  _A    :				List[int] = None)   ->     int:
					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  _lowercase					(self    :				Union[str, Any]     ,  _A    :				List[int]     ,  _A    :				Optional[List[int]] = None     ,  _A    :				bool = False)   ->     int:

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






"""simple docstring"""







import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch

class 		UpperCamelCase      (   lowercase						):



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Optional[Any]:
					# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
					# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
					# while running an external program

					# python one-liner segments

					# this must be loaded before socket.socket is monkey-patched
					__snake_case      :							Any						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Tuple						=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

					__snake_case      :							int						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

					# Force fetching the files so that we can use the cache
					__snake_case      :							int						=			'hf-internal-testing/tiny-random-bert'
					BertConfig.from_pretrained(_A)
					BertModel.from_pretrained(_A)
					BertTokenizer.from_pretrained(_A)
					pipeline(task='fill-mask'     ,  model=_A)

					# baseline - just load from_pretrained with normal network
					__snake_case      :							Union[str, Any]						=			[sys.executable, '-c', '\n'.join([load, run, mock])]

					# should succeed
					__snake_case      :							Union[str, Any]						=			self.get_env()
					# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
					__snake_case      :							str						=			'1'
					__snake_case      :							Union[str, Any]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Union[str, Any]:
					# python one-liner segments
					# this must be loaded before socket.socket is monkey-patched
					__snake_case      :							Optional[Any]						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Optional[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

					__snake_case      :							Union[str, Any]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

					# Force fetching the files so that we can use the cache
					__snake_case      :							str						=			'hf-internal-testing/tiny-random-bert'
					BertConfig.from_pretrained(_A)
					BertModel.from_pretrained(_A)
					BertTokenizer.from_pretrained(_A)
					pipeline(task='fill-mask'     ,  model=_A)

					# baseline - just load from_pretrained with normal network
					__snake_case      :							Any						=			[sys.executable, '-c', '\n'.join([load, run, mock])]

					# should succeed
					__snake_case      :							int						=			self.get_env()
					__snake_case      :							Tuple						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				int)   ->     Any:
					# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
					# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
					# while running an external program

					# python one-liner segments

					# this must be loaded before socket.socket is monkey-patched
					__snake_case      :							int						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

					__snake_case      :							int						=			'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

					__snake_case      :							Optional[int]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

					# baseline - just load from_pretrained with normal network
					__snake_case      :							Optional[Any]						=			[sys.executable, '-c', '\n'.join([load, run])]

					# should succeed
					__snake_case      :							Optional[int]						=			self.get_env()
					__snake_case      :							Dict						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())

					# next emulate no network
					__snake_case      :							Optional[Any]						=			[sys.executable, '-c', '\n'.join([load, mock, run])]

					# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
					# env["TRANSFORMERS_OFFLINE"] = "0"
					# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
					# self.assertEqual(result.returncode, 1, result.stderr)

					# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
					__snake_case      :							Union[str, Any]						=			'1'
					__snake_case      :							str						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				str)   ->     Dict:
					__snake_case      :							Dict						=			'\nfrom transformers import pipeline\n        '
					__snake_case      :							List[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

					__snake_case      :							List[str]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Tuple						=			'1'
					__snake_case      :							Tuple						=			[sys.executable, '-c', '\n'.join([load, mock, run])]
					__snake_case      :							Optional[Any]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  1     ,  result.stderr)
					self.assertIn(
					    'You cannot infer task automatically within `pipeline` when using offline mode'     ,  result.stderr.decode().replace('\n'     ,  '')     ,  )






				@require_torch
				def  _lowercase					(self    :				int)   ->     Optional[Any]:
					__snake_case      :							int						=			'\nfrom transformers import AutoModel\n        '
					__snake_case      :							str						=			'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

					# baseline - just load from_pretrained with normal network
					__snake_case      :							str						=			[sys.executable, '-c', '\n'.join([load, run])]

					# should succeed
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Dict						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())

					# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
					__snake_case      :							List[str]						=			'1'
					__snake_case      :							Optional[int]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())





