


import os
import shutil
import sys
import tempfile
import unittest
from pathlib import Path

import pytest

import transformers
from transformers import (
    BERT_PRETRAINED_CONFIG_ARCHIVE_MAP,
    GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP,
    AutoTokenizer,
    BertConfig,
    BertTokenizer,
    BertTokenizerFast,
    CTRLTokenizer,
    GPTaTokenizer,
    GPTaTokenizerFast,
    PreTrainedTokenizerFast,
    RobertaTokenizer,
    RobertaTokenizerFast,
    is_tokenizers_available,
)
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.auto.tokenization_auto import (
    TOKENIZER_MAPPING,
    get_tokenizer_config,
    tokenizer_class_from_name,
)
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import (
    DUMMY_DIFF_TOKENIZER_IDENTIFIER,
    DUMMY_UNKNOWN_IDENTIFIER,
    SMALL_MODEL_IDENTIFIER,
    RequestCounter,
    require_tokenizers,
    slow,
)


sys.path.append(str(Path(__file__).parent.parent.parent.parent / "utils"))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
			from test_module.custom_tokenization_fast import CustomTokenizerFast

class        a	(unittest.TestCase   ):




							"""simple docstring"""


							def 					__snake_case      (							self :						Any       )			->					Optional[Any]:
														__snake_case	:    Optional[int]											=     0

							@slow
							def 					__snake_case      (							self :						int       )			->					Optional[Any]:
														for model_name in (x for x in BERT_PRETRAINED_CONFIG_ARCHIVE_MAP.keys() if "japanese" not in x):
																					__snake_case	:    Optional[int]											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
																					self.assertIsNotNone(lowerCamelCase       )
																					self.assertIsInstance(lowerCamelCase					,			(BertTokenizer, BertTokenizerFast)       )
																					self.assertGreater(len(lowerCamelCase       )					,			0       )

														for model_name in GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP.keys():
																					__snake_case	:    List[str]											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
																					self.assertIsNotNone(lowerCamelCase       )
																					self.assertIsInstance(lowerCamelCase					,			(GPTaTokenizer, GPTaTokenizerFast)       )
																					self.assertGreater(len(lowerCamelCase       )					,			0       )

							def 					__snake_case      (							self :						Dict       )			->					str:
														__snake_case	:    List[str]											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
														self.assertIsInstance(lowerCamelCase					,			(BertTokenizer, BertTokenizerFast)       )
														self.assertEqual(tokenizer.vocab_size					,			12       )

							def 					__snake_case      (							self :						Dict       )			->					Optional[Any]:
														__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
														self.assertIsInstance(lowerCamelCase					,			(RobertaTokenizer, RobertaTokenizerFast)       )
														self.assertEqual(tokenizer.vocab_size					,			20       )

							def 					__snake_case      (							self :						Any       )			->					List[Any]:
														__snake_case	:    Dict											=     AutoConfig.from_pretrained(lowerCamelCase       )
														self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )
														# Check that tokenizer_type ≠ model_type
														__snake_case	:    int											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			config=lowerCamelCase       )
														self.assertIsInstance(lowerCamelCase					,			(BertTokenizer, BertTokenizerFast)       )
														self.assertEqual(tokenizer.vocab_size					,			12       )

							def 					__snake_case      (							self :						List[Any]       )			->					Any:
														with tempfile.TemporaryDirectory() as tmp_dir:
																					shutil.copy("./tests/fixtures/vocab.txt"					,			os.path.join(lowerCamelCase					,			"vocab.txt"       )       )

																					__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			tokenizer_type="bert"					,			use_fast=lowerCamelCase       )
																					self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

														with tempfile.TemporaryDirectory() as tmp_dir:
																					shutil.copy("./tests/fixtures/vocab.json"					,			os.path.join(lowerCamelCase					,			"vocab.json"       )       )
																					shutil.copy("./tests/fixtures/merges.txt"					,			os.path.join(lowerCamelCase					,			"merges.txt"       )       )

																					__snake_case	:    Union[str, Any]											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			tokenizer_type="gpt2"					,			use_fast=lowerCamelCase       )
																					self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

							@require_tokenizers
							def 					__snake_case      (							self :						List[str]       )			->					Dict:
														with tempfile.TemporaryDirectory() as tmp_dir:
																					shutil.copy("./tests/fixtures/vocab.txt"					,			os.path.join(lowerCamelCase					,			"vocab.txt"       )       )

																					__snake_case	:    Any											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			tokenizer_type="bert"       )
																					self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

														with tempfile.TemporaryDirectory() as tmp_dir:
																					shutil.copy("./tests/fixtures/vocab.json"					,			os.path.join(lowerCamelCase					,			"vocab.json"       )       )
																					shutil.copy("./tests/fixtures/merges.txt"					,			os.path.join(lowerCamelCase					,			"merges.txt"       )       )

																					__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			tokenizer_type="gpt2"       )
																					self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

							def 					__snake_case      (							self :						Optional[int]       )			->					Dict:
														with pytest.raises(lowerCamelCase       ):
																					AutoTokenizer.from_pretrained("./"					,			tokenizer_type="xxx"       )

							@require_tokenizers
							def 					__snake_case      (							self :						List[str]       )			->					Optional[Any]:
														for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
																					__snake_case	:    str											=     tokenizer_class.from_pretrained("wietsedv/bert-base-dutch-cased"       )
																					self.assertIsInstance(lowerCamelCase					,			(BertTokenizer, BertTokenizerFast)       )

																					if isinstance(lowerCamelCase					,			lowerCamelCase       ):
																												self.assertEqual(tokenizer.basic_tokenizer.do_lower_case					,			lowerCamelCase       )
																					else:
																												self.assertEqual(tokenizer.do_lower_case					,			lowerCamelCase       )

																					self.assertEqual(tokenizer.model_max_length					,			512       )

							@require_tokenizers
							def 					__snake_case      (							self :						Dict       )			->					List[Any]:
														for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
																					with self.assertRaisesRegex(
																					    lowerCamelCase					,			"julien-c/herlolip-not-exists is not a local folder and is not a valid model identifier"					,			):
																												__snake_case	:    Union[str, Any]											=     tokenizer_class.from_pretrained("julien-c/herlolip-not-exists"       )

							def 					__snake_case      (							self :						List[str]       )			->					List[str]:
														# tests: https://github.com/huggingface/transformers/pull/13251
														# 1. models with `-`, e.g. xlm-roberta -> xlm_roberta
														# 2. models that don't remap 1-1 from model-name to model file, e.g., openai-gpt -> openai
														__snake_case	:    Union[str, Any]											=     TOKENIZER_MAPPING.values()
														__snake_case	:    List[str]											=     []

														for slow_tok, fast_tok in tokenizers:
																					if slow_tok is not None:
																												tokenizer_names.append(slow_tok.__name__       )

																					if fast_tok is not None:
																												tokenizer_names.append(fast_tok.__name__       )

														for tokenizer_name in tokenizer_names:
																					# must find the right class
																					tokenizer_class_from_name(lowerCamelCase       )

							@require_tokenizers
							def 					__snake_case      (							self :						str       )			->					Optional[int]:
														self.assertIsInstance(AutoTokenizer.from_pretrained("bert-base-cased"					,			use_fast=lowerCamelCase       )					,			lowerCamelCase       )
														self.assertIsInstance(AutoTokenizer.from_pretrained("bert-base-cased"       )					,			lowerCamelCase       )

							@require_tokenizers
							def 					__snake_case      (							self :						Optional[Any]       )			->					Optional[Any]:
														__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained("distilbert-base-uncased"					,			do_lower_case=lowerCamelCase       )
														__snake_case	:    Dict											=     "Hello, world. How are you?"
														__snake_case	:    Tuple											=     tokenizer.tokenize(lowerCamelCase       )
														self.assertEqual("[UNK]"					,			tokens[0]       )

														__snake_case	:    Union[str, Any]											=     AutoTokenizer.from_pretrained("microsoft/mpnet-base"					,			do_lower_case=lowerCamelCase       )
														__snake_case	:    List[str]											=     tokenizer.tokenize(lowerCamelCase       )
														self.assertEqual("[UNK]"					,			tokens[0]       )

							@require_tokenizers
							def 					__snake_case      (							self :						Union[str, Any]       )			->					Dict:
														__snake_case	:    Dict											=     AutoTokenizer.from_pretrained("robot-test/dummy-tokenizer-fast-with-model-config"       )
														self.assertEqual(type(lowerCamelCase       )					,			lowerCamelCase       )
														self.assertEqual(tokenizer.model_max_length					,			512       )
														self.assertEqual(tokenizer.vocab_size					,			30000       )
														self.assertEqual(tokenizer.unk_token					,			"[UNK]"       )
														self.assertEqual(tokenizer.padding_side					,			"right"       )
														self.assertEqual(tokenizer.truncation_side					,			"right"       )

							def 					__snake_case      (							self :						Dict       )			->					List[Any]:
														__snake_case	:    List[str]											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
														self.assertIsInstance(lowerCamelCase					,			(BertTokenizer, BertTokenizerFast)       )
														with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer.save_pretrained(lowerCamelCase       )
																					__snake_case	:    Optional[int]											=     AutoTokenizer.from_pretrained(lowerCamelCase       )

														self.assertIsInstance(lowerCamelCase					,			tokenizer.__class__       )
														self.assertEqual(tokenizera.vocab_size					,			12       )

							def 					__snake_case      (							self :						List[Any]       )			->					Tuple:
														__snake_case	:    str											=     AutoTokenizer.from_pretrained("ctrl"       )
														# There is no fast CTRL so this always gives us a slow tokenizer.
														self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

							def 					__snake_case      (							self :						Optional[Any]       )			->					Optional[int]:
														# Check we can load the tokenizer config of an online model.
														__snake_case	:    Optional[Any]											=     get_tokenizer_config("bert-base-cased"       )
														__snake_case	:    str											=     config.pop("_commit_hash"					,			lowerCamelCase       )
														# If we ever update bert-base-cased tokenizer config, this dict here will need to be updated.
														self.assertEqual(lowerCamelCase					,			{"do_lower_case": False}       )

														# This model does not have a tokenizer_config so we get back an empty dict.
														__snake_case	:    Optional[Any]											=     get_tokenizer_config(lowerCamelCase       )
														self.assertDictEqual(lowerCamelCase					,			{}       )

														# A tokenizer saved with `save_pretrained` always creates a tokenizer config.
														__snake_case	:    str											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
														with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer.save_pretrained(lowerCamelCase       )
																					__snake_case	:    Optional[Any]											=     get_tokenizer_config(lowerCamelCase       )

														# Check the class of the tokenizer was properly saved (note that it always saves the slow class).
														self.assertEqual(config["tokenizer_class"]					,			"BertTokenizer"       )

							def 					__snake_case      (							self :						Optional[int]       )			->					Optional[Any]:
														try:
																					AutoConfig.register("custom"					,			lowerCamelCase       )

																					AutoTokenizer.register(lowerCamelCase					,			slow_tokenizer_class=lowerCamelCase       )
																					# Trying to register something existing in the Transformers library will raise an error
																					with self.assertRaises(lowerCamelCase       ):
																												AutoTokenizer.register(lowerCamelCase					,			slow_tokenizer_class=lowerCamelCase       )

																					__snake_case	:    Optional[int]											=     CustomTokenizer.from_pretrained(lowerCamelCase       )
																					with tempfile.TemporaryDirectory() as tmp_dir:
																												tokenizer.save_pretrained(lowerCamelCase       )

																												__snake_case	:    Tuple											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
																												self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

														finally:
																					if "custom" in CONFIG_MAPPING._extra_content:
																												del CONFIG_MAPPING._extra_content["custom"]
																					if CustomConfig in TOKENIZER_MAPPING._extra_content:
																												del TOKENIZER_MAPPING._extra_content[CustomConfig]

							@require_tokenizers
							def 					__snake_case      (							self :						List[str]       )			->					Dict:
														try:
																					AutoConfig.register("custom"					,			lowerCamelCase       )

																					# Can register in two steps
																					AutoTokenizer.register(lowerCamelCase					,			slow_tokenizer_class=lowerCamelCase       )
																					self.assertEqual(TOKENIZER_MAPPING[CustomConfig]					,			(CustomTokenizer, None)       )
																					AutoTokenizer.register(lowerCamelCase					,			fast_tokenizer_class=lowerCamelCase       )
																					self.assertEqual(TOKENIZER_MAPPING[CustomConfig]					,			(CustomTokenizer, CustomTokenizerFast)       )

																					del TOKENIZER_MAPPING._extra_content[CustomConfig]
																					# Can register in one step
																					AutoTokenizer.register(
																					    lowerCamelCase					,			slow_tokenizer_class=lowerCamelCase					,			fast_tokenizer_class=lowerCamelCase       )
																					self.assertEqual(TOKENIZER_MAPPING[CustomConfig]					,			(CustomTokenizer, CustomTokenizerFast)       )

																					# Trying to register something existing in the Transformers library will raise an error
																					with self.assertRaises(lowerCamelCase       ):
																												AutoTokenizer.register(lowerCamelCase					,			fast_tokenizer_class=lowerCamelCase       )

																					# We pass through a bert tokenizer fast cause there is no converter slow to fast for our new toknizer
																					# and that model does not have a tokenizer.json
																					with tempfile.TemporaryDirectory() as tmp_dir:
																												__snake_case	:    Optional[Any]											=     BertTokenizerFast.from_pretrained(lowerCamelCase       )
																												bert_tokenizer.save_pretrained(lowerCamelCase       )
																												__snake_case	:    Union[str, Any]											=     CustomTokenizerFast.from_pretrained(lowerCamelCase       )

																					with tempfile.TemporaryDirectory() as tmp_dir:
																												tokenizer.save_pretrained(lowerCamelCase       )

																												__snake_case	:    str											=     AutoTokenizer.from_pretrained(lowerCamelCase       )
																												self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

																												__snake_case	:    Dict											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			use_fast=lowerCamelCase       )
																												self.assertIsInstance(lowerCamelCase					,			lowerCamelCase       )

														finally:
																					if "custom" in CONFIG_MAPPING._extra_content:
																												del CONFIG_MAPPING._extra_content["custom"]
																					if CustomConfig in TOKENIZER_MAPPING._extra_content:
																												del TOKENIZER_MAPPING._extra_content[CustomConfig]

							def 					__snake_case      (							self :						str       )			->					int:
														# If remote code is not set, we will time out when asking whether to load the model.
														with self.assertRaises(lowerCamelCase       ):
																					__snake_case	:    Optional[Any]											=     AutoTokenizer.from_pretrained("hf-internal-testing/test_dynamic_tokenizer"       )
														# If remote code is disabled, we can't load this config.
														with self.assertRaises(lowerCamelCase       ):
																					__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase       )

														__snake_case	:    Union[str, Any]											=     AutoTokenizer.from_pretrained("hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase       )
														self.assertTrue(tokenizer.special_attribute_present       )
														# Test tokenizer can be reloaded.
														with tempfile.TemporaryDirectory() as tmp_dir:
																					tokenizer.save_pretrained(lowerCamelCase       )
																					__snake_case	:    Optional[Any]											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			trust_remote_code=lowerCamelCase       )
														self.assertTrue(reloaded_tokenizer.special_attribute_present       )

														if is_tokenizers_available():
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizerFast"       )
																					self.assertEqual(reloaded_tokenizer.__class__.__name__					,			"NewTokenizerFast"       )

																					# Test we can also load the slow version
																					__snake_case	:    Union[str, Any]											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase					,			use_fast=lowerCamelCase       )
																					self.assertTrue(tokenizer.special_attribute_present       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )
																					# Test tokenizer can be reloaded.
																					with tempfile.TemporaryDirectory() as tmp_dir:
																												tokenizer.save_pretrained(lowerCamelCase       )
																												__snake_case	:    Dict											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			trust_remote_code=lowerCamelCase					,			use_fast=lowerCamelCase       )
																					self.assertEqual(reloaded_tokenizer.__class__.__name__					,			"NewTokenizer"       )
																					self.assertTrue(reloaded_tokenizer.special_attribute_present       )
														else:
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )
																					self.assertEqual(reloaded_tokenizer.__class__.__name__					,			"NewTokenizer"       )

							@require_tokenizers
							def 					__snake_case      (							self :						int       )			->					Dict:

														class        a	(_lowerCAmelCase   ):




																					"""simple docstring"""


																					__UpperCAmelCase					:							List[str]  						=						False

														class        a	(_lowerCAmelCase   ):




																					"""simple docstring"""


																					__UpperCAmelCase					:							Optional[Any]  						=						NewTokenizer
																					__UpperCAmelCase					:							List[Any]  						=						False

														try:
																					AutoConfig.register("custom"					,			lowerCamelCase       )
																					AutoTokenizer.register(lowerCamelCase					,			slow_tokenizer_class=lowerCamelCase       )
																					AutoTokenizer.register(lowerCamelCase					,			fast_tokenizer_class=lowerCamelCase       )
																					# If remote code is not set, the default is to use local
																					__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained("hf-internal-testing/test_dynamic_tokenizer"       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizerFast"       )
																					self.assertFalse(tokenizer.special_attribute_present       )
																					__snake_case	:    Any											=     AutoTokenizer.from_pretrained("hf-internal-testing/test_dynamic_tokenizer"					,			use_fast=lowerCamelCase       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )
																					self.assertFalse(tokenizer.special_attribute_present       )

																					# If remote code is disabled, we load the local one.
																					__snake_case	:    str											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizerFast"       )
																					self.assertFalse(tokenizer.special_attribute_present       )
																					__snake_case	:    Union[str, Any]											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase					,			use_fast=lowerCamelCase       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )
																					self.assertFalse(tokenizer.special_attribute_present       )

																					# If remote is enabled, we load from the Hub
																					__snake_case	:    Any											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizerFast"       )
																					self.assertTrue(tokenizer.special_attribute_present       )
																					__snake_case	:    int											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer"					,			trust_remote_code=lowerCamelCase					,			use_fast=lowerCamelCase       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )
																					self.assertTrue(tokenizer.special_attribute_present       )

														finally:
																					if "custom" in CONFIG_MAPPING._extra_content:
																												del CONFIG_MAPPING._extra_content["custom"]
																					if CustomConfig in TOKENIZER_MAPPING._extra_content:
																												del TOKENIZER_MAPPING._extra_content[CustomConfig]

							def 					__snake_case      (							self :						Dict       )			->					List[Any]:
														__snake_case	:    Optional[int]											=     AutoTokenizer.from_pretrained(
														    "hf-internal-testing/test_dynamic_tokenizer_legacy"					,			trust_remote_code=lowerCamelCase       )
														self.assertTrue(tokenizer.special_attribute_present       )
														if is_tokenizers_available():
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizerFast"       )

																					# Test we can also load the slow version
																					__snake_case	:    List[Any]											=     AutoTokenizer.from_pretrained(
																					    "hf-internal-testing/test_dynamic_tokenizer_legacy"					,			trust_remote_code=lowerCamelCase					,			use_fast=lowerCamelCase       )
																					self.assertTrue(tokenizer.special_attribute_present       )
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )
														else:
																					self.assertEqual(tokenizer.__class__.__name__					,			"NewTokenizer"       )

							def 					__snake_case      (							self :						Dict       )			->					int:
														with self.assertRaisesRegex(
														    lowerCamelCase					,			"bert-base is not a local folder and is not a valid model identifier"       ):
																					__snake_case	:    Any											=     AutoTokenizer.from_pretrained("bert-base"       )

							def 					__snake_case      (							self :						str       )			->					Union[str, Any]:
														with self.assertRaisesRegex(
														    lowerCamelCase					,			R"aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"       ):
																					__snake_case	:    Optional[int]											=     AutoTokenizer.from_pretrained(lowerCamelCase					,			revision="aaaaaa"       )






							def 					__snake_case      (							self :						Optional[Any]       )			->					Optional[int]:
														# Make sure we have cached the tokenizer.
														__snake_case	:    int											=     AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"       )
														with RequestCounter() as counter:
																					__snake_case	:    Optional[int]											=     AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"       )
																					self.assertEqual(counter.get_request_count					,			0       )
																					self.assertEqual(counter.head_request_count					,			1       )
																					self.assertEqual(counter.other_request_count					,			0       )





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_  (      __lowerCamelCase			):
							if is_torch_version("<"							,			"2.0.0"			) or not hasattr(__lowerCamelCase							,			"_dynamo"			):
														return False
							return isinstance(__lowerCamelCase							,			torch._dynamo.eval_frame.OptimizedModule			)







