

"""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


a					:List[str]    = 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      __a							(UpperCamelCase_):






       '''simple docstring'''

       def __init__(						self	,    _a = 101  )  ->      Any:


             """simple docstring"""


             SCREAMING_SNAKE_CASE__   :      Any      		=	length

       def __len__(						self  )  ->      Tuple:


             """simple docstring"""


             return self.length





       def __getitem__(						self	,    _a  )  ->      int:


             """simple docstring"""


             return i





  class      __a							:






       '''simple docstring'''

       def __call__(						self	,    _a  )  ->      Optional[Any]:


             """simple docstring"""


             return {"input_ids": torch.tensor(_a  ), "labels": torch.tensor(_a  )}





  class      __a							(nn.Module):






       '''simple docstring'''

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


             """simple docstring"""


             super().__init__()
             # Add some (unused) params otherwise DDP will complain.
             SCREAMING_SNAKE_CASE__   :      int      		=	nn.Linear(120	,    80  )





       def 					_a			(						self	,    _a	,    _a=None  )  ->      Any:


             """simple docstring"""


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

class      __a							(UpperCamelCase_):






     '''simple docstring'''

     @require_torch_neuroncore
     def 					_a			(						self  )  ->      Dict:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      List[str]      		=	f'''--nproc_per_node=2
            --master_port={get_torch_dist_unique_port()}
            {self.test_file_dir}/test_trainer_distributed.py
        '''.split()
           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	self.get_auto_remove_tmp_dir()
           SCREAMING_SNAKE_CASE__   :      Dict      		=	f'''--output_dir {output_dir}'''.split()
           SCREAMING_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





class      __a							(UpperCamelCase_):






     '''simple docstring'''

     @require_torch_multi_gpu
     def 					_a			(						self  )  ->      Any:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      List[Any]      		=	f'''--nproc_per_node={torch.cuda.device_count()}
            --master_port={get_torch_dist_unique_port()}
            {self.test_file_dir}/test_trainer_distributed.py
        '''.split()
           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	self.get_auto_remove_tmp_dir()
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	f'''--output_dir {output_dir}'''.split()
           SCREAMING_SNAKE_CASE__   :      int      		=	["""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

  a					:Tuple    = HfArgumentParser((TrainingArguments,))
  a					:List[Any]    = 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 [101, 40, 7]:
    a					:Union[str, Any]    = DummyDataset(dataset_length)




    def  _lowercase					(    __lowerCAmelCase							)       ->				Dict:
          SCREAMING_SNAKE_CASE__   :      Tuple      		=	list(range(len(__lowerCAmelCase							)							)							)
          SCREAMING_SNAKE_CASE__   :      Tuple      		=	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}

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

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

    a					:Tuple    = 2

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

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



    a					:List[str]    = None







"""simple docstring"""


from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


a					:List[Any]    = logging.get_logger(__name__)

a					:Optional[int]    = {
    "microsoft/focalnet-tiny": "https://huggingface.co/microsoft/focalnet-tiny/resolve/main/config.json",
}

class      __a							(UpperCamelCase_   ,	UpperCamelCase_):






     '''simple docstring'''
     _SCREAMING_SNAKE_CASE						:Any       = """focalnet"""

     def __init__(						self	,    _a=224	,    _a=4	,    _a=3	,    _a=96	,    _a=False	,    _a=[192, 384, 768, 768]	,    _a=[2, 2, 6, 2]	,    _a=[2, 2, 2, 2]	,    _a=[3, 3, 3, 3]	,    _a="gelu"	,    _a=4.0	,    _a=0.0	,    _a=0.1	,    _a=False	,    _a=1E-4	,    _a=False	,    _a=False	,    _a=False	,    _a=0.02	,    _a=1E-5	,    _a=32	,    _a=None	,    _a=None	,    **_a	,    )  ->      Optional[Any]:


           """simple docstring"""


           super().__init__(**_a  )

           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	image_size
           SCREAMING_SNAKE_CASE__   :      str      		=	patch_size
           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	num_channels
           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	embed_dim
           SCREAMING_SNAKE_CASE__   :      List[str]      		=	use_conv_embed
           SCREAMING_SNAKE_CASE__   :      List[str]      		=	hidden_sizes
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	depths
           SCREAMING_SNAKE_CASE__   :      Any      		=	focal_levels
           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	focal_windows
           SCREAMING_SNAKE_CASE__   :      Any      		=	hidden_act
           SCREAMING_SNAKE_CASE__   :      Tuple      		=	mlp_ratio
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	hidden_dropout_prob
           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	drop_path_rate
           SCREAMING_SNAKE_CASE__   :      str      		=	use_layerscale
           SCREAMING_SNAKE_CASE__   :      int      		=	layerscale_value
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	use_post_layernorm
           SCREAMING_SNAKE_CASE__   :      Any      		=	use_post_layernorm_in_modulation
           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	normalize_modulator
           SCREAMING_SNAKE_CASE__   :      str      		=	initializer_range
           SCREAMING_SNAKE_CASE__   :      Any      		=	layer_norm_eps
           SCREAMING_SNAKE_CASE__   :      Any      		=	encoder_stride
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	["""stem"""] + [f'''stage{idx}''' for idx in range(1	,    len(self.depths  ) + 1  )]
           SCREAMING_SNAKE_CASE__		,		SCREAMING_SNAKE_CASE__   :      Dict      		=	get_aligned_output_features_output_indices(
               out_features=_a	,    out_indices=_a	,    stage_names=self.stage_names  )




