

import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import BatchEncoding, PreTrainedTokenizer
from ...utils import logging


A_       : Dict         =			logging.get_logger(__name__)

A_       : Tuple         =			'▁'

A_       : Optional[Any]         =			{
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
    'tokenizer_config_file': 'tokenizer_config.json',
}

A_       : Optional[Any]         =			{
    'vocab_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json',
    },
    'spm_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model',
    },
    'tokenizer_config_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json',
    },
}

A_       : List[Any]         =			{
    'facebook/m2m100_418M': 1024,
}

# fmt: off
A_       : int         =			{
    'm2m100': ['af', 'am', 'ar', 'ast', 'az', 'ba', 'be', 'bg', 'bn', 'br', 'bs', 'ca', 'ceb', 'cs', 'cy', 'da', 'de', 'el', 'en', 'es', 'et', 'fa', 'ff', 'fi', 'fr', 'fy', 'ga', 'gd', 'gl', 'gu', 'ha', 'he', 'hi', 'hr', 'ht', 'hu', 'hy', 'id', 'ig', 'ilo', 'is', 'it', 'ja', 'jv', 'ka', 'kk', 'km', 'kn', 'ko', 'lb', 'lg', 'ln', 'lo', 'lt', 'lv', 'mg', 'mk', 'ml', 'mn', 'mr', 'ms', 'my', 'ne', 'nl', 'no', 'ns', 'oc', 'or', 'pa', 'pl', 'ps', 'pt', 'ro', 'ru', 'sd', 'si', 'sk', 'sl', 'so', 'sq', 'sr', 'ss', 'su', 'sv', 'sw', 'ta', 'th', 'tl', 'tn', 'tr', 'uk', 'ur', 'uz', 'vi', 'wo', 'xh', 'yi', 'yo', 'zh', 'zu'],
    'wmt21': ['en', 'ha', 'is', 'ja', 'cs', 'ru', 'zh', 'de']
}



