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






"""simple docstring"""







import json
from typing import List, Optional, Tuple

from tokenizers import normalizers
from tokenizers.pre_tokenizers import BertPreTokenizer, PreTokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roformer import RoFormerTokenizer
from .tokenization_utils import JiebaPreTokenizer


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

_a					:					Any=			{"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

_a					:					int=			{
    "vocab_file": {
        "junnyu/roformer_chinese_small": "https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/vocab.txt",
        "junnyu/roformer_chinese_base": "https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/vocab.txt",
        "junnyu/roformer_chinese_char_small": (
            "https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_chinese_char_base": (
            "https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_small_discriminator": (
            "https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_small_generator": (
            "https://huggingface.co/junnyu/roformer_small_generator/resolve/main/vocab.txt"
        ),
    }
}

_a					:					Optional[Any]=			{
    "junnyu/roformer_chinese_small": 1_536,
    "junnyu/roformer_chinese_base": 1_536,
    "junnyu/roformer_chinese_char_small": 512,
    "junnyu/roformer_chinese_char_base": 512,
    "junnyu/roformer_small_discriminator": 128,
    "junnyu/roformer_small_generator": 128,
}


_a					:					str=			{
    "junnyu/roformer_chinese_small": {"do_lower_case": True},
    "junnyu/roformer_chinese_base": {"do_lower_case": True},
    "junnyu/roformer_chinese_char_small": {"do_lower_case": True},
    "junnyu/roformer_chinese_char_base": {"do_lower_case": True},
    "junnyu/roformer_small_discriminator": {"do_lower_case": True},
    "junnyu/roformer_small_generator": {"do_lower_case": True},
}

class 		UpperCamelCase      (   lowercase						):
				UpperCAmelCase    :				Union[str, Any]   =		VOCAB_FILES_NAMES
				UpperCAmelCase    :				Optional[int]   =		PRETRAINED_VOCAB_FILES_MAP
				UpperCAmelCase    :				List[str]   =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				UpperCAmelCase    :				int   =		PRETRAINED_INIT_CONFIGURATION
				UpperCAmelCase    :				Dict   =		RoFormerTokenizer



				def __init__(self    :				List[Any]     ,  _A    :				Any=None     ,  _A    :				int=None     ,  _A    :				Dict=True     ,  _A    :				List[Any]="[UNK]"     ,  _A    :				Tuple="[SEP]"     ,  _A    :				List[Any]="[PAD]"     ,  _A    :				str="[CLS]"     ,  _A    :				int="[MASK]"     ,  _A    :				Optional[int]=True     ,  _A    :				List[str]=None     ,  **_A    :				int     ,  )   ->     Dict:
					super().__init__(
					    _A     ,  tokenizer_file=_A     ,  do_lower_case=_A     ,  unk_token=_A     ,  sep_token=_A     ,  pad_token=_A     ,  cls_token=_A     ,  mask_token=_A     ,  tokenize_chinese_chars=_A     ,  strip_accents=_A     ,  **_A     ,  )

					__snake_case      :							List[Any]						=			json.loads(self.backend_tokenizer.normalizer.__getstate__())
					if (
					    pre_tok_state.get('lowercase'     ,  _A) != do_lower_case
					    or pre_tok_state.get('strip_accents'     ,  _A) != strip_accents
					):
						__snake_case      :							Union[str, Any]						=			getattr(_A     ,  pre_tok_state.pop('type'))
						__snake_case      :							Union[str, Any]						=			do_lower_case
						__snake_case      :							str						=			strip_accents
						__snake_case      :							Optional[int]						=			pre_tok_class(**_A)

					__snake_case      :							int						=			do_lower_case



				def __getstate__(self    :				Optional[Any])   ->     Dict:
					__snake_case      :							Optional[int]						=			self.__dict__.copy()
					__snake_case      :							int						=			BertPreTokenizer()
					return state



				def __setstate__(self    :				Optional[Any]     ,  _A    :				Optional[Any])   ->     Dict:
					__snake_case      :							List[str]						=			d
					__snake_case      :							str						=			self.__dict__['_tokenizer'].get_vocab()
					__snake_case      :							int						=			PreTokenizer.custom(JiebaPreTokenizer(_A))



				def  _lowercase					(self    :				int     ,  _A    :				Tuple     ,  _A    :				Any=None)   ->     str:
					__snake_case      :							Dict						=			[self.cls_token_id] + token_ids_a + [self.sep_token_id]

					if token_ids_a:
						output += token_ids_a + [self.sep_token_id]

					return output



				def  _lowercase					(self    :				List[str]     ,  _A    :				List[int]     ,  _A    :				Optional[List[int]] = None)   ->     List[int]:
					__snake_case      :							Tuple						=			[self.sep_token_id]
					__snake_case      :							Optional[Any]						=			[self.cls_token_id]
					if token_ids_a is None:
						return len(cls + token_ids_a + sep) * [0]
					return len(cls + token_ids_a + sep) * [0] + len(token_ids_a + sep) * [1]



				def  _lowercase					(self    :				List[Any]     ,  _A    :				str     ,  _A    :				Optional[str] = None)   ->     Tuple[str]:
					__snake_case      :							List[Any]						=			self._tokenizer.model.save(_A     ,  name=_A)
					return tuple(_A)






				def  _lowercase					(self    :				int     ,  _A    :				Optional[int]     ,  _A    :				Tuple=None     ,  _A    :				Tuple=None     ,  _A    :				Dict=False     ,  **_A    :				Optional[int]     ,  )   ->     Optional[Any]:
					__snake_case      :							Optional[Any]						=			BertPreTokenizer()
					return super().save_pretrained(_A     ,  _A     ,  _A     ,  _A     ,  **_A)





