



"""simple docstring"""

from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    get_torch_dist_unique_port,
    require_torch_multi_gpu,
    require_torch_neuroncore,
)
from transformers.training_args import ParallelMode
from transformers.utils import logging


lowercase__					:						Any												=			logging.get_logger(__name__)


if is_torch_available():
     import torch
     from torch import nn
     from torch.utils.data import Dataset

     from transformers import Trainer




     class  _UpperCAmelCase (    UpperCamelCase_):

      def __init__(  self	: Dict						,  lowercase_	: Optional[int] = 101       ):
          snake_case_				:  Union[str, Any]       =			length

      def __len__(  self	: Any       ):
          return self.length






      def __getitem__(  self	: Optional[int]						,  lowercase_	: Dict       ):
          return i





     class  _UpperCAmelCase :

      def __call__(  self	: Any						,  lowercase_	: Tuple       ):
          return {"input_ids": torch.tensor(__a       ), "labels": torch.tensor(__a       )}





     class  _UpperCAmelCase (    nn.Module):

      def __init__(  self	: Dict       ):
          super().__init__()
          # Add some (unused) params otherwise DDP will complain.
          snake_case_				:  List[str]       =			nn.Linear(120						,  80       )






      def 		_snake_case      (  self	: Dict						,  lowercase_	: Any						,  lowercase_	: str=None       ):






          if labels is not None:
              return torch.tensor(0.0						,  device=input_ids.device       ), input_ids
          else:
              return input_ids




class  _UpperCAmelCase (    UpperCamelCase_):

 @require_torch_neuroncore
 def 		_snake_case      (  self	: int       ):
     snake_case_				:  Optional[Any]       =			f"--nproc_per_node=2\n            --master_port={get_torch_dist_unique_port()}\n            {self.test_file_dir}/test_trainer_distributed.py\n        ".split()
     snake_case_				:  List[Any]       =			self.get_auto_remove_tmp_dir()
     snake_case_				:  Any       =			f"--output_dir {output_dir}".split()
     snake_case_				:  Union[str, Any]       =			['torchrun'] + distributed_args + args
     execute_subprocess_async(__a						,  env=self.get_env()       )
     # successful return here == success - any errors would have caused an error in the sub-call





class  _UpperCAmelCase (    UpperCamelCase_):

 @require_torch_multi_gpu
 def 		_snake_case      (  self	: Optional[Any]       ):
     snake_case_				:  Dict       =			f"--nproc_per_node={torch.cuda.device_count()}\n            --master_port={get_torch_dist_unique_port()}\n            {self.test_file_dir}/test_trainer_distributed.py\n        ".split()
     snake_case_				:  Dict       =			self.get_auto_remove_tmp_dir()
     snake_case_				:  Optional[Any]       =			f"--output_dir {output_dir}".split()
     snake_case_				:  str       =			['torchrun'] + distributed_args + args
     execute_subprocess_async(__a						,  env=self.get_env()       )
     # successful return here == success - any errors would have caused an error in the sub-call