class     _a						(__magic_name__				):


					'''simple docstring'''






					UpperCAmelCase__:   List[str]  							=       VOCAB_FILES_NAMES
					UpperCAmelCase__:   Optional[int]  							=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					UpperCAmelCase__:   Tuple  							=       PRETRAINED_VOCAB_FILES_MAP
					UpperCAmelCase__:   Optional[int]  							=       ['''input_ids''', '''attention_mask''']

					UpperCAmelCase__:   List[int]  							=       []
					UpperCAmelCase__:   List[int]  							=       []






					def __init__( self    ,       A__    ,       A__    ,       A__=None    ,       A__=None    ,       A__="<s>"    ,       A__="</s>"    ,       A__="</s>"    ,       A__="<pad>"    ,       A__="<unk>"    ,       A__="m2m100"    ,       A__ = None    ,       A__=8    ,       **A__    ,       ):
						A__							:    Dict        =		{} if sp_model_kwargs is None else sp_model_kwargs

						A__							:    str        =		language_codes
						A__							:    str        =		FAIRSEQ_LANGUAGE_CODES[language_codes]
						A__							:    Dict        =		{lang_code: F"""__{lang_code}__""" for lang_code in fairseq_language_code}

						A__							:    Dict        =		kwargs.get("""additional_special_tokens"""    ,       []			)
						kwargs["additional_special_tokens"] += [
						    self.get_lang_token(A__			)
						    for lang_code in fairseq_language_code
						    if self.get_lang_token(A__			) not in kwargs["additional_special_tokens"]
						]

						super().__init__(
						    src_lang=A__    ,       tgt_lang=A__    ,       bos_token=A__    ,       eos_token=A__    ,       sep_token=A__    ,       unk_token=A__    ,       pad_token=A__    ,       language_codes=A__    ,       sp_model_kwargs=self.sp_model_kwargs    ,       num_madeup_words=A__    ,       **A__    ,       )

						A__							:    Optional[Any]        =		vocab_file
						A__							:    Optional[Any]        =		load_json(A__			)
						A__							:    Union[str, Any]        =		{v: k for k, v in self.encoder.items()}
						A__							:    Dict        =		spm_file
						A__							:    Optional[int]        =		load_spm(A__    ,       self.sp_model_kwargs			)

						A__							:    str        =		len(self.encoder			)

						A__							:    Any        =		{
						    self.get_lang_token(A__			): self.encoder_size + i for i, lang_code in enumerate(A__			)
						}
						A__							:    int        =		{lang_code: self.encoder_size + i for i, lang_code in enumerate(A__			)}
						A__							:    Optional[Any]        =		{v: k for k, v in self.lang_token_to_id.items()}

						A__							:    List[str]        =		src_lang if src_lang is not None else """en"""
						A__							:    Dict        =		tgt_lang
						A__							:    Dict        =		self.get_lang_id(self._src_lang			)
						self.set_src_lang_special_tokens(self._src_lang			)

						A__							:    str        =		num_madeup_words






					@property
					def   __A			( self			):
						return len(self.encoder			) + len(self.lang_token_to_id			)






					@property
					def   __A			( self			):
						return self._src_lang






					@src_lang.setter
					def   __A			( self    ,       A__			):
						A__							:    Optional[int]        =		new_src_lang
						self.set_src_lang_special_tokens(self._src_lang			)






					def   __A			( self    ,       A__			):
						return self.sp_model.encode(A__    ,       out_type=A__			)






					def   __A			( self    ,       A__			):
						if token in self.lang_token_to_id:
							return self.lang_token_to_id[token]
						return self.encoder.get(A__    ,       self.encoder[self.unk_token]			)






					def   __A			( self    ,       A__			):
						if index in self.id_to_lang_token:
							return self.id_to_lang_token[index]
						return self.decoder.get(A__    ,       self.unk_token			)






					def   __A			( self    ,       A__			):
						A__							:    Optional[int]        =		[]
						A__							:    str        =		""""""
						for token in tokens:
							# make sure that special tokens are not decoded using sentencepiece model
							if token in self.all_special_tokens:
								out_string += self.sp_model.decode(A__			) + token
								A__							:    Tuple        =		[]
							else:
								current_sub_tokens.append(A__			)
						out_string += self.sp_model.decode(A__			)
						return out_string.strip()






					def   __A			( self    ,       A__    ,       A__ = None    ,       A__ = False			):

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

						A__							:    List[str]        =		[1] * len(self.prefix_tokens			)
						A__							:    int        =		[1] * len(self.suffix_tokens			)
						if token_ids_a is None:
							return prefix_ones + ([0] * len(A__			)) + suffix_ones
						return prefix_ones + ([0] * len(A__			)) + ([0] * len(A__			)) + suffix_ones






					def   __A			( self    ,       A__    ,       A__ = None			):
						if token_ids_a is None:
							return self.prefix_tokens + token_ids_a + self.suffix_tokens
						# We don't expect to process pairs, but leave the pair logic for API consistency
						return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens






					def   __A			( self			):
						A__							:    Optional[Any]        =		{self.convert_ids_to_tokens(A__			): i for i in range(self.vocab_size			)}
						vocab.update(self.added_tokens_encoder			)
						return vocab






					def __getstate__( self			):
						A__							:    List[Any]        =		self.__dict__.copy()
						A__							:    Dict        =		None
						return state






					def __setstate__( self    ,       A__			):
						A__							:    Any        =		d

						# for backward compatibility
						if not hasattr(self    ,       """sp_model_kwargs"""			):
							A__							:    Optional[Any]        =		{}

						A__							:    List[Any]        =		load_spm(self.spm_file    ,       self.sp_model_kwargs			)






					def   __A			( self    ,       A__    ,       A__ = None			):
						A__							:    Union[str, Any]        =		Path(A__			)
						if not save_dir.is_dir():
							raise OSError(F"""{save_directory} should be a directory"""			)
						A__							:    Any        =		save_dir / (
						    (filename_prefix + """-""" if filename_prefix else """""") + self.vocab_files_names["""vocab_file"""]
						)
						A__							:    Union[str, Any]        =		save_dir / (
						    (filename_prefix + """-""" if filename_prefix else """""") + self.vocab_files_names["""spm_file"""]
						)

						save_json(self.encoder    ,       A__			)

						if os.path.abspath(self.spm_file			) != os.path.abspath(A__			) and os.path.isfile(self.spm_file			):
							copyfile(self.spm_file    ,       A__			)
						elif not os.path.isfile(self.spm_file			):
							with open(A__    ,       """wb"""			) as fi:
								A__							:    Dict        =		self.sp_model.serialized_model_proto()
								fi.write(A__			)

						return (str(A__			), str(A__			))






					def   __A			( self    ,       A__    ,       A__ = "en"    ,       A__ = None    ,       A__ = "ro"    ,       **A__    ,       ):
						A__							:    List[str]        =		src_lang
						A__							:    Union[str, Any]        =		tgt_lang
						self.set_src_lang_special_tokens(self.src_lang			)
						return super().prepare_seqaseq_batch(A__    ,       A__    ,       **A__			)






					def   __A			( self    ,       A__    ,       A__    ,       A__    ,       **A__			):
						if src_lang is None or tgt_lang is None:
							raise ValueError("""Translation requires a `src_lang` and a `tgt_lang` for this model"""			)
						A__							:    Optional[int]        =		src_lang
						A__							:    Optional[Any]        =		self(A__    ,       add_special_tokens=A__    ,       **A__			)
						A__							:    Union[str, Any]        =		self.get_lang_id(A__			)
						A__							:    Any        =		tgt_lang_id
						return inputs






					def   __A			( self			):
						self.set_src_lang_special_tokens(self.src_lang			)






					def   __A			( self			):
						self.set_tgt_lang_special_tokens(self.tgt_lang			)






					def   __A			( self    ,       A__			):
						A__							:    str        =		self.get_lang_token(A__			)
						A__							:    Optional[Any]        =		self.lang_token_to_id[lang_token]
						A__							:    Optional[Any]        =		[self.cur_lang_id]
						A__							:    Tuple        =		[self.eos_token_id]






					def   __A			( self    ,       A__			):
						A__							:    str        =		self.get_lang_token(A__			)
						A__							:    Optional[Any]        =		self.lang_token_to_id[lang_token]
						A__							:    str        =		[self.cur_lang_id]
						A__							:    Optional[int]        =		[self.eos_token_id]






					def   __A			( self    ,       A__			):
						return self.lang_code_to_token[lang]






					def   __A			( self    ,       A__			):
						A__							:    List[str]        =		self.get_lang_token(A__			)
						return self.lang_token_to_id[lang_token]



