
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


lowercase__		:Tuple			  =   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		( SCREAMING_SNAKE_CASE__		):




														def __init__(    self       ,A__ = 1_0_1):
																			lowercase						     =						length




														def __len__(    self):
																			return self.length





														def __getitem__(    self       ,A__):
																			return i







							class     lowercase		:




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







							class     lowercase		( nn.Module		):




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





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







class     lowercase		( SCREAMING_SNAKE_CASE__		):




							@require_torch_neuroncore
							def 			A__				(    self):
												lowercase						     =						f'--nproc_per_node=2\n            --master_port={get_torch_dist_unique_port()}\n            {self.test_file_dir}/test_trainer_distributed.py\n        '.split()
												lowercase						     =						self.get_auto_remove_tmp_dir()
												lowercase						     =						f'--output_dir {output_dir}'.split()
												lowercase						     =						['''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     lowercase		( SCREAMING_SNAKE_CASE__		):




							@require_torch_multi_gpu
							def 			A__				(    self):
												lowercase						     =						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()
												lowercase						     =						self.get_auto_remove_tmp_dir()
												lowercase						     =						f'--output_dir {output_dir}'.split()
												lowercase						     =						['''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

							lowercase__		:List[str]			  =   HfArgumentParser((TrainingArguments,))
							lowercase__		:Any			  =   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]:
														lowercase__		:Tuple			  =   DummyDataset(dataset_length)



														def 	UpperCamelCase   (							lowerCAmelCase__		):
																			'''simple docstring'''
																			lowercase						     =						list(range(len(lowerCAmelCase__		)		)		)
																			lowercase						     =						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}

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

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

														lowercase__		:Union[str, Any]			  =   2

														lowercase__		:Union[str, Any]			  =   trainer.evaluate()
														logger.info(metrics)
														if metrics["eval_success"] is not True:
																					logger.error(metrics)
																					exit(1)

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






import unittest

from transformers import TrOCRConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch

							from transformers.models.trocr.modeling_trocr import TrOCRDecoder, TrOCRForCausalLM







@require_torch
class     lowercase		:




							def __init__(    self       ,A__       ,A__=9_9       ,A__=1_3       ,A__=1_6       ,A__=7       ,A__=True       ,A__=True       ,A__=True       ,A__=False       ,A__=True       ,A__=2       ,A__=3_2       ,A__=4       ,A__=4       ,A__=3_0       ,A__=0       ,A__=1       ,A__=2       ,A__=None       ,):
												lowercase						     =						parent
												lowercase						     =						batch_size
												lowercase						     =						decoder_seq_length
												# For common tests
												lowercase						     =						self.decoder_seq_length
												lowercase						     =						is_training
												lowercase						     =						use_attention_mask
												lowercase						     =						use_labels

												lowercase						     =						vocab_size
												lowercase						     =						d_model
												lowercase						     =						d_model
												lowercase						     =						decoder_layers
												lowercase						     =						decoder_layers
												lowercase						     =						decoder_ffn_dim
												lowercase						     =						decoder_attention_heads
												lowercase						     =						decoder_attention_heads
												lowercase						     =						eos_token_id
												lowercase						     =						bos_token_id
												lowercase						     =						pad_token_id
												lowercase						     =						decoder_start_token_id
												lowercase						     =						use_cache
												lowercase						     =						max_position_embeddings

												lowercase						     =						None
												lowercase						     =						decoder_seq_length
												lowercase						     =						2
												lowercase						     =						1




							def 			A__				(    self):
												lowercase						     =						ids_tensor([self.batch_size, self.decoder_seq_length]       ,self.vocab_size)

												lowercase						     =						None
												if self.use_attention_mask:
																	lowercase						     =						ids_tensor([self.batch_size, self.decoder_seq_length]       ,vocab_size=2)

												lowercase						     =						None
												if self.use_labels:
																	lowercase						     =						ids_tensor([self.batch_size, self.decoder_seq_length]       ,self.vocab_size)

												lowercase						     =						TrOCRConfig(
												    vocab_size=self.vocab_size       ,d_model=self.d_model       ,decoder_layers=self.decoder_layers       ,decoder_ffn_dim=self.decoder_ffn_dim       ,decoder_attention_heads=self.decoder_attention_heads       ,eos_token_id=self.eos_token_id       ,bos_token_id=self.bos_token_id       ,use_cache=self.use_cache       ,pad_token_id=self.pad_token_id       ,decoder_start_token_id=self.decoder_start_token_id       ,max_position_embeddings=self.max_position_embeddings       ,)

												return (config, input_ids, attention_mask, lm_labels)




							def 			A__				(    self       ,A__       ,A__       ,A__       ,A__       ,):
												lowercase						     =						True
												lowercase						     =						TrOCRDecoder(config=A__).to(A__).eval()
												lowercase						     =						input_ids[:2]

												input_ids[input_ids == 0] += 1
												# first forward pass
												lowercase						     =						model(A__       ,use_cache=A__)
												lowercase						     =						model(A__)
												lowercase						     =						model(A__       ,use_cache=A__)

												self.parent.assertTrue(len(A__) == len(A__))
												self.parent.assertTrue(len(A__) == len(A__) + 1)

												lowercase						     =						outputs['''past_key_values''']

												# create hypothetical next token and extent to next_input_ids
												lowercase						     =						ids_tensor((2, 1)       ,config.vocab_size - 1) + 1

												# append to next input_ids and
												lowercase						     =						torch.cat([input_ids, next_tokens]       ,dim=-1)

												lowercase						     =						model(A__)['''last_hidden_state''']
												lowercase						     =						model(A__       ,past_key_values=A__)['''last_hidden_state''']

												# select random slice
												lowercase						     =						ids_tensor((1,)       ,output_from_past.shape[-1]).item()
												lowercase						     =						output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
												lowercase						     =						output_from_past[:, 0, random_slice_idx].detach()

												# test that outputs are equal for slice
												assert torch.allclose(A__       ,A__       ,atol=1E-3)





							def 			A__				(    self):
												lowercase						     =						self.prepare_config_and_inputs()
												lowercase     ,						lowercase     ,						lowercase     ,						lowercase						     =						config_and_inputs

												lowercase						     =						{'''input_ids''': input_ids, '''attention_mask''': attention_mask}
												return config, inputs_dict







@require_torch
class     lowercase		( SCREAMING_SNAKE_CASE__	,  SCREAMING_SNAKE_CASE__	,  SCREAMING_SNAKE_CASE__	,  unittest.TestCase		):
							lowercase_	:	Any	 =(TrOCRDecoder, TrOCRForCausalLM) if is_torch_available() else ()
							lowercase_	:	Dict	 =(TrOCRForCausalLM,) if is_torch_available() else ()
							lowercase_	:	int	 ={'''text-generation''': TrOCRForCausalLM} if is_torch_available() else {}
							lowercase_	:	List[Any]	 =True
							lowercase_	:	int	 =False




							def 			A__				(    self):
												lowercase						     =						TrOCRStandaloneDecoderModelTester(self       ,is_training=A__)
												lowercase						     =						ConfigTester(self       ,config_class=A__)




							def 			A__				(    self):
												pass




							def 			A__				(    self):
												pass




							def 			A__				(    self):
												pass




							def 			A__				(    self):
												self.config_tester.run_common_tests()




							def 			A__				(    self):
												lowercase						     =						self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_decoder_model_past(*A__)




							def 			A__				(    self):
												return





							@unittest.skip('''The model doesn\'t support left padding''')  # and it's not used enough to be worth fixing :)
							def 			A__				(    self):
												pass




