


"""simple docstring"""






import multiprocessing
import os
from typing import BinaryIO, Optional, Union

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.csv.csv import Csv
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader


class 	A_       (     _UpperCAmelCase   ):





  """simple docstring"""

  def __init__(						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase = None  ,					__UpperCAmelCase = None  ,					__UpperCAmelCase = None  ,					__UpperCAmelCase = False  ,					__UpperCAmelCase = False  ,					__UpperCAmelCase = None  ,					**__UpperCAmelCase  ,					)				->					Union[str, Any]:
     super().__init__(
         __UpperCAmelCase  ,					split=__UpperCAmelCase  ,					features=__UpperCAmelCase  ,					cache_dir=__UpperCAmelCase  ,					keep_in_memory=__UpperCAmelCase  ,					streaming=__UpperCAmelCase  ,					num_proc=__UpperCAmelCase  ,					**__UpperCAmelCase  ,					)
     a   :	List[str]						=    path_or_paths if isinstance(__UpperCAmelCase  ,					__UpperCAmelCase						) else {self.split: path_or_paths}
     a   :	Union[str, Any]						=    Csv(
         cache_dir=__UpperCAmelCase  ,					data_files=__UpperCAmelCase  ,					features=__UpperCAmelCase  ,					**__UpperCAmelCase  ,					)






  def 							lowercase_     (						self						)				->					Any:
     # Build iterable dataset
     if self.streaming:
        a   :	str						=    self.builder.as_streaming_dataset(split=self.split						)
     # Build regular (map-style) dataset
     else:
        a   :	str						=    None
        a   :	Dict						=    None
        a   :	List[Any]						=    None
        a   :	List[str]						=    None

        self.builder.download_and_prepare(
            download_config=__UpperCAmelCase  ,					download_mode=__UpperCAmelCase  ,					verification_mode=__UpperCAmelCase  ,					base_path=__UpperCAmelCase  ,					num_proc=self.num_proc  ,					)
        a   :	Tuple						=    self.builder.as_dataset(
            split=self.split  ,					verification_mode=__UpperCAmelCase  ,					in_memory=self.keep_in_memory						)
     return dataset



class 	A_       :





  """simple docstring"""

  def __init__(						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase = None  ,					__UpperCAmelCase = None  ,					**__UpperCAmelCase  ,					)				->					int:
     if num_proc is not None and num_proc <= 0:
        raise ValueError(f'num_proc {num_proc} must be an integer > 0.'						)

     a   :	Any						=    dataset
     a   :	str						=    path_or_buf
     a   :	int						=    batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
     a   :	Dict						=    num_proc
     a   :	Tuple						=    'utf-8'
     a   :	int						=    to_csv_kwargs
  def 							lowercase_     (						self						)				->					int:
     a   :	int						=    self.to_csv_kwargs.pop('path_or_buf'  ,					__UpperCAmelCase						)
     a   :	str						=    self.to_csv_kwargs.pop('header'  ,					__UpperCAmelCase						)
     a   :	int						=    self.to_csv_kwargs.pop('index'  ,					__UpperCAmelCase						)

     if isinstance(self.path_or_buf  ,					(str, bytes, os.PathLike)						):
        with open(self.path_or_buf  ,					'wb+'						) as buffer:
           a   :	Any						=    self._write(file_obj=__UpperCAmelCase  ,					header=__UpperCAmelCase  ,					index=__UpperCAmelCase  ,					**self.to_csv_kwargs						)
     else:
        a   :	Tuple						=    self._write(file_obj=self.path_or_buf  ,					header=__UpperCAmelCase  ,					index=__UpperCAmelCase  ,					**self.to_csv_kwargs						)
     return written
  def 							lowercase_     (						self  ,					__UpperCAmelCase						)				->					Optional[Any]:
     a	, a	, a	, a   :	Dict						=    args

     a   :	str						=    query_table(
         table=self.dataset.data  ,					key=slice(__UpperCAmelCase  ,					offset + self.batch_size						)  ,					indices=self.dataset._indices  ,					)
     a   :	Any						=    batch.to_pandas().to_csv(
         path_or_buf=__UpperCAmelCase  ,					header=header if (offset == 0) else False  ,					index=__UpperCAmelCase  ,					**__UpperCAmelCase						)
     return csv_str.encode(self.encoding						)






  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					**__UpperCAmelCase						)				->					int:
     a   :	Union[str, Any]						=    0

     if self.num_proc is None or self.num_proc == 1:
        for offset in logging.tqdm(
            range(0  ,					len(self.dataset						)  ,					self.batch_size						)  ,					unit='ba'  ,					disable=not logging.is_progress_bar_enabled()  ,					desc='Creating CSV from Arrow format'  ,					):
           a   :	List[Any]						=    self._batch_csv((offset, header, index, to_csv_kwargs)						)
           written += file_obj.write(__UpperCAmelCase						)

     else:
        a	, a   :	Any						=    len(self.dataset						), self.batch_size
        with multiprocessing.Pool(self.num_proc						) as pool:
           for csv_str in logging.tqdm(
               pool.imap(
                   self._batch_csv  ,					[(offset, header, index, to_csv_kwargs) for offset in range(0  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)]  ,					)  ,					total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size  ,					unit='ba'  ,					disable=not logging.is_progress_bar_enabled()  ,					desc='Creating CSV from Arrow format'  ,					):
              written += file_obj.write(__UpperCAmelCase						)

     return written









