



"""simple docstring"""


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


snake_case_         = logging.get_logger(__name__)

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







class    A_    (  SCREAMING_SNAKE_CASE_       ):



						"""simple docstring"""
						__UpperCamelCase	       =							RetriBertConfig
						__UpperCamelCase	       =							None
						__UpperCamelCase	       =							"""retribert"""






						def 			UpperCAmelCase__  (      self				:Optional[int]     ,						lowercase_				:Dict      )       ->    Dict:
												if isinstance(lowercase_     ,						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(lowercase_     ,						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(lowercase_     ,						nn.LayerNorm      ):
																		module.bias.data.zero_()
																		module.weight.data.fill_(1.0      )


snake_case_         = R"""

    This model inherits from [`PreTrainedModel`]. Check the superclass documentation for the generic methods the
    library implements for all its model (such as downloading or saving, resizing the input embeddings, pruning heads
    etc.)

    This model is also a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass.
    Use it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage
    and behavior.

    Parameters:
        config ([`RetriBertConfig`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.
"""







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



						"""simple docstring"""






						def __init__(      self				:List[str]     ,						lowercase_				:RetriBertConfig      )       ->    None:
												super().__init__(lowercase_      )
												UpperCAmelCase    =				config.projection_dim

												UpperCAmelCase    =				BertModel(lowercase_      )
												UpperCAmelCase    =				None if config.share_encoders else BertModel(lowercase_      )
												UpperCAmelCase    =				nn.Dropout(config.hidden_dropout_prob      )
												UpperCAmelCase    =				nn.Linear(config.hidden_size     ,						config.projection_dim     ,						bias=lowercase_      )
												UpperCAmelCase    =				nn.Linear(config.hidden_size     ,						config.projection_dim     ,						bias=lowercase_      )

												UpperCAmelCase    =				nn.CrossEntropyLoss(reduction='mean'      )

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






						def 			UpperCAmelCase__  (      self				:List[str]     ,						lowercase_				:Optional[int]     ,						lowercase_				:Tuple     ,						lowercase_				:List[str]     ,						lowercase_				:int=-1     ,						)       ->    int:
												# reproduces BERT forward pass with checkpointing
												if checkpoint_batch_size < 0 or input_ids.shape[0] < checkpoint_batch_size:
																		return sent_encoder(lowercase_     ,						attention_mask=lowercase_      )[1]
												else:
																		# prepare implicit variables
																		UpperCAmelCase    =				input_ids.device
																		UpperCAmelCase    =				input_ids.size()
																		UpperCAmelCase    =				torch.zeros(lowercase_     ,						dtype=torch.long     ,						device=lowercase_      )
																		UpperCAmelCase    =				[None] * sent_encoder.config.num_hidden_layers
																		UpperCAmelCase    =				sent_encoder.get_extended_attention_mask(
																		    lowercase_     ,						lowercase_      )

																		# define function for checkpointing
																		def partial_encode(*lowercase_				:Any      ):
																								UpperCAmelCase    =				sent_encoder.encoder(
																								    inputs[0]     ,						attention_mask=inputs[1]     ,						head_mask=lowercase_     ,						)
																								UpperCAmelCase    =				encoder_outputs[0]
																								UpperCAmelCase    =				sent_encoder.pooler(lowercase_      )
																								return pooled_output

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






						def 			UpperCAmelCase__  (      self				:Any     ,						lowercase_				:Any     ,						lowercase_				:Optional[int]=None     ,						lowercase_				:Union[str, Any]=-1     ,						)       ->    str:
												UpperCAmelCase    =				self.embed_sentences_checkpointed(
												    lowercase_     ,						lowercase_     ,						self.bert_query     ,						lowercase_     ,						)
												return self.project_query(lowercase_      )






						def 			UpperCAmelCase__  (      self				:Union[str, Any]     ,						lowercase_				:str     ,						lowercase_				:Union[str, Any]=None     ,						lowercase_				:Optional[int]=-1     ,						)       ->    str:
												UpperCAmelCase    =				self.embed_sentences_checkpointed(
												    lowercase_     ,						lowercase_     ,						self.bert_query if self.bert_doc is None else self.bert_doc     ,						lowercase_     ,						)
												return self.project_doc(lowercase_      )






						def 			UpperCAmelCase__  (      self				:Dict     ,						lowercase_				:torch.LongTensor     ,						lowercase_				:Optional[torch.FloatTensor]     ,						lowercase_				:torch.LongTensor     ,						lowercase_				:Optional[torch.FloatTensor]     ,						lowercase_				:int = -1     ,						)       ->    torch.FloatTensor:
												UpperCAmelCase    =				input_ids_query.device
												UpperCAmelCase    =				self.embed_questions(lowercase_     ,						lowercase_     ,						lowercase_      )
												UpperCAmelCase    =				self.embed_answers(lowercase_     ,						lowercase_     ,						lowercase_      )
												UpperCAmelCase    =				torch.mm(lowercase_     ,						a_reps.t()      )
												UpperCAmelCase    =				self.ce_loss(lowercase_     ,						torch.arange(compare_scores.shape[1]      ).to(lowercase_      )      )
												UpperCAmelCase    =				self.ce_loss(compare_scores.t()     ,						torch.arange(compare_scores.shape[0]      ).to(lowercase_      )      )
												UpperCAmelCase    =				(loss_qa + loss_aq) / 2
												return loss










"""simple docstring"""


from typing import Any

import numpy as np







class    A_    :



						"""simple docstring"""






						def __init__(      self				:List[str]     ,						lowercase_				:np.ndarray     ,						lowercase_				:int      )       ->    None:
												# Check if RHS is negative
												if np.any(tableau[:, -1]     ,						where=tableau[:, -1] < 0      ):
																		raise ValueError('RHS must be > 0'      )

												UpperCAmelCase    =				tableau
												UpperCAmelCase   ,					UpperCAmelCase    =				tableau.shape

												# Number of decision variables x1, x2, x3...
												UpperCAmelCase    =				n_vars

												# Number of artificial variables to be minimised
												UpperCAmelCase    =				len(np.where(tableau[self.n_vars : -1] == -1      )[0]      )

												# 2 if there are >= or == constraints (nonstandard), 1 otherwise (std)
												UpperCAmelCase    =				(self.n_art_vars > 0) + 1

												# Number of slack variables added to make inequalities into equalities
												UpperCAmelCase    =				self.n_rows - self.n_stages

												# Objectives for each stage
												UpperCAmelCase    =				['max']

												# In two stage simplex, first minimise then maximise
												if self.n_art_vars:
																		self.objectives.append('min'      )

												UpperCAmelCase    =				['']

												# Index of current pivot row and column
												UpperCAmelCase    =				None
												UpperCAmelCase    =				None

												# Does objective row only contain (non)-negative values?
												UpperCAmelCase    =				False






						@staticmethod
						def 			UpperCAmelCase__  (      *lowercase_				:int      )       ->    list[str]:
												if len(lowercase_      ) != 3:
																		raise ValueError('Must provide n_vars, n_slack, and n_art_vars'      )

												if not all(x >= 0 and isinstance(lowercase_     ,						lowercase_      ) for x in args      ):
																		raise ValueError('All arguments must be non-negative integers'      )

												# decision | slack | artificial
												UpperCAmelCase    =				['x', 's', 'a']
												UpperCAmelCase    =				[]
												for i in range(3      ):
																		for j in range(args[i]      ):
																								titles.append(string_starts[i] + str(j + 1      )      )
												titles.append('RHS'      )
												return titles






						def 			UpperCAmelCase__  (      self				:List[str]     ,						lowercase_				:np.ndarray      )       ->    tuple[Any, Any]:
												UpperCAmelCase    =				self.objectives[-1]

												# Find entries of highest magnitude in objective rows
												UpperCAmelCase    =				(objective == 'min') - (objective == 'max')
												UpperCAmelCase    =				np.argmax(sign * tableau[0, : self.n_vars]      )

												# Choice is only valid if below 0 for maximise, and above for minimise
												if sign * self.tableau[0, col_idx] <= 0:
																		UpperCAmelCase    =				True
																		return 0, 0

												# Pivot row is chosen as having the lowest quotient when elements of
												# the pivot column divide the right-hand side

												# Slice excluding the objective rows
												UpperCAmelCase    =				slice(self.n_stages     ,						self.n_rows      )

												# RHS
												UpperCAmelCase    =				tableau[s, -1]

												# Elements of pivot column within slice
												UpperCAmelCase    =				tableau[s, col_idx]

												# Array filled with nans
												UpperCAmelCase    =				np.full(self.n_rows - self.n_stages     ,						np.nan      )

												# If element in pivot column is greater than zeron_stages, return
												# quotient or nan otherwise
												UpperCAmelCase    =				np.divide(lowercase_     ,						lowercase_     ,						out=lowercase_     ,						where=divisor > 0      )

												# Arg of minimum quotient excluding the nan values. n_stages is added
												# to compensate for earlier exclusion of objective columns
												UpperCAmelCase    =				np.nanargmin(lowercase_      ) + self.n_stages
												return row_idx, col_idx






						def 			UpperCAmelCase__  (      self				:Union[str, Any]     ,						lowercase_				:np.ndarray     ,						lowercase_				:int     ,						lowercase_				:int      )       ->    np.ndarray:
												UpperCAmelCase    =				tableau[row_idx].copy()

												UpperCAmelCase    =				piv_row[col_idx]

												# Entry becomes 1
												piv_row *= 1 / piv_val

												# Variable in pivot column becomes basic, ie the only non-zero entry
												for idx, coeff in enumerate(tableau[:, col_idx]      ):
																		tableau[idx] += -coeff * piv_row
												UpperCAmelCase    =				piv_row
												return tableau






						def 			UpperCAmelCase__  (      self				:Optional[Any]     ,						lowercase_				:np.ndarray      )       ->    np.ndarray:
												self.objectives.pop()

												if not self.objectives:
																		return tableau

												# Slice containing ids for artificial columns
												UpperCAmelCase    =				slice(-self.n_art_vars - 1     ,						-1      )

												# Delete the artificial variable columns
												UpperCAmelCase    =				np.delete(lowercase_     ,						lowercase_     ,						axis=1      )

												# Delete the objective row of the first stage
												UpperCAmelCase    =				np.delete(lowercase_     ,						0     ,						axis=0      )

												UpperCAmelCase    =				1
												self.n_rows -= 1
												UpperCAmelCase    =				0
												UpperCAmelCase    =				False
												return tableau






						def 			UpperCAmelCase__  (      self				:List[Any]      )       ->    dict[Any, Any]:
												# Stop simplex algorithm from cycling.
												for _ in range(1_00      ):
																		# Completion of each stage removes an objective. If both stages
																		# are complete, then no objectives are left
																		if not self.objectives:
																								UpperCAmelCase    =				self.generate_col_titles(
																								    self.n_vars     ,						self.n_slack     ,						self.n_art_vars      )

																								# Find the values of each variable at optimal solution
																								return self.interpret_tableau(self.tableau     ,						self.col_titles      )

																		UpperCAmelCase   ,					UpperCAmelCase    =				self.find_pivot(self.tableau      )

																		# If there are no more negative values in objective row
																		if self.stop_iter:
																								# Delete artificial variable columns and rows. Update attributes
																								UpperCAmelCase    =				self.change_stage(self.tableau      )
																		else:
																								UpperCAmelCase    =				self.pivot(self.tableau     ,						lowercase_     ,						lowercase_      )
												return {}






						def 			UpperCAmelCase__  (      self				:Union[str, Any]     ,						lowercase_				:np.ndarray     ,						lowercase_				:list[str]      )       ->    dict[str, float]:
												UpperCAmelCase    =				{'P': abs(tableau[0, -1]      )}

												for i in range(self.n_vars      ):
																		# Gives ids of nonzero entries in the ith column
																		UpperCAmelCase    =				np.nonzero(tableau[:, i]      )
																		UpperCAmelCase    =				len(nonzero[0]      )

																		# First entry in the nonzero ids
																		UpperCAmelCase    =				nonzero[0][0]
																		UpperCAmelCase    =				tableau[nonzero_rowidx, i]

																		# If there is only one nonzero value in column, which is one
																		if n_nonzero == nonzero_val == 1:
																								UpperCAmelCase    =				tableau[nonzero_rowidx, -1]
																								UpperCAmelCase    =				rhs_val

        # Check for basic variables
												for title in col_titles:
																		# Don't add RHS or slack variables to output dict
																		if title[0] not in "R-s-a":
																								output_dict.setdefault(lowercase_     ,						0      )
												return output_dict


if __name__ == "__main__":
				import doctest

				doctest.testmod()





