


"""simple docstring"""




import logging
import os
from typing import List, Tuple

import numpy as np
import psutil
import torch
import torch.distributed as dist

from transformers import RagRetriever


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

class     a		(_a				):




							"""simple docstring"""


							def __init__(							self	:  List[str]   ,     lowerCamelCase	:  Tuple   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Tuple   ,     lowerCamelCase	:  Union[str, Any]=None    )     ->   List[Any]:
														super().__init__(
														    __lowerCamelCase   ,     question_encoder_tokenizer=__lowerCamelCase   ,     generator_tokenizer=__lowerCamelCase   ,     index=__lowerCamelCase   ,     init_retrieval=__lowerCamelCase   ,     )
														__snake_case		:      Dict										=       None

							def    __snake_case (							self	:  str   ,     lowerCamelCase	:  int    )     ->   Any:
														logger.info("initializing retrieval"    )

														# initializing a separate process group for retrieval as the default
														# nccl backend doesn't support gather/scatter operations while gloo
														# is too slow to replace nccl for the core gpu communication
														if dist.is_initialized():
																					logger.info("dist initialized"    )
																					# needs to be set manually
																					__snake_case		:      Union[str, Any]										=       self._infer_socket_ifname()
																					# avoid clash with the NCCL port
																					__snake_case		:      List[Any]										=       str(distributed_port + 1    )
																					__snake_case		:      List[str]										=       dist.new_group(ranks=__lowerCamelCase   ,     backend="gloo"    )

														# initialize retriever only on the main worker
														if not dist.is_initialized() or self._is_main():
																					logger.info("dist not initialized / main"    )
																					self.index.init_index()

														# all processes wait untill the retriever is initialized by the main process
														if dist.is_initialized():
																					torch.distributed.barrier(group=self.process_group    )

							def    __snake_case (							self	:  Dict    )     ->   Dict:
														return dist.get_rank(group=self.process_group    ) == 0

							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  Optional[Any]   ,     lowerCamelCase	:  List[Any]=torch.floataa    )     ->   Tuple:
														__snake_case		:      Tuple										=       torch.empty(__lowerCamelCase   ,     dtype=__lowerCamelCase    )
														dist.scatter(__lowerCamelCase   ,     src=0   ,     scatter_list=__lowerCamelCase   ,     group=self.process_group    )
														return target_tensor

							def    __snake_case (							self	:  str    )     ->   Union[str, Any]:
														__snake_case		:      int										=       psutil.net_if_addrs()
														# a hacky way to deal with varying network interface names
														__snake_case		:      List[Any]										=       next((addr for addr in addrs if addr.startswith("e"    ))   ,     __lowerCamelCase    )
														return ifname






							def    __snake_case (							self	:  Optional[int]   ,     lowerCamelCase	:  np.ndarray   ,     lowerCamelCase	:  int    )     ->   Optional[Any]:

														# single GPU training
														if not dist.is_initialized():
																					__snake_case		:      List[str]										=       self._main_retrieve(__lowerCamelCase   ,     __lowerCamelCase    )
																					return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__lowerCamelCase    )

														# distributed training
														__snake_case		:      List[Any]										=       dist.get_world_size(group=self.process_group    )

														# gather logic
														__snake_case		:      Tuple										=       None
														if self._is_main():
																					__snake_case		:      List[str]										=       [torch.empty(question_hidden_states.shape   ,     dtype=torch.floataa    ) for _ in range(__lowerCamelCase    )]
														dist.gather(torch.tensor(__lowerCamelCase    )   ,     dst=0   ,     gather_list=__lowerCamelCase   ,     group=self.process_group    )

														# scatter logic
														__snake_case		:      str										=       question_hidden_states.shape[0]
														__snake_case		:      Tuple										=       []
														__snake_case		:      int										=       []
														if self._is_main():
																					assert len(__lowerCamelCase    ) == world_size
																					__snake_case		:      Dict										=       self._main_retrieve(torch.cat(__lowerCamelCase    ).numpy()   ,     __lowerCamelCase    )
																					__snake_case		:      int										=       torch.tensor(__lowerCamelCase    ), torch.tensor(__lowerCamelCase    )
																					__snake_case		:      Optional[Any]										=       self._chunk_tensor(__lowerCamelCase   ,     __lowerCamelCase    )
																					__snake_case		:      Dict										=       self._chunk_tensor(__lowerCamelCase   ,     __lowerCamelCase    )
														__snake_case		:      Tuple										=       self._scattered(__lowerCamelCase   ,     [n_queries, n_docs]   ,     target_type=torch.intaa    )
														__snake_case		:      str										=       self._scattered(__lowerCamelCase   ,     [n_queries, n_docs, question_hidden_states.shape[1]]    )

														return retrieved_doc_embeds.numpy(), doc_ids.numpy(), self.index.get_doc_dicts(__lowerCamelCase    )





