



'''simple docstring'''






from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


A_				:					List[Any]							=      logging.get_logger(__name__)



class 				lowercase			(    _lowerCamelCase							):


 """simple docstring"""


 UpperCAmelCase		      =    ["""input_values""", """attention_mask"""]







 def __init__(				self    ,a_=1    ,a_=16_000    ,a_=0.0    ,a_=False    ,a_=True    ,**a_    ,)						->  Dict:
        super().__init__(feature_size=a_    ,sampling_rate=a_    ,padding_value=a_    ,**a_				)
        _UpperCAmelCase    :		str     	=					return_attention_mask
        _UpperCAmelCase    :		Union[str, Any]     	=					do_normalize







 @staticmethod
 def _snake_case					(				a_    ,a_    ,a_ = 0.0				)						->  List[np.ndarray]:
        if attention_mask is not None:
               _UpperCAmelCase    :		Dict     	=					np.array(a_    ,np.intaa				)
               _UpperCAmelCase    :		Optional[int]     	=					[]

               for vector, length in zip(a_    ,attention_mask.sum(-1				)				):
                      _UpperCAmelCase    :		Dict     	=					(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7				)
                      if length < normed_slice.shape[0]:
                             _UpperCAmelCase    :		Union[str, Any]     	=					padding_value

                      normed_input_values.append(a_				)
        else:
               _UpperCAmelCase    :		int     	=					[(x - x.mean()) / np.sqrt(x.var() + 1E-7				) for x in input_values]

        return normed_input_values







 def __call__(				self    ,a_    ,a_ = False    ,a_ = None    ,a_ = False    ,a_ = None    ,a_ = None    ,a_ = None    ,a_ = None    ,**a_    ,)						->  BatchFeature:

        if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                      raise ValueError(
                          f'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
                          f''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
                          f''' {self.sampling_rate} and not {sampling_rate}.'''				)
        else:
               logger.warning(
                   """It is strongly recommended to pass the ``sampling_rate`` argument to this function. """
                   """Failing to do so can result in silent errors that might be hard to debug."""				)

        _UpperCAmelCase    :		Union[str, Any]     	=					isinstance(a_    ,np.ndarray				) and len(raw_speech.shape				) > 1
        if is_batched_numpy and len(raw_speech.shape				) > 2:
               raise ValueError(f'''Only mono-channel audio is supported for input to {self}'''				)
        _UpperCAmelCase    :		Optional[int]     	=					is_batched_numpy or (
            isinstance(a_    ,(list, tuple)				) and (isinstance(raw_speech[0]    ,(np.ndarray, tuple, list)				))
        )

        # always return batch
        if not is_batched:
               _UpperCAmelCase    :		List[Any]     	=					[raw_speech]

        # convert into correct format for padding
        _UpperCAmelCase    :		Union[str, Any]     	=					BatchFeature({"""input_values""": raw_speech}				)

        _UpperCAmelCase    :		str     	=					self.pad(
            a_    ,padding=a_    ,max_length=a_    ,truncation=a_    ,pad_to_multiple_of=a_    ,return_attention_mask=a_    ,)

        # convert input values to correct format
        _UpperCAmelCase    :		Any     	=					padded_inputs["""input_values"""]
        if not isinstance(input_values[0]    ,np.ndarray				):
               _UpperCAmelCase    :		str     	=					[np.asarray(a_    ,dtype=np.floataa				) for array in input_values]
        elif (
            not isinstance(a_    ,np.ndarray				)
            and isinstance(input_values[0]    ,np.ndarray				)
            and input_values[0].dtype is np.dtype(np.floataa				)
        ):
               _UpperCAmelCase    :		int     	=					[array.astype(np.floataa				) for array in input_values]
        elif isinstance(a_    ,np.ndarray				) and input_values.dtype is np.dtype(np.floataa				):
               _UpperCAmelCase    :		List[str]     	=					input_values.astype(np.floataa				)

        # convert attention_mask to correct format
        _UpperCAmelCase    :		Any     	=					padded_inputs.get("""attention_mask"""				)
        if attention_mask is not None:
               _UpperCAmelCase    :		Dict     	=					[np.asarray(a_    ,dtype=np.intaa				) for array in attention_mask]

        # zero-mean and unit-variance normalization
        if self.do_normalize:
               _UpperCAmelCase    :		List[Any]     	=					(
                   attention_mask
                   if self._get_padding_strategies(a_    ,max_length=a_				) is not PaddingStrategy.DO_NOT_PAD
                   else None
               )
               _UpperCAmelCase    :		Dict     	=					self.zero_mean_unit_var_norm(
                   padded_inputs["""input_values"""]    ,attention_mask=a_    ,padding_value=self.padding_value				)

        if return_tensors is not None:
               _UpperCAmelCase    :		Union[str, Any]     	=					padded_inputs.convert_to_tensors(a_				)

        return padded_inputs








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__			:


						def __init__(			self    :	Optional[Any]      ):
													a__:							Tuple									=False


						def 	_lowerCamelCase      (			self    :	Dict   ,							_a    :	Any   ,							_a    :	Union[str, Any]   ,							_a    :	List[str]   ,							_a    :	Optional[int]      ):
													if not self.initialized:
																				a__:							Any									=RagRetriever(
																				    _a   ,							question_encoder_tokenizer=_a   ,							generator_tokenizer=_a   ,							index=_a   ,							init_retrieval=_a   ,							)
																				a__:							str									=True


						def 	_lowerCamelCase      (			self    :	List[str]      ):
													self.retriever.index.init_index()


						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													# delete the old self.retriever object before assigning the new index
													del self.retriever
													a__:							Dict									=False







						def 	_lowerCamelCase      (			self    :	str   ,							_a    :	List[Any]   ,							_a    :	Any      ):
													a__				,      a__:							List[str]									=self.retriever._main_retrieve(_a   ,							_a      )
													a__:							List[str]									=self.retriever.index.get_doc_dicts(_a      )
													return doc_ids, retrieved_doc_embeds, doc_dicts


