

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


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

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





class 			__UpperCamelCase (   _A						):
		SCREAMING_SNAKE_CASE								=    RetriBertConfig
		SCREAMING_SNAKE_CASE								=    None
		SCREAMING_SNAKE_CASE								=    "retribert"


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

__A     :    List[Any]    =		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."     ,		_A     ,		)
class 			__UpperCamelCase (   _A						):


		def __init__(self :		Optional[Any]     ,     __SCREAMING_SNAKE_CASE :		RetriBertConfig):
							super().__init__(__SCREAMING_SNAKE_CASE)
							A			=					config.projection_dim

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

							A			=					nn.CrossEntropyLoss(reduction="mean")

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


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

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

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


		def 					SCREAMING_SNAKE_CASE__   (self :		Optional[int]     ,     __SCREAMING_SNAKE_CASE :		List[str]     ,     __SCREAMING_SNAKE_CASE :		Optional[Any]=None     ,     __SCREAMING_SNAKE_CASE :		Any=-1     ,     ):
							A			=					self.embed_sentences_checkpointed(
							    __SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE     ,     self.bert_query     ,     __SCREAMING_SNAKE_CASE     ,     )
							return self.project_query(__SCREAMING_SNAKE_CASE)


		def 					SCREAMING_SNAKE_CASE__   (self :		List[Any]     ,     __SCREAMING_SNAKE_CASE :		int     ,     __SCREAMING_SNAKE_CASE :		Union[str, Any]=None     ,     __SCREAMING_SNAKE_CASE :		Tuple=-1     ,     ):
							A			=					self.embed_sentences_checkpointed(
							    __SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE     ,     self.bert_query if self.bert_doc is None else self.bert_doc     ,     __SCREAMING_SNAKE_CASE     ,     )
							return self.project_doc(__SCREAMING_SNAKE_CASE)


		def 					SCREAMING_SNAKE_CASE__   (self :		Union[str, Any]     ,     __SCREAMING_SNAKE_CASE :		torch.LongTensor     ,     __SCREAMING_SNAKE_CASE :		Optional[torch.FloatTensor]     ,     __SCREAMING_SNAKE_CASE :		torch.LongTensor     ,     __SCREAMING_SNAKE_CASE :		Optional[torch.FloatTensor]     ,     __SCREAMING_SNAKE_CASE :		int = -1     ,     ):
							A			=					input_ids_query.device
							A			=					self.embed_questions(__SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE)
							A			=					self.embed_answers(__SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE)
							A			=					torch.mm(__SCREAMING_SNAKE_CASE     ,     a_reps.t())
							A			=					self.ce_loss(__SCREAMING_SNAKE_CASE     ,     torch.arange(compare_scores.shape[1]).to(__SCREAMING_SNAKE_CASE))
							A			=					self.ce_loss(compare_scores.t()     ,     torch.arange(compare_scores.shape[0]).to(__SCREAMING_SNAKE_CASE))
							A			=					(loss_qa + loss_aq) / 2
							return loss







from typing import Any

import numpy as np







class      lowerCamelCase		:





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

         snake_case      =					tableau
         snake_case     ,							snake_case      =					tableau.shape

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

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

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

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

         # Objectives for each stage
         snake_case      =					["max"]

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

         snake_case      =					[""]

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

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





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

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

         # decision | slack | artificial
         snake_case      =					["x", "s", "a"]
         snake_case      =					[]
         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      :    int				,    _A      :    np.ndarray   )						->						tuple[Any, Any]:
         snake_case      =					self.objectives[-1]

         # Find entries of highest magnitude in objective rows
         snake_case      =					(objective == "min") - (objective == "max")
         snake_case      =					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:
             snake_case      =					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
         snake_case      =					slice(self.n_stages				,    self.n_rows   )

         # RHS
         snake_case      =					tableau[s, -1]

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

         # Array filled with nans
         snake_case      =					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
         snake_case      =					np.divide(_A				,    _A				,    out=_A				,    where=divisor > 0   )

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





     def 		UpperCAmelCase(self      :    int				,    _A      :    np.ndarray				,    _A      :    int				,    _A      :    int   )						->						np.ndarray:
         snake_case      =					tableau[row_idx].copy()

         snake_case      =					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
         snake_case      =					piv_row
         return tableau





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

         if not self.objectives:
             return tableau

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

         # Delete the artificial variable columns
         snake_case      =					np.delete(_A				,    _A				,    axis=1   )

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

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





     def 		UpperCAmelCase(self      :    Optional[Any]   )						->						dict[Any, Any]:
         # Stop simplex algorithm from cycling.
         for _ in range(1_0_0   ):
             # Completion of each stage removes an objective. If both stages
             # are complete, then no objectives are left
             if not self.objectives:
                 snake_case      =					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   )

             snake_case     ,							snake_case      =					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
                 snake_case      =					self.change_stage(self.tableau   )
             else:
                 snake_case      =					self.pivot(self.tableau				,    _A				,    _A   )
         return {}





     def 		UpperCAmelCase(self      :    Optional[Any]				,    _A      :    np.ndarray				,    _A      :    list[str]   )						->						dict[str, float]:
         snake_case      =					{"P": abs(tableau[0, -1]   )}

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

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

             # If there is only one nonzero value in column, which is one
             if n_nonzero == nonzero_val == 1:
                 snake_case      =					tableau[nonzero_rowidx, -1]
                 snake_case      =					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(_A				,    0   )
         return output_dict


if __name__ == "__main__":
   import doctest

   doctest.testmod()

