


import math
from typing import Optional

import torch
import torch.utils.checkpoint as checkpoint
from torch import nn

from ....modeling_utils import PreTrainedModel
from ....utils import add_start_docstrings, logging
from ...bert.modeling_bert import BertModel
from .configuration_retribert import RetriBertConfig


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

lowerCAmelCase__       :	Tuple     				=[
    'yjernite/retribert-base-uncased',
    # See all RetriBert models at https://huggingface.co/models?filter=retribert
]







class      __lowercase  (__SCREAMING_SNAKE_CASE   ):





							"""simple docstring"""







							_UpperCAmelCase     =		RetriBertConfig
							_UpperCAmelCase     =		None
							_UpperCAmelCase     =		"""retribert"""






							def  UpperCamelCase__			(	self       ,		lowerCAmelCase__    ):





										"""simple docstring"""




										if isinstance(lowerCAmelCase__       ,		nn.Linear    ):
													module.weight.data.normal_(mean=0.0       ,		std=self.config.initializer_range    )
													if module.bias is not None:
																module.bias.data.zero_()
										elif isinstance(lowerCAmelCase__       ,		nn.Embedding    ):
													module.weight.data.normal_(mean=0.0       ,		std=self.config.initializer_range    )
													if module.padding_idx is not None:
																module.weight.data[module.padding_idx].zero_()
										elif isinstance(lowerCAmelCase__       ,		nn.LayerNorm    ):
													module.bias.data.zero_()
													module.weight.data.fill_(1.0    )







lowerCAmelCase__       :	int     				=R'\n\n    This model inherits from [`PreTrainedModel`]. Check the superclass documentation for the generic methods the\n    library implements for all its model (such as downloading or saving, resizing the input embeddings, pruning heads\n    etc.)\n\n    This model is also a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass.\n    Use it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage\n    and behavior.\n\n    Parameters:\n        config ([`RetriBertConfig`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n'







@add_start_docstrings(
    """Bert Based model to embed queries or document for document retrieval."""     ,    __SCREAMING_SNAKE_CASE     ,    )
