
import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


SCREAMING_SNAKE_CASE__     :    List[str]      							=					namedtuple("ExampleNamedTuple", "a b c")

class      lowerCAmelCase__  (			unittest.TestCase    ):






		def 				__A (	self	:							Union[str, Any]						)  ->							List[str]:
						__lowerCamelCase					      =			torch.randn(5							,    2						)
						__lowerCamelCase					      =			torch.device('''cuda'''						) if torch.cuda.is_available() else torch.device('''cpu'''						)

						__lowerCamelCase					      =			send_to_device(lowerCamelCase_							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta.cpu()							,    lowerCamelCase_						)						)

						__lowerCamelCase					      =			send_to_device((tensor, [tensor, tensor], 1)							,    lowerCamelCase_						)
						self.assertIsInstance(lowerCamelCase_							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta[0].cpu()							,    lowerCamelCase_						)						)
						self.assertIsInstance(resulta[1]							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta[1][0].cpu()							,    lowerCamelCase_						)						)
						self.assertTrue(torch.equal(resulta[1][1].cpu()							,    lowerCamelCase_						)						)
						self.assertEqual(resulta[2]							,    1						)

						__lowerCamelCase					      =			send_to_device({'''a''': tensor, '''b''': [tensor, tensor], '''c''': 1}							,    lowerCamelCase_						)
						self.assertIsInstance(lowerCamelCase_							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta['''a'''].cpu()							,    lowerCamelCase_						)						)
						self.assertIsInstance(resulta['''b''']							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta['''b'''][0].cpu()							,    lowerCamelCase_						)						)
						self.assertTrue(torch.equal(resulta['''b'''][1].cpu()							,    lowerCamelCase_						)						)
						self.assertEqual(resulta['''c''']							,    1						)

						__lowerCamelCase					      =			send_to_device(ExampleNamedTuple(a=lowerCamelCase_							,    b=[tensor, tensor]							,    c=1						)							,    lowerCamelCase_						)
						self.assertIsInstance(lowerCamelCase_							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta.a.cpu()							,    lowerCamelCase_						)						)
						self.assertIsInstance(resulta.b							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta.b[0].cpu()							,    lowerCamelCase_						)						)
						self.assertTrue(torch.equal(resulta.b[1].cpu()							,    lowerCamelCase_						)						)
						self.assertEqual(resulta.c							,    1						)

						__lowerCamelCase					      =			send_to_device(UserDict({'''a''': tensor, '''b''': [tensor, tensor], '''c''': 1}						)							,    lowerCamelCase_						)
						self.assertIsInstance(lowerCamelCase_							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta['''a'''].cpu()							,    lowerCamelCase_						)						)
						self.assertIsInstance(resulta['''b''']							,    lowerCamelCase_						)
						self.assertTrue(torch.equal(resulta['''b'''][0].cpu()							,    lowerCamelCase_						)						)
						self.assertTrue(torch.equal(resulta['''b'''][1].cpu()							,    lowerCamelCase_						)						)
						self.assertEqual(resulta['''c''']							,    1						)






		def 				__A (	self	:							List[str]						)  ->							Dict:

						with self.assertRaises(lowerCamelCase_						) as cm:
										__lowerCamelCase					      =			recursively_apply(torch.tensor							,    (torch.tensor(1						), 1)							,    error_on_other_type=lowerCamelCase_						)
						self.assertEqual(
						    str(cm.exception						)							,    '''Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'''							,    )






		def 				__A (	self	:							Dict						)  ->							List[str]:
						__lowerCamelCase					      =			torch.tensor([1, 2, 3, 4, 5]						)
						self.assertEqual(listify(lowerCamelCase_						)							,    [1, 2, 3, 4, 5]						)

						__lowerCamelCase					      =			torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]						)
						self.assertEqual(listify(lowerCamelCase_						)							,    [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]						)

						__lowerCamelCase					      =			torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]						)
						self.assertEqual(
						    listify(lowerCamelCase_						)							,    [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]						)






		def 				__A (	self	:							int						)  ->							Dict:

						with patch_environment(aa=1							,    BB=2						):
										self.assertEqual(os.environ.get('''AA'''						)							,    '''1'''						)
										self.assertEqual(os.environ.get('''BB'''						)							,    '''2'''						)

						self.assertNotIn('''AA'''							,    os.environ						)
						self.assertNotIn('''BB'''							,    os.environ						)






		def 				__A (	self	:							Optional[int]						)  ->							str:
						__lowerCamelCase					      =			RegressionModel()
						__lowerCamelCase					      =			model.forward
						__lowerCamelCase					      =			convert_outputs_to_fpaa(model.forward						)
						__lowerCamelCase					      =			extract_model_from_parallel(lowerCamelCase_							,    keep_fpaa_wrapper=lowerCamelCase_						)
						__lowerCamelCase					      =			pickle.dumps(lowerCamelCase_						)






		@require_cuda
		def 				__A (	self	:							List[Any]						)  ->							Dict:
						__lowerCamelCase					      =			RegressionModel()
						__lowerCamelCase					      =			model.forward
						__lowerCamelCase					      =			torch.cuda.amp.autocast(dtype=torch.floataa						)(model.forward						)
						__lowerCamelCase					      =			convert_outputs_to_fpaa(model.forward						)
						__lowerCamelCase					      =			extract_model_from_parallel(lowerCamelCase_							,    keep_fpaa_wrapper=lowerCamelCase_						)
						__lowerCamelCase					      =			pickle.dumps(lowerCamelCase_						)






		@require_cuda
		@require_torch_min_version(version='''2.0'''						)
		def 				__A (	self	:							str						)  ->							Optional[int]:
						__lowerCamelCase					      =			RegressionModel()
						__lowerCamelCase					      =			model.forward
						__lowerCamelCase					      =			torch.cuda.amp.autocast(dtype=torch.floataa						)(model.forward						)
						__lowerCamelCase					      =			convert_outputs_to_fpaa(model.forward						)
						__lowerCamelCase					      =			torch.compile(model.forward							,    backend='''inductor'''						)
						__lowerCamelCase					      =			torch.randn(4							,    10						).cuda()
						__lowerCamelCase					      =			model(lowerCamelCase_						)






		def 				__A (	self	:							List[Any]						)  ->							Any:
						__lowerCamelCase					      =			RegressionModel()
						# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
						__lowerCamelCase					      =			torch.nn.parallel.DataParallel(lowerCamelCase_						)
						__lowerCamelCase					      =			extract_model_from_parallel(lowerCamelCase_						)

						self.assertEqual(lowerCamelCase_							,    lowerCamelCase_						)






		@require_torch_min_version(version='''2.0'''						)
		def 				__A (	self	:							Tuple						)  ->							Dict:
						__lowerCamelCase					      =			RegressionModel()
						__lowerCamelCase					      =			torch.compile(lowerCamelCase_						)

						# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
						__lowerCamelCase					      =			torch.nn.parallel.DataParallel(lowerCamelCase_						)
						__lowerCamelCase					      =			torch.compile(lowerCamelCase_						)
						__lowerCamelCase					      =			extract_model_from_parallel(lowerCamelCase_						)

						self.assertEqual(compiled_model._orig_mod							,    compiled_model_unwrapped._orig_mod						)







		def 				__A (	self	:							Optional[Any]						)  ->							Tuple:
						self.assertEqual(find_device([1, '''a''', torch.tensor([1, 2, 3]						)]						)							,    torch.device('''cpu'''						)						)
						self.assertEqual(find_device({'''a''': 1, '''b''': torch.tensor([1, 2, 3]						)}						)							,    torch.device('''cpu'''						)						)
						self.assertIsNone(find_device([1, '''a''']						)						)





