





import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


_SCREAMING_SNAKE_CASE      	=		logging.get_logger(__name__)


_SCREAMING_SNAKE_CASE      	=		{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt"""}

# See all LED models at https://huggingface.co/models?filter=LED
_SCREAMING_SNAKE_CASE      	=		{
    """vocab_file""": {
        """allenai/led-base-16384""": """https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """allenai/led-base-16384""": """https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt""",
    },
    """tokenizer_file""": {
        """allenai/led-base-16384""": """https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json""",
    },
}

_SCREAMING_SNAKE_CASE      	=		{
    """allenai/led-base-16384""": 1_63_84,
}


@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def 							SCREAMING_SNAKE_CASE__						(    ):
	snake_case_		:			Optional[int]											=	(
	    list(range(ord('!'   )   , ord('~'   ) + 1   )   ) + list(range(ord('¡'   )   , ord('¬'   ) + 1   )   ) + list(range(ord('®'   )   , ord('ÿ'   ) + 1   )   )
	)
	snake_case_		:			Optional[int]											=	bs[:]
	snake_case_		:			Any											=	0
	for b in range(2**8   ):
		if b not in bs:
			bs.append(snake_case_   )
			cs.append(2**8 + n   )
			n += 1
	snake_case_		:			Dict											=	[chr(snake_case_   ) for n in cs]
	return dict(zip(snake_case_   , snake_case_   )   )


def 							SCREAMING_SNAKE_CASE__						(    __a   ):
	snake_case_		:			Dict											=	set()
	snake_case_		:			Tuple											=	word[0]
	for char in word[1:]:
		pairs.add((prev_char, char)   )
		snake_case_		:			List[Any]											=	char
	return pairs


