


'''simple docstring'''

import math
import os
from copy import deepcopy

import datasets
import evaluate
import torch
import transformers
from datasets import load_dataset
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer

from accelerate import Accelerator
from accelerate.test_utils import RegressionDataset, RegressionModel
from accelerate.utils import is_tpu_available, set_seed


a_  :				Optional[int]						= """true"""





def __snake_case					(      UpperCAmelCase_						:						Optional[Any]					,				UpperCAmelCase_						:						str=82					,				UpperCAmelCase_						:						int=16		):
 set_seed(42		)
 lowerCamelCase_      =  RegressionModel()
 lowerCamelCase_      =  deepcopy(UpperCAmelCase_		)
 lowerCamelCase_      =  RegressionDataset(length=UpperCAmelCase_		)
 lowerCamelCase_      =  DataLoader(UpperCAmelCase_					,				batch_size=UpperCAmelCase_		)
 model.to(accelerator.device		)
 lowerCamelCase_ ,lowerCamelCase_      =  accelerator.prepare(UpperCAmelCase_					,				UpperCAmelCase_		)
 return model, ddp_model, dataloader





def __snake_case					(      UpperCAmelCase_						:						Accelerator					,				UpperCAmelCase_						:						str=False		):
 lowerCamelCase_      =  AutoTokenizer.from_pretrained("hf-internal-testing/mrpc-bert-base-cased"		)
 lowerCamelCase_      =  load_dataset("glue"					,				"mrpc"					,				split="validation"		)

 def tokenize_function(UpperCAmelCase_						:						Union[str, Any]		):
  lowerCamelCase_      =  tokenizer(examples["sentence1"]					,				examples["sentence2"]					,				truncation=UpperCAmelCase_					,				max_length=UpperCAmelCase_		)
  return outputs

 with accelerator.main_process_first():
  lowerCamelCase_      =  dataset.map(
      UpperCAmelCase_					,				batched=UpperCAmelCase_					,				remove_columns=["idx", "sentence1", "sentence2"]					,				)

 lowerCamelCase_      =  tokenized_datasets.rename_column("label"					,				"labels"		)

 def collate_fn(UpperCAmelCase_						:						int		):
  if use_longest:
   return tokenizer.pad(UpperCAmelCase_					,				padding="longest"					,				return_tensors="pt"		)
  return tokenizer.pad(UpperCAmelCase_					,				padding="max_length"					,				max_length=128					,				return_tensors="pt"		)

 return DataLoader(UpperCAmelCase_					,				shuffle=UpperCAmelCase_					,				collate_fn=UpperCAmelCase_					,				batch_size=16		)





def __snake_case					(      UpperCAmelCase_						:						int					,				UpperCAmelCase_						:						Any		):
 lowerCamelCase_      =  Accelerator(dispatch_batches=UpperCAmelCase_					,				split_batches=UpperCAmelCase_		)
 lowerCamelCase_      =  get_dataloader(UpperCAmelCase_					,				not dispatch_batches		)
 lowerCamelCase_      =  AutoModelForSequenceClassification.from_pretrained(
     "hf-internal-testing/mrpc-bert-base-cased"					,				return_dict=UpperCAmelCase_		)
 lowerCamelCase_ ,lowerCamelCase_      =  accelerator.prepare(UpperCAmelCase_					,				UpperCAmelCase_		)
 return {"ddp": [ddp_model, ddp_dataloader, "cuda:0"], "no": [model, dataloader, accelerator.device]}, accelerator





def __snake_case					(      UpperCAmelCase_						:						Optional[int]					,				UpperCAmelCase_						:						Optional[int]					,				UpperCAmelCase_						:						Any		):
 lowerCamelCase_      =  []
 for batch in dataloader:
  lowerCamelCase_ ,lowerCamelCase_      =  batch.values()
  with torch.no_grad():
   lowerCamelCase_      =  model(UpperCAmelCase_		)
   lowerCamelCase_ ,lowerCamelCase_      =  accelerator.gather_for_metrics((logit, target)		)
   logits_and_targets.append((logit, target)		)
 lowerCamelCase_ ,lowerCamelCase_      =  [], []
 for logit, targ in logits_and_targets:
  logits.append(UpperCAmelCase_		)
  targs.append(UpperCAmelCase_		)
 lowerCamelCase_ ,lowerCamelCase_      =  torch.cat(UpperCAmelCase_		), torch.cat(UpperCAmelCase_		)
 return logits, targs





def __snake_case					(      UpperCAmelCase_						:						Accelerator					,				UpperCAmelCase_						:						Optional[Any]=82					,				UpperCAmelCase_						:						Tuple=False					,				UpperCAmelCase_						:						str=False					,				UpperCAmelCase_						:						Union[str, Any]=16		):
 lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_      =  get_basic_setup(UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_		)
 lowerCamelCase_ ,lowerCamelCase_      =  generate_predictions(UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_		)
 assert (
     len(UpperCAmelCase_		) == num_samples
 ), F'''Unexpected number of inputs:\n    Expected: {num_samples}\n    Actual: {len(UpperCAmelCase_		)}'''