from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


SCREAMING_SNAKE_CASE__     :    int      							=					logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__     :    List[str]      							=					{
    "facebook/m2m100_418M": "https://huggingface.co/facebook/m2m100_418M/resolve/main/config.json",
    # See all M2M100 models at https://huggingface.co/models?filter=m2m_100
}

class      lowerCAmelCase__  (			__lowercase    ):
		a__ :		Union[str, Any]   						=	"""m2m_100"""
		a__ :		Optional[int]   						=	["""past_key_values"""]
		a__ :		List[Any]   						=	{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}






		def __init__(	self	:							int							,    SCREAMING_SNAKE_CASE__	:							Tuple=12_81_12							,    SCREAMING_SNAKE_CASE__	:							Any=10_24							,    SCREAMING_SNAKE_CASE__	:							int=12							,    SCREAMING_SNAKE_CASE__	:							List[str]=40_96							,    SCREAMING_SNAKE_CASE__	:							Optional[int]=16							,    SCREAMING_SNAKE_CASE__	:							Optional[int]=12							,    SCREAMING_SNAKE_CASE__	:							int=40_96							,    SCREAMING_SNAKE_CASE__	:							Optional[Any]=16							,    SCREAMING_SNAKE_CASE__	:							Dict=0.05							,    SCREAMING_SNAKE_CASE__	:							List[Any]=0.05							,    SCREAMING_SNAKE_CASE__	:							Dict=True							,    SCREAMING_SNAKE_CASE__	:							Tuple=True							,    SCREAMING_SNAKE_CASE__	:							Optional[Any]="relu"							,    SCREAMING_SNAKE_CASE__	:							Dict=10_24							,    SCREAMING_SNAKE_CASE__	:							Optional[int]=0.1							,    SCREAMING_SNAKE_CASE__	:							Tuple=0.1							,    SCREAMING_SNAKE_CASE__	:							str=0.0							,    SCREAMING_SNAKE_CASE__	:							Any=0.02							,    SCREAMING_SNAKE_CASE__	:							int=2							,    SCREAMING_SNAKE_CASE__	:							List[Any]=True							,    SCREAMING_SNAKE_CASE__	:							Any=1							,    SCREAMING_SNAKE_CASE__	:							Union[str, Any]=0							,    SCREAMING_SNAKE_CASE__	:							str=2							,    **SCREAMING_SNAKE_CASE__	:							Tuple							,    )  ->							Tuple:
						__lowerCamelCase					      =			vocab_size
						__lowerCamelCase					      =			max_position_embeddings
						__lowerCamelCase					      =			d_model
						__lowerCamelCase					      =			encoder_ffn_dim
						__lowerCamelCase					      =			encoder_layers
						__lowerCamelCase					      =			encoder_attention_heads
						__lowerCamelCase					      =			decoder_ffn_dim
						__lowerCamelCase					      =			decoder_layers
						__lowerCamelCase					      =			decoder_attention_heads
						__lowerCamelCase					      =			dropout
						__lowerCamelCase					      =			attention_dropout
						__lowerCamelCase					      =			activation_dropout
						__lowerCamelCase					      =			activation_function
						__lowerCamelCase					      =			init_std
						__lowerCamelCase					      =			encoder_layerdrop
						__lowerCamelCase					      =			decoder_layerdrop
						__lowerCamelCase					      =			use_cache
						__lowerCamelCase					      =			encoder_layers
						__lowerCamelCase					      =			scale_embedding  # scale factor will be sqrt(d_model) if True

						super().__init__(
						    pad_token_id=SCREAMING_SNAKE_CASE__							,    bos_token_id=SCREAMING_SNAKE_CASE__							,    eos_token_id=SCREAMING_SNAKE_CASE__							,    is_encoder_decoder=SCREAMING_SNAKE_CASE__							,    decoder_start_token_id=SCREAMING_SNAKE_CASE__							,    **SCREAMING_SNAKE_CASE__							,    )





