




import json
import os
import pickle
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers import is_faiss_available
from transformers.models.bart.configuration_bart import BartConfig
from transformers.models.bart.tokenization_bart import BartTokenizer
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.dpr.configuration_dpr import DPRConfig
from transformers.models.dpr.tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer
from transformers.models.rag.configuration_rag import RagConfig
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_faiss, require_sentencepiece, require_tokenizers, require_torch


if is_faiss_available():
						import faiss







@require_faiss
class 				lowerCAmelCase_       (					lowercase_							):






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : List[Any]					  =      tempfile.mkdtemp()
							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      8

							# DPR tok
							SCREAMING_SNAKE_CASE_   : List[str]					  =      [
							    '[UNK]',
							    '[CLS]',
							    '[SEP]',
							    '[PAD]',
							    '[MASK]',
							    'want',
							    '##want',
							    '##ed',
							    'wa',
							    'un',
							    'runn',
							    '##ing',
							    ',',
							    'low',
							    'lowest',
							]
							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      os.path.join(self.tmpdirname      ,'dpr_tokenizer' )
							os.makedirs(snake_case__      ,exist_ok=snake_case__ )
							SCREAMING_SNAKE_CASE_   : List[Any]					  =      os.path.join(snake_case__      ,DPR_VOCAB_FILES_NAMES['vocab_file'] )
							with open(self.vocab_file      ,'w'      ,encoding='utf-8' ) as vocab_writer:
										vocab_writer.write(''.join([x + '\n' for x in vocab_tokens] ) )

							# BART tok
							SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      [
							    'l',
							    'o',
							    'w',
							    'e',
							    'r',
							    's',
							    't',
							    'i',
							    'd',
							    'n',
							    '\u0120',
							    '\u0120l',
							    '\u0120n',
							    '\u0120lo',
							    '\u0120low',
							    'er',
							    '\u0120lowest',
							    '\u0120newer',
							    '\u0120wider',
							    '<unk>',
							]
							SCREAMING_SNAKE_CASE_   : int					  =      dict(zip(snake_case__      ,range(len(snake_case__ ) ) ) )
							SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      ['#version: 0.2', '\u0120 l', '\u0120l o', '\u0120lo w', 'e r', '']
							SCREAMING_SNAKE_CASE_   : Tuple					  =      {'unk_token': '<unk>'}

							SCREAMING_SNAKE_CASE_   : Optional[int]					  =      os.path.join(self.tmpdirname      ,'bart_tokenizer' )
							os.makedirs(snake_case__      ,exist_ok=snake_case__ )
							SCREAMING_SNAKE_CASE_   : Tuple					  =      os.path.join(snake_case__      ,BART_VOCAB_FILES_NAMES['vocab_file'] )
							SCREAMING_SNAKE_CASE_   : str					  =      os.path.join(snake_case__      ,BART_VOCAB_FILES_NAMES['merges_file'] )
							with open(self.vocab_file      ,'w'      ,encoding='utf-8' ) as fp:
										fp.write(json.dumps(snake_case__ ) + '\n' )
							with open(self.merges_file      ,'w'      ,encoding='utf-8' ) as fp:
										fp.write('\n'.join(snake_case__ ) )






				def 		snake_case	(			self ):
							return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,'dpr_tokenizer' ) )






				def 		snake_case	(			self ):
							return DPRContextEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,'dpr_tokenizer' ) )






				def 		snake_case	(			self ):
							return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,'bart_tokenizer' ) )






				def 		snake_case	(			self ):
							shutil.rmtree(self.tmpdirname )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : int					  =      Dataset.from_dict(
							    {
							        'id': ['0', '1'],
							        'text': ['foo', 'bar'],
							        'title': ['Foo', 'Bar'],
							        'embeddings': [np.ones(self.retrieval_vector_size ), 2 * np.ones(self.retrieval_vector_size )],
							    } )
							dataset.add_faiss_index('embeddings'      ,string_factory='Flat'      ,metric_type=faiss.METRIC_INNER_PRODUCT )
							return dataset






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Tuple					  =      self.get_dummy_dataset()
							SCREAMING_SNAKE_CASE_   : Tuple					  =      RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size      ,question_encoder=DPRConfig().to_dict()      ,generator=BartConfig().to_dict()      ,)
							with patch('transformers.models.rag.retrieval_rag.load_dataset' ) as mock_load_dataset:
										SCREAMING_SNAKE_CASE_   : Any					  =      dataset
										SCREAMING_SNAKE_CASE_   : str					  =      RagRetriever(
										    snake_case__      ,question_encoder_tokenizer=self.get_dpr_tokenizer()      ,generator_tokenizer=self.get_bart_tokenizer()      ,)
							return retriever






				def 		snake_case	(			self      ,snake_case__ ):
							SCREAMING_SNAKE_CASE_   : Dict					  =      self.get_dummy_dataset()
							SCREAMING_SNAKE_CASE_   : List[str]					  =      RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size      ,question_encoder=DPRConfig().to_dict()      ,generator=BartConfig().to_dict()      ,index_name='custom'      ,)
							if from_disk:
										SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      os.path.join(self.tmpdirname      ,'dataset' )
										SCREAMING_SNAKE_CASE_   : Tuple					  =      os.path.join(self.tmpdirname      ,'index.faiss' )
										dataset.get_index('embeddings' ).save(os.path.join(self.tmpdirname      ,'index.faiss' ) )
										dataset.drop_index('embeddings' )
										dataset.save_to_disk(os.path.join(self.tmpdirname      ,'dataset' ) )
										del dataset
										SCREAMING_SNAKE_CASE_   : List[str]					  =      RagRetriever(
										    snake_case__      ,question_encoder_tokenizer=self.get_dpr_tokenizer()      ,generator_tokenizer=self.get_bart_tokenizer()      ,)
							else:
										SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      RagRetriever(
										    snake_case__      ,question_encoder_tokenizer=self.get_dpr_tokenizer()      ,generator_tokenizer=self.get_bart_tokenizer()      ,index=CustomHFIndex(config.retrieval_vector_size      ,snake_case__ )      ,)
							return retriever






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Tuple					  =      Dataset.from_dict(
							    {
							        'id': ['0', '1'],
							        'text': ['foo', 'bar'],
							        'title': ['Foo', 'Bar'],
							        'embeddings': [np.ones(self.retrieval_vector_size + 1 ), 2 * np.ones(self.retrieval_vector_size + 1 )],
							    } )
							dataset.add_faiss_index('embeddings'      ,string_factory='Flat'      ,metric_type=faiss.METRIC_INNER_PRODUCT )

							SCREAMING_SNAKE_CASE_   : List[str]					  =      os.path.join(self.tmpdirname      ,'hf_bert_base.hnswSQ8_correct_phi_128.c_index' )
							dataset.save_faiss_index('embeddings'      ,index_file_name + '.index.dpr' )
							pickle.dump(dataset['id']      ,open(index_file_name + '.index_meta.dpr'      ,'wb' ) )

							SCREAMING_SNAKE_CASE_   : Optional[int]					  =      os.path.join(self.tmpdirname      ,'psgs_w100.tsv.pkl' )
							SCREAMING_SNAKE_CASE_   : List[Any]					  =      {sample['id']: [sample['text'], sample['title']] for sample in dataset}
							pickle.dump(snake_case__      ,open(snake_case__      ,'wb' ) )

							SCREAMING_SNAKE_CASE_   : List[str]					  =      RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size      ,question_encoder=DPRConfig().to_dict()      ,generator=BartConfig().to_dict()      ,index_name='legacy'      ,index_path=self.tmpdirname      ,)
							SCREAMING_SNAKE_CASE_   : Tuple					  =      RagRetriever(
							    snake_case__      ,question_encoder_tokenizer=self.get_dpr_tokenizer()      ,generator_tokenizer=self.get_bart_tokenizer() )
							return retriever






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      1
							SCREAMING_SNAKE_CASE_   : List[str]					  =      self.get_dummy_canonical_hf_index_retriever()
							SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      np.array(
							    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
							SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_   : Any					  =      retriever.retrieve(snake_case__      ,n_docs=snake_case__ )
							self.assertEqual(retrieved_doc_embeds.shape      ,(2, n_docs, self.retrieval_vector_size) )
							self.assertEqual(len(snake_case__ )      ,2 )
							self.assertEqual(sorted(doc_dicts[0] )      ,['embeddings', 'id', 'text', 'title'] )
							self.assertEqual(len(doc_dicts[0]['id'] )      ,snake_case__ )
							self.assertEqual(doc_dicts[0]['id'][0]      ,'1' )  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['id'][0]      ,'0' )  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()      ,[[1], [0]] )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : str					  =      self.get_dummy_canonical_hf_index_retriever()
							with tempfile.TemporaryDirectory() as tmp_dirname:
										with patch('transformers.models.rag.retrieval_rag.load_dataset' ) as mock_load_dataset:
													SCREAMING_SNAKE_CASE_   : Tuple					  =      self.get_dummy_dataset()
													retriever.save_pretrained(snake_case__ )
													SCREAMING_SNAKE_CASE_   : Optional[int]					  =      RagRetriever.from_pretrained(snake_case__ )
													self.assertIsInstance(snake_case__      ,snake_case__ )
													SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      np.array(
													    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
													SCREAMING_SNAKE_CASE_   : str					  =      retriever.retrieve(snake_case__      ,n_docs=1 )
													self.assertTrue(out is not None )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : str					  =      1
							SCREAMING_SNAKE_CASE_   : Tuple					  =      self.get_dummy_custom_hf_index_retriever(from_disk=snake_case__ )
							SCREAMING_SNAKE_CASE_   : Tuple					  =      np.array(
							    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
							SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_   : List[Any]					  =      retriever.retrieve(snake_case__      ,n_docs=snake_case__ )
							self.assertEqual(retrieved_doc_embeds.shape      ,(2, n_docs, self.retrieval_vector_size) )
							self.assertEqual(len(snake_case__ )      ,2 )
							self.assertEqual(sorted(doc_dicts[0] )      ,['embeddings', 'id', 'text', 'title'] )
							self.assertEqual(len(doc_dicts[0]['id'] )      ,snake_case__ )
							self.assertEqual(doc_dicts[0]['id'][0]      ,'1' )  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['id'][0]      ,'0' )  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()      ,[[1], [0]] )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : str					  =      self.get_dummy_custom_hf_index_retriever(from_disk=snake_case__ )
							with tempfile.TemporaryDirectory() as tmp_dirname:
										retriever.save_pretrained(snake_case__ )
										SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      RagRetriever.from_pretrained(snake_case__ )
										self.assertIsInstance(snake_case__      ,snake_case__ )
										SCREAMING_SNAKE_CASE_   : int					  =      np.array(
										    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
										SCREAMING_SNAKE_CASE_   : Tuple					  =      retriever.retrieve(snake_case__      ,n_docs=1 )
										self.assertTrue(out is not None )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Any					  =      1
							SCREAMING_SNAKE_CASE_   : Any					  =      self.get_dummy_custom_hf_index_retriever(from_disk=snake_case__ )
							SCREAMING_SNAKE_CASE_   : Tuple					  =      np.array(
							    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
							SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_   : Dict					  =      retriever.retrieve(snake_case__      ,n_docs=snake_case__ )
							self.assertEqual(retrieved_doc_embeds.shape      ,(2, n_docs, self.retrieval_vector_size) )
							self.assertEqual(len(snake_case__ )      ,2 )
							self.assertEqual(sorted(doc_dicts[0] )      ,['embeddings', 'id', 'text', 'title'] )
							self.assertEqual(len(doc_dicts[0]['id'] )      ,snake_case__ )
							self.assertEqual(doc_dicts[0]['id'][0]      ,'1' )  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['id'][0]      ,'0' )  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()      ,[[1], [0]] )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Tuple					  =      self.get_dummy_custom_hf_index_retriever(from_disk=snake_case__ )
							with tempfile.TemporaryDirectory() as tmp_dirname:
										retriever.save_pretrained(snake_case__ )
										SCREAMING_SNAKE_CASE_   : int					  =      RagRetriever.from_pretrained(snake_case__ )
										self.assertIsInstance(snake_case__      ,snake_case__ )
										SCREAMING_SNAKE_CASE_   : List[Any]					  =      np.array(
										    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
										SCREAMING_SNAKE_CASE_   : int					  =      retriever.retrieve(snake_case__      ,n_docs=1 )
										self.assertTrue(out is not None )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : int					  =      1
							SCREAMING_SNAKE_CASE_   : Tuple					  =      self.get_dummy_legacy_index_retriever()
							SCREAMING_SNAKE_CASE_   : str					  =      np.array(
							    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
							SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_   : List[Any]					  =      retriever.retrieve(snake_case__      ,n_docs=snake_case__ )
							self.assertEqual(retrieved_doc_embeds.shape      ,(2, n_docs, self.retrieval_vector_size) )
							self.assertEqual(len(snake_case__ )      ,2 )
							self.assertEqual(sorted(doc_dicts[0] )      ,['text', 'title'] )
							self.assertEqual(len(doc_dicts[0]['text'] )      ,snake_case__ )
							self.assertEqual(doc_dicts[0]['text'][0]      ,'bar' )  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['text'][0]      ,'foo' )  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()      ,[[1], [0]] )






				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      self.get_dummy_legacy_index_retriever()
							with tempfile.TemporaryDirectory() as tmp_dirname:
										retriever.save_pretrained(snake_case__ )
										SCREAMING_SNAKE_CASE_   : List[Any]					  =      RagRetriever.from_pretrained(snake_case__ )
										self.assertIsInstance(snake_case__      ,snake_case__ )
										SCREAMING_SNAKE_CASE_   : Tuple					  =      np.array(
										    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
										SCREAMING_SNAKE_CASE_   : List[str]					  =      retriever.retrieve(snake_case__      ,n_docs=1 )
										self.assertTrue(out is not None )






				@require_torch
				@require_tokenizers
				@require_sentencepiece
				def 		snake_case	(			self ):
							import torch

							SCREAMING_SNAKE_CASE_   : List[Any]					  =      1
							SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      self.get_dummy_canonical_hf_index_retriever()
							SCREAMING_SNAKE_CASE_   : Optional[int]					  =      [[5, 7], [10, 11]]
							SCREAMING_SNAKE_CASE_   : Tuple					  =      np.array(
							    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
							SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      retriever(snake_case__      ,snake_case__      ,prefix=retriever.config.generator.prefix      ,n_docs=snake_case__ )
							SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_   : str					  =      (
							    out['context_input_ids'],
							    out['context_attention_mask'],
							    out['retrieved_doc_embeds'],
							)
							self.assertEqual(retrieved_doc_embeds.shape      ,(2, n_docs, self.retrieval_vector_size) )
							self.assertIsInstance(snake_case__      ,snake_case__ )
							self.assertIsInstance(snake_case__      ,snake_case__ )
							self.assertIsInstance(snake_case__      ,np.ndarray )

							SCREAMING_SNAKE_CASE_   : List[Any]					  =      retriever(
							    snake_case__      ,snake_case__      ,prefix=retriever.config.generator.prefix      ,n_docs=snake_case__      ,return_tensors='pt'      ,)
							SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_  ,    SCREAMING_SNAKE_CASE_   : Tuple					  =      (  # noqa: F841
							    out['context_input_ids'],
							    out['context_attention_mask'],
							    out['retrieved_doc_embeds'],
							    out['doc_ids'],
							)
							self.assertEqual(retrieved_doc_embeds.shape      ,(2, n_docs, self.retrieval_vector_size) )
							self.assertIsInstance(snake_case__      ,torch.Tensor )
							self.assertIsInstance(snake_case__      ,torch.Tensor )
							self.assertIsInstance(snake_case__      ,torch.Tensor )






				@require_torch
				@require_tokenizers
				@require_sentencepiece
				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Optional[int]					  =      self.get_dpr_ctx_encoder_tokenizer()
							SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      1
							SCREAMING_SNAKE_CASE_   : Tuple					  =      self.get_dummy_custom_hf_index_retriever(from_disk=snake_case__ )
							retriever.set_ctx_encoder_tokenizer(snake_case__ )

							SCREAMING_SNAKE_CASE_   : Tuple					  =      [[5, 7], [10, 11]]
							SCREAMING_SNAKE_CASE_   : List[Any]					  =      np.array(
							    [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]      ,dtype=np.floataa )
							SCREAMING_SNAKE_CASE_   : Any					  =      retriever(snake_case__      ,snake_case__      ,prefix=retriever.config.generator.prefix      ,n_docs=snake_case__ )

							self.assertEqual(
							    len(snake_case__ )      ,6 )  # check whether the retriever output consist of 6 attributes including tokenized docs
							self.assertEqual(
							    all(k in out for k in ('tokenized_doc_ids', 'tokenized_doc_attention_mask') )      ,snake_case__ )  # check for doc token related keys in dictionary.







'''simple docstring'''

def _SCREAMING_SNAKE_CASE (A			)			->			bool:






       """simple docstring"""







       lowercase__      =					[int(A			) for i in ip_va_address.split('''.'''			) if i.isdigit()]
       return len(A			) == 4 and all(0 <= int(A			) <= 254 for octet in octets			)


if __name__ == "__main__":
 lowerCamelCase				:  Optional[Any]    		=      input().strip()
 lowerCamelCase				:  Union[str, Any]    		=      'valid' if is_ip_va_address_valid(ip) else 'invalid'
 print(f"""{ip} is a {valid_or_invalid} IP v4 address.""")