def 						UpperCamelCase					(lowercase_: str    ,      lowercase_: Dict[str, Any]					)     ->      sentencepiece.SentencePieceProcessor:
	A__							:    Dict        =		sentencepiece.SentencePieceProcessor(**lowercase_					)
	spm.Load(str(lowercase_					)					)
	return spm

def 						UpperCamelCase					(lowercase_: str					)     ->      Union[Dict, List]:
	with open(lowercase_    ,      """r"""					) as f:
		return json.load(lowercase_					)


def 						UpperCamelCase					(lowercase_: List[Any]    ,      lowercase_: str					)     ->      None:
	with open(lowercase_    ,      """w"""					) as f:
		json.dump(lowercase_    ,      lowercase_    ,      indent=2					)






import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pytest

from datasets.arrow_dataset import Dataset
from datasets.search import ElasticSearchIndex, FaissIndex, MissingIndex

from .utils import require_elasticsearch, require_faiss


A_       : Any         =			pytest.mark.integration



@require_faiss
class     _a						(__magic_name__				):


					'''simple docstring'''






					def   __A			( self			):
						A__							:    Tuple        =		Dataset.from_dict({"""filename""": ["""my_name-train""" + """_""" + str(A__			) for x in np.arange(30			).tolist()]}			)
						return dset






					def   __A			( self			):
						import faiss

						A__							:    Dataset        =		self._create_dummy_dataset()
						A__							:    Union[str, Any]        =		dset.map(
						    lambda A__    ,       A__			: {"vecs": i * np.ones(5    ,       dtype=np.floataa			)}    ,       with_indices=A__    ,       keep_in_memory=A__			)
						A__							:    int        =		dset.add_faiss_index("""vecs"""    ,       batch_size=100    ,       metric_type=faiss.METRIC_INNER_PRODUCT			)
						A__   ,   A__							:    Optional[Any]        =		dset.get_nearest_examples("""vecs"""    ,       np.ones(5    ,       dtype=np.floataa			)			)
						self.assertEqual(examples["""filename"""][0]    ,       """my_name-train_29"""			)
						dset.drop_index("""vecs"""			)






					def   __A			( self			):
						import faiss

						A__							:    Dataset        =		self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((30, 5)			) * np.arange(30			).reshape(-1    ,       1			)    ,       index_name="""vecs"""    ,       batch_size=100    ,       metric_type=faiss.METRIC_INNER_PRODUCT    ,       )
						A__   ,   A__							:    str        =		dset.get_nearest_examples("""vecs"""    ,       np.ones(5    ,       dtype=np.floataa			)			)
						self.assertEqual(examples["""filename"""][0]    ,       """my_name-train_29"""			)






					def   __A			( self			):
						import faiss

						A__							:    Dataset        =		self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((30, 5)			) * np.arange(30			).reshape(-1    ,       1			)    ,       index_name="""vecs"""    ,       metric_type=faiss.METRIC_INNER_PRODUCT    ,       )

						# Setting delete=False and unlinking manually is not pretty... but it is required on Windows to
						# ensure somewhat stable behaviour. If we don't, we get PermissionErrors. This is an age-old issue.
						# see https://bugs.python.org/issue14243 and
						# https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file/23212515
						with tempfile.NamedTemporaryFile(delete=A__			) as tmp_file:
							dset.save_faiss_index("""vecs"""    ,       tmp_file.name			)
							dset.load_faiss_index("""vecs2"""    ,       tmp_file.name			)
						os.unlink(tmp_file.name			)

						A__   ,   A__							:    Tuple        =		dset.get_nearest_examples("""vecs2"""    ,       np.ones(5    ,       dtype=np.floataa			)			)
						self.assertEqual(examples["""filename"""][0]    ,       """my_name-train_29"""			)






					def   __A			( self			):
						A__							:    Dataset        =		self._create_dummy_dataset()
						dset.add_faiss_index_from_external_arrays(
						    external_arrays=np.ones((30, 5)			) * np.arange(30			).reshape(-1    ,       1			)    ,       index_name="""vecs"""			)
						dset.drop_index("""vecs"""			)
						self.assertRaises(A__    ,       partial(dset.get_nearest_examples    ,       """vecs2"""    ,       np.ones(5    ,       dtype=np.floataa			)			)			)






					def   __A			( self			):
						from elasticsearch import Elasticsearch

						A__							:    Dataset        =		self._create_dummy_dataset()
						with patch("""elasticsearch.Elasticsearch.search"""			) as mocked_search, patch(
						    """elasticsearch.client.IndicesClient.create"""			) as mocked_index_create, patch("""elasticsearch.helpers.streaming_bulk"""			) as mocked_bulk:
							A__							:    List[Any]        =		{"""acknowledged""": True}
							mocked_bulk.return_value([(True, None)] * 30			)
							A__							:    List[Any]        =		{"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 29}]}}
							A__							:    Any        =		Elasticsearch()

							dset.add_elasticsearch_index("""filename"""    ,       es_client=A__			)
							A__   ,   A__							:    Any        =		dset.get_nearest_examples("""filename"""    ,       """my_name-train_29"""			)
							self.assertEqual(examples["""filename"""][0]    ,       """my_name-train_29"""			)



@require_faiss
class     _a						(__magic_name__				):


					'''simple docstring'''






					def   __A			( self			):
						import faiss

						A__							:    Optional[int]        =		FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT			)

						# add vectors
						index.add_vectors(np.eye(5    ,       dtype=np.floataa			)			)
						self.assertIsNotNone(index.faiss_index			)
						self.assertEqual(index.faiss_index.ntotal    ,       5			)
						index.add_vectors(np.zeros((5, 5)    ,       dtype=np.floataa			)			)
						self.assertEqual(index.faiss_index.ntotal    ,       10			)

						# single query
						A__							:    Any        =		np.zeros(5    ,       dtype=np.floataa			)
						A__							:    str        =		1
						A__   ,   A__							:    Optional[Any]        =		index.search(A__			)
						self.assertRaises(A__    ,       index.search    ,       query.reshape(-1    ,       1			)			)
						self.assertGreater(scores[0]    ,       0			)
						self.assertEqual(indices[0]    ,       1			)

						# batched queries
						A__							:    Optional[int]        =		np.eye(5    ,       dtype=np.floataa			)[::-1]
						A__   ,   A__							:    str        =		index.search_batch(A__			)
						self.assertRaises(A__    ,       index.search_batch    ,       queries[0]			)
						A__							:    str        =		[scores[0] for scores in total_scores]
						A__							:    Tuple        =		[indices[0] for indices in total_indices]
						self.assertGreater(np.min(A__			)    ,       0			)
						self.assertListEqual([4, 3, 2, 1, 0]    ,       A__			)






					def   __A			( self			):
						import faiss

						A__							:    Dict        =		FaissIndex(string_factory="""Flat"""			)
						index.add_vectors(np.eye(5    ,       dtype=np.floataa			)			)
						self.assertIsInstance(index.faiss_index    ,       faiss.IndexFlat			)
						A__							:    Dict        =		FaissIndex(string_factory="""LSH"""			)
						index.add_vectors(np.eye(5    ,       dtype=np.floataa			)			)
						self.assertIsInstance(index.faiss_index    ,       faiss.IndexLSH			)
						with self.assertRaises(A__			):
							A__							:    List[Any]        =		FaissIndex(string_factory="""Flat"""    ,       custom_index=faiss.IndexFlat(5			)			)






					def   __A			( self			):
						import faiss

						A__							:    List[Any]        =		faiss.IndexFlat(5			)
						A__							:    Union[str, Any]        =		FaissIndex(custom_index=A__			)
						index.add_vectors(np.eye(5    ,       dtype=np.floataa			)			)
						self.assertIsInstance(index.faiss_index    ,       faiss.IndexFlat			)






					def   __A			( self			):
						import faiss

						A__							:    Any        =		FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT			)
						index.add_vectors(np.eye(5    ,       dtype=np.floataa			)			)

						# Setting delete=False and unlinking manually is not pretty... but it is required on Windows to
						# ensure somewhat stable behaviour. If we don't, we get PermissionErrors. This is an age-old issue.
						# see https://bugs.python.org/issue14243 and
						# https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file/23212515
						with tempfile.NamedTemporaryFile(delete=A__			) as tmp_file:
							index.save(tmp_file.name			)
							A__							:    int        =		FaissIndex.load(tmp_file.name			)
						os.unlink(tmp_file.name			)

						A__							:    Optional[Any]        =		np.zeros(5    ,       dtype=np.floataa			)
						A__							:    Optional[int]        =		1
						A__   ,   A__							:    List[Any]        =		index.search(A__			)
						self.assertGreater(scores[0]    ,       0			)
						self.assertEqual(indices[0]    ,       1			)



