



"""simple docstring"""
import unittest

from transformers import ErnieMConfig, 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, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch

						from transformers import (
						    ErnieMForInformationExtraction,
						    ErnieMForMultipleChoice,
						    ErnieMForQuestionAnswering,
						    ErnieMForSequenceClassification,
						    ErnieMForTokenClassification,
						    ErnieMModel,
						)
						from transformers.models.ernie_m.modeling_ernie_m import ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST
class     lowerCamelCase  :






					'''simple docstring'''





					def __init__(	self:							Optional[Any]						,					snake_case:							Tuple						,					snake_case:							Any=13						,					snake_case:							List[str]=7						,					snake_case:							Dict=True						,					snake_case:							List[str]=True						,					snake_case:							str=True						,					snake_case:							int=99						,					snake_case:							int=32						,					snake_case:							Optional[int]=5						,					snake_case:							List[Any]=4						,					snake_case:							Any=37						,					snake_case:							List[Any]="gelu"						,					snake_case:							int=0.1						,					snake_case:							Dict=0.1						,					snake_case:							List[str]=512						,					snake_case:							str=16						,					snake_case:							List[str]=2						,					snake_case:							str=0.0_2						,					snake_case:							Dict=3						,					snake_case:							Tuple=4						,					snake_case:							List[str]=None						,					)       ->     Optional[int]:
											snake_case_			:List[str]						=      parent
											snake_case_			:int						=      batch_size
											snake_case_			:List[str]						=      seq_length
											snake_case_			:List[str]						=      is_training
											snake_case_			:Union[str, Any]						=      use_input_mask
											snake_case_			:Union[str, Any]						=      use_labels
											snake_case_			:Optional[int]						=      vocab_size
											snake_case_			:Any						=      hidden_size
											snake_case_			:List[Any]						=      num_hidden_layers
											snake_case_			:Optional[int]						=      num_attention_heads
											snake_case_			:Optional[Any]						=      intermediate_size
											snake_case_			:int						=      hidden_act
											snake_case_			:int						=      hidden_dropout_prob
											snake_case_			:Dict						=      attention_probs_dropout_prob
											snake_case_			:Any						=      max_position_embeddings
											snake_case_			:Optional[Any]						=      type_vocab_size
											snake_case_			:Optional[Any]						=      type_sequence_label_size
											snake_case_			:Optional[Any]						=      initializer_range
											snake_case_			:Any						=      num_labels
											snake_case_			:str						=      num_choices
											snake_case_			:str						=      scope






					def        lowerCAmelCase_    (	self:							int						)       ->     Union[str, Any]:
											snake_case_			:str						=      ids_tensor([self.batch_size, self.seq_length]						,					self.vocab_size						)

											snake_case_			:Dict						=      None
											if self.use_input_mask:
																	snake_case_			:Tuple						=      random_attention_mask([self.batch_size, self.seq_length]						)

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

											snake_case_			:int						=      self.get_config()

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






					def        lowerCAmelCase_    (	self:							int						)       ->     List[str]:
											snake_case_			:Tuple						=      ids_tensor([self.batch_size, self.seq_length]						,					self.vocab_size						)

											snake_case_			:Union[str, Any]						=      None
											if self.use_input_mask:
																	snake_case_			:str						=      random_attention_mask([self.batch_size, self.seq_length]						)
											snake_case_			:str						=      self.get_config()

											return config, input_ids, input_mask






					def        lowerCAmelCase_    (	self:							Optional[int]						)       ->     List[str]:
											return ErnieMConfig(
											    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						,					snake_case:							int						,					snake_case:							Dict						,					snake_case:							Dict						,					snake_case:							Optional[Any]						,					snake_case:							List[Any]						,					snake_case:							str						)       ->     Tuple:
											snake_case_			:Any						=      ErnieMModel(config=snake_case						)
											model.to(snake_case						)
											model.eval()
											snake_case_			:Any						=      model(snake_case						,					return_dict=snake_case						)
											self.parent.assertEqual(result.last_hidden_state.shape						,					(self.batch_size, self.seq_length, self.hidden_size)						)






					def        lowerCAmelCase_    (	self:							List[str]						,					snake_case:							List[Any]						,					snake_case:							List[str]						,					snake_case:							str						,					snake_case:							Optional[Any]						,					snake_case:							List[Any]						,					snake_case:							Tuple						)       ->     str:
											snake_case_			:Dict						=      ErnieMForQuestionAnswering(config=snake_case						)
											model.to(snake_case						)
											model.eval()
											snake_case_			:Optional[int]						=      model(
											    snake_case						,					attention_mask=snake_case						,					start_positions=snake_case						,					end_positions=snake_case						,					)
											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]						,					snake_case:							Dict						,					snake_case:							List[str]						,					snake_case:							Union[str, Any]						,					snake_case:							List[Any]						,					snake_case:							str						,					snake_case:							Optional[int]						)       ->     Optional[int]:
											snake_case_			:List[Any]						=      ErnieMForInformationExtraction(config=snake_case						)
											model.to(snake_case						)
											model.eval()
											snake_case_			:Any						=      torch.ones_like(snake_case						,					dtype=torch.floataa						)
											snake_case_			:List[Any]						=      model(
											    snake_case						,					attention_mask=snake_case						,					start_positions=snake_case						,					end_positions=snake_case						,					)
											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:							int						,					snake_case:							List[Any]						,					snake_case:							str						,					snake_case:							Any						,					snake_case:							int						,					snake_case:							Dict						,					snake_case:							Tuple						)       ->     Dict:
											snake_case_			:Any						=      self.num_labels
											snake_case_			:Tuple						=      ErnieMForSequenceClassification(snake_case						)
											model.to(snake_case						)
											model.eval()
											snake_case_			:Dict						=      model(snake_case						,					attention_mask=snake_case						,					labels=snake_case						)
											self.parent.assertEqual(result.logits.shape						,					(self.batch_size, self.num_labels)						)






					def        lowerCAmelCase_    (	self:							Optional[int]						,					snake_case:							Dict						,					snake_case:							Any						,					snake_case:							str						,					snake_case:							List[Any]						,					snake_case:							Dict						,					snake_case:							int						)       ->     Optional[Any]:
											snake_case_			:Union[str, Any]						=      self.num_labels
											snake_case_			:List[Any]						=      ErnieMForTokenClassification(config=snake_case						)
											model.to(snake_case						)
											model.eval()
											input_ids.to(snake_case						)
											input_mask.to(snake_case						)
											token_labels.to(snake_case						)

											snake_case_			:Tuple						=      model(snake_case						,					attention_mask=snake_case						,					labels=snake_case						)

											self.parent.assertEqual(result.logits.shape						,					(self.batch_size, self.seq_length, self.num_labels)						)






					def        lowerCAmelCase_    (	self:							Tuple						,					snake_case:							Dict						,					snake_case:							Tuple						,					snake_case:							Dict						,					snake_case:							Optional[Any]						,					snake_case:							List[str]						,					snake_case:							List[str]						)       ->     List[str]:
											snake_case_			:Optional[Any]						=      self.num_choices
											snake_case_			:Any						=      ErnieMForMultipleChoice(config=snake_case						)
											model.to(snake_case						)
											model.eval()
											snake_case_			:Tuple						=      input_ids.unsqueeze(1						).expand(-1						,					self.num_choices						,					-1						).contiguous()
											snake_case_			:int						=      input_mask.unsqueeze(1						).expand(-1						,					self.num_choices						,					-1						).contiguous()
											snake_case_			:List[str]						=      model(
											    snake_case						,					attention_mask=snake_case						,					labels=snake_case						,					)
											self.parent.assertEqual(result.logits.shape						,					(self.batch_size, self.num_choices)						)






					def        lowerCAmelCase_    (	self:							List[str]						)       ->     List[str]:
											snake_case_			:Union[str, Any]						=      self.prepare_config_and_inputs()
											(
											    (
											    snake_case_
											),	(
											    snake_case_
											),	(
											    snake_case_
											),	(
											    snake_case_
											),	(
											    snake_case_
											),	(
											    snake_case_
											),	
											)			:List[Any]						=      config_and_inputs
											snake_case_			:List[str]						=      {"""input_ids""": input_ids, """attention_mask""": input_mask}
											return config, inputs_dict

