






'''simple docstring'''




import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
		import torch_xla.core.xla_model as xm
		import torch_xla.debug.metrics as met







class  A_						(__SCREAMING_SNAKE_CASE				):
							"""simple docstring"""




							def __init__(		self      :List[Any]							,				*lowerCAmelCase__      :Optional[int]							,				lowerCAmelCase__      :Optional[Any]=None							,				lowerCAmelCase__      :int=None							,				**lowerCAmelCase__      :str					)					->	Dict:


										'''simple docstring'''




										super().__init__(*__lowerCamelCase							,				**__lowerCamelCase					)
										snake_case_					:       Optional[int] =							eval_examples
										snake_case_					:       Optional[Any] =							post_process_function

							def 				_A						(		self      :int							,				lowerCAmelCase__      :Optional[Dataset] = None							,				lowerCAmelCase__      :List[Any]=None							,				lowerCAmelCase__      :Optional[List[str]] = None							,				lowerCAmelCase__      :str = "eval"							,				**lowerCAmelCase__      :int							,				)					->	Dict:


										'''simple docstring'''




										snake_case_					:       Union[str, Any] =							gen_kwargs.copy()
										snake_case_					:       Optional[int] =							(
										    gen_kwargs["max_length"] if gen_kwargs.get("max_length"					) is not None else self.args.generation_max_length
										)
										snake_case_					:       str =							(
										    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"					) is not None else self.args.generation_num_beams
										)
										snake_case_					:       Optional[Any] =							gen_kwargs

										snake_case_					:       List[str] =							self.eval_dataset if eval_dataset is None else eval_dataset
										snake_case_					:       Optional[Any] =							self.get_eval_dataloader(__lowerCamelCase					)
										snake_case_					:       Optional[int] =							self.eval_examples if eval_examples is None else eval_examples

										# Temporarily disable metric computation, we will do it in the loop here.
										snake_case_					:       Dict =							self.compute_metrics
										snake_case_					:       Tuple =							None
										snake_case_					:       Tuple =							time.time()
										snake_case_					:       Optional[int] =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
													snake_case_					:       str =							eval_loop(
													    __lowerCamelCase							,				description="Evaluation"							,				prediction_loss_only=True if compute_metrics is None else None							,				ignore_keys=__lowerCamelCase							,				metric_key_prefix=__lowerCamelCase							,				)
										finally:
													snake_case_					:       Optional[int] =							compute_metrics
										snake_case_					:       Optional[int] =							self.args.eval_batch_size * self.args.world_size
										if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
													start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        __lowerCamelCase							,				__lowerCamelCase							,				num_samples=output.num_samples							,				num_steps=math.ceil(output.num_samples / total_batch_size					)							,				)					)

										if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
													# Only the main node write the results by default
													snake_case_					:       List[Any] =							self.post_process_function(__lowerCamelCase							,				__lowerCamelCase							,				__lowerCamelCase					)
													snake_case_					:       Optional[Any] =							self.compute_metrics(__lowerCamelCase					)

													# Prefix all keys with metric_key_prefix + '_'
													for key in list(metrics.keys()					):
																if not key.startswith(F'''{metric_key_prefix}_'''					):
																			snake_case_					:       Dict =							metrics.pop(__lowerCamelCase					)

													metrics.update(output.metrics					)
										else:
													snake_case_					:       str =							output.metrics

										if self.args.should_log:
													# Only the main node log the results by default
													self.log(__lowerCamelCase					)

										if self.args.tpu_metrics_debug or self.args.debug:
													# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
													xm.master_print(met.metrics_report()					)

										snake_case_					:       Any =							self.callback_handler.on_evaluate(self.args							,				self.state							,				self.control							,				__lowerCamelCase					)
										return metrics






							def 				_A						(		self      :Union[str, Any]							,				lowerCAmelCase__      :List[Any]							,				lowerCAmelCase__      :Dict							,				lowerCAmelCase__      :Dict=None							,				lowerCAmelCase__      :str = "test"							,				**lowerCAmelCase__      :int					)					->	int:


										'''simple docstring'''




										snake_case_					:       Any =							gen_kwargs.copy()

										snake_case_					:       Tuple =							self.get_test_dataloader(__lowerCamelCase					)

										# Temporarily disable metric computation, we will do it in the loop here.
										snake_case_					:       str =							self.compute_metrics
										snake_case_					:       Optional[Any] =							None
										snake_case_					:       Dict =							time.time()
										snake_case_					:       List[str] =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
													snake_case_					:       Optional[Any] =							eval_loop(
													    __lowerCamelCase							,				description="Prediction"							,				prediction_loss_only=True if compute_metrics is None else None							,				ignore_keys=__lowerCamelCase							,				metric_key_prefix=__lowerCamelCase							,				)
										finally:
													snake_case_					:       Optional[Any] =							compute_metrics

										snake_case_					:       List[str] =							self.args.eval_batch_size * self.args.world_size
										if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
													start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        __lowerCamelCase							,				__lowerCamelCase							,				num_samples=output.num_samples							,				num_steps=math.ceil(output.num_samples / total_batch_size					)							,				)					)
										if self.post_process_function is None or self.compute_metrics is None:
													return output

										snake_case_					:       Union[str, Any] =							self.post_process_function(__lowerCamelCase							,				__lowerCamelCase							,				__lowerCamelCase							,				"predict"					)
										snake_case_					:       str =							self.compute_metrics(__lowerCamelCase					)

										# Prefix all keys with metric_key_prefix + '_'
										for key in list(metrics.keys()					):
													if not key.startswith(F'''{metric_key_prefix}_'''					):
																snake_case_					:       List[str] =							metrics.pop(__lowerCamelCase					)
										metrics.update(output.metrics					)
										return PredictionOutput(predictions=predictions.predictions							,				label_ids=predictions.label_ids							,				metrics=__lowerCamelCase					)