class SCREAMING_SNAKE_CASE_						(					UpperCamelCase_					):
						__magic_name__:			Optional[int]       =       VOCAB_FILES_NAMES
						__magic_name__:			List[str]       =       PRETRAINED_VOCAB_FILES_MAP
						__magic_name__:			Union[str, Any]       =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						__magic_name__:			Tuple       =       ["input_ids", "attention_mask"]






						def __init__(      self      :      List[str]		,   _A      :      Any		,   _A      :      List[str]		,   _A      :      Optional[Any]="replace"		,   _A      :      Optional[int]="<s>"		,   _A      :      Union[str, Any]="</s>"		,   _A      :      Tuple="</s>"		,   _A      :      List[Any]="<s>"		,   _A      :      Dict="<unk>"		,   _A      :      Any="<pad>"		,   _A      :      Optional[int]="<mask>"		,   _A      :      List[str]=False		,   **_A      :      Union[str, Any]		,   )							->    List[str]:







							"""simple docstring"""


							snake_case_		:			List[Any]											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else bos_token
							snake_case_		:			List[str]											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else eos_token
							snake_case_		:			Any											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else sep_token
							snake_case_		:			List[Any]											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else cls_token
							snake_case_		:			Tuple											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else unk_token
							snake_case_		:			List[Any]											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else pad_token

							# Mask token behave like a normal word, i.e. include the space before it
							snake_case_		:			List[str]											=	AddedToken(__A		,   lstrip=__A		,   rstrip=__A	) if isinstance(__A		,   __A	) else mask_token

							super().__init__(
							    errors=__A		,   bos_token=__A		,   eos_token=__A		,   unk_token=__A		,   sep_token=__A		,   cls_token=__A		,   pad_token=__A		,   mask_token=__A		,   add_prefix_space=__A		,   **__A		,   )

							with open(__A		,   encoding='utf-8'	) as vocab_handle:
								snake_case_		:			Any											=	json.load(__A	)
							snake_case_		:			Optional[Any]											=	{v: k for k, v in self.encoder.items()}
							snake_case_		:			Union[str, Any]											=	errors  # how to handle errors in decoding
							snake_case_		:			Any											=	bytes_to_unicode()
							snake_case_		:			Optional[Any]											=	{v: k for k, v in self.byte_encoder.items()}
							with open(__A		,   encoding='utf-8'	) as merges_handle:
								snake_case_		:			str											=	merges_handle.read().split('\n'	)[1:-1]
							snake_case_		:			int											=	[tuple(merge.split()	) for merge in bpe_merges]
							snake_case_		:			str											=	dict(zip(__A		,   range(len(__A	)	)	)	)
							snake_case_		:			Optional[int]											=	{}
							snake_case_		:			Any											=	add_prefix_space

							# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
							snake_case_		:			Union[str, Any]											=	re.compile(R'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'	)






						@property
						# Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
						def UpperCAmelCase_				(      self      :      List[Any]	)							->    Optional[Any]:







							"""simple docstring"""


							return len(self.encoder	)






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







							"""simple docstring"""


							return dict(self.encoder		,   **self.added_tokens_encoder	)






						def UpperCAmelCase_				(      self      :      Optional[Any]		,   _A      :      Optional[int]	)							->    str:







							"""simple docstring"""


							if token in self.cache:
								return self.cache[token]
							snake_case_		:			Union[str, Any]											=	tuple(__A	)
							snake_case_		:			List[Any]											=	get_pairs(__A	)

							if not pairs:
								return token

							while True:
								snake_case_		:			Tuple											=	min(__A		,   key=lambda _A	: self.bpe_ranks.get(__A		,   float('inf'	)	)	)
								if bigram not in self.bpe_ranks:
									break
								snake_case_		:			Dict											=	bigram
								snake_case_		:			str											=	[]
								snake_case_		:			Union[str, Any]											=	0
								while i < len(__A	):
									try:
										snake_case_		:			Dict											=	word.index(__A		,   __A	)
									except ValueError:
										new_word.extend(word[i:]	)
										break
									else:
										new_word.extend(word[i:j]	)
										snake_case_		:			str											=	j

									if word[i] == first and i < len(__A	) - 1 and word[i + 1] == second:
										new_word.append(first + second	)
										i += 2
									else:
										new_word.append(word[i]	)
										i += 1
								snake_case_		:			str											=	tuple(__A	)
								snake_case_		:			int											=	new_word
								if len(__A	) == 1:
									break
								else:
									snake_case_		:			List[str]											=	get_pairs(__A	)
							snake_case_		:			List[Any]											=	" ".join(__A	)
							snake_case_		:			Optional[int]											=	word
							return word






						def UpperCAmelCase_				(      self      :      Optional[Any]		,   _A      :      Optional[Any]	)							->    str:







							"""simple docstring"""


							snake_case_		:			List[str]											=	[]
							for token in re.findall(self.pat		,   __A	):
								snake_case_		:			Dict											=	"".join(
								    self.byte_encoder[b] for b in token.encode('utf-8'	)	)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
								bpe_tokens.extend(bpe_token for bpe_token in self.bpe(__A	).split(' '	)	)
							return bpe_tokens






						def UpperCAmelCase_				(      self      :      Union[str, Any]		,   _A      :      Optional[int]	)							->    Dict:







							"""simple docstring"""


							return self.encoder.get(__A		,   self.encoder.get(self.unk_token	)	)






						def UpperCAmelCase_				(      self      :      Optional[int]		,   _A      :      Optional[Any]	)							->    Tuple:







							"""simple docstring"""


							return self.decoder.get(__A	)






						def UpperCAmelCase_				(      self      :      Union[str, Any]		,   _A      :      Dict	)							->    List[Any]:







							"""simple docstring"""


							snake_case_		:			Optional[Any]											=	"".join(__A	)
							snake_case_		:			int											=	bytearray([self.byte_decoder[c] for c in text]	).decode('utf-8'		,   errors=self.errors	)
							return text






						def UpperCAmelCase_				(      self      :      Optional[int]		,   _A      :      str		,   _A      :      Optional[str] = None	)							->    Optional[int]:







							"""simple docstring"""


							if not os.path.isdir(__A	):
								logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""	)
								return
							snake_case_		:			List[Any]											=	os.path.join(
							    __A		,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']	)
							snake_case_		:			str											=	os.path.join(
							    __A		,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']	)

							with open(__A		,   'w'		,   encoding='utf-8'	) as f:
								f.write(json.dumps(self.encoder		,   indent=2		,   sort_keys=__A		,   ensure_ascii=__A	) + '\n'	)

							snake_case_		:			str											=	0
							with open(__A		,   'w'		,   encoding='utf-8'	) as writer:
								writer.write('#version: 0.2\n'	)
								for bpe_tokens, token_index in sorted(self.bpe_ranks.items()		,   key=lambda _A	: kv[1]	):
									if index != token_index:
										logger.warning(
										    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
										    ' Please check that the tokenizer is not corrupted!'	)
										snake_case_		:			int											=	token_index
									writer.write(' '.join(__A	) + '\n'	)
									index += 1

							return vocab_file, merge_file






						def UpperCAmelCase_				(      self      :      int		,   _A      :      List[int]		,   _A      :      Optional[List[int]] = None	)							->    Any:







							"""simple docstring"""


							if token_ids_a is None:
								return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
							snake_case_		:			Tuple											=	[self.cls_token_id]
							snake_case_		:			List[Any]											=	[self.sep_token_id]
							return cls + token_ids_a + sep + sep + token_ids_a + sep






						def UpperCAmelCase_				(      self      :      Optional[Any]		,   _A      :      List[int]		,   _A      :      Optional[List[int]] = None		,   _A      :      bool = False	)							->    int:







							"""simple docstring"""


							if already_has_special_tokens:
								return super().get_special_tokens_mask(
								    token_ids_a=__A		,   token_ids_a=__A		,   already_has_special_tokens=__A	)

							if token_ids_a is None:
								return [1] + ([0] * len(__A	)) + [1]
							return [1] + ([0] * len(__A	)) + [1, 1] + ([0] * len(__A	)) + [1]






						def UpperCAmelCase_				(      self      :      List[Any]		,   _A      :      List[int]		,   _A      :      Optional[List[int]] = None	)							->    List[Any]:







							"""simple docstring"""


							snake_case_		:			Any											=	[self.sep_token_id]
							snake_case_		:			List[Any]											=	[self.cls_token_id]

							if token_ids_a is None:
								return len(cls + token_ids_a + sep	) * [0]
							return len(cls + token_ids_a + sep + sep + token_ids_a + sep	) * [0]






						def UpperCAmelCase_				(      self      :      Optional[Any]		,   _A      :      int		,   _A      :      int=False		,   **_A      :      Dict	)							->    Any:







							"""simple docstring"""


							snake_case_		:			Optional[int]											=	kwargs.pop('add_prefix_space'		,   self.add_prefix_space	)
							if (is_split_into_words or add_prefix_space) and (len(__A	) > 0 and not text[0].isspace()):
								snake_case_		:			Optional[int]											=	" " + text
							return (text, kwargs)






						def UpperCAmelCase_				(      self      :      Any		,   _A      :      Union[Dict[str, EncodedInput], BatchEncoding]		,   _A      :      Optional[int] = None		,   _A      :      PaddingStrategy = PaddingStrategy.DO_NOT_PAD		,   _A      :      Optional[int] = None		,   _A      :      Optional[bool] = None		,   )							->    int:







							"""simple docstring"""


							snake_case_		:			Optional[Any]											=	super()._pad(
							    encoded_inputs=__A		,   max_length=__A		,   padding_strategy=__A		,   pad_to_multiple_of=__A		,   return_attention_mask=__A		,   )

							# Load from model defaults
							if return_attention_mask is None:
								snake_case_		:			Union[str, Any]											=	"attention_mask" in self.model_input_names

							if return_attention_mask and "global_attention_mask" in encoded_inputs:
								snake_case_		:			Union[str, Any]											=	encoded_inputs[self.model_input_names[0]]
								# `global_attention_mask` need to have the same length as other (sequential) inputs.
								snake_case_		:			Tuple											=	len(encoded_inputs['global_attention_mask']	) != len(__A	)

								if needs_to_be_padded:
									snake_case_		:			int											=	len(__A	) - len(encoded_inputs['global_attention_mask']	)

									if self.padding_side == "right":
										# Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
										snake_case_		:			int											=	(
										    encoded_inputs["global_attention_mask"] + [-1] * difference
										)
									elif self.padding_side == "left":
										snake_case_		:			Tuple											=	[-1] * difference + encoded_inputs[
										    "global_attention_mask"
										]
									else:
										raise ValueError('Invalid padding strategy:' + str(self.padding_side	)	)

							return encoded_inputs








import argparse
import json

import numpy
import torch

from transformers.models.xlm.tokenization_xlm import VOCAB_FILES_NAMES
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


logging.set_verbosity_info()


def 							SCREAMING_SNAKE_CASE__						(    __a   , __a   ):
	# Load checkpoint
	snake_case_		:			Union[str, Any]											=	torch.load(__a   , map_location='cpu'   )

	snake_case_		:			Union[str, Any]											=	chkpt['model']

	# We have the base model one level deeper than the original XLM repository
	snake_case_		:			str											=	{}
	for k, v in state_dict.items():
		if "pred_layer" in k:
			snake_case_		:			Tuple											=	v
		else:
			snake_case_		:			Dict											=	v

	snake_case_		:			Tuple											=	chkpt['params']
	snake_case_		:			List[Any]											=	{n: v for n, v in config.items() if not isinstance(__a   , (torch.FloatTensor, numpy.ndarray)   )}

	snake_case_		:			Optional[int]											=	chkpt['dico_word2id']
	snake_case_		:			List[str]											=	{s + '</w>' if s.find('@@'   ) == -1 and i > 13 else s.replace('@@'   , ''   ): i for s, i in vocab.items()}

	# Save pytorch-model
	snake_case_		:			List[str]											=	pytorch_dump_folder_path + '/' + WEIGHTS_NAME
	snake_case_		:			Dict											=	pytorch_dump_folder_path + '/' + CONFIG_NAME
	snake_case_		:			Any											=	pytorch_dump_folder_path + '/' + VOCAB_FILES_NAMES['vocab_file']

	print(f"""Save PyTorch model to {pytorch_weights_dump_path}"""   )
	torch.save(__a   , __a   )

	print(f"""Save configuration file to {pytorch_config_dump_path}"""   )
	with open(__a   , 'w'   , encoding='utf-8'   ) as f:
		f.write(json.dumps(__a   , indent=2   ) + '\n'   )

	print(f"""Save vocab file to {pytorch_config_dump_path}"""   )
	with open(__a   , 'w'   , encoding='utf-8'   ) as f:
		f.write(json.dumps(__a   , indent=2   ) + '\n'   )


if __name__ == "__main__":
						_SCREAMING_SNAKE_CASE      	=		argparse.ArgumentParser()
						# Required parameters
						parser.add_argument(
						    """--xlm_checkpoint_path""", default=None, type=str, required=True, help="""Path the official PyTorch dump."""
						)
						parser.add_argument(
						    """--pytorch_dump_folder_path""", default=None, type=str, required=True, help="""Path to the output PyTorch model."""
						)
						_SCREAMING_SNAKE_CASE      	=		parser.parse_args()
						convert_xlm_checkpoint_to_pytorch(args.xlm_checkpoint_path, args.pytorch_dump_folder_path)