def __snake_case					(      UpperCAmelCase_						:						bool = False					,				UpperCAmelCase_						:						bool = False		):
 lowerCamelCase_      =  evaluate.load("glue"					,				"mrpc"		)
 lowerCamelCase_ ,lowerCamelCase_      =  get_mrpc_setup(UpperCAmelCase_					,				UpperCAmelCase_		)
 # First do baseline
 lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_      =  setup["no"]
 model.to(UpperCAmelCase_		)
 model.eval()
 for batch in dataloader:
  batch.to(UpperCAmelCase_		)
  with torch.inference_mode():
   lowerCamelCase_      =  model(**UpperCAmelCase_		)
  lowerCamelCase_      =  outputs.logits.argmax(dim=-1		)
  metric.add_batch(predictions=UpperCAmelCase_					,				references=batch["labels"]		)
 lowerCamelCase_      =  metric.compute()

 # Then do distributed
 lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_      =  setup["ddp"]
 model.eval()
 for batch in dataloader:
  with torch.inference_mode():
   lowerCamelCase_      =  model(**UpperCAmelCase_		)
  lowerCamelCase_      =  outputs.logits.argmax(dim=-1		)
  lowerCamelCase_      =  batch["labels"]
  lowerCamelCase_ ,lowerCamelCase_      =  accelerator.gather_for_metrics((preds, references)		)
  metric.add_batch(predictions=UpperCAmelCase_					,				references=UpperCAmelCase_		)
 lowerCamelCase_      =  metric.compute()

 for key in "accuracy f1".split():
  assert math.isclose(
      baseline[key]					,				distributed[key]		), F'''Baseline and Distributed are not the same for key {key}:\n\tBaseline: {baseline[key]}\n\tDistributed: {distributed[key]}\n'''





def __snake_case					(      ):
 lowerCamelCase_      =  Accelerator(split_batches=UpperCAmelCase_					,				dispatch_batches=UpperCAmelCase_		)
 if accelerator.is_local_main_process:
  datasets.utils.logging.set_verbosity_warning()
  transformers.utils.logging.set_verbosity_warning()
 else:
  datasets.utils.logging.set_verbosity_error()
  transformers.utils.logging.set_verbosity_error()
 # These are a bit slower so they should only be ran on the GPU or TPU
 if torch.cuda.is_available() or is_tpu_available():
  if accelerator.is_local_main_process:
   print("**Testing gather_for_metrics**"		)
  for split_batches in [True, False]:
   for dispatch_batches in [True, False]:
    if accelerator.is_local_main_process:
     print(F'''With: `split_batches={split_batches}`, `dispatch_batches={dispatch_batches}`'''		)
    test_mrpc(UpperCAmelCase_					,				UpperCAmelCase_		)
    accelerator.state._reset_state()
 if accelerator.is_local_main_process:
  print("**Test torch metrics**"		)
 for split_batches in [True, False]:
  for dispatch_batches in [True, False]:
   lowerCamelCase_      =  Accelerator(split_batches=UpperCAmelCase_					,				dispatch_batches=UpperCAmelCase_		)
   if accelerator.is_local_main_process:
    print(F'''With: `split_batches={split_batches}`, `dispatch_batches={dispatch_batches}`, length=99'''		)
   test_torch_metrics(UpperCAmelCase_					,				99		)
   accelerator.state._reset_state()
 if accelerator.is_local_main_process:
  print("**Test last batch is not dropped when perfectly divisible**"		)
 lowerCamelCase_      =  Accelerator()
 test_torch_metrics(UpperCAmelCase_					,				512		)
 accelerator.state._reset_state()





def __snake_case					(      UpperCAmelCase_						:						int		):
 # For xla_spawn (TPUs)
 main()


if __name__ == "__main__":
     main()







import inspect
import unittest

from transformers import RegNetConfig
from transformers.file_utils import cached_property, is_torch_available, is_vision_available
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch
     from torch import nn

     from transformers import RegNetForImageClassification, RegNetModel
     from transformers.models.regnet.modeling_regnet import REGNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from PIL import Image

     from transformers import AutoImageProcessor