@require_torch
class     lowerCamelCase  (      _lowerCAmelCase   ,						_lowerCAmelCase   ,						unittest.TestCase    ):






					'''simple docstring'''





					_A		:      Tuple						=					(
					    (
					        ErnieMModel,
					        ErnieMForMultipleChoice,
					        ErnieMForQuestionAnswering,
					        ErnieMForSequenceClassification,
					        ErnieMForTokenClassification,
					    )
					    if is_torch_available()
					    else ()
					)
					_A		:      List[Any]						=					()
					_A		:      Union[str, Any]						=					(
					    {
					        """feature-extraction""": ErnieMModel,
					        """question-answering""": ErnieMForQuestionAnswering,
					        """text-classification""": ErnieMForSequenceClassification,
					        """token-classification""": ErnieMForTokenClassification,
					        """zero-shot""": ErnieMForSequenceClassification,
					    }
					    if is_torch_available()
					    else {}
					)
					_A		:      str						=					False






					def        lowerCAmelCase_    (	self:							List[str]						,					snake_case:							Optional[Any]						,					snake_case:							List[str]						,					snake_case:							List[Any]						,					snake_case:							Tuple						,					snake_case:							List[Any]						)       ->     int:
											if pipeline_test_casse_name == "QAPipelineTests":
																	return True

											return False






					def        lowerCAmelCase_    (	self:							Any						)       ->     Optional[int]:
											snake_case_			:Tuple						=      ErnieMModelTester(self						)
											snake_case_			:Tuple						=      ConfigTester(self						,					config_class=snake_case						,					hidden_size=37						)






					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     List[Any]:
											self.config_tester.run_common_tests()






					def        lowerCAmelCase_    (	self:							Optional[int]						)       ->     List[Any]:
											snake_case_			:int						=      self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_model(*snake_case						)






					def        lowerCAmelCase_    (	self:							Any						)       ->     Tuple:
											snake_case_			:int						=      self.model_tester.prepare_config_and_inputs()
											for type in ["absolute", "relative_key", "relative_key_query"]:
																	snake_case_			:List[str]						=      type
																	self.model_tester.create_and_check_model(*snake_case						)






					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     Optional[Any]:
											snake_case_			:str						=      self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_multiple_choice(*snake_case						)






					def        lowerCAmelCase_    (	self:							Optional[Any]						)       ->     Optional[Any]:
											snake_case_			:List[str]						=      self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_question_answering(*snake_case						)






					def        lowerCAmelCase_    (	self:							Tuple						)       ->     int:
											snake_case_			:List[str]						=      self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_sequence_classification(*snake_case						)






					def        lowerCAmelCase_    (	self:							int						)       ->     Dict:
											snake_case_			:List[Any]						=      self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_information_extraction(*snake_case						)






					def        lowerCAmelCase_    (	self:							Any						)       ->     List[Any]:
											snake_case_			:int						=      self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_token_classification(*snake_case						)






					@slow
					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     List[str]:
											for model_name in ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	snake_case_			:List[Any]						=      ErnieMModel.from_pretrained(snake_case						)
																	self.assertIsNotNone(snake_case						)

