
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


__lowerCAmelCase			:     Union[str, Any]           =   get_tests_dir('fixtures')
__lowerCAmelCase			:     Union[str, Any]           =   get_tests_dir('fixtures/dummy_feature_extractor_config.json')
__lowerCAmelCase			:     int           =   get_tests_dir('fixtures/dummy-config.json')




class        UpperCAmelCase_			( unittest.TestCase     ):


  '''simple docstring'''







  def 							_lowercase							( self			: Optional[int]    )			->	Optional[int]:







    """simple docstring"""






    __magic_name__				     =       0
  def 							_lowercase							( self			: Optional[Any]    )			->	str:







    """simple docstring"""






    __magic_name__				     =       AutoFeatureExtractor.from_pretrained("""facebook/wav2vec2-base-960h"""    )
    self.assertIsInstance(UpperCamelCase__			,							UpperCamelCase__    )
  def 							_lowercase							( self			: List[str]    )			->	Optional[Any]:







    """simple docstring"""






    __magic_name__				     =       AutoFeatureExtractor.from_pretrained(UpperCamelCase__    )
    self.assertIsInstance(UpperCamelCase__			,							UpperCamelCase__    )
  def 							_lowercase							( self			: Tuple    )			->	Optional[Any]:







    """simple docstring"""






    with tempfile.TemporaryDirectory() as tmpdirname:
      __magic_name__				     =       WavaVecaConfig()

      # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(UpperCamelCase__    ).to_dict()

      config_dict.pop("""feature_extractor_type"""    )
      __magic_name__				     =       WavaVecaFeatureExtractor(**UpperCamelCase__    )

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

      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(UpperCamelCase__    )

      # make sure private variable is not incorrectly saved
      __magic_name__				     =       json.loads(config.to_json_string()    )
      self.assertTrue("""_processor_class""" not in dict_as_saved    )

    self.assertIsInstance(UpperCamelCase__			,							UpperCamelCase__    )
  def 							_lowercase							( self			: List[str]    )			->	List[str]:







    """simple docstring"""






    __magic_name__				     =       AutoFeatureExtractor.from_pretrained(UpperCamelCase__    )
    self.assertIsInstance(UpperCamelCase__			,							UpperCamelCase__    )
  def 							_lowercase							( self			: List[str]    )			->	Any:







    """simple docstring"""






    with self.assertRaisesRegex(
        UpperCamelCase__			,							"""bert-base is not a local folder and is not a valid model identifier"""    ):
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained("""bert-base"""    )
  def 							_lowercase							( self			: Any    )			->	Dict:







    """simple docstring"""






    with self.assertRaisesRegex(
        UpperCamelCase__			,							R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""    ):
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(UpperCamelCase__			,							revision="""aaaaaa"""    )
  def 							_lowercase							( self			: Any    )			->	Dict:







    """simple docstring"""






    with self.assertRaisesRegex(
        UpperCamelCase__			,							"""hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""			,							):
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained("""hf-internal-testing/config-no-model"""    )
  def 							_lowercase							( self			: Optional[int]    )			->	Union[str, Any]:







    """simple docstring"""






    with self.assertRaises(UpperCamelCase__    ):
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(
          """hf-internal-testing/test_dynamic_feature_extractor"""    )
    # If remote code is disabled, we can't load this config.
    with self.assertRaises(UpperCamelCase__    ):
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(
          """hf-internal-testing/test_dynamic_feature_extractor"""			,							trust_remote_code=UpperCamelCase__    )

    __magic_name__				     =       AutoFeatureExtractor.from_pretrained(
        """hf-internal-testing/test_dynamic_feature_extractor"""			,							trust_remote_code=UpperCamelCase__    )
    self.assertEqual(feature_extractor.__class__.__name__			,							"""NewFeatureExtractor"""    )

    # Test feature extractor can be reloaded.
    with tempfile.TemporaryDirectory() as tmp_dir:
      feature_extractor.save_pretrained(UpperCamelCase__    )
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(UpperCamelCase__			,							trust_remote_code=UpperCamelCase__    )
    self.assertEqual(reloaded_feature_extractor.__class__.__name__			,							"""NewFeatureExtractor"""    )
  def 							_lowercase							( self			: List[Any]    )			->	Optional[int]:







    """simple docstring"""






    try:
      AutoConfig.register("""custom"""			,							UpperCamelCase__    )
      AutoFeatureExtractor.register(UpperCamelCase__			,							UpperCamelCase__    )
      # Trying to register something existing in the Transformers library will raise an error
      with self.assertRaises(UpperCamelCase__    ):
        AutoFeatureExtractor.register(UpperCamelCase__			,							UpperCamelCase__    )

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

    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 							_lowercase							( self			: Optional[int]    )			->	Any:







    """simple docstring"""






    class        UpperCAmelCase_			( _A     ):


      '''simple docstring'''







      a__           =		True

    try:
      AutoConfig.register("""custom"""			,							UpperCamelCase__    )
      AutoFeatureExtractor.register(UpperCamelCase__			,							UpperCamelCase__    )
      # If remote code is not set, the default is to use local
      __magic_name__				     =       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.
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(
          """hf-internal-testing/test_dynamic_feature_extractor"""			,							trust_remote_code=UpperCamelCase__    )
      self.assertEqual(feature_extractor.__class__.__name__			,							"""NewFeatureExtractor"""    )
      self.assertTrue(feature_extractor.is_local    )

      # If remote is enabled, we load from the Hub
      __magic_name__				     =       AutoFeatureExtractor.from_pretrained(
          """hf-internal-testing/test_dynamic_feature_extractor"""			,							trust_remote_code=UpperCamelCase__    )
      self.assertEqual(feature_extractor.__class__.__name__			,							"""NewFeatureExtractor"""    )
      self.assertTrue(not hasattr(UpperCamelCase__			,							"""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]






