




"""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	     =					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     (     _lowerCamelCase							):







							'''simple docstring'''




							def __init__(  self		,     _UpperCAmelCase = 101       ):
											__a	:  Optional[int]         =					length







							def __len__(  self       ):
											return self.length







							def __getitem__(  self		,     _UpperCAmelCase       ):
											return i






		class 			__lowercase     :







							'''simple docstring'''




							def __call__(  self		,     _UpperCAmelCase       ):
											return {"input_ids": torch.tensor(lowercase_       ), "labels": torch.tensor(lowercase_       )}






		class 			__lowercase     (     nn.Module							):







							'''simple docstring'''




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







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




class 			__lowercase     (     _lowerCamelCase							):







					'''simple docstring'''




					@require_torch_neuroncore
					def 	_lowerCamelCase						(  self       ):
									__a	:  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()
									__a	:  List[str]         =					self.get_auto_remove_tmp_dir()
									__a	:  Tuple         =					f"""--output_dir {output_dir}""".split()
									__a	:  Optional[Any]         =					['''torchrun'''] + distributed_args + args
									execute_subprocess_async(lowercase_		,     env=self.get_env()       )
									# successful return here == success - any errors would have caused an error in the sub-call






class 			__lowercase     (     _lowerCamelCase							):







					'''simple docstring'''




					@require_torch_multi_gpu
					def 	_lowerCamelCase						(  self       ):
									__a	:  int         =					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()
									__a	:  Any         =					self.get_auto_remove_tmp_dir()
									__a	:  List[str]         =					f"""--output_dir {output_dir}""".split()
									__a	:  Optional[int]         =					['''torchrun'''] + distributed_args + args
									execute_subprocess_async(lowercase_		,     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	     =					HfArgumentParser((TrainingArguments,))
		A	     =					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	     =					DummyDataset(dataset_length)

				def       __A (     a_  :List[Any])						->						Dict:
								__a	:  List[str]         =					list(range(len(__UpperCAmelCase)))
								__a	:  Optional[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}

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

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

				A	     =					2

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

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






				A	     =					None






"""simple docstring"""






import os
import string
import sys


A	     =					1 << 8

A	     =					{
    '''tab''': ord('''\t'''),
    '''newline''': ord('''\r'''),
    '''esc''': 27,
    '''up''': 65 + ARROW_KEY_FLAG,
    '''down''': 66 + ARROW_KEY_FLAG,
    '''right''': 67 + ARROW_KEY_FLAG,
    '''left''': 68 + ARROW_KEY_FLAG,
    '''mod_int''': 91,
    '''undefined''': sys.maxsize,
    '''interrupt''': 3,
    '''insert''': 50,
    '''delete''': 51,
    '''pg_up''': 53,
    '''pg_down''': 54,
}

A	     =					KEYMAP['''up''']
A	     =					KEYMAP['''left''']

if sys.platform == "win32":
		A	     =					[]
		A	     =					{
		    B'''\xe0H''': KEYMAP['''up'''] - ARROW_KEY_FLAG,
		    B'''\x00H''': KEYMAP['''up'''] - ARROW_KEY_FLAG,
		    B'''\xe0P''': KEYMAP['''down'''] - ARROW_KEY_FLAG,
		    B'''\x00P''': KEYMAP['''down'''] - ARROW_KEY_FLAG,
		    B'''\xe0M''': KEYMAP['''right'''] - ARROW_KEY_FLAG,
		    B'''\x00M''': KEYMAP['''right'''] - ARROW_KEY_FLAG,
		    B'''\xe0K''': KEYMAP['''left'''] - ARROW_KEY_FLAG,
		    B'''\x00K''': KEYMAP['''left'''] - ARROW_KEY_FLAG,
		}

for i in range(10):
		A	     =					ord(str(i))

def       __A (     )						->						Dict:
				if os.name == "nt":
								import msvcrt

								__a	:  Optional[Any]         =					'''mbcs'''
								# Flush the keyboard buffer
								while msvcrt.kbhit():
												msvcrt.getch()
								if len(a_) == 0:
												# Read the keystroke
												__a	:  Optional[Any]         =					msvcrt.getch()

												# If it is a prefix char, get second part
												if ch in (b"\x00", b"\xe0"):
																__a	:  Optional[Any]         =					ch + msvcrt.getch()
																# Translate actual Win chars to bullet char types
																try:
																				__a	:  Union[str, Any]         =					chr(WIN_KEYMAP[cha])
																				WIN_CH_BUFFER.append(chr(KEYMAP['''mod_int''']))
																				WIN_CH_BUFFER.append(a_)
																				if ord(a_) in (
																				    KEYMAP["insert"] - 1 << 9,
																				    KEYMAP["delete"] - 1 << 9,
																				    KEYMAP["pg_up"] - 1 << 9,
																				    KEYMAP["pg_down"] - 1 << 9,
																				):
																								WIN_CH_BUFFER.append(chr(1_26))
																				__a	:  str         =					chr(KEYMAP['''esc'''])
																except KeyError:
																				__a	:  str         =					cha[1]
												else:
																__a	:  Optional[Any]         =					ch.decode(a_)
								else:
												__a	:  Union[str, Any]         =					WIN_CH_BUFFER.pop(0)
				elif os.name == "posix":
								import termios
								import tty

								__a	:  Any         =					sys.stdin.fileno()
								__a	:  List[str]         =					termios.tcgetattr(a_)
								try:
												tty.setraw(a_)
												__a	:  int         =					sys.stdin.read(1)
								finally:
												termios.tcsetattr(a_				,    termios.TCSADRAIN				,    a_)
				return ch






def       __A (     )						->						str:
				__a	:  Any         =					get_raw_chars()
				if ord(a_) in [KEYMAP["interrupt"], KEYMAP["newline"]]:
								return char

				elif ord(a_) == KEYMAP["esc"]:
								__a	:  str         =					get_raw_chars()
								if ord(a_) == KEYMAP["mod_int"]:
												__a	:  List[str]         =					get_raw_chars()
												if ord(a_) >= KEYMAP["arrow_begin"] - ARROW_KEY_FLAG and ord(a_) <= KEYMAP["arrow_end"] - ARROW_KEY_FLAG:
																return chr(ord(a_) + ARROW_KEY_FLAG)
												else:
																return KEYMAP["undefined"]
								else:
												return get_raw_chars()

				else:
								if char in string.printable:
												return char
								else:
												return KEYMAP["undefined"]