





'''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 			__A       (      __lowerCamelCase  ):




      """simple docstring"""







      def __init__(		self      ,							a__      ,							a__      ,							a__      ,							a__=None):



             """simple docstring"""





             super().__init__(
                 UpperCamelCase_      ,							question_encoder_tokenizer=UpperCamelCase_      ,							generator_tokenizer=UpperCamelCase_      ,							index=UpperCamelCase_      ,							init_retrieval=UpperCamelCase_      ,							)
             _lowerCamelCase							:     List[str]	=		None


      def 				__snake_case			(		self      ,							a__):



             """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							:     Tuple	=		self._infer_socket_ifname()
                    # avoid clash with the NCCL port
                    _lowerCamelCase							:     Union[str, Any]	=		str(distributed_port + 1)
                    _lowerCamelCase							:     Dict	=		dist.new_group(ranks=UpperCamelCase_      ,							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):



             """simple docstring"""





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


      def 				__snake_case			(		self      ,							a__      ,							a__      ,							a__=torch.floataa):



             """simple docstring"""





             _lowerCamelCase							:     str	=		torch.empty(UpperCamelCase_      ,							dtype=UpperCamelCase_)
             dist.scatter(UpperCamelCase_      ,							src=0      ,							scatter_list=UpperCamelCase_      ,							group=self.process_group)
             return target_tensor


      def 				__snake_case			(		self):



             """simple docstring"""





             _lowerCamelCase							:     int	=		psutil.net_if_addrs()
             # a hacky way to deal with varying network interface names
             _lowerCamelCase							:     Tuple	=		next((addr for addr in addrs if addr.startswith('''e'''))      ,							UpperCamelCase_)
             return ifname



      def 				__snake_case			(		self      ,							a__      ,							a__):



             """simple docstring"""





             if not dist.is_initialized():
                    _lowerCamelCase,  _lowerCamelCase							:     Union[str, Any]	=		self._main_retrieve(UpperCamelCase_      ,							UpperCamelCase_)
                    return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(UpperCamelCase_)

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

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

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

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








import qiskit




def    __UpperCAmelCase(      lowercase_ = 2      ):
       _lowerCamelCase							:     Optional[Any]	=		qubits

       # Using Aer's simulator
       _lowerCamelCase							:     Optional[int]	=		qiskit.Aer.get_backend('''aer_simulator'''      )

       # Creating a Quantum Circuit acting on the q register
       _lowerCamelCase							:     List[str]	=		qiskit.QuantumCircuit(lowercase_ ,    lowercase_      )

       # Adding a H gate on qubit 0 (now q0 in superposition)
       circuit.h(0      )

       for i in range(1 ,    lowercase_      ):
              # Adding CX (CNOT) gate
              circuit.cx(i - 1 ,    lowercase_      )

       # Mapping the quantum measurement to the classical bits
       circuit.measure(list(range(lowercase_      )      ) ,    list(range(lowercase_      )      )      )

       # Now measuring any one qubit would affect other qubits to collapse
       # their super position and have same state as the measured one.

       # Executing the circuit on the simulator
       _lowerCamelCase							:     str	=		qiskit.execute(lowercase_ ,    lowercase_ ,    shots=10_00      )

       return job.result().get_counts(lowercase_      )


if __name__ == "__main__":
       print(F'''Total count for various states are: {quantum_entanglement(3)}''')