"""simple docstring"""






import logging
import random

import ray

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


SCREAMING_SNAKE_CASE__				:  Optional[Any]						= logging.getLogger(__name__)


class 	A_       :





  """simple docstring"""

  def __init__(						self						)				->					Dict:
     a   :	List[str]						=    False
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					List[str]:
     if not self.initialized:
        a   :	Dict						=    RagRetriever(
            __UpperCAmelCase  ,					question_encoder_tokenizer=__UpperCAmelCase  ,					generator_tokenizer=__UpperCAmelCase  ,					index=__UpperCAmelCase  ,					init_retrieval=__UpperCAmelCase  ,					)
        a   :	Dict						=    True
  def 							lowercase_     (						self						)				->					List[Any]:
     self.retriever.index.init_index()
  def 							lowercase_     (						self						)				->					List[Any]:
     # delete the old self.retriever object before assigning the new index
     del self.retriever
     a   :	Dict						=    False






  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Optional[int]:
     a	, a   :	List[str]						=    self.retriever._main_retrieve(__UpperCAmelCase  ,					__UpperCAmelCase						)
     a   :	List[Any]						=    self.retriever.index.get_doc_dicts(__UpperCAmelCase						)
     return doc_ids, retrieved_doc_embeds, doc_dicts



class 	A_       (     _UpperCAmelCase   ):





  """simple docstring"""

  def __init__(						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase=None						)				->					Tuple:
     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  ,					)

     a   :	Tuple						=    retrieval_workers
     a   :	List[str]						=    question_encoder_tokenizer
     a   :	Union[str, Any]						=    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 							lowercase_     (						self						)				->					Tuple:
     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 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Tuple:
     if len(self.retrieval_workers						) > 0:
        # Select a random retrieval actor.
        a   :	Dict						=    self.retrieval_workers[random.randint(0  ,					len(self.retrieval_workers						) - 1						)]
        a	, a	, a   :	Any						=    ray.get(
            random_worker.retrieve.remote(__UpperCAmelCase  ,					__UpperCAmelCase						)						)
     else:
        a	, a   :	str						=    self._main_retrieve(__UpperCAmelCase  ,					__UpperCAmelCase						)
        a   :	str						=    self.index.get_doc_dicts(__UpperCAmelCase						)
     return retrieved_doc_embeds, doc_ids, doc_dicts
  @classmethod
  def 							lowercase_     (						cls  ,					__UpperCAmelCase  ,					__UpperCAmelCase=None  ,					**__UpperCAmelCase						)				->					int:
     return super(__UpperCAmelCase  ,					cls						).get_tokenizers(__UpperCAmelCase  ,					__UpperCAmelCase  ,					**__UpperCAmelCase						)
  @classmethod
  def 							lowercase_     (						cls  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase=None  ,					**__UpperCAmelCase						)				->					str:
     a   :	Union[str, Any]						=    kwargs.pop('config'  ,					__UpperCAmelCase						) or RagConfig.from_pretrained(__UpperCAmelCase  ,					**__UpperCAmelCase						)
     a   :	Optional[Any]						=    RagTokenizer.from_pretrained(__UpperCAmelCase  ,					config=__UpperCAmelCase						)
     a   :	List[str]						=    rag_tokenizer.question_encoder
     a   :	int						=    rag_tokenizer.generator

     if indexed_dataset is not None:
        a   :	Optional[int]						=    'custom'
        a   :	str						=    CustomHFIndex(config.retrieval_vector_size  ,					__UpperCAmelCase						)
     else:
        a   :	Dict						=    cls._build_index(__UpperCAmelCase						)

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






  def 							lowercase_     (						self						)				->					List[Any]:
     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
     a   :	Optional[Any]						=    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
         ]						)





