



from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase__      =    logging.get_logger(__name__)

lowerCAmelCase__      =    {
    '''caidas/swin2sr-classicalsr-x2-64''': (
        '''https://huggingface.co/caidas/swin2sr-classicalsr-x2-64/resolve/main/config.json'''
    ),
}






class 	__snake_case	(     A__):
  snake_case__    :       Dict        =						"swin2sr"

  snake_case__    :       Tuple        =						{
      "hidden_size": "embed_dim",
      "num_attention_heads": "num_heads",
      "num_hidden_layers": "num_layers",
  }




  def __init__(			self					:						int						, __lowerCAmelCase					:						str=6_4						, __lowerCAmelCase					:						List[Any]=1						, __lowerCAmelCase					:						Optional[Any]=3						, __lowerCAmelCase					:						Optional[int]=1_8_0						, __lowerCAmelCase					:						Dict=[6, 6, 6, 6, 6, 6]						, __lowerCAmelCase					:						Tuple=[6, 6, 6, 6, 6, 6]						, __lowerCAmelCase					:						Dict=8						, __lowerCAmelCase					:						str=2.0						, __lowerCAmelCase					:						Any=True						, __lowerCAmelCase					:						List[Any]=0.0						, __lowerCAmelCase					:						List[Any]=0.0						, __lowerCAmelCase					:						Any=0.1						, __lowerCAmelCase					:						Tuple="gelu"						, __lowerCAmelCase					:						Optional[int]=False						, __lowerCAmelCase					:						Any=0.02						, __lowerCAmelCase					:						int=1E-5						, __lowerCAmelCase					:						Dict=2						, __lowerCAmelCase					:						Tuple=1.0						, __lowerCAmelCase					:						Dict="1conv"						, __lowerCAmelCase					:						Dict="pixelshuffle"						, **__lowerCAmelCase					:						Dict						, ):






         """simple docstring"""






         super().__init__(**__A		)

         _lowerCamelCase						:    List[str]         =			image_size
         _lowerCamelCase						:    Union[str, Any]         =			patch_size
         _lowerCamelCase						:    Dict         =			num_channels
         _lowerCamelCase						:    Any         =			embed_dim
         _lowerCamelCase						:    List[Any]         =			depths
         _lowerCamelCase						:    Tuple         =			len(__A		)
         _lowerCamelCase						:    int         =			num_heads
         _lowerCamelCase						:    Optional[int]         =			window_size
         _lowerCamelCase						:    str         =			mlp_ratio
         _lowerCamelCase						:    List[Any]         =			qkv_bias
         _lowerCamelCase						:    Tuple         =			hidden_dropout_prob
         _lowerCamelCase						:    str         =			attention_probs_dropout_prob
         _lowerCamelCase						:    Optional[int]         =			drop_path_rate
         _lowerCamelCase						:    Tuple         =			hidden_act
         _lowerCamelCase						:    int         =			use_absolute_embeddings
         _lowerCamelCase						:    Dict         =			layer_norm_eps
         _lowerCamelCase						:    Optional[Any]         =			initializer_range
         _lowerCamelCase						:    List[str]         =			upscale
         _lowerCamelCase						:    Union[str, Any]         =			img_range
         _lowerCamelCase						:    int         =			resi_connection
         _lowerCamelCase						:    Dict         =			upsampler







"""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


lowerCAmelCase__      =    logging.getLogger(__name__)






class 	__snake_case	(     _lowercase):




  def __init__(			self					:						Union[str, Any]						, __lowerCAmelCase					:						Optional[int]						, __lowerCAmelCase					:						Optional[Any]						, __lowerCAmelCase					:						List[str]						, __lowerCAmelCase					:						int=None		):






         """simple docstring"""






         super().__init__(
             __lowerCAmelCase						, question_encoder_tokenizer=__lowerCAmelCase						, generator_tokenizer=__lowerCAmelCase						, index=__lowerCAmelCase						, init_retrieval=__lowerCAmelCase						, )
         _lowerCamelCase						:    Dict         =			None




  def 						SCREAMING_SNAKE_CASE    (			self					:						Dict						, __lowerCAmelCase					:						int		):






         """simple docstring"""






         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
                _lowerCamelCase						:    List[str]         =			self._infer_socket_ifname()
                # avoid clash with the NCCL port
                _lowerCamelCase						:    Dict         =			str(distributed_port + 1		)
                _lowerCamelCase						:    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 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]		):






         """simple docstring"""






         return dist.get_rank(group=self.process_group		) == 0




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]						, __lowerCAmelCase					:						Union[str, Any]						, __lowerCAmelCase					:						List[Any]						, __lowerCAmelCase					:						Tuple=torch.floataa		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			torch.empty(__lowerCAmelCase						, dtype=__lowerCAmelCase		)
         dist.scatter(__lowerCAmelCase						, src=0						, scatter_list=__lowerCAmelCase						, group=self.process_group		)
         return target_tensor




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			psutil.net_if_addrs()
         # a hacky way to deal with varying network interface names
         _lowerCamelCase						:    str         =			next((addr for addr in addrs if addr.startswith('''e'''		))						, __lowerCAmelCase		)
         return ifname




  def 						SCREAMING_SNAKE_CASE    (			self					:						str						, __lowerCAmelCase					:						np.ndarray						, __lowerCAmelCase					:						int		):






         """simple docstring"""






         if not dist.is_initialized():
                _lowerCamelCase      ,   _lowerCamelCase						:    Any         =			self._main_retrieve(__lowerCAmelCase						, __lowerCAmelCase		)
                return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__lowerCAmelCase		)

         # distributed training
         _lowerCamelCase						:    Dict         =			dist.get_world_size(group=self.process_group		)

         # gather logic
         _lowerCamelCase						:    str         =			None
         if self._is_main():
                _lowerCamelCase						:    List[Any]         =			[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
         _lowerCamelCase						:    int         =			question_hidden_states.shape[0]
         _lowerCamelCase						:    str         =			[]
         _lowerCamelCase						:    Optional[int]         =			[]
         if self._is_main():
                assert len(__lowerCAmelCase		) == world_size
                _lowerCamelCase      ,   _lowerCamelCase						:    Tuple         =			self._main_retrieve(torch.cat(__lowerCAmelCase		).numpy()						, __lowerCAmelCase		)
                _lowerCamelCase      ,   _lowerCamelCase						:    List[str]         =			torch.tensor(__lowerCAmelCase		), torch.tensor(__lowerCAmelCase		)
                _lowerCamelCase						:    List[Any]         =			self._chunk_tensor(__lowerCAmelCase						, __lowerCAmelCase		)
                _lowerCamelCase						:    Optional[Any]         =			self._chunk_tensor(__lowerCAmelCase						, __lowerCAmelCase		)
         _lowerCamelCase						:    int         =			self._scattered(__lowerCAmelCase						, [n_queries, n_docs]						, target_type=torch.intaa		)
         _lowerCamelCase						:    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		)


