
# 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


UpperCAmelCase__				:  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 			UpperCamelCase_			(						__UpperCamelCase   ):




														'''simple docstring'''

														def __init__(				self     ,						UpperCamelCase = 1_01) ->	Union[str, Any]:
																	UpperCamelCase__    :		str     =				length
														def __len__(				self) ->	Optional[Any]:
																	return self.length






														def __getitem__(				self     ,						UpperCamelCase) ->	int:
																	return i



							class 			UpperCamelCase_			:




														'''simple docstring'''

														def __call__(				self     ,						UpperCamelCase) ->	str:
																	return {"input_ids": torch.tensor(UpperCamelCase), "labels": torch.tensor(UpperCamelCase)}



							class 			UpperCamelCase_			(						nn.Module   ):




														'''simple docstring'''

														def __init__(				self) ->	str:
																	super().__init__()
																	# Add some (unused) params otherwise DDP will complain.
																	UpperCamelCase__    :		int     =				nn.Linear(1_20     ,						80)






														def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase=None) ->	Optional[Any]:
																	if labels is not None:
																				return torch.tensor(0.0     ,						device=input_ids.device), input_ids
																	else:
																				return input_ids



def       _lowercase					(      )	->     int:
			UpperCamelCase__    :		List[Any]     =				HfArgumentParser((TrainingArguments,) )
			sys.argv += ["--output_dir", "./examples"]
			UpperCamelCase__    :		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]:
						UpperCamelCase__    :		Tuple     =				DummyDataset(__SCREAMING_SNAKE_CASE )

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

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

						UpperCamelCase__    :		Optional[Any]     =				trainer.predict(__SCREAMING_SNAKE_CASE )
						logger.info(p.metrics )
						if p.metrics["test_success"] is not True:
									logger.error(p.metrics )
									exit(1 )

						UpperCamelCase__    :		List[Any]     =				2

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

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

						UpperCamelCase__    :		List[Any]     =				None

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



def       _lowercase					(      __SCREAMING_SNAKE_CASE )	->     Any:
			# For xla_spawn (TPUs)
			main()


if __name__ == "__main__":
							main()





import enum
import inspect
from typing import Iterable, List, Optional, Tuple, Union



class 			UpperCamelCase_			(						enum.Enum   ):




							'''simple docstring'''

							UpperCamelCase_					   =							"""timm"""
							UpperCamelCase_					   =							"""transformers"""



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )	->     Dict:
			if stage_names is None:
						raise ValueError('Stage_names must be set for transformers backbones' )

			if out_features is not None:
						if not isinstance(__SCREAMING_SNAKE_CASE				,    (list,) ):
									raise ValueError(F"""out_features must be a list {type(__SCREAMING_SNAKE_CASE )}""" )
						if any(feat not in stage_names for feat in out_features ):
									raise ValueError(F"""out_features must be a subset of stage_names: {stage_names} got {out_features}""" )

			if out_indices is not None:
						if not isinstance(__SCREAMING_SNAKE_CASE				,    (list, tuple) ):
									raise ValueError(F"""out_indices must be a list or tuple, got {type(__SCREAMING_SNAKE_CASE )}""" )
						if any(idx >= len(__SCREAMING_SNAKE_CASE ) for idx in out_indices ):
									raise ValueError('out_indices must be valid indices for stage_names {stage_names}, got {out_indices}' )

			if out_features is not None and out_indices is not None:
						if len(__SCREAMING_SNAKE_CASE ) != len(__SCREAMING_SNAKE_CASE ):
									raise ValueError('out_features and out_indices should have the same length if both are set' )
						if out_features != [stage_names[idx] for idx in out_indices]:
									raise ValueError('out_features and out_indices should correspond to the same stages if both are set' )



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    )	->     List[Any]:
			if out_indices is None and out_features is None:
						UpperCamelCase__    :		List[Any]     =				[len(__SCREAMING_SNAKE_CASE ) - 1]
						UpperCamelCase__    :		Tuple     =				[stage_names[-1]]
			elif out_indices is None and out_features is not None:
						UpperCamelCase__    :		str     =				[stage_names.index(__SCREAMING_SNAKE_CASE ) for layer in out_features]
			elif out_features is None and out_indices is not None:
						UpperCamelCase__    :		int     =				[stage_names[idx] for idx in out_indices]
			return out_features, out_indices



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    )	->     Tuple[List[str], List[int]]:
			verify_out_features_out_indices(out_features=__SCREAMING_SNAKE_CASE				,    out_indices=__SCREAMING_SNAKE_CASE				,    stage_names=__SCREAMING_SNAKE_CASE )
			UpperCamelCase__   ,			UpperCamelCase__    :		str     =				_align_output_features_output_indices(
			    out_features=__SCREAMING_SNAKE_CASE				,    out_indices=__SCREAMING_SNAKE_CASE				,    stage_names=__SCREAMING_SNAKE_CASE )
			# Verify that the aligned out_features and out_indices are valid
			verify_out_features_out_indices(out_features=__SCREAMING_SNAKE_CASE				,    out_indices=__SCREAMING_SNAKE_CASE				,    stage_names=__SCREAMING_SNAKE_CASE )
			return output_features, output_indices



