




# 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


__a  :		Dict		     =   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(					snake_case_					):



							"""simple docstring"""






							def __init__(					self	,				lowerCamelCase = 101    ) ->      Optional[Any]:





														"""simple docstring"""


														lowercase__  :	List[Any]        =							length

							def __len__(					self    ) ->      str:





														"""simple docstring"""


														return self.length







							def __getitem__(					self	,				lowerCamelCase    ) ->      int:





														"""simple docstring"""


														return i







						class  UpperCAmelCase:



							"""simple docstring"""






							def __call__(					self	,				lowerCamelCase    ) ->      List[Any]:





														"""simple docstring"""


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







						class  UpperCAmelCase(					nn.Module					):



							"""simple docstring"""






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





														"""simple docstring"""


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







							def    __a     (					self	,				lowerCamelCase	,				lowerCamelCase=None    ) ->      Dict:





														"""simple docstring"""


														if labels is not None:
																					return torch.tensor(0.0	,				device=input_ids.device    ), input_ids
														else:
																					return input_ids
def  snake_case_					(		)      ->			Tuple:
							lowercase__  :	str        =							HfArgumentParser((TrainingArguments,)    )
							sys.argv += ["--output_dir", "./examples"]
							lowercase__  :	Tuple        =							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 [10_01, 2_56, 15]:
														lowercase__  :	str        =							DummyDataset(SCREAMING_SNAKE_CASE_    )

														def compute_metrics(SCREAMING_SNAKE_CASE_    ) -> Dict:
																					lowercase__  :	List[Any]        =							list(range(len(SCREAMING_SNAKE_CASE_    )    )    )
																					lowercase__  :	Union[str, Any]        =							p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
																					return {"success": success}

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

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

														lowercase__  :	int        =							2

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

														lowercase__  :	List[Any]        =							trainer.predict(SCREAMING_SNAKE_CASE_    )
														logger.info(p.metrics    )
														if p.metrics["test_success"] is not True:
																					logger.error(p.metrics    )
																					exit(1    )

														lowercase__  :	Tuple        =							None

							logger.info("🔥 All distributed tests successful"    )
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			List[Any]:
							# For xla_spawn (TPUs)
							main()


if __name__ == "__main__":
						main()





import hashlib
import importlib.util
import json
import os
import pathlib
from types import ModuleType

import pytest
import requests

__a  :		str		     =   pathlib.Path.cwd().joinpath('''project_euler''')
__a  :		Tuple		     =   pathlib.Path.cwd().joinpath(
    '''scripts''', '''project_euler_answers.json'''
)

with open(PROJECT_EULER_ANSWERS_PATH) as file_handle:
						__a  :		dict[str, str]		     =   json.load(file_handle)
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			ModuleType:
							lowercase__  :	List[str]        =							importlib.util.spec_from_file_location(file_path.name    ,str(SCREAMING_SNAKE_CASE_    )    )
							lowercase__  :	int        =							importlib.util.module_from_spec(SCREAMING_SNAKE_CASE_    )  # type: ignore
							spec.loader.exec_module(SCREAMING_SNAKE_CASE_    )  # type: ignore
							return module
def  snake_case_					(		)      ->			list[pathlib.Path]:
							lowercase__  :	List[str]        =							[]
							for problem_dir_path in PROJECT_EULER_DIR_PATH.iterdir():
														if problem_dir_path.is_file() or problem_dir_path.name.startswith("_"    ):
																					continue
														for file_path in problem_dir_path.iterdir():
																					if file_path.suffix != ".py" or file_path.name.startswith(("_", "test")    ):
																												continue
																					solution_file_paths.append(SCREAMING_SNAKE_CASE_    )
							return solution_file_paths
def  snake_case_					(		)      ->			str:
							with open(os.environ["GITHUB_EVENT_PATH"]    ) as file:
														lowercase__  :	str        =							json.load(SCREAMING_SNAKE_CASE_    )
							return event["pull_request"]["url"] + "/files"
def  snake_case_					(		)      ->			list[pathlib.Path]:
							lowercase__  :	Optional[Any]        =							[]
							lowercase__  :	Dict        =							{
							    "Accept": "application/vnd.github.v3+json",
							    "Authorization": "token " + os.environ["GITHUB_TOKEN"],
							}
							lowercase__  :	Union[str, Any]        =							requests.get(get_files_url()    ,headers=SCREAMING_SNAKE_CASE_    ).json()
							for file in files:
														lowercase__  :	Optional[Any]        =							pathlib.Path.cwd().joinpath(file["filename"]    )
														if (
														    filepath.suffix != ".py"
														    or filepath.name.startswith(("_", "test")    )
														    or not filepath.name.startswith("sol"    )
														):
																					continue
														solution_file_paths.append(SCREAMING_SNAKE_CASE_    )
							return solution_file_paths
def  snake_case_					(		)      ->			list[pathlib.Path]:
							if os.environ.get("CI"    ) and os.environ.get("GITHUB_EVENT_NAME"    ) == "pull_request":
														# Return only if there are any, otherwise default to all solutions
														if filepaths := added_solution_file_path():
																					return filepaths
							return all_solution_file_paths()




@pytest.mark.parametrize(
    "solution_path"    ,collect_solution_file_paths()    ,ids=lambda SCREAMING_SNAKE_CASE_    : F"""{path.parent.name}/{path.name}"""    ,)
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			None:
							lowercase__  :	str        =							solution_path.parent.name[8:].zfill(3    )
							lowercase__  :	str        =							PROBLEM_ANSWERS[problem_number]
							lowercase__  :	Optional[Any]        =							convert_path_to_module(SCREAMING_SNAKE_CASE_    )
							lowercase__  :	str        =							str(solution_module.solution()    )  # type: ignore
							lowercase__  :	List[str]        =							hashlib.shaaaa(answer.encode()    ).hexdigest()
							assert (
							    answer == expected
							), F"""Expected solution to {problem_number} to have hash {expected}, got {answer}"""