

"""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__					(					__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE = 5_1_2				):
		__lowerCAmelCase: int					   =    x.shape[1] if len(x.shape				) > 1 else 1
		__lowerCAmelCase: Dict					   =    1 + np.arange(0      ,				feature_num * offset      ,				__SCREAMING_SNAKE_CASE      ,				dtype=np.intaa				)
		__lowerCAmelCase: Any					   =    x + feature_offset
		return x






def 					a__					(					__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE=True				):
		requires_backends(__SCREAMING_SNAKE_CASE      ,				["cython"]				)

		if keep_features and "edge_attr" in item.keys():  # edge_attr
				__lowerCAmelCase: Any					   =    np.asarray(item["edge_attr"]      ,				dtype=np.intaa				)
		else:
				__lowerCAmelCase: List[Any]					   =    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
				__lowerCAmelCase: int					   =    np.asarray(item["node_feat"]      ,				dtype=np.intaa				)
		else:
				__lowerCAmelCase: str					   =    np.ones((item["num_nodes"], 1)      ,				dtype=np.intaa				)  # same embedding for all

		__lowerCAmelCase: int					   =    np.asarray(item["edge_index"]      ,				dtype=np.intaa				)

		__lowerCAmelCase: Union[str, Any]					   =    convert_to_single_emb(__SCREAMING_SNAKE_CASE				) + 1
		__lowerCAmelCase: List[str]					   =    item["num_nodes"]

		if len(edge_attr.shape				) == 1:
				__lowerCAmelCase: Tuple					   =    edge_attr[:, None]
		__lowerCAmelCase: Tuple					   =    np.zeros([num_nodes, num_nodes, edge_attr.shape[-1]]      ,				dtype=np.intaa				)
		__lowerCAmelCase: Union[str, Any]					   =    convert_to_single_emb(__SCREAMING_SNAKE_CASE				) + 1

		# node adj matrix [num_nodes, num_nodes] bool
		__lowerCAmelCase: Any					   =    np.zeros([num_nodes, num_nodes]      ,				dtype=__SCREAMING_SNAKE_CASE				)
		__lowerCAmelCase: int					   =    True

		__lowerCAmelCase: Optional[int]					   =    algos_graphormer.floyd_warshall(__SCREAMING_SNAKE_CASE				)
		__lowerCAmelCase: List[Any]					   =    np.amax(__SCREAMING_SNAKE_CASE				)

		__lowerCAmelCase: Optional[int]					   =    algos_graphormer.gen_edge_input(__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE      ,				__SCREAMING_SNAKE_CASE				)
		__lowerCAmelCase: List[Any]					   =    np.zeros([num_nodes + 1, num_nodes + 1]      ,				dtype=np.single				)  # with graph token

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

		return item





class 					snake_case			:

				def __init__(			self							:       List[Any]  ,						UpperCamelCase__							:       Optional[Any]=2_0  ,						UpperCamelCase__							:       List[str]=False)->     Any:



						'''simple docstring'''


						if not is_cython_available():
								raise ImportError("Graphormer preprocessing needs Cython (pyximport)")

						__lowerCAmelCase: Dict					   =    spatial_pos_max
						__lowerCAmelCase: int					   =    on_the_fly_processing




				def __call__(			self							:       Any  ,						UpperCamelCase__							:       List[dict])->     Dict[str, Any]:



						'''simple docstring'''


						if self.on_the_fly_processing:
								__lowerCAmelCase: List[str]					   =    [preprocess_item(UpperCamelCase__) for i in features]

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

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

						__lowerCAmelCase: Dict					   =    torch.zeros(UpperCamelCase__  ,						max_node_num + 1  ,						max_node_num + 1  ,						dtype=torch.float)
						__lowerCAmelCase: Dict					   =    torch.zeros(UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						dtype=torch.long)
						__lowerCAmelCase: List[str]					   =    torch.zeros(UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						dtype=torch.long)
						__lowerCAmelCase: Optional[int]					   =    torch.zeros(UpperCamelCase__  ,						UpperCamelCase__  ,						dtype=torch.long)
						__lowerCAmelCase: List[str]					   =    torch.zeros(UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						dtype=torch.long)
						__lowerCAmelCase: List[Any]					   =    torch.zeros(
						    UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						dtype=torch.long)

						for ix, f in enumerate(UpperCamelCase__):
								for k in ["attn_bias", "attn_edge_type", "spatial_pos", "in_degree", "input_nodes", "input_edges"]:
										__lowerCAmelCase: Any					   =    torch.tensor(f[k])

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

								__lowerCAmelCase: Optional[Any]					   =    f["attn_bias"]
								__lowerCAmelCase: Any					   =    f[
								    "attn_edge_type"
								]
								__lowerCAmelCase: Optional[Any]					   =    f["spatial_pos"]
								__lowerCAmelCase: List[str]					   =    f["in_degree"]
								__lowerCAmelCase: str					   =    f["input_nodes"]
								__lowerCAmelCase: Union[str, Any]					   =    f["input_edges"]

						__lowerCAmelCase: Tuple					   =    batch["in_degree"]

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

						return batch





"""simple docstring"""


from __future__ import annotations

import unittest

from transformers import OpenAIGPTConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
			import tensorflow as tf

			from transformers.models.openai.modeling_tf_openai import (
			    TF_OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST,
			    TFOpenAIGPTDoubleHeadsModel,
			    TFOpenAIGPTForSequenceClassification,
			    TFOpenAIGPTLMHeadModel,
			    TFOpenAIGPTModel,
			)





class 					snake_case			:

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



						'''simple docstring'''


						__lowerCAmelCase: str					   =    parent
						__lowerCAmelCase: Union[str, Any]					   =    1_3
						__lowerCAmelCase: Union[str, Any]					   =    7
						__lowerCAmelCase: List[Any]					   =    True
						__lowerCAmelCase: Any					   =    True
						__lowerCAmelCase: List[Any]					   =    True
						__lowerCAmelCase: str					   =    True
						__lowerCAmelCase: List[Any]					   =    True
						__lowerCAmelCase: Tuple					   =    9_9
						__lowerCAmelCase: Dict					   =    3_2
						__lowerCAmelCase: Optional[Any]					   =    2
						__lowerCAmelCase: Union[str, Any]					   =    4
						__lowerCAmelCase: Union[str, Any]					   =    3_7
						__lowerCAmelCase: List[str]					   =    "gelu"
						__lowerCAmelCase: List[str]					   =    0.1
						__lowerCAmelCase: List[str]					   =    0.1
						__lowerCAmelCase: Union[str, Any]					   =    5_1_2
						__lowerCAmelCase: int					   =    1_6
						__lowerCAmelCase: int					   =    2
						__lowerCAmelCase: str					   =    0.02
						__lowerCAmelCase: Union[str, Any]					   =    3
						__lowerCAmelCase: str					   =    4
						__lowerCAmelCase: List[Any]					   =    None
						__lowerCAmelCase: int					   =    self.vocab_size - 1

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



						'''simple docstring'''


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

						__lowerCAmelCase: str					   =    None
						if self.use_input_mask:
								__lowerCAmelCase: List[Any]					   =    random_attention_mask([self.batch_size, self.seq_length])

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

						__lowerCAmelCase: Dict					   =    None
						if self.use_mc_token_ids:
								__lowerCAmelCase: Union[str, Any]					   =    ids_tensor([self.batch_size, self.num_choices]  ,						self.seq_length)

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

						__lowerCAmelCase: int					   =    OpenAIGPTConfig(
						    vocab_size=self.vocab_size  ,						n_embd=self.hidden_size  ,						n_layer=self.num_hidden_layers  ,						n_head=self.num_attention_heads  ,						n_positions=self.max_position_embeddings  ,						pad_token_id=self.pad_token_id  ,						)

						__lowerCAmelCase: Union[str, Any]					   =    ids_tensor([self.num_hidden_layers, self.num_attention_heads]  ,						2)

						return (
						    config,
						    input_ids,
						    input_mask,
						    head_mask,
						    token_type_ids,
						    mc_token_ids,
						    sequence_labels,
						    token_labels,
						    choice_labels,
						)

				def 							lowercase_  (			self							:       Optional[int]  ,						UpperCamelCase__							:       int  ,						UpperCamelCase__							:       Any  ,						UpperCamelCase__							:       List[Any]  ,						UpperCamelCase__							:       int  ,						UpperCamelCase__							:       Dict  ,						*UpperCamelCase__							:       Dict)->     Union[str, Any]:



						'''simple docstring'''


						__lowerCAmelCase: List[str]					   =    TFOpenAIGPTModel(config=UpperCamelCase__)
						__lowerCAmelCase: Tuple					   =    {"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}
						__lowerCAmelCase: List[Any]					   =    model(UpperCamelCase__)

						__lowerCAmelCase: List[Any]					   =    [input_ids, input_mask]
						__lowerCAmelCase: Tuple					   =    model(UpperCamelCase__)

						__lowerCAmelCase: List[Any]					   =    model(UpperCamelCase__)

						self.parent.assertEqual(result.last_hidden_state.shape  ,						(self.batch_size, self.seq_length, self.hidden_size))

				def 							lowercase_  (			self							:       Dict  ,						UpperCamelCase__							:       List[str]  ,						UpperCamelCase__							:       Union[str, Any]  ,						UpperCamelCase__							:       List[Any]  ,						UpperCamelCase__							:       Optional[int]  ,						UpperCamelCase__							:       int  ,						*UpperCamelCase__							:       Optional[Any])->     Union[str, Any]:



						'''simple docstring'''


						__lowerCAmelCase: Any					   =    TFOpenAIGPTLMHeadModel(config=UpperCamelCase__)
						__lowerCAmelCase: int					   =    {"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}
						__lowerCAmelCase: Any					   =    model(UpperCamelCase__)
						self.parent.assertEqual(result.logits.shape  ,						(self.batch_size, self.seq_length, self.vocab_size))

				def 							lowercase_  (			self							:       Tuple  ,						UpperCamelCase__							:       str  ,						UpperCamelCase__							:       Union[str, Any]  ,						UpperCamelCase__							:       Tuple  ,						UpperCamelCase__							:       Optional[int]  ,						UpperCamelCase__							:       List[str]  ,						UpperCamelCase__							:       Optional[Any]  ,						*UpperCamelCase__							:       Optional[Any])->     Optional[int]:



						'''simple docstring'''


						__lowerCAmelCase: Union[str, Any]					   =    TFOpenAIGPTDoubleHeadsModel(config=UpperCamelCase__)

						__lowerCAmelCase: List[str]					   =    tf.tile(tf.expand_dims(UpperCamelCase__  ,						1)  ,						(1, self.num_choices, 1))
						__lowerCAmelCase: Union[str, Any]					   =    tf.tile(tf.expand_dims(UpperCamelCase__  ,						1)  ,						(1, self.num_choices, 1))
						__lowerCAmelCase: Optional[Any]					   =    tf.tile(tf.expand_dims(UpperCamelCase__  ,						1)  ,						(1, self.num_choices, 1))

						__lowerCAmelCase: str					   =    {
						    "input_ids": multiple_choice_inputs_ids,
						    "mc_token_ids": mc_token_ids,
						    "attention_mask": multiple_choice_input_mask,
						    "token_type_ids": multiple_choice_token_type_ids,
						}
						__lowerCAmelCase: Optional[int]					   =    model(UpperCamelCase__)
						self.parent.assertEqual(
						    result.logits.shape  ,						(self.batch_size, self.num_choices, self.seq_length, self.vocab_size))
						self.parent.assertEqual(result.mc_logits.shape  ,						(self.batch_size, self.num_choices))

				def 							lowercase_  (			self							:       List[str]  ,						UpperCamelCase__							:       List[Any]  ,						UpperCamelCase__							:       Optional[Any]  ,						UpperCamelCase__							:       Any  ,						UpperCamelCase__							:       str  ,						UpperCamelCase__							:       Optional[Any]  ,						*UpperCamelCase__							:       List[str])->     Any:



						'''simple docstring'''


						__lowerCAmelCase: str					   =    self.num_labels
						__lowerCAmelCase: Dict					   =    ids_tensor([self.batch_size]  ,						self.type_sequence_label_size)
						__lowerCAmelCase: List[str]					   =    {
						    "input_ids": input_ids,
						    "attention_mask": input_mask,
						    "token_type_ids": token_type_ids,
						    "labels": sequence_labels,
						}
						__lowerCAmelCase: str					   =    TFOpenAIGPTForSequenceClassification(UpperCamelCase__)
						__lowerCAmelCase: Optional[int]					   =    model(UpperCamelCase__)
						self.parent.assertEqual(result.logits.shape  ,						(self.batch_size, self.num_labels))




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



						'''simple docstring'''


						__lowerCAmelCase: List[str]					   =    self.prepare_config_and_inputs()

						(
						    (
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				(
						    __lowerCAmelCase
						)   ,				
						): List[Any]					   =    config_and_inputs

						__lowerCAmelCase: List[Any]					   =    {"input_ids": input_ids, "token_type_ids": token_type_ids, "attention_mask": input_mask}
						return config, inputs_dict





@require_tf
class 					snake_case			(       __snake_case,							__snake_case,							unittest.TestCase						):
				SCREAMING_SNAKE_CASE_     :   Dict        =						(
				    (TFOpenAIGPTModel, TFOpenAIGPTLMHeadModel, TFOpenAIGPTDoubleHeadsModel, TFOpenAIGPTForSequenceClassification)
				    if is_tf_available()
				    else ()
				)
				SCREAMING_SNAKE_CASE_     :   List[str]        =						(
				    (TFOpenAIGPTLMHeadModel,) if is_tf_available() else ()
				)  # TODO (PVP): Add Double HeadsModel when generate() function is changed accordingly
				SCREAMING_SNAKE_CASE_     :   int        =						(
				    {
				        """feature-extraction""": TFOpenAIGPTModel,
				        """text-classification""": TFOpenAIGPTForSequenceClassification,
				        """text-generation""": TFOpenAIGPTLMHeadModel,
				        """zero-shot""": TFOpenAIGPTForSequenceClassification,
				    }
				    if is_tf_available()
				    else {}
				)
				SCREAMING_SNAKE_CASE_     :   List[Any]        =						False
				SCREAMING_SNAKE_CASE_     :   Tuple        =						False

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



						'''simple docstring'''


						if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
								# Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
								# `OpenAIGPTConfig` was never used in pipeline tests, either because of a missing checkpoint or because a
								# tiny config could not be created.
								return True

						return False

				def 							lowercase_  (			self							:       str)->     Tuple:



						'''simple docstring'''


						__lowerCAmelCase: Optional[int]					   =    TFOpenAIGPTModelTester(self)
						__lowerCAmelCase: List[Any]					   =    ConfigTester(self  ,						config_class=UpperCamelCase__  ,						n_embd=3_7)

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



						'''simple docstring'''


						self.config_tester.run_common_tests()

				def 							lowercase_  (			self							:       Tuple)->     Optional[int]:



						'''simple docstring'''


						__lowerCAmelCase: Optional[Any]					   =    self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_openai_gpt_model(*UpperCamelCase__)

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



						'''simple docstring'''


						__lowerCAmelCase: Optional[Any]					   =    self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_openai_gpt_lm_head(*UpperCamelCase__)

				def 							lowercase_  (			self							:       Optional[Any])->     Dict:



						'''simple docstring'''


						__lowerCAmelCase: List[Any]					   =    self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_openai_gpt_double_head(*UpperCamelCase__)

				def 							lowercase_  (			self							:       List[str])->     Tuple:



						'''simple docstring'''


						__lowerCAmelCase: Optional[int]					   =    self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_openai_gpt_for_sequence_classification(*UpperCamelCase__)




				@slow
				def 							lowercase_  (			self							:       str)->     Dict:



						'''simple docstring'''


						for model_name in TF_OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
								__lowerCAmelCase: str					   =    TFOpenAIGPTModel.from_pretrained(UpperCamelCase__)
								self.assertIsNotNone(UpperCamelCase__)





@require_tf
class 					snake_case			(       unittest.TestCase						):

				@slow
				def 							lowercase_  (			self							:       Union[str, Any])->     List[str]:



						'''simple docstring'''


						__lowerCAmelCase: Optional[Any]					   =    TFOpenAIGPTLMHeadModel.from_pretrained("openai-gpt")
						__lowerCAmelCase: List[str]					   =    tf.convert_to_tensor([[4_8_1, 4_7_3_5, 5_4_4]]  ,						dtype=tf.intaa)  # the president is
						__lowerCAmelCase: Union[str, Any]					   =    [
						    4_8_1,
						    4_7_3_5,
						    5_4_4,
						    2_4_6,
						    9_6_3,
						    8_7_0,
						    7_6_2,
						    2_3_9,
						    2_4_4,
						    4_0_4_7_7,
						    2_4_4,
						    2_4_9,
						    7_1_9,
						    8_8_1,
						    4_8_7,
						    5_4_4,
						    2_4_0,
						    2_4_4,
						    6_0_3,
						    4_8_1,
						]  # the president is a very good man. " \n " i\'m sure he is, " said the

						__lowerCAmelCase: str					   =    model.generate(UpperCamelCase__  ,						do_sample=UpperCamelCase__)
						self.assertListEqual(output_ids[0].numpy().tolist()  ,						UpperCamelCase__)