class   _lowerCamelCase :




       """simple docstring"""



       def __init__(      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE=3				,       _SCREAMING_SNAKE_CASE=32				,       _SCREAMING_SNAKE_CASE=3				,       _SCREAMING_SNAKE_CASE=10				,       _SCREAMING_SNAKE_CASE=[10, 20, 30, 40]				,       _SCREAMING_SNAKE_CASE=[1, 1, 2, 1]				,       _SCREAMING_SNAKE_CASE=True				,       _SCREAMING_SNAKE_CASE=True				,       _SCREAMING_SNAKE_CASE="relu"				,       _SCREAMING_SNAKE_CASE=3				,       _SCREAMING_SNAKE_CASE=None				,       )->List[str]:




         '''simple docstring'''





         A_      :				str		     =					parent
         A_      :				int		     =					batch_size
         A_      :				List[str]		     =					image_size
         A_      :				Dict		     =					num_channels
         A_      :				Tuple		     =					embeddings_size
         A_      :				Union[str, Any]		     =					hidden_sizes
         A_      :				Dict		     =					depths
         A_      :				str		     =					is_training
         A_      :				Union[str, Any]		     =					use_labels
         A_      :				Union[str, Any]		     =					hidden_act
         A_      :				Optional[Any]		     =					num_labels
         A_      :				Tuple		     =					scope
         A_      :				Optional[int]		     =					len(_SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self		)->Optional[Any]:




         '''simple docstring'''





         A_      :				List[str]		     =					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]		)

         A_      :				str		     =					None
         if self.use_labels:
           A_      :				Union[str, Any]		     =					ids_tensor([self.batch_size]				,       self.num_labels		)

         A_      :				Optional[Any]		     =					self.get_config()

         return config, pixel_values, labels



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




         '''simple docstring'''





         return RegNetConfig(
             num_channels=self.num_channels				,       embeddings_size=self.embeddings_size				,       hidden_sizes=self.hidden_sizes				,       depths=self.depths				,       hidden_act=self.hidden_act				,       num_labels=self.num_labels				,       )



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




         '''simple docstring'''





         A_      :				Dict		     =					RegNetModel(config=_SCREAMING_SNAKE_CASE		)
         model.to(_SCREAMING_SNAKE_CASE		)
         model.eval()
         A_      :				Any		     =					model(_SCREAMING_SNAKE_CASE		)
         # expected last hidden states: B, C, H // 32, W // 32
         self.parent.assertEqual(
             result.last_hidden_state.shape				,       (self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32)				,       )



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




         '''simple docstring'''





         A_      :				Union[str, Any]		     =					self.num_labels
         A_      :				Dict		     =					RegNetForImageClassification(_SCREAMING_SNAKE_CASE		)
         model.to(_SCREAMING_SNAKE_CASE		)
         model.eval()
         A_      :				int		     =					model(_SCREAMING_SNAKE_CASE				,       labels=_SCREAMING_SNAKE_CASE		)
         self.parent.assertEqual(result.logits.shape				,       (self.batch_size, self.num_labels)		)







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




         '''simple docstring'''





         A_      :				Tuple		     =					self.prepare_config_and_inputs()
         A_     ,       A_     ,       A_      :				str		     =					config_and_inputs
         A_      :				Any		     =					{'''pixel_values''': pixel_values}
         return config, inputs_dict