@require_torch
class     lowerCamelCase  (      unittest.TestCase    ):






					'''simple docstring'''





					@slow
					def        lowerCAmelCase_    (	self:							List[str]						)       ->     Optional[Any]:
											snake_case_			:Tuple						=      ErnieMModel.from_pretrained("""susnato/ernie-m-base_pytorch"""						)
											model.eval()
											snake_case_			:str						=      torch.tensor([[0, 1, 2, 3, 4, 5]]						)
											snake_case_			:Dict						=      model(snake_case						)[0]

											# TODO Replace vocab size
											snake_case_			:str						=      768

											snake_case_			:int						=      torch.Size((1, 6, hidden_size)						)
											self.assertEqual(output.shape						,					snake_case						)

											snake_case_			:List[str]						=      torch.tensor(
											    [[[-0.0_0_1_2, 0.1_2_4_5, -0.0_2_1_4], [-0.0_7_4_2, 0.0_2_4_4, -0.0_7_7_1], [-0.0_3_3_3, 0.1_1_6_4, -0.1_5_5_4]]]						)

											self.assertTrue(torch.allclose(output[:, :3, :3]						,					snake_case						,					atol=1E-3						)						)





"""simple docstring"""
from typing import Any, Dict, List, Mapping

import numpy as np
import torch