import argparse
import os

import evaluate
import torch
from datasets import load_dataset
from torch.optim import AdamW
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer, get_linear_schedule_with_warmup, set_seed

from accelerate import Accelerator, DistributedType


########################################################################
# This is a fully working simple example to use Accelerate,
# specifically showcasing how to properly calculate the metrics on the
# validation dataset when in a distributed system, and builds off the
# `nlp_example.py` script.
#
# This example trains a Bert base model on GLUE MRPC
# in any of the following settings (with the same script):
#   - single CPU or single GPU
#   - multi GPUS (using PyTorch distributed mode)
#   - (multi) TPUs
#   - fp16 (mixed-precision) or fp32 (normal precision)
#
# To help focus on the differences in the code, building `DataLoaders`
# was refactored into its own function.
# New additions from the base script can be found quickly by
# looking for the # New Code # tags
#
# To run it in each of these various modes, follow the instructions
# in the readme for examples:
# https://github.com/huggingface/accelerate/tree/main/examples
#
########################################################################


__A   			=1_6
__A   			=3_2

def        a						(      _UpperCAmelCase							:					Accelerator ,   _UpperCAmelCase							:					int = 16						):







		'''simple docstring'''





		__UpperCAmelCase    :     Union[str, Any]													=	AutoTokenizer.from_pretrained('''bert-base-cased'''						)
		__UpperCAmelCase    :     Any													=	load_dataset('''glue''' ,   '''mrpc'''						)

		def tokenize_function(_UpperCAmelCase							:					List[Any]						):
				# max_length=None => use the model max length (it's actually the default)
				__UpperCAmelCase    :     str													=	tokenizer(examples['''sentence1'''] ,   examples['''sentence2'''] ,   truncation=_UpperCAmelCase ,   max_length=_UpperCAmelCase						)
				return outputs

		# Apply the method we just defined to all the examples in all the splits of the dataset
		# starting with the main process first:
		with accelerator.main_process_first():
				__UpperCAmelCase    :     Optional[int]													=	datasets.map(
				    _UpperCAmelCase ,   batched=_UpperCAmelCase ,   remove_columns=['''idx''', '''sentence1''', '''sentence2'''] ,   )

		# We also rename the 'label' column to 'labels' which is the expected name for labels by the models of the
		# transformers library
		__UpperCAmelCase    :     Optional[Any]													=	tokenized_datasets.rename_column('''label''' ,   '''labels'''						)

		def collate_fn(_UpperCAmelCase							:					str						):
				# On TPU it's best to pad everything to the same length or training will be very slow.
				__UpperCAmelCase    :     List[Any]													=	1_28 if accelerator.distributed_type == DistributedType.TPU else None
				# When using mixed precision we want round multiples of 8/16
				if accelerator.mixed_precision == "fp8":
						__UpperCAmelCase    :     Any													=	16
				elif accelerator.mixed_precision != "no":
						__UpperCAmelCase    :     Tuple													=	8
				else:
						__UpperCAmelCase    :     Optional[int]													=	None

				return tokenizer.pad(
				    _UpperCAmelCase ,   padding='''longest''' ,   max_length=_UpperCAmelCase ,   pad_to_multiple_of=_UpperCAmelCase ,   return_tensors='''pt''' ,   )

		# Instantiate dataloaders.
		__UpperCAmelCase    :     Optional[Any]													=	DataLoader(
		    tokenized_datasets['''train'''] ,   shuffle=_UpperCAmelCase ,   collate_fn=_UpperCAmelCase ,   batch_size=_UpperCAmelCase						)
		__UpperCAmelCase    :     str													=	DataLoader(
		    tokenized_datasets['''validation'''] ,   shuffle=_UpperCAmelCase ,   collate_fn=_UpperCAmelCase ,   batch_size=_UpperCAmelCase						)

		return train_dataloader, eval_dataloader


