






import unittest

from transformers.testing_utils import is_torch_available, require_torch, tooslow

from ...generation.test_utils import 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 (
		    CpmAntConfig,
		    CpmAntForCausalLM,
		    CpmAntModel,
		    CpmAntTokenizer,
		)





@require_torch
class  snake_case__      :
	def __init__(       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__=2    ,     lowerCAmelCase__=8    ,     lowerCAmelCase__=True    ,     lowerCAmelCase__=False    ,     lowerCAmelCase__=False    ,     lowerCAmelCase__=False    ,     lowerCAmelCase__=False    ,     lowerCAmelCase__=99    ,     lowerCAmelCase__=32    ,     lowerCAmelCase__=3    ,     lowerCAmelCase__=4    ,     lowerCAmelCase__=37    ,     lowerCAmelCase__=32    ,     lowerCAmelCase__=1_28    ,     lowerCAmelCase__=8    ,     lowerCAmelCase__=8    ,     lowerCAmelCase__=8    ,     lowerCAmelCase__=1.0    ,     lowerCAmelCase__=True    ,     )							->   Optional[Any]:
							__magic_name__    :     List[Any]     		=      parent
							__magic_name__    :     Optional[Any]     		=      batch_size
							__magic_name__    :     Dict     		=      seq_length
							__magic_name__    :     List[Any]     		=      is_training
							__magic_name__    :     Tuple     		=      use_token_type_ids
							__magic_name__    :     Optional[int]     		=      use_input_mask
							__magic_name__    :     Optional[Any]     		=      use_labels
							__magic_name__    :     int     		=      use_mc_token_ids
							__magic_name__    :     Union[str, Any]     		=      vocab_size
							__magic_name__    :     str     		=      hidden_size
							__magic_name__    :     Union[str, Any]     		=      num_hidden_layers
							__magic_name__    :     List[str]     		=      num_attention_heads
							__magic_name__    :     Optional[Any]     		=      intermediate_size
							__magic_name__    :     Tuple     		=      num_buckets
							__magic_name__    :     Any     		=      max_distance
							__magic_name__    :     List[Any]     		=      prompt_length
							__magic_name__    :     int     		=      prompt_types
							__magic_name__    :     Optional[Any]     		=      segment_types
							__magic_name__    :     Optional[int]     		=      init_std
							__magic_name__    :     str     		=      return_dict
	def     __magic_name__       (       self				)							->   Tuple:
							__magic_name__    :     List[str]     		=      {}
							__magic_name__    :     Optional[int]     		=      ids_tensor([self.batch_size, self.seq_length]    ,     self.vocab_size				).type(torch.intaa				)
							__magic_name__    :     Tuple     		=      False

							__magic_name__    :     Union[str, Any]     		=      self.get_config()

							return (config, input_ids)
	def     __magic_name__       (       self				)							->   Dict:
							return CpmAntConfig(
							    vocab_size=self.vocab_size    ,     hidden_size=self.hidden_size    ,     num_hidden_layers=self.num_hidden_layers    ,     num_attention_heads=self.num_attention_heads    ,     dim_ff=self.intermediate_size    ,     position_bias_num_buckets=self.num_buckets    ,     position_bias_max_distance=self.max_distance    ,     prompt_types=self.prompt_types    ,     prompt_length=self.prompt_length    ,     segment_types=self.segment_types    ,     use_cache=lowerCAmelCase__    ,     init_std=self.init_std    ,     return_dict=self.return_dict    ,     )
	def     __magic_name__       (       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     *lowerCAmelCase__				)							->   str:
							__magic_name__    :     Dict     		=      CpmAntModel(config=lowerCAmelCase__				)
							model.to(lowerCAmelCase__				)
							model.eval()

							__magic_name__    :     str     		=      model(**lowerCAmelCase__				).last_hidden_state

							self.parent.assertEqual(hidden_states.shape    ,     (self.batch_size, self.seq_length, config.hidden_size)				)
	def     __magic_name__       (       self    ,     lowerCAmelCase__    ,     lowerCAmelCase__    ,     *lowerCAmelCase__				)							->   Tuple:
							__magic_name__    :     Any     		=      CpmAntForCausalLM(lowerCAmelCase__				)
							model.to(lowerCAmelCase__				)
							__magic_name__    :     Any     		=      input_ids["""input_ids"""].to(lowerCAmelCase__				)
							model.eval()

							__magic_name__    :     Union[str, Any]     		=      model(**lowerCAmelCase__				)
							self.parent.assertEqual(
							    model_output.logits.shape    ,     (self.batch_size, self.seq_length, config.vocab_size + config.prompt_types * config.prompt_length)    ,     )







	def     __magic_name__       (       self				)							->   Union[str, Any]:
							__magic_name__		,__magic_name__    :     Optional[int]     		=      self.prepare_config_and_inputs()
							return config, inputs_dict