"""simple docstring"""






import importlib
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
import transformers.models.auto
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.bert.configuration_bert import BertConfig
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


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

from test_module.custom_configuration import CustomConfig  # noqa E402


snake_case   					=				get_tests_dir('''fixtures/dummy-config.json''')



class     UpperCAmelCase    (	unittest.TestCase  ):



   def    __UpperCAmelCase    (       self    :	Any				):



     """simple docstring"""
     _snake_case   	=      0



   def    __UpperCAmelCase    (       self    :	Optional[Any]				):



     """simple docstring"""
     self.assertIsNotNone(transformers.models.auto.__spec__				)
     self.assertIsNotNone(importlib.util.find_spec('''transformers.models.auto'''				)				)



   def    __UpperCAmelCase    (       self    :	List[str]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained('''bert-base-uncased'''				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	int				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.for_model('''roberta'''				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmp_dir:
       # This model name contains bert and roberta, but roberta ends up being picked.
       _snake_case   	=      os.path.join(__lowerCamelCase							,       '''fake-roberta'''				)
       os.makedirs(__lowerCamelCase							,       exist_ok=__lowerCamelCase				)
       with open(os.path.join(__lowerCamelCase							,       '''config.json'''				)							,       '''w'''				) as f:
         f.write(json.dumps({}				)				)
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
       self.assertEqual(type(__lowerCamelCase				)							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Optional[Any]				):



     """simple docstring"""
     try:
       AutoConfig.register('''custom'''							,       __lowerCamelCase				)
       # Wrong model type will raise an error
       with self.assertRaises(__lowerCamelCase				):
         AutoConfig.register('''model'''							,       __lowerCamelCase				)
       # Trying to register something existing in the Transformers library will raise an error
       with self.assertRaises(__lowerCamelCase				):
         AutoConfig.register('''bert'''							,       __lowerCamelCase				)

       # Now that the config is registered, it can be used as any other config with the auto-API
       _snake_case   	=      CustomConfig()
       with tempfile.TemporaryDirectory() as tmp_dir:
         config.save_pretrained(__lowerCamelCase				)
         _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
         self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)

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



   def    __UpperCAmelCase    (       self    :	Optional[int]				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       '''bert-base is not a local folder and is not a valid model identifier'''				):
       _snake_case   	=      AutoConfig.from_pretrained('''bert-base'''				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''				):
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase							,       revision='''aaaaaa'''				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       '''hf-internal-testing/no-config-test-repo does not appear to have a file named config.json.'''							,       ):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/no-config-test-repo'''				)



   def    __UpperCAmelCase    (       self    :	str				):



     """simple docstring"""
     # If remote code is not set, we will time out when asking whether to load the model.
     with self.assertRaises(__lowerCamelCase				):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''				)
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(__lowerCamelCase				):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)

     _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
     self.assertEqual(config.__class__.__name__							,       '''NewModelConfig'''				)

     # Test config can be reloaded.
     with tempfile.TemporaryDirectory() as tmp_dir:
       config.save_pretrained(__lowerCamelCase				)
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase							,       trust_remote_code=__lowerCamelCase				)
     self.assertEqual(reloaded_config.__class__.__name__							,       '''NewModelConfig'''				)





   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



     """simple docstring"""



     class     UpperCAmelCase    (	__SCREAMING_SNAKE_CASE  ):
        A__		:	List[Any]           =  '''new-model'''

     try:
       AutoConfig.register('''new-model'''							,       __lowerCamelCase				)
       # If remote code is not set, the default is to use local
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfigLocal'''				)

       # If remote code is disabled, we load the local one.
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfigLocal'''				)

       # If remote is enabled, we load from the Hub
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfig'''				)

     finally:
       if "new-model" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["new-model"]





