




'''simple docstring'''




#
# Sort large text files in a minimum amount of memory
#
import argparse
import os

class 						lowerCAmelCase_   :







  '''simple docstring'''







  lowerCAmelCase_				:					Optional[Any]				     =						"""block_{0}.dat"""


  def __init__(						self    :	Dict						,		_UpperCAmelCase    :	Optional[Any] ):






   """simple docstring"""






   UpperCAmelCase__										=    filename
   UpperCAmelCase__										=    []


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[Any]						,		_UpperCAmelCase    :	Optional[Any]						,		_UpperCAmelCase    :	Optional[Any] ):






   """simple docstring"""






   UpperCAmelCase__										=    self.BLOCK_FILENAME_FORMAT.format(lowerCAmelCase_ )
   with open(lowerCAmelCase_						,		"""w""" ) as file:
    file.write(lowerCAmelCase_ )
   self.block_filenames.append(lowerCAmelCase_ )


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Tuple ):






   """simple docstring"""






   return self.block_filenames


  def 							SCREAMING_SNAKE_CASE__			(						self    :	List[Any]						,		_UpperCAmelCase    :	Tuple						,		_UpperCAmelCase    :	Optional[int]=None ):






   """simple docstring"""






   UpperCAmelCase__										=    0
   with open(self.filename ) as file:
    while True:
     UpperCAmelCase__										=    file.readlines(lowerCAmelCase_ )

     if lines == []:
      break

     if sort_key is None:
      lines.sort()
     else:
      lines.sort(key=lowerCAmelCase_ )

     self.write_block("""""".join(lowerCAmelCase_ )						,		lowerCAmelCase_ )
     i += 1



  def 							SCREAMING_SNAKE_CASE__			(						self    :	Any ):






   """simple docstring"""






   map(os.remove						,		self.block_filenames )

class 						lowerCAmelCase_   :







  '''simple docstring'''







  def 							SCREAMING_SNAKE_CASE__			(						self    :	Any						,		_UpperCAmelCase    :	str ):






   """simple docstring"""






   UpperCAmelCase__										=    -1
   UpperCAmelCase__										=    None

   for i in range(len(lowerCAmelCase_ ) ):
    if min_str is None or choices[i] < min_str:
     UpperCAmelCase__										=    i

   return min_index

class 						lowerCAmelCase_   :







  '''simple docstring'''







  def __init__(						self    :	Tuple						,		_UpperCAmelCase    :	Tuple ):






   """simple docstring"""






   UpperCAmelCase__										=    files
   UpperCAmelCase__										=    set()
   UpperCAmelCase__										=    len(lowerCAmelCase_ )
   UpperCAmelCase__										=    {i: None for i in range(self.num_buffers )}


  def 							SCREAMING_SNAKE_CASE__			(						self    :	int ):






   """simple docstring"""






   return {
       i: self.buffers[i] for i in range(self.num_buffers ) if i not in self.empty
   }


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Dict ):






   """simple docstring"""






   for i in range(self.num_buffers ):
    if self.buffers[i] is None and i not in self.empty:
     UpperCAmelCase__										=    self.files[i].readline()

     if self.buffers[i] == "":
      self.empty.add(lowerCAmelCase_ )
      self.files[i].close()

   if len(self.empty ) == self.num_buffers:
    return False

   return True



  def 							SCREAMING_SNAKE_CASE__			(						self    :	int						,		_UpperCAmelCase    :	Union[str, Any] ):






   """simple docstring"""






   UpperCAmelCase__										=    self.buffers[index]
   UpperCAmelCase__										=    None

   return value

class 						lowerCAmelCase_   :







  '''simple docstring'''







  def __init__(						self    :	Any						,		_UpperCAmelCase    :	int ):






   """simple docstring"""






   UpperCAmelCase__										=    merge_strategy


  def 							SCREAMING_SNAKE_CASE__			(						self    :	List[str]						,		_UpperCAmelCase    :	Dict						,		_UpperCAmelCase    :	Union[str, Any]						,		_UpperCAmelCase    :	Union[str, Any] ):






   """simple docstring"""






   UpperCAmelCase__										=    FilesArray(self.get_file_handles(lowerCAmelCase_						,		lowerCAmelCase_ ) )
   with open(lowerCAmelCase_						,		"""w"""						,		lowerCAmelCase_ ) as outfile:
    while buffers.refresh():
     UpperCAmelCase__										=    self.merge_strategy.select(buffers.get_dict() )
     outfile.write(buffers.unshift(lowerCAmelCase_ ) )



  def 							SCREAMING_SNAKE_CASE__			(						self    :	List[str]						,		_UpperCAmelCase    :	Union[str, Any]						,		_UpperCAmelCase    :	str ):






   """simple docstring"""






   UpperCAmelCase__										=    {}

   for i in range(len(lowerCAmelCase_ ) ):
    UpperCAmelCase__										=    open(filenames[i]						,		"""r"""						,		lowerCAmelCase_ )  # noqa: UP015

   return files