from ...utils import is_cython_available, requires_backends


if is_cython_available():
						import pyximport

						pyximport.install(setup_args={"include_dirs": np.get_include()})
						from . import algos_graphormer  # noqa E402
def        A_  (      _lowercase, _lowercase = 512     ):







						'''simple docstring'''


						snake_case_			:int						=      x.shape[1] if len(x.shape     ) > 1 else 1
						snake_case_			:List[Any]						=      1 + np.arange(0, feature_num * offset, _lowercase, dtype=np.intaa     )
						snake_case_			:int						=      x + feature_offset
						return x
def        A_  (      _lowercase, _lowercase=True     ):







						'''simple docstring'''


						requires_backends(_lowercase, ["""cython"""]     )

						if keep_features and "edge_attr" in item.keys():  # edge_attr
												snake_case_			:Tuple						=      np.asarray(item["""edge_attr"""], dtype=np.intaa     )
						else:
												snake_case_			:List[str]						=      np.ones((len(item["""edge_index"""][0]     ), 1), dtype=np.intaa     )  # same embedding for all

						if keep_features and "node_feat" in item.keys():  # input_nodes
												snake_case_			:Dict						=      np.asarray(item["""node_feat"""], dtype=np.intaa     )
						else:
												snake_case_			:Tuple						=      np.ones((item["""num_nodes"""], 1), dtype=np.intaa     )  # same embedding for all

						snake_case_			:Any						=      np.asarray(item["""edge_index"""], dtype=np.intaa     )

						snake_case_			:str						=      convert_to_single_emb(_lowercase     ) + 1
						snake_case_			:str						=      item["""num_nodes"""]

						if len(edge_attr.shape     ) == 1:
												snake_case_			:Any						=      edge_attr[:, None]
						snake_case_			:Dict						=      np.zeros([num_nodes, num_nodes, edge_attr.shape[-1]], dtype=np.intaa     )
						snake_case_			:str						=      convert_to_single_emb(_lowercase     ) + 1

						# node adj matrix [num_nodes, num_nodes] bool
						snake_case_			:List[str]						=      np.zeros([num_nodes, num_nodes], dtype=_lowercase     )
						snake_case_			:int						=      True

						snake_case_,	snake_case_			:Tuple						=      algos_graphormer.floyd_warshall(_lowercase     )
						snake_case_			:List[str]						=      np.amax(_lowercase     )

						snake_case_			:Dict						=      algos_graphormer.gen_edge_input(_lowercase, _lowercase, _lowercase     )
						snake_case_			:Tuple						=      np.zeros([num_nodes + 1, num_nodes + 1], dtype=np.single     )  # with graph token

						# combine
						snake_case_			:str						=      input_nodes + 1  # we shift all indices by one for padding
						snake_case_			:str						=      attn_bias
						snake_case_			:Optional[Any]						=      attn_edge_type
						snake_case_			:Any						=      shortest_path_result.astype(np.intaa     ) + 1  # we shift all indices by one for padding
						snake_case_			:int						=      np.sum(_lowercase, axis=1     ).reshape(-1     ) + 1  # we shift all indices by one for padding
						snake_case_			:List[str]						=      item["""in_degree"""]  # for undirected graph
						snake_case_			:List[str]						=      input_edges + 1  # we shift all indices by one for padding
						if "labels" not in item:
												snake_case_			:Optional[Any]						=      item["""y"""]

						return item