if __name__ == "__main__":
     # The script below is meant to be run under torch.distributed, on a machine with multiple GPUs:
     #
     # PYTHONPATH="src" python -m torch.distributed.run --nproc_per_node 2 --output_dir output_dir ./tests/test_trainer_distributed.py

     lowercase__					:						List[str]												=			HfArgumentParser((TrainingArguments,))
     lowercase__					:						int												=			parser.parse_args_into_dataclasses()[0]

     logger.warning(
         f'Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}, '
         f'distributed training: {training_args.parallel_mode != ParallelMode.NOT_DISTRIBUTED}'
     )

     # Essentially, what we want to verify in the distributed case is that we get all samples back,
     # in the right order. (this is crucial for prediction for instance)
     for dataset_length in [1_01, 40, 7]:
          lowercase__					:						str												=			DummyDataset(dataset_length)




          def __lowercase    (							_a					):

              '''simple docstring'''





              snake_case_				:  str       =			list(range(len(_a					)					)					)
              snake_case_				:  Dict       =			p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
              if not success and training_args.local_rank == 0:
                  logger.warning(
                      '''Predictions and/or labels do not match expected results:\n  - predictions: '''
                      f"{p.predictions.tolist()}\n  - labels: {p.label_ids.tolist()}\n  - expected: {sequential}"					)
              return {"success": success}

          lowercase__					:						Optional[Any]												=			Trainer(
              model=DummyModel(),
              args=training_args,
              data_collator=DummyDataCollator(),
              eval_dataset=dataset,
              compute_metrics=compute_metrics,
          )
          lowercase__					:						Optional[int]												=			trainer.evaluate()
          logger.info(metrics)
          if metrics["eval_success"] is not True:
               logger.error(metrics)
               exit(1)

          lowercase__					:						Optional[Any]												=			trainer.predict(dataset)
          logger.info(p.metrics)
          if p.metrics["test_success"] is not True:
               logger.error(p.metrics)
               exit(1)

          lowercase__					:						int												=			2

          lowercase__					:						List[str]												=			trainer.evaluate()
          logger.info(metrics)
          if metrics["eval_success"] is not True:
               logger.error(metrics)
               exit(1)

          lowercase__					:						Tuple												=			trainer.predict(dataset)
          logger.info(p.metrics)
          if p.metrics["test_success"] is not True:
               logger.error(p.metrics)
               exit(1)






          lowercase__					:						Optional[int]												=			None










"""simple docstring"""

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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
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
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowercase__					:						Tuple												=			get_tests_dir('''fixtures''')
lowercase__					:						List[str]												=			get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
lowercase__					:						Dict												=			get_tests_dir('''fixtures/dummy-config.json''')