class 						lowerCamelCase__			(		_a ):


						def __init__(			self    :	Optional[int]   ,							_a    :	List[Any]   ,							_a    :	Any   ,							_a    :	Optional[Any]   ,							_a    :	Union[str, Any]   ,							_a    :	str=None      ):
													if index is not None and index.is_initialized() and len(_a      ) > 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__(
													    _a   ,							question_encoder_tokenizer=_a   ,							generator_tokenizer=_a   ,							index=_a   ,							init_retrieval=_a   ,							)

													a__:							Optional[Any]									=retrieval_workers
													a__:							str									=question_encoder_tokenizer
													a__:							Optional[int]									=generator_tokenizer
													if len(self.retrieval_workers      ) > 0:
																				ray.get(
																				    [
																				        worker.create_rag_retriever.remote(_a   ,							_a   ,							_a   ,							_a      )
																				        for worker in self.retrieval_workers
																				    ]      )


						def 	_lowerCamelCase      (			self    :	List[Any]      ):
													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 	_lowerCamelCase      (			self    :	List[Any]   ,							_a    :	Union[str, Any]   ,							_a    :	Union[str, Any]      ):
													if len(self.retrieval_workers      ) > 0:
																				# Select a random retrieval actor.
																				a__:							Tuple									=self.retrieval_workers[random.randint(0   ,							len(self.retrieval_workers      ) - 1      )]
																				a__				,      a__				,      a__:							Union[str, Any]									=ray.get(
																				    random_worker.retrieve.remote(_a   ,							_a      )      )
													else:
																				a__				,      a__:							List[Any]									=self._main_retrieve(_a   ,							_a      )
																				a__:							List[str]									=self.index.get_doc_dicts(_a      )
													return retrieved_doc_embeds, doc_ids, doc_dicts


						@classmethod
						def 	_lowerCamelCase      (			cls    :	str   ,							_a    :	Union[str, Any]   ,							_a    :	Any=None   ,							**_a    :	Dict      ):
													return super(_a   ,							cls      ).get_tokenizers(_a   ,							_a   ,							**_a      )


						@classmethod
						def 	_lowerCamelCase      (			cls    :	Dict   ,							_a    :	Optional[int]   ,							_a    :	Tuple   ,							_a    :	int=None   ,							**_a    :	List[str]      ):
													a__:							Union[str, Any]									=kwargs.pop("config"   ,							_a      ) or RagConfig.from_pretrained(_a   ,							**_a      )
													a__:							List[Any]									=RagTokenizer.from_pretrained(_a   ,							config=_a      )
													a__:							str									=rag_tokenizer.question_encoder
													a__:							Any									=rag_tokenizer.generator

													if indexed_dataset is not None:
																				a__:							Dict									="custom"
																				a__:							Dict									=CustomHFIndex(config.retrieval_vector_size   ,							_a      )
													else:
																				a__:							List[str]									=cls._build_index(_a      )

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







						def 	_lowerCamelCase      (			self    :	Union[str, 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__:							str									=self._build_index(self.config      )

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