class 						lowerCAmelCase_   :







  '''simple docstring'''







  def __init__(						self    :	int						,		_UpperCAmelCase    :	List[Any] ):






   """simple docstring"""






   UpperCAmelCase__										=    block_size


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[Any]						,		_UpperCAmelCase    :	Optional[int]						,		_UpperCAmelCase    :	int=None ):






   """simple docstring"""






   UpperCAmelCase__										=    self.get_number_blocks(lowerCAmelCase_						,		self.block_size )
   UpperCAmelCase__										=    FileSplitter(lowerCAmelCase_ )
   splitter.split(self.block_size						,		lowerCAmelCase_ )

   UpperCAmelCase__										=    FileMerger(NWayMerge() )
   UpperCAmelCase__										=    self.block_size / (num_blocks + 1)
   merger.merge(splitter.get_block_filenames()						,		filename + """.out"""						,		lowerCAmelCase_ )

   splitter.cleanup()



  def 							SCREAMING_SNAKE_CASE__			(						self    :	int						,		_UpperCAmelCase    :	Dict						,		_UpperCAmelCase    :	Tuple ):






   """simple docstring"""






   return (os.stat(lowerCAmelCase_ ).st_size / block_size) + 1

def    _UpperCamelCase      (				SCREAMING_SNAKE_CASE__			:      Dict       ):
 '''simple docstring'''





 if string[-1].lower() == "k":
  return int(string[:-1]       ) * 1024
 elif string[-1].lower() == "m":
  return int(string[:-1]       ) * 1024 * 1024
 elif string[-1].lower() == "g":
  return int(string[:-1]       ) * 1024 * 1024 * 1024
 else:
  return int(SCREAMING_SNAKE_CASE__       )

def    _UpperCamelCase      (				):
 '''simple docstring'''





 UpperCAmelCase__										=    argparse.ArgumentParser()
 parser.add_argument(
     """-m"""       ,						"""--mem"""       ,						help="""amount of memory to use for sorting"""       ,						default="""100M"""       )
 parser.add_argument(
     """filename"""       ,						metavar="""<filename>"""       ,						nargs=1       ,						help="""name of file to sort"""       )
 UpperCAmelCase__										=    parser.parse_args()

 UpperCAmelCase__										=    ExternalSort(parse_memory(args.mem       )       )
 sorter.sort(args.filename[0]       )


if __name__ == "__main__":
    main()






'''simple docstring'''




import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


UpperCAmelCase_       				=      logging.getLogger(__name__)

class 						lowerCAmelCase_   :







  '''simple docstring'''







  def __init__(						self    :	int ):






   """simple docstring"""






   UpperCAmelCase__										=    False


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Any						,		_UpperCAmelCase    :	Optional[Any]						,		_UpperCAmelCase    :	str						,		_UpperCAmelCase    :	Tuple						,		_UpperCAmelCase    :	Optional[Any] ):






   """simple docstring"""






   if not self.initialized:
    UpperCAmelCase__										=    RagRetriever(
        _UpperCAmelCase						,		question_encoder_tokenizer=_UpperCAmelCase						,		generator_tokenizer=_UpperCAmelCase						,		index=_UpperCAmelCase						,		init_retrieval=_UpperCAmelCase						,		)
    UpperCAmelCase__										=    True


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Union[str, Any] ):






   """simple docstring"""






   self.retriever.index.init_index()


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Union[str, Any] ):






   """simple docstring"""






   del self.retriever
   UpperCAmelCase__										=    False



  def 							SCREAMING_SNAKE_CASE__			(						self    :	Any						,		_UpperCAmelCase    :	Union[str, Any]						,		_UpperCAmelCase    :	Optional[int] ):






   """simple docstring"""






   UpperCAmelCase__							,			UpperCAmelCase__										=    self.retriever._main_retrieve(_UpperCAmelCase						,		_UpperCAmelCase )
   UpperCAmelCase__										=    self.retriever.index.get_doc_dicts(_UpperCAmelCase )
   return doc_ids, retrieved_doc_embeds, doc_dicts

