
import os
import unittest

from transformers.models.phobert.tokenization_phobert import VOCAB_FILES_NAMES, PhobertTokenizer

from ...test_tokenization_common import TokenizerTesterMixin





class       __A(				UpperCAmelCase     ,		unittest.TestCase       ):
					SCREAMING_SNAKE_CASE			    =    PhobertTokenizer
					SCREAMING_SNAKE_CASE			    =    False

					def    lowercase__      (	self					:     List[str]   ):
												super().setUp()

												# Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
												lowerCamelCase_							  =					["""T@@""", """i""", """I""", """R@@""", """r""", """e@@"""]
												lowerCamelCase_							  =					dict(zip(__UpperCamelCase   ,					range(len(__UpperCamelCase   )   )   )   )
												lowerCamelCase_							  =					["""#version: 0.2""", """l à</w>"""]
												lowerCamelCase_							  =					{"""unk_token""": """<unk>"""}

												lowerCamelCase_							  =					os.path.join(self.tmpdirname   ,					VOCAB_FILES_NAMES["""vocab_file"""]   )
												lowerCamelCase_							  =					os.path.join(self.tmpdirname   ,					VOCAB_FILES_NAMES["""merges_file"""]   )

												with open(self.vocab_file   ,					"""w"""   ,					encoding="""utf-8"""   ) as fp:
																			for token in vocab_tokens:
																										fp.write(F'''{token} {vocab_tokens[token]}\n'''   )
												with open(self.merges_file   ,					"""w"""   ,					encoding="""utf-8"""   ) as fp:
																			fp.write("""\n""".join(__UpperCamelCase   )   )

					def    lowercase__      (	self					:     List[Any]   ,					**__UpperCamelCase					:     Dict   ):
												kwargs.update(self.special_tokens_map   )
												return PhobertTokenizer.from_pretrained(self.tmpdirname   ,					**__UpperCamelCase   )

					def    lowercase__      (	self					:     Union[str, Any]   ,					__UpperCamelCase					:     List[str]   ):
												lowerCamelCase_							  =					"""Tôi là VinAI Research"""
												lowerCamelCase_							  =					"""T<unk> i <unk> <unk> <unk> <unk> <unk> <unk> I Re<unk> e<unk> <unk> <unk> <unk>"""
												return input_text, output_text



					def    lowercase__      (	self					:     Dict   ):
												lowerCamelCase_							  =					PhobertTokenizer(self.vocab_file   ,					self.merges_file   ,					**self.special_tokens_map   )
												lowerCamelCase_							  =					"""Tôi là VinAI Research"""
												lowerCamelCase_							  =					"""T@@ ô@@ i l@@ à V@@ i@@ n@@ A@@ I R@@ e@@ s@@ e@@ a@@ r@@ c@@ h""".split()
												lowerCamelCase_							  =					tokenizer.tokenize(__UpperCamelCase   )
												print(__UpperCamelCase   )
												self.assertListEqual(__UpperCamelCase   ,					__UpperCamelCase   )

												lowerCamelCase_							  =					tokens + [tokenizer.unk_token]

												lowerCamelCase_							  =					[4, 3, 5, 3, 3, 3, 3, 3, 3, 6, 7, 9, 3, 9, 3, 3, 3, 3, 3]
												self.assertListEqual(tokenizer.convert_tokens_to_ids(__UpperCamelCase   )   ,					__UpperCamelCase   )







import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
		from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
		import torch_xla.core.xla_model as xm



def      __lowerCAmelCase						(							UpperCAmelCase__				:      Any       )  ->    List[str]:
							if is_torch_version("""<"""			,				"""2.0.0"""       ) or not hasattr(UpperCAmelCase__			,				"""_dynamo"""       ):
														return False
							return isinstance(UpperCAmelCase__			,				torch._dynamo.eval_frame.OptimizedModule       )



