




import unittest

from transformers import LayoutLMConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

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 import (
	    LayoutLMForMaskedLM,
	    LayoutLMForQuestionAnswering,
	    LayoutLMForSequenceClassification,
	    LayoutLMForTokenClassification,
	    LayoutLMModel,
	)


class       SCREAMING_SNAKE_CASE_							:



	"""simple docstring"""


	def __init__(				self		:	int				,    lowerCAmelCase		:	Tuple				,    lowerCAmelCase		:	Optional[int]=13				,    lowerCAmelCase		:	int=7				,    lowerCAmelCase		:	List[str]=True				,    lowerCAmelCase		:	List[Any]=True				,    lowerCAmelCase		:	Optional[int]=True				,    lowerCAmelCase		:	Any=True				,    lowerCAmelCase		:	Tuple=99				,    lowerCAmelCase		:	Dict=32				,    lowerCAmelCase		:	Tuple=5				,    lowerCAmelCase		:	Union[str, Any]=4				,    lowerCAmelCase		:	str=37				,    lowerCAmelCase		:	Optional[int]="gelu"				,    lowerCAmelCase		:	List[str]=0.1				,    lowerCAmelCase		:	int=0.1				,    lowerCAmelCase		:	Optional[Any]=512				,    lowerCAmelCase		:	str=16				,    lowerCAmelCase		:	int=2				,    lowerCAmelCase		:	List[Any]=0.02				,    lowerCAmelCase		:	List[str]=3				,    lowerCAmelCase		:	Tuple=4				,    lowerCAmelCase		:	str=None				,    lowerCAmelCase		:	Tuple=1000				,    )  ->					Tuple:







						"""simple docstring"""







						__UpperCamelCase				:  Tuple								=			parent
						__UpperCamelCase				:  Optional[int]								=			batch_size
						__UpperCamelCase				:  Dict								=			seq_length
						__UpperCamelCase				:  List[Any]								=			is_training
						__UpperCamelCase				:  Union[str, Any]								=			use_input_mask
						__UpperCamelCase				:  List[str]								=			use_token_type_ids
						__UpperCamelCase				:  int								=			use_labels
						__UpperCamelCase				:  Tuple								=			vocab_size
						__UpperCamelCase				:  List[Any]								=			hidden_size
						__UpperCamelCase				:  str								=			num_hidden_layers
						__UpperCamelCase				:  Union[str, Any]								=			num_attention_heads
						__UpperCamelCase				:  int								=			intermediate_size
						__UpperCamelCase				:  Tuple								=			hidden_act
						__UpperCamelCase				:  List[Any]								=			hidden_dropout_prob
						__UpperCamelCase				:  Union[str, Any]								=			attention_probs_dropout_prob
						__UpperCamelCase				:  List[Any]								=			max_position_embeddings
						__UpperCamelCase				:  List[Any]								=			type_vocab_size
						__UpperCamelCase				:  Any								=			type_sequence_label_size
						__UpperCamelCase				:  Any								=			initializer_range
						__UpperCamelCase				:  int								=			num_labels
						__UpperCamelCase				:  Optional[Any]								=			num_choices
						__UpperCamelCase				:  Dict								=			scope
						__UpperCamelCase				:  str								=			range_bbox






	def 				lowerCamelCase__    (				self		:	Dict  )  ->					str:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[Any]								=			ids_tensor([self.batch_size, self.seq_length]				,    self.vocab_size  )

						__UpperCamelCase				:  int								=			ids_tensor([self.batch_size, self.seq_length, 4]				,    self.range_bbox  )
						# Ensure that bbox is legal
						for i in range(bbox.shape[0]  ):
											for j in range(bbox.shape[1]  ):
																if bbox[i, j, 3] < bbox[i, j, 1]:
																					__UpperCamelCase				:  List[Any]								=			bbox[i, j, 3]
																					__UpperCamelCase				:  List[Any]								=			bbox[i, j, 1]
																					__UpperCamelCase				:  int								=			t
																if bbox[i, j, 2] < bbox[i, j, 0]:
																					__UpperCamelCase				:  List[str]								=			bbox[i, j, 2]
																					__UpperCamelCase				:  str								=			bbox[i, j, 0]
																					__UpperCamelCase				:  List[str]								=			t

						__UpperCamelCase				:  List[str]								=			None
						if self.use_input_mask:
											__UpperCamelCase				:  str								=			ids_tensor([self.batch_size, self.seq_length]				,    vocab_size=2  )

						__UpperCamelCase				:  Optional[Any]								=			None
						if self.use_token_type_ids:
											__UpperCamelCase				:  int								=			ids_tensor([self.batch_size, self.seq_length]				,    self.type_vocab_size  )

						__UpperCamelCase				:  int								=			None
						__UpperCamelCase				:  Tuple								=			None
						__UpperCamelCase				:  int								=			None
						if self.use_labels:
											__UpperCamelCase				:  int								=			ids_tensor([self.batch_size]				,    self.type_sequence_label_size  )
											__UpperCamelCase				:  int								=			ids_tensor([self.batch_size, self.seq_length]				,    self.num_labels  )
											__UpperCamelCase				:  Tuple								=			ids_tensor([self.batch_size]				,    self.num_choices  )

						__UpperCamelCase				:  Dict								=			self.get_config()

						return config, input_ids, bbox, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels






	def 				lowerCamelCase__    (				self		:	Union[str, Any]  )  ->					Tuple:







						"""simple docstring"""







						return LayoutLMConfig(
						    vocab_size=self.vocab_size				,    hidden_size=self.hidden_size				,    num_hidden_layers=self.num_hidden_layers				,    num_attention_heads=self.num_attention_heads				,    intermediate_size=self.intermediate_size				,    hidden_act=self.hidden_act				,    hidden_dropout_prob=self.hidden_dropout_prob				,    attention_probs_dropout_prob=self.attention_probs_dropout_prob				,    max_position_embeddings=self.max_position_embeddings				,    type_vocab_size=self.type_vocab_size				,    initializer_range=self.initializer_range				,    )






	def 				lowerCamelCase__    (				self		:	int				,    lowerCAmelCase		:	Dict				,    lowerCAmelCase		:	Optional[Any]				,    lowerCAmelCase		:	List[str]				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	Optional[int]				,    lowerCAmelCase		:	Dict  )  ->					Tuple:







						"""simple docstring"""







						__UpperCamelCase				:  List[str]								=			LayoutLMModel(config=lowerCAmelCase  )
						model.to(lowerCAmelCase  )
						model.eval()
						__UpperCamelCase				:  Any								=			model(lowerCAmelCase				,    lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase  )
						__UpperCamelCase				:  Optional[int]								=			model(lowerCAmelCase				,    lowerCAmelCase				,    token_type_ids=lowerCAmelCase  )
						__UpperCamelCase				:  Union[str, Any]								=			model(lowerCAmelCase				,    lowerCAmelCase  )
						self.parent.assertEqual(result.last_hidden_state.shape				,    (self.batch_size, self.seq_length, self.hidden_size)  )
						self.parent.assertEqual(result.pooler_output.shape				,    (self.batch_size, self.hidden_size)  )






	def 				lowerCamelCase__    (				self		:	int				,    lowerCAmelCase		:	Union[str, Any]				,    lowerCAmelCase		:	List[Any]				,    lowerCAmelCase		:	str				,    lowerCAmelCase		:	Optional[Any]				,    lowerCAmelCase		:	List[Any]				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	str				,    lowerCAmelCase		:	List[str]  )  ->					List[str]:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[int]								=			LayoutLMForMaskedLM(config=lowerCAmelCase  )
						model.to(lowerCAmelCase  )
						model.eval()
						__UpperCamelCase				:  int								=			model(lowerCAmelCase				,    lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase				,    labels=lowerCAmelCase  )
						self.parent.assertEqual(result.logits.shape				,    (self.batch_size, self.seq_length, self.vocab_size)  )






	def 				lowerCamelCase__    (				self		:	int				,    lowerCAmelCase		:	Optional[Any]				,    lowerCAmelCase		:	Optional[int]				,    lowerCAmelCase		:	Any				,    lowerCAmelCase		:	Union[str, Any]				,    lowerCAmelCase		:	Optional[int]				,    lowerCAmelCase		:	Dict				,    lowerCAmelCase		:	Any				,    lowerCAmelCase		:	List[Any]  )  ->					List[str]:







						"""simple docstring"""







						__UpperCamelCase				:  List[str]								=			self.num_labels
						__UpperCamelCase				:  str								=			LayoutLMForSequenceClassification(lowerCAmelCase  )
						model.to(lowerCAmelCase  )
						model.eval()
						__UpperCamelCase				:  List[Any]								=			model(
						    lowerCAmelCase				,    lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase				,    labels=lowerCAmelCase  )
						self.parent.assertEqual(result.logits.shape				,    (self.batch_size, self.num_labels)  )






	def 				lowerCamelCase__    (				self		:	str				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	List[Any]				,    lowerCAmelCase		:	List[str]				,    lowerCAmelCase		:	List[Any]				,    lowerCAmelCase		:	List[str]				,    lowerCAmelCase		:	List[str]				,    lowerCAmelCase		:	Union[str, Any]				,    lowerCAmelCase		:	List[Any]  )  ->					int:







						"""simple docstring"""







						__UpperCamelCase				:  Union[str, Any]								=			self.num_labels
						__UpperCamelCase				:  Optional[Any]								=			LayoutLMForTokenClassification(config=lowerCAmelCase  )
						model.to(lowerCAmelCase  )
						model.eval()
						__UpperCamelCase				:  Tuple								=			model(lowerCAmelCase				,    lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase				,    labels=lowerCAmelCase  )
						self.parent.assertEqual(result.logits.shape				,    (self.batch_size, self.seq_length, self.num_labels)  )






	def 				lowerCamelCase__    (				self		:	List[Any]				,    lowerCAmelCase		:	Any				,    lowerCAmelCase		:	List[str]				,    lowerCAmelCase		:	List[Any]				,    lowerCAmelCase		:	List[str]				,    lowerCAmelCase		:	Optional[Any]				,    lowerCAmelCase		:	Union[str, Any]				,    lowerCAmelCase		:	Optional[int]				,    lowerCAmelCase		:	Union[str, Any]  )  ->					List[str]:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[Any]								=			LayoutLMForQuestionAnswering(config=lowerCAmelCase  )
						model.to(lowerCAmelCase  )
						model.eval()
						__UpperCamelCase				:  Tuple								=			model(
						    lowerCAmelCase				,    bbox=lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase				,    start_positions=lowerCAmelCase				,    end_positions=lowerCAmelCase				,    )
						self.parent.assertEqual(result.start_logits.shape				,    (self.batch_size, self.seq_length)  )
						self.parent.assertEqual(result.end_logits.shape				,    (self.batch_size, self.seq_length)  )






	def 				lowerCamelCase__    (				self		:	Optional[Any]  )  ->					Any:







						"""simple docstring"""







						__UpperCamelCase				:  str								=			self.prepare_config_and_inputs()
						(
						    (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, (
						    __UpperCamelCase
						)			, 
						)				:  str								=			config_and_inputs
						__UpperCamelCase				:  Optional[Any]								=			{
						    """input_ids""": input_ids,
						    """bbox""": bbox,
						    """token_type_ids""": token_type_ids,
						    """attention_mask""": input_mask,
						}
						return config, inputs_dict






@require_torch
class       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase  ,     _UpperCamelCase  ,     unittest.TestCase      ):



	"""simple docstring"""


	__magic_name__      :					str			  =		(
	    (
	        LayoutLMModel,
	        LayoutLMForMaskedLM,
	        LayoutLMForSequenceClassification,
	        LayoutLMForTokenClassification,
	        LayoutLMForQuestionAnswering,
	    )
	    if is_torch_available()
	    else None
	)
	__magic_name__      :					List[Any]			  =		(
	    {
	        'document-question-answering': LayoutLMForQuestionAnswering,
	        'feature-extraction': LayoutLMModel,
	        'fill-mask': LayoutLMForMaskedLM,
	        'text-classification': LayoutLMForSequenceClassification,
	        'token-classification': LayoutLMForTokenClassification,
	        'zero-shot': LayoutLMForSequenceClassification,
	    }
	    if is_torch_available()
	    else {}
	)
	__magic_name__      :					List[str]			  =		True






	def 				lowerCamelCase__    (				self		:	Optional[int]  )  ->					Any:







						"""simple docstring"""







						__UpperCamelCase				:  Tuple								=			LayoutLMModelTester(self  )
						__UpperCamelCase				:  Optional[int]								=			ConfigTester(self				,    config_class=lowerCAmelCase				,    hidden_size=37  )






	def 				lowerCamelCase__    (				self		:	str  )  ->					Union[str, Any]:







						"""simple docstring"""







						self.config_tester.run_common_tests()






	def 				lowerCamelCase__    (				self		:	List[Any]  )  ->					List[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  List[Any]								=			self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*lowerCAmelCase  )






	def 				lowerCamelCase__    (				self		:	str  )  ->					List[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  str								=			self.model_tester.prepare_config_and_inputs()
						for type in ["absolute", "relative_key", "relative_key_query"]:
											__UpperCamelCase				:  Optional[Any]								=			type
											self.model_tester.create_and_check_model(*lowerCAmelCase  )






	def 				lowerCamelCase__    (				self		:	Optional[Any]  )  ->					List[str]:







						"""simple docstring"""







						__UpperCamelCase				:  Dict								=			self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_masked_lm(*lowerCAmelCase  )






	def 				lowerCamelCase__    (				self		:	List[str]  )  ->					List[str]:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[Any]								=			self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_sequence_classification(*lowerCAmelCase  )






	def 				lowerCamelCase__    (				self		:	Optional[int]  )  ->					List[str]:







						"""simple docstring"""







						__UpperCamelCase				:  int								=			self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_token_classification(*lowerCAmelCase  )






	def 				lowerCamelCase__    (				self		:	Tuple  )  ->					List[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[int]								=			self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_question_answering(*lowerCAmelCase  )






	@unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""  )
	def 				lowerCamelCase__    (				self		:	List[Any]  )  ->					List[str]:







						"""simple docstring"""







						pass






def 							A__   ()    ->	Union[str, Any]:
					# Here we prepare a batch of 2 sequences to test a LayoutLM forward pass on:
					# fmt: off
					__UpperCamelCase				:  Optional[Any]								=			torch.tensor([[1_01,10_19,10_14,10_16,10_37,1_28_49,47_47,10_04,1_42_46,22_78,54_39,45_24,50_02,29_30,21_93,29_30,43_41,32_08,10_05,10_55,21_71,28_48,1_13_00,35_31,1_02],[1_01,40_70,40_34,70_20,10_24,30_58,10_15,10_13,28_61,10_13,60_70,1_92_74,27_72,62_05,2_78_14,1_61_47,1_61_47,43_43,20_47,1_02_83,1_09_69,1_43_89,10_12,23_38,1_02]]		,  device=snake_case							)  # noqa: E231
					__UpperCamelCase				:  str								=			torch.tensor([[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],]		,  device=snake_case							)  # noqa: E231
					__UpperCamelCase				:  int								=			torch.tensor([[[0,0,0,0],[4_23,2_37,4_40,2_51],[4_27,2_72,4_41,2_87],[4_19,1_15,4_37,1_29],[9_61,8_85,9_92,9_12],[2_56,38,3_30,58],[2_56,38,3_30,58],[3_36,42,3_53,57],[3_60,39,4_01,56],[3_60,39,4_01,56],[4_11,39,4_71,59],[4_79,41,5_28,59],[5_33,39,6_30,60],[67,1_13,1_34,1_31],[1_41,1_15,2_09,1_32],[68,1_49,1_33,1_66],[1_41,1_49,1_87,1_64],[1_95,1_48,2_87,1_65],[1_95,1_48,2_87,1_65],[1_95,1_48,2_87,1_65],[2_95,1_48,3_49,1_65],[4_41,1_49,4_92,1_66],[4_97,1_49,5_46,1_64],[64,2_01,1_25,2_18],[10_00,10_00,10_00,10_00]],[[0,0,0,0],[6_62,1_50,7_54,1_66],[6_65,1_99,7_42,2_11],[5_19,2_13,5_54,2_28],[5_19,2_13,5_54,2_28],[1_34,4_33,1_87,4_54],[1_30,4_67,2_04,4_80],[1_30,4_67,2_04,4_80],[1_30,4_67,2_04,4_80],[1_30,4_67,2_04,4_80],[1_30,4_67,2_04,4_80],[3_14,4_69,3_76,4_82],[5_04,6_84,5_82,7_06],[9_41,8_25,9_73,9_00],[9_41,8_25,9_73,9_00],[9_41,8_25,9_73,9_00],[9_41,8_25,9_73,9_00],[6_10,7_49,6_52,7_65],[1_30,6_59,1_68,6_72],[1_76,6_57,2_37,6_72],[2_38,6_57,3_12,6_72],[4_43,6_53,6_28,6_72],[4_43,6_53,6_28,6_72],[7_16,3_01,8_25,3_17],[10_00,10_00,10_00,10_00]]]		,  device=snake_case							)  # noqa: E231
					__UpperCamelCase				:  Dict								=			torch.tensor([[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]]		,  device=snake_case							)  # noqa: E231
					# these are sequence labels (i.e. at the token level)
					__UpperCamelCase				:  str								=			torch.tensor([[-1_00,10,10,10,9,1,-1_00,7,7,-1_00,7,7,4,2,5,2,8,8,-1_00,-1_00,5,0,3,2,-1_00],[-1_00,12,12,12,-1_00,12,10,-1_00,-1_00,-1_00,-1_00,10,12,9,-1_00,-1_00,-1_00,10,10,10,9,12,-1_00,10,-1_00]]		,  device=snake_case							)  # noqa: E231
					# fmt: on

					return input_ids, attention_mask, bbox, token_type_ids, labels







@require_torch
class       SCREAMING_SNAKE_CASE_							(					unittest.TestCase      ):



	"""simple docstring"""


	@slow
	def 				lowerCamelCase__    (				self		:	int  )  ->					Union[str, Any]:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[int]								=			LayoutLMModel.from_pretrained("""microsoft/layoutlm-base-uncased"""  ).to(lowerCAmelCase  )

						__UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase				:  List[Any]								=			prepare_layoutlm_batch_inputs()

						# forward pass
						__UpperCamelCase				:  Dict								=			model(input_ids=lowerCAmelCase				,    bbox=lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase  )

						# test the sequence output on [0, :3, :3]
						__UpperCamelCase				:  int								=			torch.tensor(
						    [[0.17_85, -0.19_47, -0.04_25], [-0.32_54, -0.28_07, 0.25_53], [-0.53_91, -0.33_22, 0.33_64]]				,    device=lowerCAmelCase				,    )

						self.assertTrue(torch.allclose(outputs.last_hidden_state[0, :3, :3]				,    lowerCAmelCase				,    atol=1E-3  )  )

						# test the pooled output on [1, :3]
						__UpperCamelCase				:  List[str]								=			torch.tensor([-0.65_80, -0.02_14, 0.85_52]				,    device=lowerCAmelCase  )

						self.assertTrue(torch.allclose(outputs.pooler_output[1, :3]				,    lowerCAmelCase				,    atol=1E-3  )  )






	@slow
	def 				lowerCamelCase__    (				self		:	Tuple  )  ->					Optional[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[int]								=			LayoutLMForSequenceClassification.from_pretrained("""microsoft/layoutlm-base-uncased"""				,    num_labels=2  ).to(
						    lowerCAmelCase  )

						__UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase				:  Dict								=			prepare_layoutlm_batch_inputs()

						# forward pass
						__UpperCamelCase				:  Tuple								=			model(
						    input_ids=lowerCAmelCase				,    bbox=lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase				,    labels=torch.tensor([1, 1]				,    device=lowerCAmelCase  )				,    )

						# test whether we get a loss as a scalar
						__UpperCamelCase				:  Tuple								=			outputs.loss
						__UpperCamelCase				:  Dict								=			torch.Size([]  )
						self.assertEqual(loss.shape				,    lowerCAmelCase  )

						# test the shape of the logits
						__UpperCamelCase				:  Optional[int]								=			outputs.logits
						__UpperCamelCase				:  List[str]								=			torch.Size((2, 2)  )
						self.assertEqual(logits.shape				,    lowerCAmelCase  )






	@slow
	def 				lowerCamelCase__    (				self		:	Dict  )  ->					Tuple:







						"""simple docstring"""







						__UpperCamelCase				:  Optional[Any]								=			LayoutLMForTokenClassification.from_pretrained("""microsoft/layoutlm-base-uncased"""				,    num_labels=13  ).to(
						    lowerCAmelCase  )

						__UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase				:  List[str]								=			prepare_layoutlm_batch_inputs()

						# forward pass
						__UpperCamelCase				:  Any								=			model(
						    input_ids=lowerCAmelCase				,    bbox=lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase				,    labels=lowerCAmelCase  )

						# test the loss calculation to be around 2.65
						# expected_loss = torch.tensor(2.65, device=torch_device)

						# The loss is currently somewhat random and can vary between 0.1-0.3 atol.
						# self.assertTrue(torch.allclose(outputs.loss, expected_loss, atol=0.1))

						# test the shape of the logits
						__UpperCamelCase				:  Any								=			outputs.logits
						__UpperCamelCase				:  Dict								=			torch.Size((2, 25, 13)  )
						self.assertEqual(logits.shape				,    lowerCAmelCase  )






	@slow
	def 				lowerCamelCase__    (				self		:	List[Any]  )  ->					Optional[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  List[Any]								=			LayoutLMForQuestionAnswering.from_pretrained("""microsoft/layoutlm-base-uncased"""  ).to(lowerCAmelCase  )

						__UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase			, __UpperCamelCase				:  Dict								=			prepare_layoutlm_batch_inputs()

						# forward pass
						__UpperCamelCase				:  int								=			model(input_ids=lowerCAmelCase				,    bbox=lowerCAmelCase				,    attention_mask=lowerCAmelCase				,    token_type_ids=lowerCAmelCase  )

						# test the shape of the logits
						__UpperCamelCase				:  Union[str, Any]								=			torch.Size((2, 25)  )
						self.assertEqual(outputs.start_logits.shape				,    lowerCAmelCase  )
						self.assertEqual(outputs.end_logits.shape				,    lowerCAmelCase  )












import importlib.util
import weakref
from functools import partialmethod

from .dependency_versions_check import dep_version_check
from .utils import is_accelerate_available, is_torch_available, logging


if is_torch_available():
	import torch

a__		    =    logging.get_logger(__name__)





def 							A__   ()    ->	int:
					return importlib.util.find_spec("""deepspeed"""							) is not None


if is_accelerate_available() and is_deepspeed_available():
	from accelerate.utils.deepspeed import HfDeepSpeedConfig as DeepSpeedConfig
else:
	# Inherits from a dummy `object` if accelerate is not available, so that python succeeds to import this file.
	# Deepspeed glue code will never inherit this dummy object as it checks if accelerate is available.
	from builtins import object as DeepSpeedConfig


class       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase      ):



	"""simple docstring"""


	def __init__(				self		:	List[Any]				,    lowerCAmelCase		:	Any  )  ->					str:







						"""simple docstring"""







						set_hf_deepspeed_config(self  )
						dep_version_check("""accelerate"""  )
						dep_version_check("""deepspeed"""  )
						super().__init__(lowerCAmelCase  )






class       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase      ):



	"""simple docstring"""


	def __init__(				self		:	Optional[Any]				,    lowerCAmelCase		:	Union[str, Any]  )  ->					int:







						"""simple docstring"""







						super().__init__(lowerCAmelCase  )
						__UpperCamelCase				:  Optional[int]								=			None
						__UpperCamelCase				:  Tuple								=			[]






	def 				lowerCamelCase__    (				self		:	List[str]  )  ->					List[Any]:







						"""simple docstring"""







						if self._dtype is None:
											raise ValueError("""trainer_config_process() wasn't called yet to tell dtype"""  )
						return self._dtype






	def 				lowerCamelCase__    (				self		:	int				,    lowerCAmelCase		:	Union[str, Any]  )  ->					List[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  List[str]								=			self.get_value(lowerCAmelCase  )
						if val is None:
											return False
						else:
											return val == "auto"






	def 				lowerCamelCase__    (				self		:	str				,    lowerCAmelCase		:	Tuple				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	List[Any]=None				,    lowerCAmelCase		:	int=True  )  ->					int:







						"""simple docstring"""







						__UpperCamelCase			, __UpperCamelCase				:  Union[str, Any]								=			self.find_config_node(lowerCAmelCase  )
						if config is None:
											return

						if config.get(lowerCAmelCase  ) == "auto":
											__UpperCamelCase				:  List[str]								=			hf_val
											return

						if not must_match:
											return

						__UpperCamelCase				:  Dict								=			config.get(lowerCAmelCase  )
						if ds_val is not None and ds_val != hf_val:
											self.mismatches.append(F'''- ds {ds_key_long}={ds_val} vs hf {hf_key}={hf_val}'''  )

	__magic_name__      :					Optional[Any]			  =		partialmethod(_UpperCamelCase  ,     must_match=_UpperCamelCase      )






	def 				lowerCamelCase__    (				self		:	Tuple				,    lowerCAmelCase		:	Optional[int]  )  ->					Union[str, Any]:







						"""simple docstring"""







						__UpperCamelCase				:  str								=			args.world_size * args.per_device_train_batch_size * args.gradient_accumulation_steps
						self.fill_match(
						    """train_micro_batch_size_per_gpu"""				,    args.per_device_train_batch_size				,    """per_device_train_batch_size"""  )
						self.fill_match("""gradient_accumulation_steps"""				,    args.gradient_accumulation_steps				,    """gradient_accumulation_steps"""  )
						self.fill_match("""train_batch_size"""				,    lowerCAmelCase				,    """train_batch_size (calculated)"""  )
						self.fill_match("""gradient_clipping"""				,    args.max_grad_norm				,    """max_grad_norm"""  )

						self.fill_match("""optimizer.params.lr"""				,    args.learning_rate				,    """learning_rate"""  )
						self.fill_match("""optimizer.params.betas"""				,    [args.adam_betaa, args.adam_betaa]				,    """adam_beta1+adam_beta2"""  )
						self.fill_match("""optimizer.params.eps"""				,    args.adam_epsilon				,    """adam_epsilon"""  )
						self.fill_match("""optimizer.params.weight_decay"""				,    args.weight_decay				,    """weight_decay"""  )

						self.fill_only("""scheduler.params.warmup_min_lr"""				,    0  )  # not a trainer arg
						self.fill_match("""scheduler.params.warmup_max_lr"""				,    args.learning_rate				,    """learning_rate"""  )
						# total_num_steps - will get set in trainer_config_finalize

						# fp16
						if args.fpaa or args.fpaa_full_eval:
											__UpperCamelCase				:  Union[str, Any]								=			"""apex""" if args.fpaa_backend == """apex""" else """amp"""
						else:
											__UpperCamelCase				:  Optional[int]								=			None

						if args.save_on_each_node:
											# deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
											__UpperCamelCase				:  List[Any]								=			self.config.get("""checkpoint"""				,    {}  )
											__UpperCamelCase				:  Tuple								=			args.save_on_each_node

						# amp: similar to the pytorch native amp - it has a bunch of optional params but we won't set
						# any here unless the user did the work
						self.fill_match(
						    """fp16.enabled"""				,    ((args.fpaa or args.fpaa_full_eval) and fpaa_backend == """amp""")				,    """fp16|fp16_full_eval+fp16_backend(amp)"""				,    )

						# apex: delegates amp work to apex (which needs to be available), but it cannot be used with any
						# ZeRO features
						self.fill_match("""amp.enabled"""				,    fpaa_backend == """apex"""				,    """fp16+fp16_backend(apex)"""  )
						self.fill_match("""amp.opt_level"""				,    args.fpaa_opt_level				,    """fp16_opt_level"""  )

						self.fill_match("""bf16.enabled"""				,    (args.bfaa or args.bfaa_full_eval)				,    """bf16|bf16_full_eval"""  )

						# deepspeed's default mode is fp16 unless there is a config that says differently
						if self.is_true("""bf16.enabled"""  ):
											__UpperCamelCase				:  List[Any]								=			torch.bfloataa
						elif self.is_false("""fp16.enabled"""  ):
											__UpperCamelCase				:  Optional[int]								=			torch.floataa
						else:
											__UpperCamelCase				:  List[str]								=			torch.floataa






	def 				lowerCamelCase__    (				self		:	Optional[Any]				,    lowerCAmelCase		:	Tuple				,    lowerCAmelCase		:	Union[str, Any]				,    lowerCAmelCase		:	Union[str, Any]  )  ->					str:







						"""simple docstring"""







						__UpperCamelCase				:  Tuple								=			[
						    """zero_optimization.reduce_bucket_size""",
						    """zero_optimization.stage3_prefetch_bucket_size""",
						    """zero_optimization.stage3_param_persistence_threshold""",
						]
						__UpperCamelCase				:  str								=			[x for x in hidden_size_based_keys if self.is_auto(lowerCAmelCase  )]

						if len(lowerCAmelCase  ) > 0:
											if hasattr(model.config				,    """hidden_size"""  ):
																__UpperCamelCase				:  Any								=			model.config.hidden_size
											elif hasattr(model.config				,    """hidden_sizes"""  ):
																# if there are many hidden sizes pick the largest one
																__UpperCamelCase				:  Optional[int]								=			max(model.config.hidden_sizes  )
											else:
																raise ValueError(
																    """The model's config file has neither `hidden_size` nor `hidden_sizes` entry, """
																    """therefore it's not possible to automatically fill out the following `auto` entries """
																    F'''in the DeepSpeed config file: {hidden_size_auto_keys}. You can fix that by replacing '''
																    """`auto` values for these keys with an integer value of your choice."""  )

											self.fill_only("""zero_optimization.reduce_bucket_size"""				,    hidden_size * hidden_size  )
											if self.is_zeroa():
																# automatically assign the optimal config values based on model config
																self.fill_only("""zero_optimization.stage3_prefetch_bucket_size"""				,    0.9 * hidden_size * hidden_size  )
																self.fill_only("""zero_optimization.stage3_param_persistence_threshold"""				,    10 * hidden_size  )

        # scheduler
						self.fill_match("""scheduler.params.total_num_steps"""				,    lowerCAmelCase				,    """num_training_steps (calculated)"""  )
						self.fill_match("""scheduler.params.warmup_num_steps"""				,    args.get_warmup_steps(lowerCAmelCase  )				,    """warmup_steps"""  )

						if len(self.mismatches  ) > 0:
											__UpperCamelCase				:  Tuple								=			"""\n""".join(self.mismatches  )
											raise ValueError(
											    """Please correct the following DeepSpeed config values that mismatch TrainingArguments"""
											    F''' values:\n{mismatches}\nThe easiest method is to set these DeepSpeed config values to \'auto\'.'''  )






a__		    =    None





def 							A__   (snake_case      :		Tuple							)    ->	Union[str, Any]:
					# this is a special weakref global object to allow us to get to Deepspeed config from APIs
					# that don't have an easy way to get to the Deepspeed config outside of the Trainer domain.
					global _hf_deepspeed_config_weak_ref
					# will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
					__UpperCamelCase				:  str								=			weakref.ref(snake_case							)





def 							A__   ()    ->	Any:
					# useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
					global _hf_deepspeed_config_weak_ref
					__UpperCamelCase				:  List[Any]								=			None





def 							A__   ()    ->	Tuple:
					if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
										return _hf_deepspeed_config_weak_ref().is_zeroa()
					else:
										return False





def 							A__   ()    ->	Any:
					if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
										return _hf_deepspeed_config_weak_ref().config
					else:
										return None





def 							A__   (snake_case      :		Dict		,  snake_case      :		Optional[int]		,  snake_case      :		List[str]		,  snake_case      :		Dict		,  snake_case      :		List[str]							)    ->	int:
					from accelerate.utils import DummyOptim, DummyScheduler

					__UpperCamelCase				:  Tuple								=			hf_deepspeed_config.config

					# Optimizer + Scheduler
					# Currently supported combos:
					# 1. DS scheduler + DS optimizer: Yes
					# 2. HF scheduler + HF optimizer: Yes
					# 3. DS scheduler + HF optimizer: Yes
					# 4. HF scheduler + DS optimizer: No
					#
					# Unless Offload is enabled in which case it's:
					# 1. DS scheduler + DS optimizer: Yes
					# 2. HF scheduler + HF optimizer: Mostly*
					# 3. DS scheduler + HF optimizer: Mostly*
					# 4. HF scheduler + DS optimizer: No
					#
					# Mostly*: All non-native DeepSpeed optimizers that have both CPU and GPU implementation should work (except LAMB)

					__UpperCamelCase				:  Dict								=			None
					if "optimizer" in config:
										if args.adafactor:
															raise ValueError(
															    """--adafactor was passed, but also found `optimizer` configured in the DeepSpeed config. """
															    """Only one optimizer can be configured."""							)
										__UpperCamelCase				:  Any								=			DummyOptim(params=snake_case							)
					else:
										if hf_deepspeed_config.is_offload():
															logger.info(
															    """Detected ZeRO Offload and non-DeepSpeed optimizers: This combination should work as long as the"""
															    """ custom optimizer has both CPU and GPU implementation (except LAMB)"""							)

										# ds supports Adam, OneBitAdam, and Lamb optimizers and can import other optimizers from torch.
										# But trainer uses AdamW by default.
										__UpperCamelCase				:  Optional[Any]								=			trainer.create_optimizer()
										# To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
										__UpperCamelCase				:  List[str]								=			True

					__UpperCamelCase				:  str								=			None
					if "scheduler" in config:
										__UpperCamelCase				:  str								=			DummyScheduler(snake_case							)
					else:
										if isinstance(snake_case		,  snake_case							):
															raise ValueError(
															    """Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. """
															    """Please configure a scheduler in the DeepSpeed config."""							)
										__UpperCamelCase				:  List[Any]								=			trainer.create_scheduler(num_training_steps=snake_case		,  optimizer=snake_case							)

					return optimizer, lr_scheduler





def 							A__   (snake_case      :		Any		,  snake_case      :		Union[str, Any]		,  snake_case      :		str=False							)    ->	int:
					from deepspeed.utils import logger as ds_logger

					__UpperCamelCase				:  List[Any]								=			trainer.model
					__UpperCamelCase				:  Tuple								=			trainer.args

					__UpperCamelCase				:  Union[str, Any]								=			trainer.accelerator.state.deepspeed_plugin.hf_ds_config

					# resume config update - some bits like `model` and `num_training_steps` only become available during train
					hf_deepspeed_config.trainer_config_finalize(snake_case		,  snake_case		,  snake_case							)

					# set the Deepspeed log level consistent with the Trainer
					ds_logger.setLevel(args.get_process_log_level()							)

					if inference:
										# only Z3 makes sense for the inference
										if not hf_deepspeed_config.is_zeroa():
															raise ValueError("""ZeRO inference only makes sense with ZeRO Stage 3 - please adjust your config"""							)

										# in case the training config is re-used for inference
										hf_deepspeed_config.del_config_sub_tree("""optimizer"""							)
										hf_deepspeed_config.del_config_sub_tree("""lr_scheduler"""							)
										__UpperCamelCase			, __UpperCamelCase				:  str								=			None, None
										__UpperCamelCase				:  Dict								=			None
					else:
										__UpperCamelCase				:  Optional[Any]								=			None  # important for when deepspeed_init is used as re-init
										__UpperCamelCase				:  Optional[int]								=			list(filter(lambda snake_case							: p.requires_grad		,  model.parameters()							)							)
										__UpperCamelCase			, __UpperCamelCase				:  str								=			deepspeed_optim_sched(
										    snake_case		,  snake_case		,  snake_case		,  snake_case		,  snake_case							)

					# keep for quick debug:
					# from pprint import pprint; pprint(config)

					return optimizer, lr_scheduler







def 							A__   (snake_case      :		Optional[Any]		,  snake_case      :		Any							)    ->	str:
					# it's possible that the user is trying to resume from model_path, which doesn't necessarily
					# contain a deepspeed checkpoint. e.g. examples just check if the dir exists and assume it's
					# a resume from a checkpoint and not just a local pretrained weight. So we check here if the
					# path contains what looks like a deepspeed checkpoint
					import glob

					__UpperCamelCase				:  Any								=			sorted(glob.glob(F'''{checkpoint_path}/global_step*'''							)							)

					if len(snake_case							) > 0:
										logger.info(F'''Attempting to resume from {checkpoint_path}'''							)
										# this magically updates self.optimizer and self.lr_scheduler
										__UpperCamelCase			, __UpperCamelCase				:  Optional[int]								=			deepspeed_engine.load_checkpoint(
										    snake_case		,  load_optimizer_states=snake_case		,  load_lr_scheduler_states=snake_case							)
										if load_path is None:
															raise ValueError(F'''[deepspeed] failed to resume from checkpoint {checkpoint_path}'''							)
					else:
										raise ValueError(F'''Can\'t find a valid checkpoint at {checkpoint_path}'''							)