class      __lowercase  (__SCREAMING_SNAKE_CASE   ):





							"""simple docstring"""







							def __init__(	self       ,		lowerCAmelCase__    ):





										"""simple docstring"""




										super().__init__(lowerCAmelCase__    )
										SCREAMING_SNAKE_CASE_						:			Tuple       						= config.projection_dim

										SCREAMING_SNAKE_CASE_						:			Optional[Any]       						= BertModel(lowerCAmelCase__    )
										SCREAMING_SNAKE_CASE_						:			List[str]       						= None if config.share_encoders else BertModel(lowerCAmelCase__    )
										SCREAMING_SNAKE_CASE_						:			Any       						= nn.Dropout(config.hidden_dropout_prob    )
										SCREAMING_SNAKE_CASE_						:			str       						= nn.Linear(config.hidden_size       ,		config.projection_dim       ,		bias=lowerCAmelCase__    )
										SCREAMING_SNAKE_CASE_						:			Tuple       						= nn.Linear(config.hidden_size       ,		config.projection_dim       ,		bias=lowerCAmelCase__    )

										SCREAMING_SNAKE_CASE_						:			int       						= nn.CrossEntropyLoss(reduction='mean'    )

										# Initialize weights and apply final processing
										self.post_init()






							def  UpperCamelCase__			(	self       ,		lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__=-1       ,		):





										"""simple docstring"""




										if checkpoint_batch_size < 0 or input_ids.shape[0] < checkpoint_batch_size:
													return sent_encoder(lowerCAmelCase__       ,		attention_mask=lowerCAmelCase__    )[1]
										else:
													# prepare implicit variables
													SCREAMING_SNAKE_CASE_						:			List[str]       						= input_ids.device
													SCREAMING_SNAKE_CASE_						:			Optional[Any]       						= input_ids.size()
													SCREAMING_SNAKE_CASE_						:			str       						= torch.zeros(lowerCAmelCase__       ,		dtype=torch.long       ,		device=lowerCAmelCase__    )
													SCREAMING_SNAKE_CASE_						:			Optional[Any]       						= [None] * sent_encoder.config.num_hidden_layers
													SCREAMING_SNAKE_CASE_						:			torch.Tensor       						= sent_encoder.get_extended_attention_mask(
													    lowerCAmelCase__       ,		lowerCAmelCase__    )

													# define function for checkpointing
													def partial_encode(*lowerCAmelCase__    ):
																SCREAMING_SNAKE_CASE_						:			List[str]       						= sent_encoder.encoder(
																    inputs[0]       ,		attention_mask=inputs[1]       ,		head_mask=lowerCAmelCase__       ,		)
																SCREAMING_SNAKE_CASE_						:			Tuple       						= encoder_outputs[0]
																SCREAMING_SNAKE_CASE_						:			List[str]       						= sent_encoder.pooler(lowerCAmelCase__    )
																return pooled_output

													# run embedding layer on everything at once
													SCREAMING_SNAKE_CASE_						:			List[Any]       						= sent_encoder.embeddings(
													    input_ids=lowerCAmelCase__       ,		position_ids=lowerCAmelCase__       ,		token_type_ids=lowerCAmelCase__       ,		inputs_embeds=lowerCAmelCase__    )
													# run encoding and pooling on one mini-batch at a time
													SCREAMING_SNAKE_CASE_						:			Tuple       						= []
													for b in range(math.ceil(input_ids.shape[0] / checkpoint_batch_size    )    ):
																SCREAMING_SNAKE_CASE_						:			str       						= embedding_output[b * checkpoint_batch_size : (b + 1) * checkpoint_batch_size]
																SCREAMING_SNAKE_CASE_						:			int       						= extended_attention_mask[b * checkpoint_batch_size : (b + 1) * checkpoint_batch_size]
																SCREAMING_SNAKE_CASE_						:			Any       						= checkpoint.checkpoint(lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__    )
																pooled_output_list.append(lowerCAmelCase__    )
													return torch.cat(lowerCAmelCase__       ,		dim=0    )






							def  UpperCamelCase__			(	self       ,		lowerCAmelCase__       ,		lowerCAmelCase__=None       ,		lowerCAmelCase__=-1       ,		):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_						:			Any       						= self.embed_sentences_checkpointed(
										    lowerCAmelCase__       ,		lowerCAmelCase__       ,		self.bert_query       ,		lowerCAmelCase__       ,		)
										return self.project_query(lowerCAmelCase__    )






							def  UpperCamelCase__			(	self       ,		lowerCAmelCase__       ,		lowerCAmelCase__=None       ,		lowerCAmelCase__=-1       ,		):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_						:			Union[str, Any]       						= self.embed_sentences_checkpointed(
										    lowerCAmelCase__       ,		lowerCAmelCase__       ,		self.bert_query if self.bert_doc is None else self.bert_doc       ,		lowerCAmelCase__       ,		)
										return self.project_doc(lowerCAmelCase__    )






							def  UpperCamelCase__			(	self       ,		lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__ = -1       ,		):





										"""simple docstring"""




										SCREAMING_SNAKE_CASE_						:			Any       						= input_ids_query.device
										SCREAMING_SNAKE_CASE_						:			str       						= self.embed_questions(lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__    )
										SCREAMING_SNAKE_CASE_						:			Tuple       						= self.embed_answers(lowerCAmelCase__       ,		lowerCAmelCase__       ,		lowerCAmelCase__    )
										SCREAMING_SNAKE_CASE_						:			Any       						= torch.mm(lowerCAmelCase__       ,		a_reps.t()    )
										SCREAMING_SNAKE_CASE_						:			List[Any]       						= self.ce_loss(lowerCAmelCase__       ,		torch.arange(compare_scores.shape[1]    ).to(lowerCAmelCase__    )    )
										SCREAMING_SNAKE_CASE_						:			Any       						= self.ce_loss(compare_scores.t()       ,		torch.arange(compare_scores.shape[0]    ).to(lowerCAmelCase__    )    )
										SCREAMING_SNAKE_CASE_						:			Tuple       						= (loss_qa + loss_aq) / 2
										return loss