def        lowerCAmelCase_  (      __lowerCamelCase							,			__lowerCamelCase = True			):
							__snake_case	:    Optional[Any]											=     (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

							__snake_case	:    Tuple											=     is_compiled_module(__lowerCamelCase			)
							if is_compiled:
														__snake_case	:    Tuple											=     model
														__snake_case	:    List[str]											=     model._orig_mod

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

							while isinstance(__lowerCamelCase							,			__lowerCamelCase			):
														__snake_case	:    List[Any]											=     model.module

							if not keep_fpaa_wrapper:
														__snake_case	:    Tuple											=     getattr(__lowerCamelCase							,			"forward"			)
														__snake_case	:    int											=     model.__dict__.pop("_original_forward"							,			__lowerCamelCase			)
														if original_forward is not None:
																					while hasattr(__lowerCamelCase							,			"__wrapped__"			):
																												__snake_case	:    str											=     forward.__wrapped__
																												if forward == original_forward:
																																			break
																					__snake_case	:    Dict											=     forward
														if getattr(__lowerCamelCase							,			"_converted_to_transformer_engine"							,			__lowerCamelCase			):
																					convert_model(__lowerCamelCase							,			to_transformer_engine=__lowerCamelCase			)

							if is_compiled:
														__snake_case	:    Tuple											=     model
														__snake_case	:    str											=     compiled_model

							return model







def        lowerCAmelCase_  (      ):
							PartialState().wait_for_everyone()







def        lowerCAmelCase_  (      __lowerCamelCase							,			__lowerCamelCase			):
							if PartialState().distributed_type == DistributedType.TPU:
														xm.save(__lowerCamelCase							,			__lowerCamelCase			)
							elif PartialState().local_process_index == 0:
														torch.save(__lowerCamelCase							,			__lowerCamelCase			)







@contextmanager
def        lowerCAmelCase_  (      **__lowerCamelCase			):
							for key, value in kwargs.items():
														__snake_case	:    Union[str, Any]											=     str(__lowerCamelCase			)

							yield

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







def        lowerCAmelCase_  (      __lowerCamelCase			):
							if not hasattr(__lowerCamelCase							,			"__qualname__"			) and not hasattr(__lowerCamelCase							,			"__name__"			):
														__snake_case	:    Dict											=     getattr(__lowerCamelCase							,			"__class__"							,			__lowerCamelCase			)
							if hasattr(__lowerCamelCase							,			"__qualname__"			):
														return obj.__qualname__
							if hasattr(__lowerCamelCase							,			"__name__"			):
														return obj.__name__
							return str(__lowerCamelCase			)







def        lowerCAmelCase_  (      __lowerCamelCase							,			__lowerCamelCase			):
							for key, value in source.items():
														if isinstance(__lowerCamelCase							,			__lowerCamelCase			):
																					__snake_case	:    Tuple											=     destination.setdefault(__lowerCamelCase							,			{}			)
																					merge_dicts(__lowerCamelCase							,			__lowerCamelCase			)
														else:
																					__snake_case	:    Union[str, Any]											=     value

							return destination







def        lowerCAmelCase_  (      __lowerCamelCase = None			):
							if port is None:
														__snake_case	:    List[str]											=     2_9_5_0_0
							with socket.socket(socket.AF_INET							,			socket.SOCK_STREAM			) as s:
														return s.connect_ex(("localhost", port)			) == 0