class     lowerCamelCase  :






					'''simple docstring'''





					def __init__(	self:							Dict						,					snake_case:							Optional[int]=20						,					snake_case:							List[Any]=False						)       ->     Union[str, Any]:
											if not is_cython_available():
																	raise ImportError("""Graphormer preprocessing needs Cython (pyximport)"""						)

											snake_case_			:List[str]						=      spatial_pos_max
											snake_case_			:int						=      on_the_fly_processing






					def __call__(	self:							str						,					snake_case:							List[dict]						)       ->     Dict[str, Any]:
											if self.on_the_fly_processing:
																	snake_case_			:int						=      [preprocess_item(snake_case						) for i in features]

											if not isinstance(features[0]						,					snake_case						):
																	snake_case_			:Tuple						=      [vars(snake_case						) for f in features]
											snake_case_			:Any						=      {}

											snake_case_			:Dict						=      max(len(i["""input_nodes"""]						) for i in features						)
											snake_case_			:List[Any]						=      len(features[0]["""input_nodes"""][0]						)
											snake_case_			:Dict						=      len(features[0]["""attn_edge_type"""][0][0]						)
											snake_case_			:List[Any]						=      max(len(i["""input_edges"""][0][0]						) for i in features						)
											snake_case_			:Any						=      len(features[0]["""input_edges"""][0][0][0]						)
											snake_case_			:List[str]						=      len(snake_case						)

											snake_case_			:Union[str, Any]						=      torch.zeros(snake_case						,					max_node_num + 1						,					max_node_num + 1						,					dtype=torch.float						)
											snake_case_			:int						=      torch.zeros(snake_case						,					snake_case						,					snake_case						,					snake_case						,					dtype=torch.long						)
											snake_case_			:Optional[Any]						=      torch.zeros(snake_case						,					snake_case						,					snake_case						,					dtype=torch.long						)
											snake_case_			:Tuple						=      torch.zeros(snake_case						,					snake_case						,					dtype=torch.long						)
											snake_case_			:Dict						=      torch.zeros(snake_case						,					snake_case						,					snake_case						,					dtype=torch.long						)
											snake_case_			:Union[str, Any]						=      torch.zeros(
											    snake_case						,					snake_case						,					snake_case						,					snake_case						,					snake_case						,					dtype=torch.long						)

											for ix, f in enumerate(snake_case						):
																	for k in ["attn_bias", "attn_edge_type", "spatial_pos", "in_degree", "input_nodes", "input_edges"]:
																							snake_case_			:List[str]						=      torch.tensor(f[k]						)

																	if len(f["""attn_bias"""][1:, 1:][f["""spatial_pos"""] >= self.spatial_pos_max]						) > 0:
																							snake_case_			:Union[str, Any]						=      float("""-inf"""						)

																	snake_case_			:Optional[Any]						=      f["""attn_bias"""]
																	snake_case_			:int						=      f[
																	    """attn_edge_type"""
																	]
																	snake_case_			:Dict						=      f["""spatial_pos"""]
																	snake_case_			:Tuple						=      f["""in_degree"""]
																	snake_case_			:List[Any]						=      f["""input_nodes"""]
																	snake_case_			:List[Any]						=      f["""input_edges"""]

											snake_case_			:int						=      batch["""in_degree"""]

											snake_case_			:Any						=      features[0]["""labels"""]
											if len(snake_case						) == 1:  # one task
																	if isinstance(sample[0]						,					snake_case						):  # regression
																							snake_case_			:Dict						=      torch.from_numpy(np.concatenate([i["""labels"""] for i in features]						)						)
																	else:  # binary classification
																							snake_case_			:str						=      torch.from_numpy(np.concatenate([i["""labels"""] for i in features]						)						)
											else:  # multi task classification, left to float to keep the NaNs
																	snake_case_			:Optional[int]						=      torch.from_numpy(np.stack([i["""labels"""] for i in features]						,					axis=0						)						)

											return batch
