






"""simple docstring"""
import argparse
import json
import os
from collections import OrderedDict

import numpy as np
import tensorflow as tf
import torch





def 				lowercase__		(					lowerCamelCase       ):
				_SCREAMING_SNAKE_CASE   :		Any       		=    os.path.join(args.tf_model_dir,						'parameters.json'       )
				_SCREAMING_SNAKE_CASE   :		Any       		=    json.loads(open(lowerCamelCase       ).read()       )
				if not params:
								raise ValueError(
								    f"""It seems that the json file at {parameter_file} is empty. Make sure you have a correct json file."""       )
				if not args.output.endswith('.pt'       ):
								_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    args.output + '.pt'
				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    OrderedDict()
				with tf.device('/CPU:0'       ):
								_SCREAMING_SNAKE_CASE   :		Any       		=    tf.train.load_checkpoint(args.tf_model_dir       )
								_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    reader.get_variable_to_shape_map()
								for key_name in shapes.keys():
												_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    reader.get_tensor(lowerCamelCase       ).astype(np.floataa       )
												if key_name.endswith('/adam_m'       ) or key_name.endswith('/adam_v'       ):
																continue
												if key_name.startswith('pasts/'       ):
																if key_name.startswith('pasts/mlp'       ):
																				_SCREAMING_SNAKE_CASE   :		Tuple       		=    int(key_name[9]       )
																elif key_name.startswith('pasts/out'       ):
																				_SCREAMING_SNAKE_CASE   :		Any       		=    8
																_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    'model.sqout.%d.weight' % (player * 2)  # enter to nn.Sequencial with Tanh, so 2 at a time
																_SCREAMING_SNAKE_CASE   :		Tuple       		=    vnp.transpose([1, 0]       ).copy()  # Mesh-Tensorflow is a diagonal matrix
																_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    torch.tensor(lowerCamelCase       )
												elif key_name.startswith('model/moe'       ):
																_SCREAMING_SNAKE_CASE   :		Tuple       		=    int(key_name[9:].split('/'       )[0]       )
																if key_name.endswith('/switch_gating/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		Any       		=    'model.blocks.%d.feed_forward.mlp.router.classifier.weight' % player
																				_SCREAMING_SNAKE_CASE   :		List[str]       		=    vnp.transpose([1, 0]       ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/softmlp/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    'model.blocks.%d.feed_forward.soft_bypass_mlp.weight' % player
																				_SCREAMING_SNAKE_CASE   :		Tuple       		=    vnp.transpose([1, 0]       ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		Any       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/wo/kernel'       ) or key_name.endswith('/wi/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    key_name[-9:-7]
																				for i in range(16       ):
																								_SCREAMING_SNAKE_CASE   :		List[Any]       		=    'model.blocks.%d.feed_forward.mlp.experts.expert_%d.%s.weight' % (player, i, nlayer)
																								_SCREAMING_SNAKE_CASE   :		int       		=    (
																								    vnp[i].transpose([1, 0]       ).copy()
																								)  # In Mesh-Tensorflow, it is one array, so it is divided
																								_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    torch.tensor(lowerCamelCase       )
												elif key_name.startswith('model/mlp'       ):
																_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    int(key_name[9:].split('/'       )[0]       )
																if key_name.endswith('/p1/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    'model.blocks.%d.feed_forward.mlp.wi.weight' % player
																				_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    vnp.transpose([1, 0]       ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		str       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/p1/bias'       ):
																				_SCREAMING_SNAKE_CASE   :		int       		=    'model.blocks.%d.feed_forward.mlp.wi.bias' % player
																				_SCREAMING_SNAKE_CASE   :		Dict       		=    vnp.copy()  # same because it is one dimensional
																				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/p2/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    'model.blocks.%d.feed_forward.mlp.wo.weight' % player
																				_SCREAMING_SNAKE_CASE   :		str       		=    vnp.transpose([1, 0]       ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		Any       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/p2/bias'       ):
																				_SCREAMING_SNAKE_CASE   :		int       		=    'model.blocks.%d.feed_forward.mlp.wo.bias' % player
																				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    vnp.copy()  # same because it is one dimensional
																				_SCREAMING_SNAKE_CASE   :		List[str]       		=    torch.tensor(lowerCamelCase       )
												elif key_name.startswith('model/ln'       ):
																_SCREAMING_SNAKE_CASE   :		Tuple       		=    int(key_name[8:].split('/'       )[0]       )
																if key_name.endswith('/b'       ):
																				_SCREAMING_SNAKE_CASE   :		Any       		=    'model.blocks.%d.feed_forward.norm.bias' % player
																				_SCREAMING_SNAKE_CASE   :		Any       		=    vnp.copy()  # same because it is one dimensional
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/g'       ):
																				_SCREAMING_SNAKE_CASE   :		Tuple       		=    'model.blocks.%d.feed_forward.norm.weight' % player
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    vnp.copy()  # same because it is one dimensional
																				_SCREAMING_SNAKE_CASE   :		Tuple       		=    torch.tensor(lowerCamelCase       )
												elif key_name.startswith('model/att'       ):
																_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    int(key_name[9:].split('/'       )[0]       )
																if key_name.endswith('/qkv/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    vnp.copy()  # Compute same dimension as Mesh-tensorflow using einsum
																				_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    state[:, 0, :, :]
																				_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    state[:, 1, :, :]
																				_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    state[:, 2, :, :]
																				_SCREAMING_SNAKE_CASE   :		int       		=    (
																				    state_q.reshape([state_q.shape[0], state_q.shape[1] * state_q.shape[2]]       )
																				    .transpose([1, 0]       )
																				    .copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    (
																				    state_k.reshape([state_k.shape[0], state_k.shape[1] * state_k.shape[2]]       )
																				    .transpose([1, 0]       )
																				    .copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		str       		=    (
																				    state_v.reshape([state_v.shape[0], state_v.shape[1] * state_v.shape[2]]       )
																				    .transpose([1, 0]       )
																				    .copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    'model.blocks.%d.self_attn.self_attn.q_proj.weight' % player
																				_SCREAMING_SNAKE_CASE   :		Dict       		=    torch.tensor(lowerCamelCase       )
																				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    'model.blocks.%d.self_attn.self_attn.k_proj.weight' % player
																				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    torch.tensor(lowerCamelCase       )
																				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    'model.blocks.%d.self_attn.self_attn.v_proj.weight' % player
																				_SCREAMING_SNAKE_CASE   :		str       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/o/kernel'       ):
																				_SCREAMING_SNAKE_CASE   :		Any       		=    'model.blocks.%d.self_attn.self_attn.out_proj.weight' % player
																				_SCREAMING_SNAKE_CASE   :		Any       		=    (
																				    vnp.reshape([vnp.shape[0] * vnp.shape[1], vnp.shape[2]]       ).transpose([1, 0]       ).copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    torch.tensor(lowerCamelCase       )
												elif key_name.startswith('model/an'       ):
																_SCREAMING_SNAKE_CASE   :		Any       		=    int(key_name[8:].split('/'       )[0]       )
																if key_name.endswith('/b'       ):
																				_SCREAMING_SNAKE_CASE   :		Any       		=    'model.blocks.%d.self_attn.norm.bias' % player
																				_SCREAMING_SNAKE_CASE   :		int       		=    vnp.copy()  # same because it is one dimensional
																				_SCREAMING_SNAKE_CASE   :		List[str]       		=    torch.tensor(lowerCamelCase       )
																elif key_name.endswith('/g'       ):
																				_SCREAMING_SNAKE_CASE   :		Dict       		=    'model.blocks.%d.self_attn.norm.weight' % player
																				_SCREAMING_SNAKE_CASE   :		Dict       		=    vnp.copy()  # same because it is one dimensional
																				_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    torch.tensor(lowerCamelCase       )
												elif (
												    key_name.startswith('model/wte'       )
												    or key_name.startswith('model/wpe'       )
												    or key_name.startswith('model/ete'       )
												):
																_SCREAMING_SNAKE_CASE   :		Dict       		=    {'wte': 'embed_tokens', 'wpe': 'position_embeddings', 'ete': 'extra_position_embeddings'}[
																    key_name[-3:]
																]
																_SCREAMING_SNAKE_CASE   :		Tuple       		=    'model.%s.weight' % nlayer
																_SCREAMING_SNAKE_CASE   :		List[Any]       		=    vnp.copy()  # same in embedded
																_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    torch.tensor(lowerCamelCase       )
																if key_name.startswith('model/wte'       ):
																				_SCREAMING_SNAKE_CASE   :		int       		=    'lm_head.weight'
																				_SCREAMING_SNAKE_CASE   :		List[Any]       		=    vnp.copy()  # same in embedded
																				_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    torch.tensor(lowerCamelCase       )
												elif key_name.startswith('model/wob'       ):
																_SCREAMING_SNAKE_CASE   :		int       		=    'final_logits_bias'
																_SCREAMING_SNAKE_CASE   :		int       		=    vnp.copy()  # same in embedded
																_SCREAMING_SNAKE_CASE   :		int       		=    state.reshape((1, -1)       )
																_SCREAMING_SNAKE_CASE   :		Tuple       		=    torch.tensor(lowerCamelCase       )
												elif key_name == "model/dense/kernel":
																_SCREAMING_SNAKE_CASE   :		List[Any]       		=    'model.last_project.weight'
																_SCREAMING_SNAKE_CASE   :		Any       		=    vnp.transpose([1, 0]       ).copy()  # Mesh-Tensorflow is a diagonal matrix
																_SCREAMING_SNAKE_CASE   :		List[Any]       		=    torch.tensor(lowerCamelCase       )
												elif key_name == "model/dense_1/bias":
																_SCREAMING_SNAKE_CASE   :		List[str]       		=    'model.last_project.bias'
																_SCREAMING_SNAKE_CASE   :		Tuple       		=    vnp.copy()  # same because it is one dimensional
																_SCREAMING_SNAKE_CASE   :		str       		=    torch.tensor(lowerCamelCase       )
				torch.save(lowerCamelCase,						args.output       )


if __name__ == "__main__":
			lowerCAmelCase__    						=			argparse.ArgumentParser(
			    description='model converter.', formatter_class=argparse.ArgumentDefaultsHelpFormatter
			)
			parser.add_argument('--tf_model_dir', metavar='PATH', type=str, required=True, help='import model')
			parser.add_argument('--output', metavar='PATH', type=str, required=True, help='output model')
			lowerCAmelCase__    						=			parser.parse_args()
			convert_tf_gptsan_to_pt(args)









"""simple docstring"""
class _lowerCAmelCase						:







						def __init__(    self   ,				lowerCAmelCase_   ,				lowerCAmelCase_=None   ,				lowerCAmelCase_=None							)		->							int:
										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    data
										_SCREAMING_SNAKE_CASE   :		Tuple       		=    previous
										_SCREAMING_SNAKE_CASE   :		Any       		=    next_node







						def __str__(    self							)		->							str:
										return F"""{self.data}"""







						def 			A      (    self							)		->							int:
										return self.data







						def 			A      (    self							)		->							Dict:
										return self.next







						def 			A      (    self							)		->							Union[str, Any]:
										return self.previous







class _lowerCAmelCase						:







						def __init__(    self   ,				lowerCAmelCase_							)		->							List[str]:
										_SCREAMING_SNAKE_CASE   :		List[Any]       		=    head







						def __iter__(    self							)		->							Any:
										return self







						def 			A      (    self							)		->							Dict:
										if not self.current:
														raise StopIteration
										else:
														_SCREAMING_SNAKE_CASE   :		Any       		=    self.current.get_data()
														_SCREAMING_SNAKE_CASE   :		Dict       		=    self.current.get_next()
														return value







class _lowerCAmelCase						:







						def __init__(    self							)		->							Dict:
										_SCREAMING_SNAKE_CASE   :		Dict       		=    None  # First node in list
										_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    None  # Last node in list







						def __str__(    self							)		->							Any:
										_SCREAMING_SNAKE_CASE   :		str       		=    self.head
										_SCREAMING_SNAKE_CASE   :		int       		=    []
										while current is not None:
														nodes.append(current.get_data()							)
														_SCREAMING_SNAKE_CASE   :		List[Any]       		=    current.get_next()
										return " ".join(str(lowerCAmelCase_							) for node in nodes							)







						def __contains__(    self   ,				lowerCAmelCase_							)		->							Any:
										_SCREAMING_SNAKE_CASE   :		Any       		=    self.head
										while current:
														if current.get_data() == value:
																		return True
														_SCREAMING_SNAKE_CASE   :		int       		=    current.get_next()
										return False







						def __iter__(    self							)		->							str:
										return LinkedListIterator(self.head							)







						def 			A      (    self							)		->							Dict:
										if self.head:
														return self.head.get_data()
										return None







						def 			A      (    self							)		->							List[Any]:
										if self.tail:
														return self.tail.get_data()
										return None







						def 			A      (    self   ,				lowerCAmelCase_							)		->							None:
										if self.head is None:
														_SCREAMING_SNAKE_CASE   :		List[Any]       		=    node
														_SCREAMING_SNAKE_CASE   :		Dict       		=    node
										else:
														self.insert_before_node(self.head   ,				lowerCAmelCase_							)







						def 			A      (    self   ,				lowerCAmelCase_							)		->							None:
										if self.head is None:
														self.set_head(lowerCAmelCase_							)
										else:
														self.insert_after_node(self.tail   ,				lowerCAmelCase_							)







						def 			A      (    self   ,				lowerCAmelCase_							)		->							None:
										_SCREAMING_SNAKE_CASE   :		Any       		=    Node(lowerCAmelCase_							)
										if self.head is None:
														self.set_head(lowerCAmelCase_							)
										else:
														self.set_tail(lowerCAmelCase_							)







						def 			A      (    self   ,				lowerCAmelCase_   ,				lowerCAmelCase_							)		->							None:
										_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    node
										_SCREAMING_SNAKE_CASE   :		List[Any]       		=    node.previous

										if node.get_previous() is None:
														_SCREAMING_SNAKE_CASE   :		List[Any]       		=    node_to_insert
										else:
														_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    node_to_insert

										_SCREAMING_SNAKE_CASE   :		Any       		=    node_to_insert







						def 			A      (    self   ,				lowerCAmelCase_   ,				lowerCAmelCase_							)		->							None:
										_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    node
										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    node.next

										if node.get_next() is None:
														_SCREAMING_SNAKE_CASE   :		Tuple       		=    node_to_insert
										else:
														_SCREAMING_SNAKE_CASE   :		Any       		=    node_to_insert

										_SCREAMING_SNAKE_CASE   :		List[str]       		=    node_to_insert







						def 			A      (    self   ,				lowerCAmelCase_   ,				lowerCAmelCase_							)		->							None:
										_SCREAMING_SNAKE_CASE   :		List[Any]       		=    1
										_SCREAMING_SNAKE_CASE   :		str       		=    Node(lowerCAmelCase_							)
										_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    self.head
										while node:
														if current_position == position:
																		self.insert_before_node(lowerCAmelCase_   ,				lowerCAmelCase_							)
																		return
														current_position += 1
														_SCREAMING_SNAKE_CASE   :		List[str]       		=    node.next
										self.insert_after_node(self.tail   ,				lowerCAmelCase_							)







						def 			A      (    self   ,				lowerCAmelCase_							)		->							Node:
										_SCREAMING_SNAKE_CASE   :		Tuple       		=    self.head
										while node:
														if node.get_data() == item:
																		return node
														_SCREAMING_SNAKE_CASE   :		Dict       		=    node.get_next()
										raise Exception('Node not found'							)







						def 			A      (    self   ,				lowerCAmelCase_							)		->							int:
										if (node := self.get_node(lowerCAmelCase_							)) is not None:
														if node == self.head:
																		_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    self.head.get_next()

														if node == self.tail:
																		_SCREAMING_SNAKE_CASE   :		int       		=    self.tail.get_previous()

														self.remove_node_pointers(lowerCAmelCase_							)







						@staticmethod
						def 			A      (    lowerCAmelCase_							)		->							None:
										if node.get_next():
														_SCREAMING_SNAKE_CASE   :		List[Any]       		=    node.previous

										if node.get_previous():
														_SCREAMING_SNAKE_CASE   :		str       		=    node.next

										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    None
										_SCREAMING_SNAKE_CASE   :		List[Any]       		=    None







						def 			A      (    self							)		->							List[str]:
										return self.head is None







def 				lowercase__		(					):
				pass


if __name__ == "__main__":
			import doctest

			doctest.testmod()