@require_faiss
def 						UpperCamelCase					(lowercase_: Dict					)     ->      Optional[Any]:
	import faiss

	A__							:    Dict        =		FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT					)
	index.add_vectors(np.eye(5    ,      dtype=np.floataa					)					)

	A__							:    Optional[Any]        =		"""index.faiss"""
	A__							:    Any        =		f"""mock://{index_name}"""
	index.save(lowercase_    ,      storage_options=mockfs.storage_options					)
	A__							:    str        =		FaissIndex.load(lowercase_    ,      storage_options=mockfs.storage_options					)

	A__							:    int        =		np.zeros(5    ,      dtype=np.floataa					)
	A__							:    Union[str, Any]        =		1
	A__   ,   A__							:    Any        =		index.search(lowercase_					)
	assert scores[0] > 0
	assert indices[0] == 1



@require_elasticsearch
class     _a						(__magic_name__				):


					'''simple docstring'''






					def   __A			( self			):
						from elasticsearch import Elasticsearch

						with patch("""elasticsearch.Elasticsearch.search"""			) as mocked_search, patch(
						    """elasticsearch.client.IndicesClient.create"""			) as mocked_index_create, patch("""elasticsearch.helpers.streaming_bulk"""			) as mocked_bulk:
							A__							:    List[str]        =		Elasticsearch()
							A__							:    List[Any]        =		{"""acknowledged""": True}
							A__							:    int        =		ElasticSearchIndex(es_client=A__			)
							mocked_bulk.return_value([(True, None)] * 3			)
							index.add_documents(["""foo""", """bar""", """foobar"""]			)

							# single query
							A__							:    Dict        =		"""foo"""
							A__							:    Tuple        =		{"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 0}]}}
							A__   ,   A__							:    Tuple        =		index.search(A__			)
							self.assertEqual(scores[0]    ,       1			)
							self.assertEqual(indices[0]    ,       0			)

							# single query with timeout
							A__							:    List[str]        =		"""foo"""
							A__							:    str        =		{"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 0}]}}
							A__   ,   A__							:    Dict        =		index.search(A__    ,       request_timeout=30			)
							self.assertEqual(scores[0]    ,       1			)
							self.assertEqual(indices[0]    ,       0			)

							# batched queries
							A__							:    Union[str, Any]        =		["""foo""", """bar""", """foobar"""]
							A__							:    Tuple        =		{"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 1}]}}
							A__   ,   A__							:    Dict        =		index.search_batch(A__			)
							A__							:    Tuple        =		[scores[0] for scores in total_scores]
							A__							:    Optional[int]        =		[indices[0] for indices in total_indices]
							self.assertGreater(np.min(A__			)    ,       0			)
							self.assertListEqual([1, 1, 1]    ,       A__			)

							# batched queries with timeout
							A__							:    int        =		["""foo""", """bar""", """foobar"""]
							A__							:    List[Any]        =		{"""hits""": {"""hits""": [{"""_score""": 1, """_id""": 1}]}}
							A__   ,   A__							:    Tuple        =		index.search_batch(A__    ,       request_timeout=30			)
							A__							:    List[Any]        =		[scores[0] for scores in total_scores]
							A__							:    Optional[int]        =		[indices[0] for indices in total_indices]
							self.assertGreater(np.min(A__			)    ,       0			)
							self.assertListEqual([1, 1, 1]    ,       A__			)