@require_torch
class   _lowerCamelCase (					UpperCamelCase					,							UpperCamelCase					,							unittest.TestCase			):




       """simple docstring"""
       snake_case      =    (RegNetModel, RegNetForImageClassification) if is_torch_available() else ()
       snake_case      =    (
           {"feature-extraction": RegNetModel, "image-classification": RegNetForImageClassification}
           if is_torch_available()
           else {}
       )

       snake_case      =    False
       snake_case      =    False
       snake_case      =    False
       snake_case      =    False



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




         '''simple docstring'''





         A_      :				Union[str, Any]		     =					RegNetModelTester(self		)
         A_      :				Union[str, Any]		     =					ConfigTester(self				,       config_class=_SCREAMING_SNAKE_CASE				,       has_text_modality=_SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self		)->Dict:




         '''simple docstring'''





         self.create_and_test_config_common_properties()
         self.config_tester.create_and_test_config_to_json_string()
         self.config_tester.create_and_test_config_to_json_file()
         self.config_tester.create_and_test_config_from_and_save_pretrained()
         self.config_tester.create_and_test_config_with_num_labels()
         self.config_tester.check_config_can_be_init_without_params()
         self.config_tester.check_config_arguments_init()



       def        _snake_case       (      self		)->Tuple:




         '''simple docstring'''





         return



       @unittest.skip(reason='''RegNet does not use inputs_embeds'''		)
       def        _snake_case       (      self		)->Dict:




         '''simple docstring'''





         pass



       @unittest.skip(reason='''RegNet does not support input and output embeddings'''		)
       def        _snake_case       (      self		)->str:




         '''simple docstring'''





         pass



       def        _snake_case       (      self		)->List[Any]:




         '''simple docstring'''





         A_     ,       A_      :				Optional[Any]		     =					self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
           A_      :				str		     =					model_class(_SCREAMING_SNAKE_CASE		)
           A_      :				Union[str, Any]		     =					inspect.signature(model.forward		)
           # signature.parameters is an OrderedDict => so arg_names order is deterministic
           A_      :				Any		     =					[*signature.parameters.keys()]

           A_      :				Any		     =					['''pixel_values''']
           self.assertListEqual(arg_names[:1]				,       _SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self		)->Any:




         '''simple docstring'''





         A_      :				Dict		     =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*_SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self		)->Optional[Any]:




         '''simple docstring'''





         A_     ,       A_      :				int		     =					self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
           A_      :				Union[str, Any]		     =					model_class(config=_SCREAMING_SNAKE_CASE		)
           for name, module in model.named_modules():
             if isinstance(_SCREAMING_SNAKE_CASE				,       (nn.BatchNormad, nn.GroupNorm)		):
               self.assertTrue(
                   torch.all(module.weight == 1		)				,       msg=F'''Parameter {name} of model {model_class} seems not properly initialized'''				,       )
               self.assertTrue(
                   torch.all(module.bias == 0		)				,       msg=F'''Parameter {name} of model {model_class} seems not properly initialized'''				,       )



       def        _snake_case       (      self		)->List[Any]:




         '''simple docstring'''





         def check_hidden_states_output(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		):
           A_      :				str		     =					model_class(_SCREAMING_SNAKE_CASE		)
           model.to(_SCREAMING_SNAKE_CASE		)
           model.eval()

           with torch.no_grad():
             A_      :				Tuple		     =					model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)		)

           A_      :				Union[str, Any]		     =					outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

           A_      :				Optional[int]		     =					self.model_tester.num_stages
           self.assertEqual(len(_SCREAMING_SNAKE_CASE		)				,       expected_num_stages + 1		)

           # RegNet's feature maps are of shape (batch_size, num_channels, height, width)
           self.assertListEqual(
               list(hidden_states[0].shape[-2:]		)				,       [self.model_tester.image_size // 2, self.model_tester.image_size // 2]				,       )

         A_     ,       A_      :				Tuple		     =					self.model_tester.prepare_config_and_inputs_for_common()
         A_      :				int		     =					['''basic''', '''bottleneck''']
         for model_class in self.all_model_classes:
           for layer_type in layers_type:
             A_      :				int		     =					layer_type
             A_      :				List[Any]		     =					True
             check_hidden_states_output(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)

             # check that output_hidden_states also work using config
             del inputs_dict["output_hidden_states"]
             A_      :				str		     =					True

             check_hidden_states_output(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self		)->Dict:




         '''simple docstring'''





         A_      :				List[str]		     =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_image_classification(*_SCREAMING_SNAKE_CASE		)







       @slow
       def        _snake_case       (      self		)->str:




         '''simple docstring'''





         for model_name in REGNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           A_      :				Dict		     =					RegNetModel.from_pretrained(_SCREAMING_SNAKE_CASE		)
           self.assertIsNotNone(_SCREAMING_SNAKE_CASE		)




def   _SCREAMING_SNAKE_CASE	( ):
  A_      :				int		     =					Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''       )
  return image







@require_torch
@require_vision
class   _lowerCamelCase (					unittest.TestCase			):




       """simple docstring"""



       @cached_property
       def        _snake_case       (      self		)->List[str]:




         '''simple docstring'''





         return (
             AutoImageProcessor.from_pretrained(REGNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]		)
             if is_vision_available()
             else None
         )







       @slow
       def        _snake_case       (      self		)->Tuple:




         '''simple docstring'''





         A_      :				List[Any]		     =					RegNetForImageClassification.from_pretrained(REGNET_PRETRAINED_MODEL_ARCHIVE_LIST[0]		).to(_SCREAMING_SNAKE_CASE		)

         A_      :				Optional[Any]		     =					self.default_image_processor
         A_      :				Any		     =					prepare_img()
         A_      :				Optional[Any]		     =					image_processor(images=_SCREAMING_SNAKE_CASE				,       return_tensors='''pt'''		).to(_SCREAMING_SNAKE_CASE		)

         # forward pass
         with torch.no_grad():
           A_      :				Union[str, Any]		     =					model(**_SCREAMING_SNAKE_CASE		)

         # verify the logits
         A_      :				Union[str, Any]		     =					torch.Size((1, 1000)		)
         self.assertEqual(outputs.logits.shape				,       _SCREAMING_SNAKE_CASE		)

         A_      :				Optional[int]		     =					torch.tensor([-0.4_1_8_0, -1.5_0_5_1, -3.4_8_3_6]		).to(_SCREAMING_SNAKE_CASE		)

         self.assertTrue(torch.allclose(outputs.logits[0, :3]				,       _SCREAMING_SNAKE_CASE				,       atol=1e-4		)		)




