code: 
  
  
  
  
  
  import json
  import os
  import shutil
  import sys
  import tempfile
  import unittest
  from unittest import TestCase
  from unittest.mock import patch
  
  import faiss
  import numpy as np
  from datasets import Dataset
  
  from transformers import BartConfig, BartTokenizer, DPRConfig, DPRQuestionEncoderTokenizer, RagConfig
  from transformers.file_utils import is_datasets_available, is_faiss_available, is_psutil_available, is_torch_available
  from transformers.integrations import is_ray_available
  from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
  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_ray
  
  
  sys.path.append(os.path.join(os.getcwd()))  # noqa: E402 # noqa: E402 # isort:skip
  
  if is_torch_available():
    from distributed_pytorch_retriever import RagPyTorchDistributedRetriever  # noqa: E402 # isort:skip
  else:
    _lowercase		:     int 					=None
  
  if is_ray_available():
    import ray  # noqa: E402 # isort:skip
    from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
  else:
    _lowercase		:     Any 					=None
    _lowercase		:     Tuple 					=None
    _lowercase		:     List[Any] 					=None
  
  
  def 		_SCREAMING_SNAKE_CASE    (   lowerCAmelCase__	):
    if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
      lowerCamelCase_  :    List[Any]   	=			unittest.skip('test requires Datasets, Faiss, psutil'	)(lowerCAmelCase__	)
    return test_case
  
  
  
  
  
  
  @require_distributed_retrieval
  class 	UpperCamelCase_     (     snake_case__       ):
  
  
  
       def      __a    (		self    :    Dict	):
         lowerCamelCase_  :    Optional[Any]   	=			tempfile.mkdtemp()
         lowerCamelCase_  :    Tuple   	=			8
  
         # DPR tok
         lowerCamelCase_  :    Any   	=			[
             '[UNK]',
             '[CLS]',
             '[SEP]',
             '[PAD]',
             '[MASK]',
             'want',
             '##want',
             '##ed',
             'wa',
             'un',
             'runn',
             '##ing',
             ',',
             'low',
             'lowest',
         ]
         lowerCamelCase_  :    List[str]   	=			os.path.join(self.tmpdirname					,			'dpr_tokenizer'	)
         os.makedirs(lowerCamelCase					,			exist_ok=lowerCamelCase	)
         lowerCamelCase_  :    int   	=			os.path.join(lowerCamelCase					,			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
         lowerCamelCase_  :    Optional[Any]   	=			[
             'l',
             'o',
             'w',
             'e',
             'r',
             's',
             't',
             'i',
             'd',
             'n',
             '\u0120',
             '\u0120l',
             '\u0120n',
             '\u0120lo',
             '\u0120low',
             'er',
             '\u0120lowest',
             '\u0120newer',
             '\u0120wider',
             '<unk>',
         ]
         lowerCamelCase_  :    Any   	=			dict(zip(lowerCamelCase					,			range(len(lowerCamelCase	)	)	)	)
         lowerCamelCase_  :    str   	=			['#version: 0.2', '\u0120 l', '\u0120l o', '\u0120lo w', 'e r', '']
         lowerCamelCase_  :    Union[str, Any]   	=			{'unk_token': '<unk>'}
  
         lowerCamelCase_  :    int   	=			os.path.join(self.tmpdirname					,			'bart_tokenizer'	)
         os.makedirs(lowerCamelCase					,			exist_ok=lowerCamelCase	)
         lowerCamelCase_  :    List[Any]   	=			os.path.join(lowerCamelCase					,			BART_VOCAB_FILES_NAMES['vocab_file']	)
         lowerCamelCase_  :    Tuple   	=			os.path.join(lowerCamelCase					,			BART_VOCAB_FILES_NAMES['merges_file']	)
         with open(self.vocab_file					,			'w'					,			encoding='utf-8'	) as fp:
           fp.write(json.dumps(lowerCamelCase	) + '\n'	)
         with open(self.merges_file					,			'w'					,			encoding='utf-8'	) as fp:
           fp.write('\n'.join(lowerCamelCase	)	)
  
  
  
       def      __a    (		self    :    List[Any]	):
         return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname					,			'dpr_tokenizer'	)	)
  
  
  
       def      __a    (		self    :    Union[str, Any]	):
         return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname					,			'bart_tokenizer'	)	)
  
  
  
       def      __a    (		self    :    str	):
         shutil.rmtree(self.tmpdirname	)
  
  
  
       def      __a    (		self    :    Union[str, Any]	):
         lowerCamelCase_  :    Optional[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      __a    (		self    :    Any					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    Optional[Any]=1_23_45	):
         lowerCamelCase_  :    Optional[Any]   	=			self.get_dummy_dataset()
         lowerCamelCase_  :    Optional[int]   	=			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:
           lowerCamelCase_  :    Optional[Any]   	=			dataset
           lowerCamelCase_  :    List[str]   	=			RagPyTorchDistributedRetriever(
               lowerCamelCase					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			)
           if init_retrieval:
             retriever.init_retrieval(lowerCamelCase	)
         return retriever
  
  
  
       def      __a    (		self    :    Dict					,			lowerCamelCase    :    bool	):
         # Have to run in local mode because sys.path modifications at top of
         # file are not propogated to remote workers.
         # https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
         ray.init(local_mode=lowerCamelCase	)
         lowerCamelCase_  :    Optional[int]   	=			RagConfig(
             retrieval_vector_size=self.retrieval_vector_size					,			question_encoder=DPRConfig().to_dict()					,			generator=BartConfig().to_dict()					,			)
         lowerCamelCase_  :    Any   	=			ray.remote(lowerCamelCase	)
         lowerCamelCase_  :    int   	=			[remote_cls.remote() for _ in range(1	)]
         with patch('transformers.models.rag.retrieval_rag.load_dataset'	) as mock_load_dataset:
           lowerCamelCase_  :    int   	=			self.get_dummy_dataset()
           lowerCamelCase_  :    Tuple   	=			RagRayDistributedRetriever(
               lowerCamelCase					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			retrieval_workers=lowerCamelCase					,			)
           if init_retrieval:
             retriever.init_retrieval()
         return retriever
  
  
  
       def      __a    (		self    :    List[str]					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    List[Any]=1_23_45	):
         lowerCamelCase_  :    int   	=			self.get_dummy_dataset()
         lowerCamelCase_  :    Any   	=			RagConfig(
             retrieval_vector_size=self.retrieval_vector_size					,			question_encoder=DPRConfig().to_dict()					,			generator=BartConfig().to_dict()					,			index_name='custom'					,			)
         if from_disk:
           lowerCamelCase_  :    Any   	=			os.path.join(self.tmpdirname					,			'dataset'	)
           lowerCamelCase_  :    str   	=			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
           lowerCamelCase_  :    str   	=			RagPyTorchDistributedRetriever(
               lowerCamelCase					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			)
         else:
           lowerCamelCase_  :    List[Any]   	=			RagPyTorchDistributedRetriever(
               lowerCamelCase					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			index=CustomHFIndex(config.retrieval_vector_size					,			lowerCamelCase	)					,			)
         if init_retrieval:
           retriever.init_retrieval(lowerCamelCase	)
         return retriever
  
  
  
       def      __a    (		self    :    Dict					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    bool	):
         # Have to run in local mode because sys.path modifications at top of
         # file are not propogated to remote workers.
         # https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
         ray.init(local_mode=lowerCamelCase	)
         lowerCamelCase_  :    Dict   	=			self.get_dummy_dataset()
         lowerCamelCase_  :    Dict   	=			RagConfig(
             retrieval_vector_size=self.retrieval_vector_size					,			question_encoder=DPRConfig().to_dict()					,			generator=BartConfig().to_dict()					,			index_name='custom'					,			)
         lowerCamelCase_  :    Any   	=			ray.remote(lowerCamelCase	)
         lowerCamelCase_  :    List[str]   	=			[remote_cls.remote() for _ in range(1	)]
         if from_disk:
           lowerCamelCase_  :    int   	=			os.path.join(self.tmpdirname					,			'dataset'	)
           lowerCamelCase_  :    Optional[int]   	=			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
           lowerCamelCase_  :    List[Any]   	=			RagRayDistributedRetriever(
               lowerCamelCase					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			retrieval_workers=lowerCamelCase					,			index=CustomHFIndex.load_from_disk(
                   vector_size=config.retrieval_vector_size					,			dataset_path=config.passages_path					,			index_path=config.index_path					,			)					,			)
         else:
           lowerCamelCase_  :    List[str]   	=			RagRayDistributedRetriever(
               lowerCamelCase					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			retrieval_workers=lowerCamelCase					,			index=CustomHFIndex(config.retrieval_vector_size					,			lowerCamelCase	)					,			)
         if init_retrieval:
           retriever.init_retrieval()
         return retriever
  
  
  
       def      __a    (		self    :    Any					,			lowerCamelCase    :    RagRetriever					,			lowerCamelCase    :    np.array					,			lowerCamelCase    :    int	):
         lowerCamelCase_  :    List[Any]   	=			retriever.retrieve(lowerCamelCase					,			n_docs=lowerCamelCase	)
         self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)	)
         self.assertEqual(len(lowerCamelCase	)					,			2	)
         self.assertEqual(sorted(doc_dicts[0]	)					,			['embeddings', 'id', 'text', 'title']	)
         self.assertEqual(len(doc_dicts[0]['id']	)					,			lowerCamelCase	)
         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      __a    (		self    :    int	):
         lowerCamelCase_  :    List[Any]   	=			1
         lowerCamelCase_  :    Union[str, Any]   	=			np.array(
             [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]					,			dtype=np.floataa	)
  
         self.distributed_retriever_check(
             self.get_dummy_pytorch_distributed_retriever(init_retrieval=lowerCamelCase	)					,			lowerCamelCase					,			lowerCamelCase	)
  
  
  
       def      __a    (		self    :    Optional[int]	):
         lowerCamelCase_  :    Optional[Any]   	=			1
         lowerCamelCase_  :    int   	=			np.array(
             [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]					,			dtype=np.floataa	)
  
         self.distributed_retriever_check(
             self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=lowerCamelCase					,			from_disk=lowerCamelCase	)					,			lowerCamelCase					,			lowerCamelCase					,			)
  
  
  
       def      __a    (		self    :    List[Any]	):
         lowerCamelCase_  :    List[str]   	=			1
         lowerCamelCase_  :    Union[str, Any]   	=			np.array(
             [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]					,			dtype=np.floataa	)
  
         self.distributed_retriever_check(
             self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=lowerCamelCase					,			from_disk=lowerCamelCase	)					,			lowerCamelCase					,			lowerCamelCase					,			)
  
  
  
       @require_ray
       def      __a    (		self    :    Optional[Any]	):
         lowerCamelCase_  :    Optional[int]   	=			1
         lowerCamelCase_  :    Tuple   	=			np.array(
             [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]					,			dtype=np.floataa	)
  
         self.distributed_retriever_check(
             self.get_dummy_ray_distributed_retriever(init_retrieval=lowerCamelCase	)					,			lowerCamelCase					,			lowerCamelCase	)
         ray.shutdown()
  
  
  
       @require_ray
       def      __a    (		self    :    List[str]	):
         lowerCamelCase_  :    str   	=			1
         lowerCamelCase_  :    Optional[int]   	=			np.array(
             [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]					,			dtype=np.floataa	)
         with self.assertRaises(lowerCamelCase	):
           self.distributed_retriever_check(
               self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=lowerCamelCase					,			from_disk=lowerCamelCase	)					,			lowerCamelCase					,			lowerCamelCase					,			)
         ray.shutdown()
  
  
  
  
  
       @require_ray
       def      __a    (		self    :    List[Any]	):
         lowerCamelCase_  :    int   	=			1
         lowerCamelCase_  :    int   	=			np.array(
             [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]					,			dtype=np.floataa	)
  
         self.distributed_retriever_check(
             self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=lowerCamelCase					,			from_disk=lowerCamelCase	)					,			lowerCamelCase					,			lowerCamelCase	)
         ray.shutdown()
  
