




"""simple docstring"""





# This test is meant to be run in on an instance with TPUs like this:
#
#   python examples/pytorch/xla_spawn.py --num_cores=8 tests/test_trainer_tpu.py
#
# Replace 8 with the number of TPU cores you have.
#

import sys
from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.utils import logging


_lowercase		 =      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  __lowercase    (     snake_case_		):


											'''simple docstring'''


											def __init__(				self				:       Any ,SCREAMING_SNAKE_CASE_				:       int = 1_0_1 ):
																	snake_case__    :   Any							     =   length

											def __len__(				self				:       Tuple ):
																	return self.length





											def __getitem__(				self				:       Any ,SCREAMING_SNAKE_CASE_				:       Union[str, Any] ):
																	return i







							class  __lowercase    :


											'''simple docstring'''


											def __call__(				self				:       Optional[int] ,SCREAMING_SNAKE_CASE_				:       Dict ):
																	return {"input_ids": torch.tensor(SCREAMING_SNAKE_CASE_ ), "labels": torch.tensor(SCREAMING_SNAKE_CASE_ )}







							class  __lowercase    (     nn.Module		):


											'''simple docstring'''


											def __init__(				self				:       Optional[Any] ):
																	super().__init__()
																	# Add some (unused) params otherwise DDP will complain.
																	snake_case__    :   Any							     =   nn.Linear(1_2_0 ,8_0 )





											def  _lowerCamelCase   (				self				:       List[Any] ,SCREAMING_SNAKE_CASE_				:       Optional[int] ,SCREAMING_SNAKE_CASE_				:       List[Any]=None ):
																	if labels is not None:
																							return torch.tensor(0.0 ,device=input_ids.device ), input_ids
																	else:
																							return input_ids

def 						lowerCAmelCase							(	):
						'''simple docstring'''


						snake_case__    :   List[Any]							     =   HfArgumentParser((TrainingArguments,)						)
						sys.argv += ["--output_dir", "./examples"]
						snake_case__    :   List[str]							     =   parser.parse_args_into_dataclasses()[0]

						logger.warning(
						    f'''Process rank: {training_args.local_rank}, device: {training_args.device}, '''
						    f'''tpu_num_cores: {training_args.tpu_num_cores}''' ,     )

						# 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 [1001, 256, 15]:
												snake_case__    :   str							     =   DummyDataset(lowerCamelCase						)

												def compute_metrics(lowerCamelCase						) -> Dict:
																		snake_case__    :   Dict							     =   list(range(len(lowerCamelCase						)						)						)
																		snake_case__    :   Dict							     =   p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
																		return {"success": success}

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

												snake_case__    :   Optional[int]							     =   trainer.predict(lowerCamelCase						)
												logger.info(p.metrics						)
												if p.metrics["test_success"] is not True:
																		logger.error(p.metrics						)
																		exit(1						)

												snake_case__    :   Dict							     =   2

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

												snake_case__    :   Optional[int]							     =   trainer.predict(lowerCamelCase						)
												logger.info(p.metrics						)
												if p.metrics["test_success"] is not True:
																		logger.error(p.metrics						)
																		exit(1						)

												snake_case__    :   Optional[int]							     =   None

						logger.info('''🔥 All distributed tests successful'''						)

def 						lowerCAmelCase							(	lowerCamelCase						):
						'''simple docstring'''


						main()


if __name__ == "__main__":
							main()








"""simple docstring"""





def 						lowerCAmelCase							(	lowerCamelCase						):
						'''simple docstring'''


						return int(lowerCamelCase						) if x - int(lowerCamelCase						) <= 0 else int(lowerCamelCase						) + 1


if __name__ == "__main__":
							import doctest

							doctest.testmod()