class  _UpperCAmelCase (    unittest.TestCase):

 def 		_snake_case      (  self	: Tuple       ):
     snake_case_				:  Optional[Any]       =			0

 def 		_snake_case      (  self	: int       ):
     snake_case_				:  str       =			AutoFeatureExtractor.from_pretrained('''facebook/wav2vec2-base-960h'''       )
     self.assertIsInstance(lowercase_						,  lowercase_       )

 def 		_snake_case      (  self	: Optional[Any]       ):
     snake_case_				:  Dict       =			AutoFeatureExtractor.from_pretrained(lowercase_       )
     self.assertIsInstance(lowercase_						,  lowercase_       )

 def 		_snake_case      (  self	: int       ):
     with tempfile.TemporaryDirectory() as tmpdirname:
         snake_case_				:  List[Any]       =			WavaVecaConfig()

         # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
         snake_case_				:  Optional[Any]       =			AutoFeatureExtractor.from_pretrained(lowercase_       ).to_dict()

         config_dict.pop('''feature_extractor_type'''       )
         snake_case_				:  Optional[Any]       =			WavaVecaFeatureExtractor(**lowercase_       )

         # save in new folder
         model_config.save_pretrained(lowercase_       )
         config.save_pretrained(lowercase_       )

         snake_case_				:  List[Any]       =			AutoFeatureExtractor.from_pretrained(lowercase_       )

         # make sure private variable is not incorrectly saved
         snake_case_				:  Dict       =			json.loads(config.to_json_string()       )
         self.assertTrue('''_processor_class''' not in dict_as_saved       )

     self.assertIsInstance(lowercase_						,  lowercase_       )

 def 		_snake_case      (  self	: Optional[Any]       ):
     snake_case_				:  str       =			AutoFeatureExtractor.from_pretrained(lowercase_       )
     self.assertIsInstance(lowercase_						,  lowercase_       )

 def 		_snake_case      (  self	: List[Any]       ):
     with self.assertRaisesRegex(
         lowercase_						,  '''bert-base is not a local folder and is not a valid model identifier'''       ):
         snake_case_				:  List[str]       =			AutoFeatureExtractor.from_pretrained('''bert-base'''       )

 def 		_snake_case      (  self	: Any       ):
     with self.assertRaisesRegex(
         lowercase_						,  r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''       ):
         snake_case_				:  Dict       =			AutoFeatureExtractor.from_pretrained(lowercase_						,  revision='''aaaaaa'''       )

 def 		_snake_case      (  self	: Optional[Any]       ):
     with self.assertRaisesRegex(
         lowercase_						,  '''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''						,  ):
         snake_case_				:  Dict       =			AutoFeatureExtractor.from_pretrained('''hf-internal-testing/config-no-model'''       )

 def 		_snake_case      (  self	: Optional[int]       ):
     # If remote code is not set, we will time out when asking whether to load the model.
     with self.assertRaises(lowercase_       ):
         snake_case_				:  str       =			AutoFeatureExtractor.from_pretrained(
             '''hf-internal-testing/test_dynamic_feature_extractor'''       )
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(lowercase_       ):
         snake_case_				:  str       =			AutoFeatureExtractor.from_pretrained(
             '''hf-internal-testing/test_dynamic_feature_extractor'''						,  trust_remote_code=lowercase_       )

     snake_case_				:  List[str]       =			AutoFeatureExtractor.from_pretrained(
         '''hf-internal-testing/test_dynamic_feature_extractor'''						,  trust_remote_code=lowercase_       )
     self.assertEqual(feature_extractor.__class__.__name__						,  '''NewFeatureExtractor'''       )

     # Test feature extractor can be reloaded.
     with tempfile.TemporaryDirectory() as tmp_dir:
         feature_extractor.save_pretrained(lowercase_       )
         snake_case_				:  Optional[int]       =			AutoFeatureExtractor.from_pretrained(lowercase_						,  trust_remote_code=lowercase_       )
     self.assertEqual(reloaded_feature_extractor.__class__.__name__						,  '''NewFeatureExtractor'''       )

 def 		_snake_case      (  self	: Optional[Any]       ):
     try:
         AutoConfig.register('''custom'''						,  lowercase_       )
         AutoFeatureExtractor.register(lowercase_						,  lowercase_       )
         # Trying to register something existing in the Transformers library will raise an error
         with self.assertRaises(lowercase_       ):
             AutoFeatureExtractor.register(lowercase_						,  lowercase_       )

         # Now that the config is registered, it can be used as any other config with the auto-API
         snake_case_				:  List[Any]       =			CustomFeatureExtractor.from_pretrained(lowercase_       )
         with tempfile.TemporaryDirectory() as tmp_dir:
             feature_extractor.save_pretrained(lowercase_       )
             snake_case_				:  Any       =			AutoFeatureExtractor.from_pretrained(lowercase_       )
             self.assertIsInstance(lowercase_						,  lowercase_       )

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






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




     class  _UpperCAmelCase (    lowerCAmelCase__):
      _lowerCAmelCase     :    str	      =	True

     try:
         AutoConfig.register('''custom'''						,  lowercase_       )
         AutoFeatureExtractor.register(lowercase_						,  lowercase_       )
         # If remote code is not set, the default is to use local
         snake_case_				:  List[Any]       =			AutoFeatureExtractor.from_pretrained(
             '''hf-internal-testing/test_dynamic_feature_extractor'''       )
         self.assertEqual(feature_extractor.__class__.__name__						,  '''NewFeatureExtractor'''       )
         self.assertTrue(feature_extractor.is_local       )

         # If remote code is disabled, we load the local one.
         snake_case_				:  Union[str, Any]       =			AutoFeatureExtractor.from_pretrained(
             '''hf-internal-testing/test_dynamic_feature_extractor'''						,  trust_remote_code=lowercase_       )
         self.assertEqual(feature_extractor.__class__.__name__						,  '''NewFeatureExtractor'''       )
         self.assertTrue(feature_extractor.is_local       )

         # If remote is enabled, we load from the Hub
         snake_case_				:  Tuple       =			AutoFeatureExtractor.from_pretrained(
             '''hf-internal-testing/test_dynamic_feature_extractor'''						,  trust_remote_code=lowercase_       )
         self.assertEqual(feature_extractor.__class__.__name__						,  '''NewFeatureExtractor'''       )
         self.assertTrue(not hasattr(lowercase_						,  '''is_local'''       )       )

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