style_context: 
  
  
  
  
  
  class 	UpperCamelCase_     :
  
  
  
  					def __init__(		self    :    List[Any]					,			lowerCamelCase    :    int	):
  							lowerCamelCase_  :    Union[str, Any]   	=			val
  							# Number of nodes in left subtree
  							lowerCamelCase_  :    Dict   	=			0
  							lowerCamelCase_  :    str   	=			None
  							lowerCamelCase_  :    int   	=			None
  							lowerCamelCase_  :    Dict   	=			None
  
  
  
  
  
  					def      __a    (		self    :    Optional[int]					,			lowerCamelCase    :    List[Any]	):
  							assert self.left_tree_size == other.left_tree_size, "Unequal Sizes of Blocks"
  
  							if self.val < other.val:
  									lowerCamelCase_  :    int   	=			self.right
  									lowerCamelCase_  :    Any   	=			None
  									if self.right:
  											lowerCamelCase_  :    Optional[Any]   	=			other
  									lowerCamelCase_  :    Union[str, Any]   	=			other
  									lowerCamelCase_  :    List[Any]   	=			self.left_tree_size * 2 + 1
  									return self
  							else:
  									lowerCamelCase_  :    str   	=			other.right
  									lowerCamelCase_  :    List[Any]   	=			None
  									if other.right:
  											lowerCamelCase_  :    Union[str, Any]   	=			self
  									lowerCamelCase_  :    Optional[int]   	=			self
  									lowerCamelCase_  :    Optional[int]   	=			other.left_tree_size * 2 + 1
  									return other
  
  
  
  
  
  
  class 	UpperCamelCase_     :
  
  
  
  					def __init__(		self    :    List[str]					,			lowerCamelCase    :    Optional[int]=None					,			lowerCamelCase    :    str=None					,			lowerCamelCase    :    List[Any]=0	):
  							lowerCamelCase_  :    Any   	=			heap_size
  							lowerCamelCase_  :    int   	=			bottom_root
  							lowerCamelCase_  :    Any   	=			min_node
  
  
  
  					def      __a    (		self    :    int					,			lowerCamelCase    :    int	):
  
  							# Empty heaps corner cases
  							if other.size == 0:
  									return None
  							if self.size == 0:
  									lowerCamelCase_  :    List[str]   	=			other.size
  									lowerCamelCase_  :    Tuple   	=			other.bottom_root
  									lowerCamelCase_  :    Tuple   	=			other.min_node
  									return None
  							# Update size
  							lowerCamelCase_  :    Optional[Any]   	=			self.size + other.size
  
  							# Update min.node
  							if self.min_node.val > other.min_node.val:
  									lowerCamelCase_  :    Optional[Any]   	=			other.min_node
  							# Merge
  
  							# Order roots by left_subtree_size
  							lowerCamelCase_  :    Dict   	=			[]
  							lowerCamelCase_	,			lowerCamelCase_  :    Optional[Any]   	=			self.bottom_root, other.bottom_root
  							while i or j:
  									if i and ((not j) or i.left_tree_size < j.left_tree_size):
  											combined_roots_list.append((i, True)	)
  											lowerCamelCase_  :    Tuple   	=			i.parent
  									else:
  											combined_roots_list.append((j, False)	)
  											lowerCamelCase_  :    Optional[Any]   	=			j.parent
          # Insert links between them
  							for i in range(len(lowerCamelCase	) - 1	):
  									if combined_roots_list[i][1] != combined_roots_list[i + 1][1]:
  											lowerCamelCase_  :    Dict   	=			combined_roots_list[i + 1][0]
  											lowerCamelCase_  :    Optional[int]   	=			combined_roots_list[i][0]
          # Consecutively merge roots with same left_tree_size
  							lowerCamelCase_  :    Any   	=			combined_roots_list[0][0]
  							while i.parent:
  									if (
  									    (i.left_tree_size == i.parent.left_tree_size) and (not i.parent.parent)
  									) or (
  									    i.left_tree_size == i.parent.left_tree_size
  									    and i.left_tree_size != i.parent.parent.left_tree_size
  									):
  											# Neighbouring Nodes
  											lowerCamelCase_  :    Dict   	=			i.left
  											lowerCamelCase_  :    int   	=			i.parent.parent
  
  											# Merging trees
  											lowerCamelCase_  :    str   	=			i.merge_trees(i.parent	)
  
  											# Updating links
  											lowerCamelCase_  :    Union[str, Any]   	=			previous_node
  											lowerCamelCase_  :    List[Any]   	=			next_node
  											if previous_node:
  													lowerCamelCase_  :    List[str]   	=			i
  											if next_node:
  													lowerCamelCase_  :    Dict   	=			i
  									else:
  											lowerCamelCase_  :    List[str]   	=			i.parent
          # Updating self.bottom_root
  							while i.left:
  									lowerCamelCase_  :    Union[str, Any]   	=			i.left
  							lowerCamelCase_  :    Any   	=			i
  
  							# Update other
  							lowerCamelCase_  :    Optional[int]   	=			self.size
  							lowerCamelCase_  :    Optional[Any]   	=			self.bottom_root
  							lowerCamelCase_  :    str   	=			self.min_node
  
  							# Return the merged heap
  							return self
  
  
  
  					def      __a    (		self    :    Any					,			lowerCamelCase    :    Dict	):
  							if self.size == 0:
  									lowerCamelCase_  :    Tuple   	=			Node(lowerCamelCase	)
  									lowerCamelCase_  :    Optional[int]   	=			1
  									lowerCamelCase_  :    Optional[int]   	=			self.bottom_root
  							else:
  									# Create new node
  									lowerCamelCase_  :    List[Any]   	=			Node(lowerCamelCase	)
  
  									# Update size
  									self.size += 1
  
  									# update min_node
  									if val < self.min_node.val:
  											lowerCamelCase_  :    List[str]   	=			new_node
  									# Put new_node as a bottom_root in heap
  									lowerCamelCase_  :    int   	=			new_node
  									lowerCamelCase_  :    Tuple   	=			self.bottom_root
  									lowerCamelCase_  :    Any   	=			new_node
  
  									# Consecutively merge roots with same left_tree_size
  									while (
  									    self.bottom_root.parent
  									    and self.bottom_root.left_tree_size
  									    == self.bottom_root.parent.left_tree_size
  									):
  											# Next node
  											lowerCamelCase_  :    Any   	=			self.bottom_root.parent.parent
  
  											# Merge
  											lowerCamelCase_  :    int   	=			self.bottom_root.merge_trees(self.bottom_root.parent	)
  
  											# Update Links
  											lowerCamelCase_  :    Dict   	=			next_node
  											lowerCamelCase_  :    Optional[Any]   	=			None
  											if next_node:
  													lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root
  
  
  
  					def      __a    (		self    :    Dict	):
  							return self.min_node.val
  
  
  
  					def      __a    (		self    :    int	):
  							return self.size == 0
  
  
  
  					def      __a    (		self    :    List[Any]	):
  							lowerCamelCase_  :    Tuple   	=			self.min_node.val
  
  							# Last element in heap corner case
  							if self.size == 1:
  									# Update size
  									lowerCamelCase_  :    Optional[Any]   	=			0
  
  									# Update bottom root
  									lowerCamelCase_  :    Any   	=			None
  
  									# Update min_node
  									lowerCamelCase_  :    List[str]   	=			None
  
  									return min_value
  							# No right subtree corner case
  							# The structure of the tree implies that this should be the bottom root
  							# and there is at least one other root
  							if self.min_node.right is None:
  									# Update size
  									self.size -= 1
  
  									# Update bottom root
  									lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root.parent
  									lowerCamelCase_  :    Union[str, Any]   	=			None
  
  									# Update min_node
  									lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root
  									lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root.parent
  									while i:
  											if i.val < self.min_node.val:
  													lowerCamelCase_  :    Union[str, Any]   	=			i
  											lowerCamelCase_  :    Optional[int]   	=			i.parent
  									return min_value
  							# General case
  							# Find the BinomialHeap of the right subtree of min_node
  							lowerCamelCase_  :    Dict   	=			self.min_node.right
  							lowerCamelCase_  :    Dict   	=			None
  							lowerCamelCase_  :    Tuple   	=			bottom_of_new
  							lowerCamelCase_  :    Dict   	=			1
  
  							# Size, min_node and bottom_root
  							while bottom_of_new.left:
  									lowerCamelCase_  :    Any   	=			size_of_new * 2 + 1
  									lowerCamelCase_  :    Tuple   	=			bottom_of_new.left
  									if bottom_of_new.val < min_of_new.val:
  											lowerCamelCase_  :    int   	=			bottom_of_new
          # Corner case of single root on top left path
  							if (not self.min_node.left) and (not self.min_node.parent):
  									lowerCamelCase_  :    Union[str, Any]   	=			size_of_new
  									lowerCamelCase_  :    Union[str, Any]   	=			bottom_of_new
  									lowerCamelCase_  :    Tuple   	=			min_of_new
  									# print("Single root, multiple nodes case")
  									return min_value
  							# Remaining cases
  							# Construct heap of right subtree
  							lowerCamelCase_  :    Optional[Any]   	=			BinomialHeap(
  							    bottom_root=lowerCamelCase					,			min_node=lowerCamelCase					,			heap_size=lowerCamelCase	)
  
  							# Update size
  							lowerCamelCase_  :    Union[str, Any]   	=			self.size - 1 - size_of_new
  
  							# Neighbour nodes
  							lowerCamelCase_  :    Union[str, Any]   	=			self.min_node.left
  							lowerCamelCase_  :    Optional[Any]   	=			self.min_node.parent
  
  							# Initialize new bottom_root and min_node
  							lowerCamelCase_  :    Optional[int]   	=			previous_node or next_node
  							lowerCamelCase_  :    Tuple   	=			next_node
  
  							# Update links of previous_node and search below for new min_node and
  							# bottom_root
  							if previous_node:
  									lowerCamelCase_  :    Dict   	=			next_node
  
  									# Update bottom_root and search for min_node below
  									lowerCamelCase_  :    int   	=			previous_node
  									lowerCamelCase_  :    Tuple   	=			previous_node
  									while self.bottom_root.left:
  											lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root.left
  											if self.bottom_root.val < self.min_node.val:
  													lowerCamelCase_  :    Any   	=			self.bottom_root
  							if next_node:
  									lowerCamelCase_  :    Any   	=			previous_node
  
  									# Search for new min_node above min_node
  									lowerCamelCase_  :    List[str]   	=			next_node
  									while i:
  											if i.val < self.min_node.val:
  													lowerCamelCase_  :    List[Any]   	=			i
  											lowerCamelCase_  :    Dict   	=			i.parent
          # Merge heaps
  							self.merge_heaps(lowerCamelCase	)
  
  							return min_value
  
  
  
  					def      __a    (		self    :    Optional[int]	):
  							lowerCamelCase_  :    List[Any]   	=			self.bottom_root
  							while top_root.parent:
  									lowerCamelCase_  :    Tuple   	=			top_root.parent
  							# preorder
  							lowerCamelCase_  :    str   	=			[]
  							self.__traversal(lowerCamelCase					,			lowerCamelCase	)
  							return heap_pre_order
  
  
  
  					def      __a    (		self    :    Dict					,			lowerCamelCase    :    Tuple					,			lowerCamelCase    :    Dict					,			lowerCamelCase    :    List[str]=0	):
  							if curr_node:
  									preorder.append((curr_node.val, level)	)
  									self.__traversal(curr_node.left					,			lowerCamelCase					,			level + 1	)
  									self.__traversal(curr_node.right					,			lowerCamelCase					,			level + 1	)
  							else:
  									preorder.append(('#', level)	)
  
  
  
  
  
  					def __str__(		self    :    List[Any]	):
  							if self.is_empty():
  									return ""
  							lowerCamelCase_  :    str   	=			self.pre_order()
  
  							return "\n".join(('-' * level + str(lowerCamelCase	)) for value, level in preorder_heap	)
  
  
  # Unit Tests
  if __name__ == "__main__":
  		import doctest
  
  		doctest.testmod()
  