

'''simple docstring'''







import copy
import inspect
import unittest

from transformers import AutoBackbone
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import require_timm, require_torch, torch_device
from transformers.utils.import_utils import is_torch_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor


if is_torch_available():
      import torch

      from transformers import TimmBackbone, TimmBackboneConfig

from ...test_pipeline_mixin import PipelineTesterMixin


class    A__ :



    def __init__(   self						,			UpperCamelCase__						,			UpperCamelCase__=None						,			UpperCamelCase__=None						,			UpperCamelCase__=None						,			UpperCamelCase__="resnet50"						,			UpperCamelCase__=3						,			UpperCamelCase__=32						,			UpperCamelCase__=3						,			UpperCamelCase__=True						,			UpperCamelCase__=True						,			)  -> Any:





           '''simple docstring'''
           A_		   = parent
           A_		   = out_indices if out_indices is not None else [4]
           A_		   = stage_names
           A_		   = out_features
           A_		   = backbone
           A_		   = batch_size
           A_		   = image_size
           A_		   = num_channels
           A_		   = use_pretrained_backbone
           A_		   = is_training



    def    snake_case_     (   self    )  -> List[str]:





           '''simple docstring'''
           A_		   = floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]    )
           A_		   = self.get_config()

           return config, pixel_values



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





           '''simple docstring'''
           return TimmBackboneConfig(
               image_size=self.image_size						,			num_channels=self.num_channels						,			out_features=self.out_features						,			out_indices=self.out_indices						,			stage_names=self.stage_names						,			use_pretrained_backbone=self.use_pretrained_backbone						,			backbone=self.backbone						,			)



    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__    )  -> Any:





           '''simple docstring'''
           A_		   = TimmBackbone(config=UpperCamelCase__    )
           model.to(UpperCamelCase__    )
           model.eval()
           with torch.no_grad():
                  A_		   = model(UpperCamelCase__    )
           self.parent.assertEqual(
               result.feature_map[-1].shape						,			(self.batch_size, model.channels[-1], 14, 14)						,			)







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





           '''simple docstring'''
           A_		   = self.prepare_config_and_inputs()
           A_   ,     A_		   = config_and_inputs
           A_		   = {"""pixel_values""": pixel_values}
           return config, inputs_dict