@require_torch
class  snake_case__      (							_lowerCAmelCase		,						_lowerCAmelCase		,						unittest.TestCase   ):
	lowercase__		:  Dict							   =					(CpmAntModel, CpmAntForCausalLM) if is_torch_available() else ()
	lowercase__		:  Tuple							   =					(
	    {'''feature-extraction''': CpmAntModel, '''text-generation''': CpmAntForCausalLM} if is_torch_available() else {}
	)

	lowercase__		:  Dict							   =					False
	lowercase__		:  Optional[Any]							   =					False
	lowercase__		:  Tuple							   =					False
	lowercase__		:  str							   =					False
	lowercase__		:  int							   =					False
	def     __magic_name__       (       self				)							->   int:
							__magic_name__    :     int     		=      CpmAntModelTester(self				)
							__magic_name__    :     List[str]     		=      ConfigTester(self    ,     config_class=lowerCAmelCase__				)
	def     __magic_name__       (       self				)							->   Tuple:
							self.config_tester.create_and_test_config_common_properties()
							self.config_tester.create_and_test_config_to_json_string()
							self.config_tester.create_and_test_config_to_json_file()
							self.config_tester.create_and_test_config_from_and_save_pretrained()
							self.config_tester.check_config_can_be_init_without_params()
							self.config_tester.check_config_arguments_init()
	def     __magic_name__       (       self				)							->   Dict:
							unittest.skip("""CPMAnt doesn't support input_embeds."""				)(self.test_inputs_embeds				)
	def     __magic_name__       (       self				)							->   Dict:
							unittest.skip(
							    """CPMAnt doesn't support retain grad in hidden_states or attentions, because prompt management will peel off the output.hidden_states from graph.\
                 So is attentions. We strongly recommand you use loss to tune model."""				)(self.test_retain_grad_hidden_states_attentions				)
	def     __magic_name__       (       self				)							->   Optional[Any]:
							__magic_name__		,__magic_name__    :     Dict     		=      self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_cpmant_model(lowerCAmelCase__    ,     lowerCAmelCase__				)







	def     __magic_name__       (       self				)							->   Union[str, Any]:
							__magic_name__		,__magic_name__    :     List[Any]     		=      self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_lm_head_model(lowerCAmelCase__    ,     lowerCAmelCase__				)





@require_torch
class  snake_case__      (							unittest.TestCase   ):
	@tooslow
	def     __magic_name__       (       self				)							->   int:
							__magic_name__    :     Optional[Any]     		=      """今天天气真好！"""
							__magic_name__    :     str     		=      """openbmb/cpm-ant-10b"""
							__magic_name__    :     List[str]     		=      CpmAntModel.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Tuple     		=      CpmAntTokenizer.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Optional[int]     		=      tokenizer(lowerCAmelCase__    ,     return_tensors="""pt"""				)
							__magic_name__    :     Any     		=      model(**lowerCAmelCase__				).last_hidden_state

							__magic_name__    :     Tuple     		=      torch.tensor(
							    [[[6.1_7_0_8, 5.9_2_4_4, 1.0_8_3_5], [6.5_2_0_7, 6.2_8_9_3, -1_1.3_3_2_4], [-1.0_1_0_7, -0.0_5_7_6, -5.9_5_7_7]]]    ,     )
							self.assertTrue(torch.allclose(hidden_states[:, :3, :3]    ,     lowerCAmelCase__    ,     atol=1e-2				)				)