import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor





class   SCREAMING_SNAKE_CASE (       unittest.TestCase    ):





							"""simple docstring"""

							def 					__A       (     self:      Optional[Any]						,   __A:      bool = False      )			-> Union[str, Any]:
												_A									=       10

												if only_cross_attention:
																	_A									=       12
												else:
																	# when only cross attention is not set, the cross attention dim must be the same as the query dim
																	_A									=       query_dim

												return {
												    "query_dim": query_dim,
												    "cross_attention_dim": cross_attention_dim,
												    "heads": 2,
												    "dim_head": 4,
												    "added_kv_proj_dim": 6,
												    "norm_num_groups": 1,
												    "only_cross_attention": only_cross_attention,
												    "processor": AttnAddedKVProcessor(),
												}

							def 					__A       (     self:      List[Any]						,   __A:      str						,   __A:      Union[str, Any]      )			-> Union[str, Any]:
												_A									=       2

												_A									=       torch.rand(__A						,   __A						,   3						,   2      )
												_A									=       torch.rand(__A						,   4						,   __A      )
												_A									=       None

												return {
												    "hidden_states": hidden_states,
												    "encoder_hidden_states": encoder_hidden_states,
												    "attention_mask": attention_mask,
												}



							def 					__A       (     self:      Optional[int]      )			-> Any:
												# self and cross attention

												torch.manual_seed(0      )

												_A									=       self.get_constructor_arguments(only_cross_attention=__A      )
												_A									=       Attention(**__A      )

												self.assertTrue(attn.to_k is not None      )
												self.assertTrue(attn.to_v is not None      )

												_A									=       self.get_forward_arguments(
												    query_dim=constructor_args['''query_dim''']						,   added_kv_proj_dim=constructor_args['''added_kv_proj_dim''']      )

												_A									=       attn(**__A      )

												# only self attention

												torch.manual_seed(0      )

												_A									=       self.get_constructor_arguments(only_cross_attention=__A      )
												_A									=       Attention(**__A      )

												self.assertTrue(attn.to_k is None      )
												self.assertTrue(attn.to_v is None      )

												_A									=       self.get_forward_arguments(
												    query_dim=constructor_args['''query_dim''']						,   added_kv_proj_dim=constructor_args['''added_kv_proj_dim''']      )

												_A									=       attn(**__A      )

												self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()      )





class   SCREAMING_SNAKE_CASE (       unittest.TestCase    ):





							"""simple docstring"""

							def 					__A       (     self:      Any      )			-> Dict:
												_A									=       DiffusionPipeline.from_pretrained('''hf-internal-testing/tiny-stable-diffusion-pipe'''						,   safety_checker=__A      )

												_A									=       pipe(
												    '''foo'''						,   num_inference_steps=2						,   generator=torch.Generator('''cpu'''      ).manual_seed(0      )						,   output_type='''np'''						,   ).images

												# the initial conversion succeeds
												_A									=       DiffusionPipeline.from_pretrained(
												    '''hf-internal-testing/tiny-stable-diffusion-pipe'''						,   device_map='''sequential'''						,   safety_checker=__A      )

												_A									=       pipe(
												    '''foo'''						,   num_inference_steps=2						,   generator=torch.Generator('''cpu'''      ).manual_seed(0      )						,   output_type='''np'''						,   ).images

												with tempfile.TemporaryDirectory() as tmpdir:
																	# save the converted model
																	pipe.save_pretrained(__A      )

																	# can also load the converted weights
																	_A									=       DiffusionPipeline.from_pretrained(__A						,   device_map='''sequential'''						,   safety_checker=__A      )

												_A									=       pipe(
												    '''foo'''						,   num_inference_steps=2						,   generator=torch.Generator('''cpu'''      ).manual_seed(0      )						,   output_type='''np'''						,   ).images

												self.assertTrue(np.allclose(__A						,   __A						,   atol=1e-5      )      )
												self.assertTrue(np.allclose(__A						,   __A						,   atol=1e-5      )      )