@require_torch
@require_timm
class    A__ ( _snake_case  ,  _snake_case  ,  _snake_case  ,  unittest.TestCase     ):
    lowercase		   =						(TimmBackbone,) if is_torch_available() else ()
    lowercase		   =						{"feature-extraction": TimmBackbone} if is_torch_available() else {}
    lowercase		   =						False
    lowercase		   =						False
    lowercase		   =						False
    lowercase		   =						False



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





           '''simple docstring'''
           A_		   = TimmBackboneModelTester(self    )
           A_		   = ConfigTester(self						,			config_class=UpperCamelCase__						,			has_text_modality=UpperCamelCase__    )



    def    snake_case_     (   self    )  -> Tuple:





           '''simple docstring'''
           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    )  -> Any:





           '''simple docstring'''
           A_		   = """resnet18"""
           A_		   = """microsoft/resnet-18"""

           A_		   = AutoBackbone.from_pretrained(UpperCamelCase__						,			use_timm_backbone=UpperCamelCase__    )
           A_		   = AutoBackbone.from_pretrained(UpperCamelCase__    )

           self.assertEqual(len(timm_model.out_features    )						,			len(transformers_model.out_features    )    )
           self.assertEqual(len(timm_model.stage_names    )						,			len(transformers_model.stage_names    )    )
           self.assertEqual(timm_model.channels						,			transformers_model.channels    )
           # Out indices are set to the last layer by default. For timm models, we don't know
           # the number of layers in advance, so we set it to (-1,), whereas for transformers
           # models, we set it to [len(stage_names) - 1] (kept for backward compatibility).
           self.assertEqual(timm_model.out_indices						,			(-1,)    )
           self.assertEqual(transformers_model.out_indices						,			[len(timm_model.stage_names    ) - 1]    )

           A_		   = AutoBackbone.from_pretrained(UpperCamelCase__						,			use_timm_backbone=UpperCamelCase__						,			out_indices=[1, 2, 3]    )
           A_		   = AutoBackbone.from_pretrained(UpperCamelCase__						,			out_indices=[1, 2, 3]    )

           self.assertEqual(timm_model.out_indices						,			transformers_model.out_indices    )
           self.assertEqual(len(timm_model.out_features    )						,			len(transformers_model.out_features    )    )
           self.assertEqual(timm_model.channels						,			transformers_model.channels    )



    @unittest.skip("""TimmBackbone doesn't support feed forward chunking"""    )
    def    snake_case_     (   self    )  -> List[str]:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone doesn't have num_hidden_layers attribute"""    )
    def    snake_case_     (   self    )  -> Dict:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone initialization is managed on the timm side"""    )
    def    snake_case_     (   self    )  -> Optional[int]:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone models doesn't have inputs_embeds"""    )
    def    snake_case_     (   self    )  -> str:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone models doesn't have inputs_embeds"""    )
    def    snake_case_     (   self    )  -> Union[str, Any]:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone model cannot be created without specifying a backbone checkpoint"""    )
    def    snake_case_     (   self    )  -> List[str]:





           '''simple docstring'''
           pass



    @unittest.skip("""Only checkpoints on timm can be loaded into TimmBackbone"""    )
    def    snake_case_     (   self    )  -> Union[str, Any]:





           '''simple docstring'''
           pass



    @unittest.skip("""model weights aren't tied in TimmBackbone."""    )
    def    snake_case_     (   self    )  -> Any:





           '''simple docstring'''
           pass



    @unittest.skip("""model weights aren't tied in TimmBackbone."""    )
    def    snake_case_     (   self    )  -> Tuple:





           '''simple docstring'''
           pass



    @unittest.skip("""Only checkpoints on timm can be loaded into TimmBackbone"""    )
    def    snake_case_     (   self    )  -> Optional[Any]:





           '''simple docstring'''
           pass



    @unittest.skip("""Only checkpoints on timm can be loaded into TimmBackbone"""    )
    def    snake_case_     (   self    )  -> Optional[int]:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone doesn't have hidden size info in its configuration."""    )
    def    snake_case_     (   self    )  -> List[Any]:





           '''simple docstring'''
           pass



    @unittest.skip("""TimmBackbone doesn't support output_attentions."""    )
    def    snake_case_     (   self    )  -> Any:





           '''simple docstring'''
           pass



    @unittest.skip("""Safetensors is not supported by timm."""    )
    def    snake_case_     (   self    )  -> Optional[Any]:





           '''simple docstring'''
           pass



    @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""    )
    def    snake_case_     (   self    )  -> str:





           '''simple docstring'''
           pass



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





           '''simple docstring'''
           A_   ,     A_		   = self.model_tester.prepare_config_and_inputs_for_common()

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

                  A_		   = ["""pixel_values"""]
                  self.assertListEqual(arg_names[:1]						,			UpperCamelCase__    )



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





           '''simple docstring'''
           A_   ,     A_		   = self.model_tester.prepare_config_and_inputs_for_common()
           A_		   = True
           A_		   = self.has_attentions

           # no need to test all models as different heads yield the same functionality
           A_		   = self.all_model_classes[0]
           A_		   = model_class(UpperCamelCase__    )
           model.to(UpperCamelCase__    )

           A_		   = self._prepare_for_class(UpperCamelCase__						,			UpperCamelCase__    )
           A_		   = model(**UpperCamelCase__    )
           A_		   = outputs[0][-1]

           # Encoder-/Decoder-only models
           A_		   = outputs.hidden_states[0]
           hidden_states.retain_grad()

           if self.has_attentions:
                  A_		   = outputs.attentions[0]
                  attentions.retain_grad()

           output.flatten()[0].backward(retain_graph=UpperCamelCase__    )

           self.assertIsNotNone(hidden_states.grad    )

           if self.has_attentions:
                  self.assertIsNotNone(attentions.grad    )







    def    snake_case_     (   self    )  -> Dict:





           '''simple docstring'''
           A_   ,     A_		   = self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
                  A_		   = model_class(UpperCamelCase__    )
                  model.to(UpperCamelCase__    )
                  model.eval()
                  A_		   = model(**UpperCamelCase__    )

                  self.assertEqual(len(result.feature_maps    )						,			len(config.out_indices    )    )
                  self.assertEqual(len(model.channels    )						,			len(config.out_indices    )    )

                  # Check output of last stage is taken if out_features=None, out_indices=None
                  A_		   = copy.deepcopy(UpperCamelCase__    )
                  A_		   = None
                  A_		   = model_class(UpperCamelCase__    )
                  model.to(UpperCamelCase__    )
                  model.eval()
                  A_		   = model(**UpperCamelCase__    )

                  self.assertEqual(len(result.feature_maps    )						,			1    )
                  self.assertEqual(len(model.channels    )						,			1    )

                  # Check backbone can be initialized with fresh weights
                  A_		   = copy.deepcopy(UpperCamelCase__    )
                  A_		   = False
                  A_		   = model_class(UpperCamelCase__    )
                  model.to(UpperCamelCase__    )
                  model.eval()
                  A_		   = model(**UpperCamelCase__    )