class      lowerCAmelCase__  (			__lowercase    ):






		@property
		def 				__A (	self	:							List[Any]						)  ->							Mapping[str, Mapping[int, str]]:
						__lowerCamelCase					      =			OrderedDict(
						    [
						        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
						        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
						    ]						)

						if self.use_past:
										__lowerCamelCase					      =			{0: '''batch'''}
										__lowerCamelCase					      =			{0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
						else:
										__lowerCamelCase					      =			{0: '''batch''', 1: '''decoder_sequence'''}
										__lowerCamelCase					      =			{0: '''batch''', 1: '''decoder_sequence'''}

						if self.use_past:
										self.fill_with_past_key_values_(SCREAMING_SNAKE_CASE__							,    direction='''inputs'''						)
						return common_inputs






		def 				__A (	self	:							Any							,    SCREAMING_SNAKE_CASE__	:							PreTrainedTokenizer							,    SCREAMING_SNAKE_CASE__	:							int = -1							,    SCREAMING_SNAKE_CASE__	:							int = -1							,    SCREAMING_SNAKE_CASE__	:							bool = False							,    SCREAMING_SNAKE_CASE__	:							Optional[TensorType] = None							,    )  ->							Mapping[str, Any]:
						# Copied from OnnxConfig.generate_dummy_inputs
						# Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
						# If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
						__lowerCamelCase					      =			compute_effective_axis_dimension(
						    SCREAMING_SNAKE_CASE__							,    fixed_dimension=OnnxConfig.default_fixed_batch							,    num_token_to_add=0						)

						# If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
						__lowerCamelCase					      =			tokenizer.num_special_tokens_to_add(SCREAMING_SNAKE_CASE__						)
						__lowerCamelCase					      =			compute_effective_axis_dimension(
						    SCREAMING_SNAKE_CASE__							,    fixed_dimension=OnnxConfig.default_fixed_sequence							,    num_token_to_add=SCREAMING_SNAKE_CASE__						)

						# Generate dummy inputs according to compute batch and sequence
						__lowerCamelCase					      =			[''' '''.join([tokenizer.unk_token]						) * seq_length] * batch_size
						__lowerCamelCase					      =			dict(tokenizer(SCREAMING_SNAKE_CASE__							,    return_tensors=SCREAMING_SNAKE_CASE__						)						)
						return common_inputs






		def 				__A (	self	:							Union[str, Any]							,    SCREAMING_SNAKE_CASE__	:							PreTrainedTokenizer							,    SCREAMING_SNAKE_CASE__	:							int = -1							,    SCREAMING_SNAKE_CASE__	:							int = -1							,    SCREAMING_SNAKE_CASE__	:							bool = False							,    SCREAMING_SNAKE_CASE__	:							Optional[TensorType] = None							,    )  ->							Mapping[str, Any]:
						__lowerCamelCase					      =			self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
						    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__						)

						# Generate decoder inputs
						__lowerCamelCase					      =			seq_length if not self.use_past else 1
						__lowerCamelCase					      =			self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
						    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__						)
						__lowerCamelCase					      =			{f'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
						__lowerCamelCase					      =			dict(**SCREAMING_SNAKE_CASE__							,    **SCREAMING_SNAKE_CASE__						)

						if self.use_past:
										if not is_torch_available():
														raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''						)
										else:
														import torch
										__lowerCamelCase      ,					__lowerCamelCase					      =			common_inputs['''input_ids'''].shape
										__lowerCamelCase					      =			common_inputs['''decoder_input_ids'''].shape[1]
										__lowerCamelCase      ,					__lowerCamelCase					      =			self.num_attention_heads
										__lowerCamelCase					      =			(
										    batch,
										    num_encoder_attention_heads,
										    encoder_seq_length,
										    self._config.hidden_size // num_encoder_attention_heads,
										)
										__lowerCamelCase					      =			decoder_seq_length + 3
										__lowerCamelCase					      =			(
										    batch,
										    num_decoder_attention_heads,
										    decoder_past_length,
										    self._config.hidden_size // num_decoder_attention_heads,
										)

										__lowerCamelCase					      =			torch.cat(
										    [common_inputs['''decoder_attention_mask'''], torch.ones(SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__						)]							,    dim=1						)

										__lowerCamelCase					      =			[]
										# If the number of encoder and decoder layers are present in the model configuration, both are considered
										__lowerCamelCase      ,					__lowerCamelCase					      =			self.num_layers
										__lowerCamelCase					      =			min(SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__						)
										__lowerCamelCase					      =			max(SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__						) - min_num_layers
										__lowerCamelCase					      =			'''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

										for _ in range(SCREAMING_SNAKE_CASE__						):
														common_inputs["past_key_values"].append(
														    (
														        torch.zeros(SCREAMING_SNAKE_CASE__						),
														        torch.zeros(SCREAMING_SNAKE_CASE__						),
														        torch.zeros(SCREAMING_SNAKE_CASE__						),
														        torch.zeros(SCREAMING_SNAKE_CASE__						),
														    )						)
										# TODO: test this.
										__lowerCamelCase					      =			encoder_shape if remaining_side_name == '''encoder''' else decoder_shape
										for _ in range(SCREAMING_SNAKE_CASE__							,    SCREAMING_SNAKE_CASE__						):
														common_inputs["past_key_values"].append((torch.zeros(SCREAMING_SNAKE_CASE__						), torch.zeros(SCREAMING_SNAKE_CASE__						))						)
						return common_inputs







		a__ :		Dict   						=	_generate_dummy_inputs_for_default_and_seqaseq_lm