class 			UpperCamelCase_			:




							'''simple docstring'''

							UpperCamelCase_					   =							None
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	None:
										if getattr(self     ,						'_backbone'     ,						UpperCamelCase) is None:
													raise ValueError('self._backbone must be set before calling _init_timm_backbone')

										# These will diagree with the defaults for the transformers models e.g. for resnet50
										# the transformer model has out_features = ['stem', 'stage1', 'stage2', 'stage3', 'stage4']
										# the timm model has out_features = ['act', 'layer1', 'layer2', 'layer3', 'layer4']
										UpperCamelCase__    :		int     =				[stage['module'] for stage in self._backbone.feature_info.info]
										UpperCamelCase__    :		str     =				[stage['num_chs'] for stage in self._backbone.feature_info.info]
										UpperCamelCase__    :		Optional[int]     =				self._backbone.feature_info.out_indices
										UpperCamelCase__    :		Optional[int]     =				self._backbone.feature_info.module_name()

										# We verify the out indices and out features are valid
										verify_out_features_out_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[int]     =				out_features, out_indices
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	None:
										UpperCamelCase__    :		List[str]     =				getattr(UpperCamelCase     ,						'stage_names')
										UpperCamelCase__    :		List[str]     =				getattr(UpperCamelCase     ,						'out_features'     ,						UpperCamelCase)
										UpperCamelCase__    :		Optional[Any]     =				getattr(UpperCamelCase     ,						'out_indices'     ,						UpperCamelCase)

										UpperCamelCase__    :		Tuple     =				stage_names
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[Any]     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=UpperCamelCase)
										# Number of channels for each stage. This is set in the transformer backbone model init
										UpperCamelCase__    :		Dict     =				None
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	None:
										UpperCamelCase__    :		List[Any]     =				config

										UpperCamelCase__    :		Dict     =				getattr(UpperCamelCase     ,						'use_timm_backbone'     ,						UpperCamelCase)
										UpperCamelCase__    :		List[str]     =				BackboneType.TIMM if self.use_timm_backbone else BackboneType.TRANSFORMERS

										if self.backbone_type == BackboneType.TIMM:
													self._init_timm_backbone(UpperCamelCase)
										elif self.backbone_type == BackboneType.TRANSFORMERS:
													self._init_transformers_backbone(UpperCamelCase)
										else:
													raise ValueError(F"""backbone_type {self.backbone_type} not supported.""")
							@property
							def        lowerCAmelCase__ (				self) ->	List[str]:
										return self._out_features
							@out_features.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	int:
										UpperCamelCase__   ,			UpperCamelCase__    :		str     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
							@property
							def        lowerCAmelCase__ (				self) ->	List[Any]:
										return self._out_indices
							@out_indices.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	int:
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[int]     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
							@property
							def        lowerCAmelCase__ (				self) ->	List[str]:
										# the current backbones will output the number of channels for each stage
										# even if that stage is not in the out_features list.
										return {stage: self.num_features[i] for i, stage in enumerate(self.stage_names)}
							@property
							def        lowerCAmelCase__ (				self) ->	Union[str, Any]:
										return [self.out_feature_channels[name] for name in self.out_features]
							def        lowerCAmelCase__ (				self     ,						*UpperCamelCase     ,						**UpperCamelCase) ->	str:
										UpperCamelCase__    :		List[Any]     =				dict(inspect.signature(self.forward).parameters)
										UpperCamelCase__    :		Any     =				{k: v for k, v in kwargs.items() if k in signature}
										return self(*UpperCamelCase     ,						**UpperCamelCase)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase = None     ,						UpperCamelCase = None     ,						UpperCamelCase = None     ,						) ->	Union[str, Any]:
										raise NotImplementedError('This method should be implemented by the derived class.')






							def        lowerCAmelCase__ (				self) ->	int:
										UpperCamelCase__    :		Any     =				super().to_dict()
										UpperCamelCase__    :		Optional[Any]     =				output.pop('_out_features')
										UpperCamelCase__    :		Any     =				output.pop('_out_indices')
										return output



class 			UpperCamelCase_			:




							'''simple docstring'''

							@property
							def        lowerCAmelCase__ (				self) ->	Tuple:
										return self._out_features
							@out_features.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	Tuple:
										UpperCamelCase__   ,			UpperCamelCase__    :		Union[str, Any]     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
							@property
							def        lowerCAmelCase__ (				self) ->	Optional[int]:
										return self._out_indices
							@out_indices.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	int:
										UpperCamelCase__   ,			UpperCamelCase__    :		str     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)






							def        lowerCAmelCase__ (				self) ->	Tuple:
										UpperCamelCase__    :		List[Any]     =				super().to_dict()
										UpperCamelCase__    :		Optional[int]     =				output.pop('_out_features')
										UpperCamelCase__    :		Tuple     =				output.pop('_out_indices')
										return output