class 						lowerCAmelCase_   ( lowerCamelCase_   ):







  '''simple docstring'''







  def __init__(						self    :	Optional[Any]						,		_UpperCAmelCase    :	List[Any]						,		_UpperCAmelCase    :	Dict						,		_UpperCAmelCase    :	str						,		_UpperCAmelCase    :	Any						,		_UpperCAmelCase    :	List[Any]=None ):






   """simple docstring"""






   if index is not None and index.is_initialized() and len(_UpperCAmelCase ) > 0:
    raise ValueError(
        """When using Ray for distributed fine-tuning, """
        """you'll need to provide the paths instead, """
        """as the dataset and the index are loaded """
        """separately. More info in examples/rag/use_own_knowledge_dataset.py """ )

   super().__init__(
       _UpperCAmelCase						,		question_encoder_tokenizer=_UpperCAmelCase						,		generator_tokenizer=_UpperCAmelCase						,		index=_UpperCAmelCase						,		init_retrieval=_UpperCAmelCase						,		)

   UpperCAmelCase__										=    retrieval_workers
   UpperCAmelCase__										=    question_encoder_tokenizer
   UpperCAmelCase__										=    generator_tokenizer
   if len(self.retrieval_workers ) > 0:
    ray.get(
        [
            worker.create_rag_retriever.remote(_UpperCAmelCase						,		_UpperCAmelCase						,		_UpperCAmelCase						,		_UpperCAmelCase )
            for worker in self.retrieval_workers
        ] )


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[int] ):






   """simple docstring"""






   logger.info("""initializing retrieval""" )

   if len(self.retrieval_workers ) > 0:
    ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers] )
   else:
    # Non-distributed training. Load index into this same process.
    self.index.init_index()


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Tuple						,		_UpperCAmelCase    :	str						,		_UpperCAmelCase    :	Dict ):






   """simple docstring"""






   if len(self.retrieval_workers ) > 0:
    # Select a random retrieval actor.
    UpperCAmelCase__										=    self.retrieval_workers[random.randint(0						,		len(self.retrieval_workers ) - 1 )]
    UpperCAmelCase__							,			UpperCAmelCase__							,			UpperCAmelCase__										=    ray.get(
        random_worker.retrieve.remote(_UpperCAmelCase						,		_UpperCAmelCase ) )
   else:
    UpperCAmelCase__							,			UpperCAmelCase__										=    self._main_retrieve(_UpperCAmelCase						,		_UpperCAmelCase )
    UpperCAmelCase__										=    self.index.get_doc_dicts(_UpperCAmelCase )
   return retrieved_doc_embeds, doc_ids, doc_dicts


  @classmethod
  def 							SCREAMING_SNAKE_CASE__			(						cls    :	str						,		_UpperCAmelCase    :	Optional[Any]						,		_UpperCAmelCase    :	str=None						,		**_UpperCAmelCase    :	Tuple ):






   """simple docstring"""






   return super(_UpperCAmelCase						,		cls ).get_tokenizers(_UpperCAmelCase						,		_UpperCAmelCase						,		**_UpperCAmelCase )


  @classmethod
  def 							SCREAMING_SNAKE_CASE__			(						cls    :	int						,		_UpperCAmelCase    :	Optional[int]						,		_UpperCAmelCase    :	Optional[int]						,		_UpperCAmelCase    :	Union[str, Any]=None						,		**_UpperCAmelCase    :	int ):






   """simple docstring"""






   UpperCAmelCase__										=    kwargs.pop("""config"""						,		_UpperCAmelCase ) or RagConfig.from_pretrained(_UpperCAmelCase						,		**_UpperCAmelCase )
   UpperCAmelCase__										=    RagTokenizer.from_pretrained(_UpperCAmelCase						,		config=_UpperCAmelCase )
   UpperCAmelCase__										=    rag_tokenizer.question_encoder
   UpperCAmelCase__										=    rag_tokenizer.generator

   if indexed_dataset is not None:
    UpperCAmelCase__										=    """custom"""
    UpperCAmelCase__										=    CustomHFIndex(config.retrieval_vector_size						,		_UpperCAmelCase )
   else:
    UpperCAmelCase__										=    cls._build_index(_UpperCAmelCase )

   return cls(
       _UpperCAmelCase						,		question_encoder_tokenizer=_UpperCAmelCase						,		generator_tokenizer=_UpperCAmelCase						,		retrieval_workers=_UpperCAmelCase						,		index=_UpperCAmelCase						,		)



  def 							SCREAMING_SNAKE_CASE__			(						self    :	Dict ):






   """simple docstring"""






   logger.info("""re-loading the new dataset with embeddings""" )
   # access from the training loop

   ray.get([worker.clear_object.remote() for worker in self.retrieval_workers] )

   # build the index object again
   UpperCAmelCase__										=    self._build_index(self.config )

   ray.get(
       [
           worker.create_rag_retriever.remote(
               self.config						,		self.question_encoder_tokenizer						,		self.generator_tokenizer						,		_UpperCAmelCase )
           for worker in self.retrieval_workers
       ] )