@require_torch
class  snake_case__      (							unittest.TestCase   ):
	@tooslow
	def     __magic_name__       (       self				)							->   List[Any]:
							__magic_name__    :     List[str]     		=      """今天天气真好！"""
							__magic_name__    :     Any     		=      """openbmb/cpm-ant-10b"""
							__magic_name__    :     Optional[Any]     		=      CpmAntForCausalLM.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     int     		=      CpmAntTokenizer.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Optional[Any]     		=      tokenizer(lowerCAmelCase__    ,     return_tensors="""pt"""				)
							__magic_name__    :     Optional[Any]     		=      model(**lowerCAmelCase__				).logits

							__magic_name__    :     List[str]     		=      torch.tensor(
							    [[[-6.4_2_6_7, -6.4_0_8_3, -6.3_9_5_8], [-5.8_8_0_2, -5.9_4_4_7, -5.7_8_1_1], [-5.3_8_9_6, -5.4_8_2_0, -5.4_2_9_5]]]    ,     )
							self.assertTrue(torch.allclose(hidden_states[:, :3, :3]    ,     lowerCAmelCase__    ,     atol=1e-2				)				)
	@tooslow
	def     __magic_name__       (       self				)							->   int:
							__magic_name__    :     List[Any]     		=      """openbmb/cpm-ant-10b"""
							__magic_name__    :     Tuple     		=      CpmAntForCausalLM.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Optional[int]     		=      CpmAntTokenizer.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Dict     		=      """今天天气不错，"""
							__magic_name__    :     Union[str, Any]     		=      """今天天气不错，阳光明媚，我和妈妈一起去超市买东西。\n在超市里，我看到了一个很好玩的玩具，它的名字叫“机器人”。它有一个圆圆的脑袋，两只圆圆的眼睛，还有一个圆圆的"""
							__magic_name__    :     int     		=      tokenizer(lowerCAmelCase__    ,     return_tensors="""pt"""				)
							__magic_name__    :     Union[str, Any]     		=      model.generate(**lowerCAmelCase__				)
							__magic_name__    :     Optional[int]     		=      tokenizer.batch_decode(lowerCAmelCase__				)
							self.assertEqual(lowerCAmelCase__    ,     lowerCAmelCase__				)







	@tooslow
	def     __magic_name__       (       self				)							->   Tuple:
							__magic_name__    :     Optional[Any]     		=      """openbmb/cpm-ant-10b"""
							__magic_name__    :     List[str]     		=      CpmAntForCausalLM.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Tuple     		=      CpmAntTokenizer.from_pretrained(lowerCAmelCase__				)
							__magic_name__    :     Dict     		=      ["""今天天气不错，""", """新年快乐，万事如意！"""]
							__magic_name__    :     Tuple     		=      [
							    """今天天气不错，阳光明媚，我和妈妈一起去超市买东西。\n在超市里，我看到了一个很好玩的玩具，它的名字叫“机器人”。它有一个圆圆的脑袋，两只圆圆的眼睛，还有一个圆圆的""",
							    """新年快乐，万事如意！在这辞旧迎新的美好时刻，我谨代表《农村新技术》杂志社全体同仁，向一直以来关心、支持《农村新技术》杂志发展的各级领导、各界朋友和广大读者致以最诚挚的""",
							]
							__magic_name__    :     Union[str, Any]     		=      tokenizer(lowerCAmelCase__    ,     return_tensors="""pt"""    ,     padding=lowerCAmelCase__				)
							__magic_name__    :     Union[str, Any]     		=      model.generate(**lowerCAmelCase__				)
							__magic_name__    :     List[str]     		=      tokenizer.batch_decode(lowerCAmelCase__				)
							self.assertEqual(lowerCAmelCase__    ,     lowerCAmelCase__				)








__magic_name__:  Tuple  	=			[
    ["ab-e1", "ac-e3", "ad-e5", "bc-e4", "bd-e2", "be-e6", "bh-e12", "cd-e2", "ce-e4",
     "de-e1", "df-e8", "dg-e5", "dh-e10", "ef-e3", "eg-e2", "fg-e6", "gh-e6", "hi-e3"],
    ["ab-e1", "ac-e3", "ad-e5", "bc-e4", "bd-e2", "be-e6", "cd-e2", "de-e1", "df-e8",
     "ef-e3", "eg-e2", "fg-e6"],
    ["ab-e1", "ac-e3", "bc-e4", "bd-e2", "de-e1", "df-e8", "dg-e5", "ef-e3", "eg-e2",
     "eh-e12", "fg-e6", "fh-e10", "gh-e6"],
    ["ab-e1", "ac-e3", "bc-e4", "bd-e2", "bh-e12", "cd-e2", "df-e8", "dh-e10"],
    ["ab-e1", "ac-e3", "ad-e5", "bc-e4", "bd-e2", "cd-e2", "ce-e4", "de-e1", "df-e8",
     "dg-e5", "ef-e3", "eg-e2", "fg-e6"]
]

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						__magic_name__    :     str     		=      set()
						for row in edge_array:
												for item in row:
																		distinct_edge.add(item[0]				)
						return list(_A				)

def 					UpperCamelCase		( _A,					_A				):






						"""simple docstring"""







						__magic_name__    :     Union[str, Any]     		=      ["""0"""] * len(_A				)
						for i, row in enumerate(_A				):
												for item in row:
																		if distinct_edge in item[0]:
																								__magic_name__    :     Optional[Any]     		=      """1"""
																								break
						return "".join(_A				)

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						__magic_name__    :     List[str]     		=      get_distinct_edge(_A				)
						__magic_name__    :     List[str]     		=      {}

						for item in distinct_edge:
												__magic_name__    :     Optional[Any]     		=      get_bitcode(_A,					_A				)
												# print('bit',bit)
												# bt=''.join(bit)
												__magic_name__    :     Union[str, Any]     		=      bit.count("""1"""				)
												__magic_name__    :     Dict     		=      [s, bit]
						# Store [Distinct edge, WT(Bitcode), Bitcode] in descending order
						__magic_name__    :     List[Any]     		=      [
						    [k, v[0], v[1]]
						    for k, v in sorted(frequency_table.items(),					key=lambda _A				: v[1][0],					reverse=_A				)
						]
						return sorted_frequency_table

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						__magic_name__    :     str     		=      {}
						for _, item in enumerate(_A				):
												nodes.setdefault(item[2],					[]				).append(item[0]				)
						return nodes

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						__magic_name__    :     Optional[int]     		=      {}
						for key, value in nodes.items():
												__magic_name__    :     Tuple     		=      value
						return cluster

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						return [i * 100 / len(_A				) for i in cluster]

