




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


UpperCamelCase_    	=				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				(			SCREAMING_SNAKE_CASE ):





             '''simple docstring'''






             def __init__(      self,							A = 101  ):







                   '''simple docstring'''







                   SCREAMING_SNAKE_CASE   :      Optional[int]    					=						length







             def __len__(      self  ):







                   '''simple docstring'''







                   return self.length







             def __getitem__(      self,							A  ):







                   '''simple docstring'''







                   return i






      class 							_a				:





             '''simple docstring'''






             def __call__(      self,							A  ):







                   '''simple docstring'''







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






      class 							_a				(			nn.Module ):





             '''simple docstring'''






             def __init__(      self  ):







                   '''simple docstring'''







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







             def        UpperCamelCase_							(      self,							A,							A=None  ):







                   '''simple docstring'''







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






class 							_a				(			SCREAMING_SNAKE_CASE ):





       '''simple docstring'''






       @require_torch_neuroncore
       def        UpperCamelCase_							(      self  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      List[str]    					=						F"--nproc_per_node=2\n            --master_port={get_torch_dist_unique_port()}\n            {self.test_file_dir}/test_trainer_distributed.py\n        ".split()
             SCREAMING_SNAKE_CASE   :      Tuple    					=						self.get_auto_remove_tmp_dir()
             SCREAMING_SNAKE_CASE   :      str    					=						F"--output_dir {output_dir}".split()
             SCREAMING_SNAKE_CASE   :      Dict    					=						['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				(			SCREAMING_SNAKE_CASE ):





       '''simple docstring'''






       @require_torch_multi_gpu
       def        UpperCamelCase_							(      self  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      List[str]    					=						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()
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						self.get_auto_remove_tmp_dir()
             SCREAMING_SNAKE_CASE   :      Optional[int]    					=						F"--output_dir {output_dir}".split()
             SCREAMING_SNAKE_CASE   :      Optional[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

      UpperCamelCase_    	=				HfArgumentParser((TrainingArguments,))
      UpperCamelCase_    	=				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_0_1, 4_0, 7]:
            UpperCamelCase_    	=				DummyDataset(dataset_length)




            def   lowercase__(			__UpperCamelCase:					EvalPrediction    ):

                  """simple docstring"""






                  SCREAMING_SNAKE_CASE   :      Dict    					=						list(range(len(__UpperCamelCase    )    )    )
                  SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						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}

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

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

            UpperCamelCase_    	=				2

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

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





            UpperCamelCase_    	=				None








'''simple docstring'''



from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


UpperCamelCase_    	=				logging.get_logger(__name__)


UpperCamelCase_    	=				TypeVar("DatasetType", Dataset, IterableDataset)




def   lowercase__(			__UpperCamelCase:					List[DatasetType]      ,__UpperCamelCase:					Optional[List[float]] = None      ,__UpperCamelCase:					Optional[int] = None      ,__UpperCamelCase:					Optional[DatasetInfo] = None      ,__UpperCamelCase:					Optional[NamedSplit] = None      ,__UpperCamelCase:					Literal["first_exhausted", "all_exhausted"] = "first_exhausted"      ,):

      """simple docstring"""






      from .arrow_dataset import Dataset
      from .iterable_dataset import IterableDataset

      if not datasets:
            raise ValueError('Unable to interleave an empty list of datasets.'    )
      for i, dataset in enumerate(__UpperCamelCase    ):
            if not isinstance(__UpperCamelCase      ,(Dataset, IterableDataset)    ):
                  if isinstance(__UpperCamelCase      ,(DatasetDict, IterableDatasetDict)    ):
                        if not dataset:
                              raise ValueError(
                                  f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
                                  'is an empty dataset dictionary.'    )
                        raise ValueError(
                            f"Dataset at position {i} has at least one split: {list(__UpperCamelCase    )}\n"
                            f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__UpperCamelCase    )    )}']"    )
                  raise ValueError(
                      f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__UpperCamelCase    ).__name__}."    )
            if i == 0:
                  SCREAMING_SNAKE_CASE					,    SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						(
                      (Dataset, IterableDataset) if isinstance(__UpperCamelCase      ,__UpperCamelCase    ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(__UpperCamelCase      ,__UpperCamelCase    ):
                  raise ValueError(
                      f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."    )
      if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
            raise ValueError(f"{stopping_strategy} is not supported. Please enter a valid stopping_strategy."    )
      if dataset_type is Dataset:
            return _interleave_map_style_datasets(
                __UpperCamelCase      ,__UpperCamelCase      ,__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,stopping_strategy=__UpperCamelCase    )
      else:
            return _interleave_iterable_datasets(
                __UpperCamelCase      ,__UpperCamelCase      ,__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,stopping_strategy=__UpperCamelCase    )





def   lowercase__(			__UpperCamelCase:					List[DatasetType]      ,__UpperCamelCase:					Optional[DatasetInfo] = None      ,__UpperCamelCase:					Optional[NamedSplit] = None      ,__UpperCamelCase:					int = 0      ,):

      """simple docstring"""






      if not dsets:
            raise ValueError('Unable to concatenate an empty list of datasets.'    )
      for i, dataset in enumerate(__UpperCamelCase    ):
            if not isinstance(__UpperCamelCase      ,(Dataset, IterableDataset)    ):
                  if isinstance(__UpperCamelCase      ,(DatasetDict, IterableDatasetDict)    ):
                        if not dataset:
                              raise ValueError(
                                  f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
                                  'is an empty dataset dictionary.'    )
                        raise ValueError(
                            f"Dataset at position {i} has at least one split: {list(__UpperCamelCase    )}\n"
                            f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__UpperCamelCase    )    )}']"    )
                  raise ValueError(
                      f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__UpperCamelCase    ).__name__}."    )
            if i == 0:
                  SCREAMING_SNAKE_CASE					,    SCREAMING_SNAKE_CASE   :      int    					=						(
                      (Dataset, IterableDataset) if isinstance(__UpperCamelCase      ,__UpperCamelCase    ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(__UpperCamelCase      ,__UpperCamelCase    ):
                  raise ValueError(
                      f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."    )
      if dataset_type is Dataset:
            return _concatenate_map_style_datasets(__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,axis=__UpperCamelCase    )
      else:
            return _concatenate_iterable_datasets(__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,axis=__UpperCamelCase    )