def      __lowerCAmelCase						(							UpperCAmelCase__				:      str			,				UpperCAmelCase__				:      bool = True       )  ->    str:
							lowerCamelCase_							  =					(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

							lowerCamelCase_							  =					is_compiled_module(UpperCAmelCase__       )
							if is_compiled:
														lowerCamelCase_							  =					model
														lowerCamelCase_							  =					model._orig_mod

							if is_deepspeed_available():
														options += (DeepSpeedEngine,)

							while isinstance(UpperCAmelCase__			,				UpperCAmelCase__       ):
														lowerCamelCase_							  =					model.module

							if not keep_fpaa_wrapper:
														lowerCamelCase_							  =					getattr(UpperCAmelCase__			,				"""forward"""       )
														lowerCamelCase_							  =					model.__dict__.pop("""_original_forward"""			,				UpperCAmelCase__       )
														if original_forward is not None:
																					while hasattr(UpperCAmelCase__			,				"""__wrapped__"""       ):
																												lowerCamelCase_							  =					forward.__wrapped__
																												if forward == original_forward:
																																			break
																					lowerCamelCase_							  =					forward
														if getattr(UpperCAmelCase__			,				"""_converted_to_transformer_engine"""			,				UpperCAmelCase__       ):
																					convert_model(UpperCAmelCase__			,				to_transformer_engine=UpperCAmelCase__       )

							if is_compiled:
														lowerCamelCase_							  =					model
														lowerCamelCase_							  =					compiled_model

							return model



def      __lowerCAmelCase						(							)  ->    Any:
							PartialState().wait_for_everyone()



def      __lowerCAmelCase						(							UpperCAmelCase__				:      Union[str, Any]			,				UpperCAmelCase__				:      Tuple       )  ->    Dict:
							if PartialState().distributed_type == DistributedType.TPU:
														xm.save(UpperCAmelCase__			,				UpperCAmelCase__       )
							elif PartialState().local_process_index == 0:
														torch.save(UpperCAmelCase__			,				UpperCAmelCase__       )



@contextmanager
def      __lowerCAmelCase						(							**UpperCAmelCase__				:      Any       )  ->    Dict:
							for key, value in kwargs.items():
														lowerCamelCase_							  =					str(UpperCAmelCase__       )

							yield

							for key in kwargs:
														if key.upper() in os.environ:
																					del os.environ[key.upper()]



def      __lowerCAmelCase						(							UpperCAmelCase__				:      Union[str, Any]       )  ->    str:
							if not hasattr(UpperCAmelCase__			,				"""__qualname__"""       ) and not hasattr(UpperCAmelCase__			,				"""__name__"""       ):
														lowerCamelCase_							  =					getattr(UpperCAmelCase__			,				"""__class__"""			,				UpperCAmelCase__       )
							if hasattr(UpperCAmelCase__			,				"""__qualname__"""       ):
														return obj.__qualname__
							if hasattr(UpperCAmelCase__			,				"""__name__"""       ):
														return obj.__name__
							return str(UpperCAmelCase__       )



def      __lowerCAmelCase						(							UpperCAmelCase__				:      str			,				UpperCAmelCase__				:      Union[str, Any]       )  ->    Dict:
							for key, value in source.items():
														if isinstance(UpperCAmelCase__			,				UpperCAmelCase__       ):
																					lowerCamelCase_							  =					destination.setdefault(UpperCAmelCase__			,				{}       )
																					merge_dicts(UpperCAmelCase__			,				UpperCAmelCase__       )
														else:
																					lowerCamelCase_							  =					value

							return destination



def      __lowerCAmelCase						(							UpperCAmelCase__				:      int = None       )  ->    bool:
							if port is None:
														lowerCamelCase_							  =					2_9_5_0_0
							with socket.socket(socket.AF_INET			,				socket.SOCK_STREAM       ) as s:
														return s.connect_ex(("""localhost""", port)       ) == 0