from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
			from PIL import Image

			from ..image_utils import load_image

if is_tf_available():
			import tensorflow as tf

			from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
			from ..tf_utils import stable_softmax

if is_torch_available():
			from ..models.auto.modeling_auto import MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

_snake_case					:		Optional[int]      =    logging.get_logger(__name__)

@add_end_docstrings(_lowerCAmelCase				)
class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							def __init__(							self	:  Tuple   ,     *lowerCamelCase	:  Any   ,     **lowerCamelCase	:  Tuple    )     ->   int:
														super().__init__(*lowerCamelCase   ,     **lowerCamelCase    )
														requires_backends(self   ,     "vision"    )
														self.check_model_type(
														    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
														    if self.framework == "tf"
														    else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING    )

							def    __snake_case (							self	:  List[str]   ,     lowerCamelCase	:  Optional[Any]=None    )     ->   Optional[int]:
														__snake_case		:      Optional[Any]										=       {}
														if top_k is not None:
																					__snake_case		:      List[Any]										=       top_k
														return {}, {}, postprocess_params

							def __call__(							self	:  List[Any]   ,     lowerCamelCase	:  Union[str, List[str], "Image.Image", List["Image.Image"]]   ,     **lowerCamelCase	:  Dict    )     ->   Optional[int]:
														return super().__call__(lowerCamelCase   ,     **lowerCamelCase    )

							def    __snake_case (							self	:  Optional[Any]   ,     lowerCamelCase	:  List[Any]    )     ->   int:
														__snake_case		:      Any										=       load_image(lowerCamelCase    )
														__snake_case		:      str										=       self.image_processor(images=lowerCamelCase   ,     return_tensors=self.framework    )
														return model_inputs

							def    __snake_case (							self	:  int   ,     lowerCamelCase	:  List[str]    )     ->   Tuple:
														__snake_case		:      List[Any]										=       self.model(**lowerCamelCase    )
														return model_outputs






							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Optional[Any]=5    )     ->   List[str]:
														if top_k > self.model.config.num_labels:
																					__snake_case		:      int										=       self.model.config.num_labels

														if self.framework == "pt":
																					__snake_case		:      Optional[Any]										=       model_outputs.logits.softmax(-1    )[0]
																					__snake_case			,      __snake_case		:      List[str]										=       probs.topk(lowerCamelCase    )
														elif self.framework == "tf":
																					__snake_case		:      Tuple										=       stable_softmax(model_outputs.logits   ,     axis=-1    )[0]
																					__snake_case		:      Optional[Any]										=       tf.math.top_k(lowerCamelCase   ,     k=lowerCamelCase    )
																					__snake_case			,      __snake_case		:      List[Any]										=       topk.values.numpy(), topk.indices.numpy()
														else:
																					raise ValueError(F'Unsupported framework: {self.framework}'    )

														__snake_case		:      Any										=       scores.tolist()
														__snake_case		:      Optional[int]										=       ids.tolist()
														return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(lowerCamelCase   ,     lowerCamelCase    )]