def 					UpperCamelCase		( ):






						"""simple docstring"""







						print("""\nNodes\n"""				)
						for key, value in nodes.items():
												print(_A,					_A				)
						print("""\nSupport\n"""				)
						print(_A				)
						print("""\n Cluster \n"""				)
						for key, value in sorted(cluster.items(),					reverse=_A				):
												print(_A,					_A				)
						print("""\n Graph\n"""				)
						for key, value in graph.items():
												print(_A,					_A				)
						print("""\n Edge List of Frequent subgraphs \n"""				)
						for edge_list in freq_subgraph_edge_list:
												print(_A				)

def 					UpperCamelCase		( _A,					_A,					_A,					_A				):






						"""simple docstring"""







						for i in cluster[ca]:
												__magic_name__    :     Any     		=      0
												__magic_name__    :     Any     		=      ca + 1
												while ca < max(cluster.keys()				):
																		for j in cluster[ca]:
																								if int(_A,					2				) & int(_A,					2				) == int(_A,					2				):
																														if tuple(nodes[i]				) in graph:
																																				graph[tuple(nodes[i]				)].append(nodes[j]				)
																														else:
																																				__magic_name__    :     int     		=      [nodes[j]]
																														count += 1
																		if count == 0:
																								__magic_name__    :     Optional[Any]     		=      ca + 1
																		else:
																								break

def 					UpperCamelCase		( _A,					_A				):






						"""simple docstring"""







						__magic_name__    :     List[str]     		=      cluster[max(cluster.keys()				)]
						__magic_name__    :     List[Any]     		=      """Header"""
						__magic_name__    :     Tuple     		=      {}
						for i in x:
												if (["Header"],) in graph:
																		graph[(["Header"],)].append(x[i]				)
												else:
																		__magic_name__    :     List[str]     		=      [x[i]]
						for i in x:
												__magic_name__    :     Tuple     		=      [["""Header"""]]
						__magic_name__    :     Union[str, Any]     		=      1
						while i < max(_A				) - 1:
												create_edge(_A,					_A,					_A,					_A				)
												__magic_name__    :     Optional[Any]     		=      i + 1
						return graph

def 					UpperCamelCase		( _A,					_A,					_A,					_A=None				):






						"""simple docstring"""







						__magic_name__    :     str     		=      (path or []) + [start]
						if start == end:
												paths.append(_A				)
						for node in graph[start]:
												if tuple(_A				) not in path:
																		my_dfs(_A,					tuple(_A				),					_A,					_A				)

def 					UpperCamelCase		( _A,					_A,					_A				):






						"""simple docstring"""







						__magic_name__    :     Union[str, Any]     		=      int(s / 100 * (len(_A				) - 1)				)
						for i in cluster[k]:
												my_dfs(_A,					tuple(cluster[k][i]				),					(["""Header"""],)				)

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						__magic_name__    :     str     		=      []
						for edges in paths:
												__magic_name__    :     List[Any]     		=      []
												for j in range(len(_A				) - 1				):
																		__magic_name__    :     Any     		=      list(edges[j]				)
																		for e in temp:
																								__magic_name__    :     Optional[Any]     		=      (e[0], e[1])
																								el.append(_A				)
												freq_sub_el.append(_A				)
						return freq_sub_el

def 					UpperCamelCase		( _A				):






						"""simple docstring"""







						for i in range(len(_A				)				):
												for j in range(len(edge_array[i]				)				):
																		__magic_name__    :     Any     		=      edge_array[i][j].split("""-"""				)
																		__magic_name__    :     Optional[int]     		=      t


if __name__ == "__main__":
		preprocess(edge_array)
		__magic_name__:  Optional[int]  	=			get_frequency_table(edge_array)
		__magic_name__:  str  	=			get_nodes(frequency_table)
		__magic_name__:  List[str]  	=			get_cluster(nodes)
		__magic_name__:  Optional[Any]  	=			get_support(cluster)
		__magic_name__:  Dict  	=			construct_graph(cluster, nodes)
		find_freq_subgraph_given_support(60, cluster, graph)
		__magic_name__:  list  	=			[]
		__magic_name__:  Dict  	=			freq_subgraphs_edge_list(paths)
		print_all()