# For testing only
if os.environ.get("TESTING_MOCKED_DATALOADERS", None) == "1":
				from accelerate.test_utils.training import mocked_dataloaders

				__A   			=mocked_dataloaders  # noqa: F811

def        a						(      _UpperCAmelCase							:					List[str] ,   _UpperCAmelCase							:					Any						):







		'''simple docstring'''





		if os.environ.get('''TESTING_MOCKED_DATALOADERS''' ,   _UpperCAmelCase						) == "1":
				__UpperCAmelCase    :     Dict													=	2
		# Initialize accelerator
		__UpperCAmelCase    :     Dict													=	Accelerator(cpu=args.cpu ,   mixed_precision=args.mixed_precision						)
		# Sample hyper-parameters for learning rate, batch size, seed and a few other HPs
		__UpperCAmelCase    :     List[Any]													=	config['''lr''']
		__UpperCAmelCase    :     Optional[Any]													=	int(config['''num_epochs''']						)
		__UpperCAmelCase    :     Optional[int]													=	int(config['''seed''']						)
		__UpperCAmelCase    :     Any													=	int(config['''batch_size''']						)

		__UpperCAmelCase    :     Optional[Any]													=	evaluate.load('''glue''' ,   '''mrpc'''						)

		# If the batch size is too big we use gradient accumulation
		__UpperCAmelCase    :     List[str]													=	1
		if batch_size > MAX_GPU_BATCH_SIZE and accelerator.distributed_type != DistributedType.TPU:
				__UpperCAmelCase    :     Tuple													=	batch_size // MAX_GPU_BATCH_SIZE
				__UpperCAmelCase    :     List[str]													=	MAX_GPU_BATCH_SIZE

		set_seed(_UpperCAmelCase						)
		__UpperCAmelCase							,					__UpperCAmelCase    :     Union[str, Any]													=	get_dataloaders(_UpperCAmelCase ,   _UpperCAmelCase						)
		# Instantiate the model (we build the model here so that the seed also control new weights initialization)
		__UpperCAmelCase    :     List[str]													=	AutoModelForSequenceClassification.from_pretrained('''bert-base-cased''' ,   return_dict=_UpperCAmelCase						)

		# We could avoid this line since the accelerator is set with `device_placement=True` (default value).
		# Note that if you are placing tensors on devices manually, this line absolutely needs to be before the optimizer
		# creation otherwise training will not work on TPU (`accelerate` will kindly throw an error to make us aware of that).
		__UpperCAmelCase    :     Tuple													=	model.to(accelerator.device						)

		# Instantiate optimizer
		__UpperCAmelCase    :     Any													=	AdamW(params=model.parameters() ,   lr=_UpperCAmelCase						)

		# Instantiate scheduler
		__UpperCAmelCase    :     str													=	get_linear_schedule_with_warmup(
		    optimizer=_UpperCAmelCase ,   num_warmup_steps=1_00 ,   num_training_steps=(len(_UpperCAmelCase						) * num_epochs) // gradient_accumulation_steps ,   )

		# Prepare everything
		# There is no specific order to remember, we just need to unpack the objects in the same order we gave them to the
		# prepare method.
		__UpperCAmelCase							,					__UpperCAmelCase							,					__UpperCAmelCase							,					__UpperCAmelCase							,					__UpperCAmelCase    :     List[str]													=	accelerator.prepare(
		    _UpperCAmelCase ,   _UpperCAmelCase ,   _UpperCAmelCase ,   _UpperCAmelCase ,   _UpperCAmelCase						)

		# Now we train the model
		for epoch in range(_UpperCAmelCase						):
				model.train()
				for step, batch in enumerate(_UpperCAmelCase						):
						# We could avoid this line since we set the accelerator with `device_placement=True`.
						batch.to(accelerator.device						)
						__UpperCAmelCase    :     int													=	model(**_UpperCAmelCase						)
						__UpperCAmelCase    :     str													=	outputs.loss
						__UpperCAmelCase    :     str													=	loss / gradient_accumulation_steps
						accelerator.backward(_UpperCAmelCase						)
						if step % gradient_accumulation_steps == 0:
								optimizer.step()
								lr_scheduler.step()
								optimizer.zero_grad()

				model.eval()
				__UpperCAmelCase    :     Tuple													=	0
				for step, batch in enumerate(_UpperCAmelCase						):
						# We could avoid this line since we set the accelerator with `device_placement=True`.
						batch.to(accelerator.device						)
						with torch.no_grad():
								__UpperCAmelCase    :     Dict													=	model(**_UpperCAmelCase						)
						__UpperCAmelCase    :     str													=	outputs.logits.argmax(dim=-1						)
						__UpperCAmelCase							,					__UpperCAmelCase    :     List[Any]													=	accelerator.gather((predictions, batch['''labels'''])						)
						# New Code #
						# First we check if it's a distributed system
						if accelerator.use_distributed:
								# Then see if we're on the last batch of our eval dataloader
								if step == len(_UpperCAmelCase						) - 1:
										# Last batch needs to be truncated on distributed systems as it contains additional samples
										__UpperCAmelCase    :     Dict													=	predictions[: len(eval_dataloader.dataset						) - samples_seen]
										__UpperCAmelCase    :     List[str]													=	references[: len(eval_dataloader.dataset						) - samples_seen]
								else:
										# Otherwise we add the number of samples seen
										samples_seen += references.shape[0]
            # All of this can be avoided if you use `Accelerator.gather_for_metrics` instead of `Accelerator.gather`:
            # accelerator.gather_for_metrics((predictions, batch["labels"]))
						metric.add_batch(
						    predictions=_UpperCAmelCase ,   references=_UpperCAmelCase ,   )

				__UpperCAmelCase    :     Tuple													=	metric.compute()
				# Use accelerator.print to print only on the main process.
				accelerator.print(f'epoch {epoch}:' ,   _UpperCAmelCase						)

def        a						(      ):







		'''simple docstring'''





		__UpperCAmelCase    :     int													=	argparse.ArgumentParser(description='''Simple example of training script.'''						)
		parser.add_argument(
		    '''--mixed_precision''' ,   type=_UpperCAmelCase ,   default=_UpperCAmelCase ,   choices=['''no''', '''fp16''', '''bf16''', '''fp8'''] ,   help='''Whether to use mixed precision. Choose'''
		    '''between fp16 and bf16 (bfloat16). Bf16 requires PyTorch >= 1.10.'''
		    '''and an Nvidia Ampere GPU.''' ,   )
		parser.add_argument('''--cpu''' ,   action='''store_true''' ,   help='''If passed, will train on the CPU.'''						)
		__UpperCAmelCase    :     int													=	parser.parse_args()
		__UpperCAmelCase    :     Union[str, Any]													=	{'''lr''': 2e-5, '''num_epochs''': 3, '''seed''': 42, '''batch_size''': 16}
		training_function(_UpperCAmelCase ,   _UpperCAmelCase						)


if __name__ == "__main__":
				main()