'''simple docstring'''







import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s''',
    datefmt='''%m/%d/%Y %H:%M:%S''',
    level=logging.INFO,
)
__lowerCamelCase						    =      logging.getLogger(__name__)
def  UpperCAmelCase__      ( UpperCAmelCase__			)     ->     int:
       A_		   = git.Repo(search_parent_directories=UpperCAmelCase__			)
       A_		   = {
           """repo_id""": str(UpperCAmelCase__			),
           """repo_sha""": str(repo.head.object.hexsha			),
           """repo_branch""": str(repo.active_branch			),
       }

       with open(os.path.join(UpperCAmelCase__,    """git_log.json"""			),    """w"""			) as f:
              json.dump(UpperCAmelCase__,    UpperCAmelCase__,    indent=4			)
def  UpperCAmelCase__      ( UpperCAmelCase__			)     ->     Dict:
       if params.n_gpu <= 0:
              A_		   = 0
              A_		   = -1
              A_		   = True
              A_		   = False
              return

       assert torch.cuda.is_available()

       logger.info("""Initializing GPUs"""			)
       if params.n_gpu > 1:
              assert params.local_rank != -1

              A_		   = int(os.environ["""WORLD_SIZE"""]			)
              A_		   = int(os.environ["""N_GPU_NODE"""]			)
              A_		   = int(os.environ["""RANK"""]			)

              # number of nodes / node ID
              A_		   = params.world_size // params.n_gpu_per_node
              A_		   = params.global_rank // params.n_gpu_per_node
              A_		   = True

              assert params.n_nodes == int(os.environ["""N_NODES"""]			)
              assert params.node_id == int(os.environ["""NODE_RANK"""]			)

       # local job (single GPU)
       else:
              assert params.local_rank == -1

              A_		   = 1
              A_		   = 0
              A_		   = 0
              A_		   = 0
              A_		   = 1
              A_		   = 1
              A_		   = False

       # sanity checks
       assert params.n_nodes >= 1
       assert 0 <= params.node_id < params.n_nodes
       assert 0 <= params.local_rank <= params.global_rank < params.world_size
       assert params.world_size == params.n_nodes * params.n_gpu_per_node

       # define whether this is the master process / if we are in multi-node distributed mode
       A_		   = params.node_id == 0 and params.local_rank == 0
       A_		   = params.n_nodes > 1

       # summary
       A_		   = F'''--- Global rank: {params.global_rank} - '''
       logger.info(PREFIX + """Number of nodes: %i""" % params.n_nodes			)
       logger.info(PREFIX + """Node ID        : %i""" % params.node_id			)
       logger.info(PREFIX + """Local rank     : %i""" % params.local_rank			)
       logger.info(PREFIX + """World size     : %i""" % params.world_size			)
       logger.info(PREFIX + """GPUs per node  : %i""" % params.n_gpu_per_node			)
       logger.info(PREFIX + """Master         : %s""" % str(params.is_master			)			)
       logger.info(PREFIX + """Multi-node     : %s""" % str(params.multi_node			)			)
       logger.info(PREFIX + """Multi-GPU      : %s""" % str(params.multi_gpu			)			)
       logger.info(PREFIX + """Hostname       : %s""" % socket.gethostname()			)

       # set GPU device
       torch.cuda.set_device(params.local_rank			)

       # initialize multi-GPU
       if params.multi_gpu:
              logger.info("""Initializing PyTorch distributed"""			)
              torch.distributed.init_process_group(
                  init_method="""env://""",    backend="""nccl""",    )







def  UpperCAmelCase__      ( UpperCAmelCase__			)     ->     Tuple:
       np.random.seed(args.seed			)
       torch.manual_seed(args.seed			)
       if args.n_gpu > 0:
              torch.cuda.manual_seed_all(args.seed			)
