


"""simple docstring"""



import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEmbeddings,
    BertLayer,
    BertPooler,
    BertPreTrainedModel,
)






def       lowerCamelCase					(a_						:int)     ->							str:
       lowercase		:List[Any]       							=				torch.exp(a_)
       lowercase		:str       							=				torch.sum(a_				,				dim=1)  # sum of exp(x_i)
       lowercase		:Tuple       							=				torch.sum(x * exp_x				,				dim=1)  # sum of x_i * exp(x_i)
       return torch.log(a_) - B / A
class 					__magic_name__		(			nn.Module     ):






       def __init__(   self  :    Tuple  ,     snake_case__  :    str				):

              '''simple docstring'''






              super().__init__()
              lowercase		:List[Any]       							=				config.output_attentions
              lowercase		:int       							=				config.output_hidden_states
              lowercase		:str       							=				nn.ModuleList([BertLayer(snake_case__				) for _ in range(config.num_hidden_layers				)]				)
              lowercase		:List[Any]       							=				nn.ModuleList([BertHighway(snake_case__				) for _ in range(config.num_hidden_layers				)]				)

              lowercase		:str       							=				[-1 for _ in range(config.num_hidden_layers				)]






       def  __snake_case						(   self  :    Any  ,     snake_case__  :    Any				):

              '''simple docstring'''






              if (type(snake_case__				) is float) or (type(snake_case__				) is int):
                     for i in range(len(self.early_exit_entropy				)				):
                            lowercase		:Union[str, Any]       							=				x
              else:
                     lowercase		:List[str]       							=				x






       def  __snake_case						(   self  :    Union[str, Any]  ,     snake_case__  :    Dict				):

              '''simple docstring'''






              lowercase		:List[str]       							=				pooler.state_dict()
              for highway in self.highway:
                     for name, param in highway.pooler.state_dict().items():
                            param.copy_(loaded_model[name]				)







       def  __snake_case						(   self  :    str  ,     snake_case__  :    List[str]  ,     snake_case__  :    Union[str, Any]=None  ,     snake_case__  :    Optional[Any]=None  ,     snake_case__  :    Dict=None  ,     snake_case__  :    Optional[Any]=None  ,     ):

              '''simple docstring'''






              lowercase		:Optional[int]       							=				()
              lowercase		:str       							=				()
              lowercase		:str       							=				()
              for i, layer_module in enumerate(self.layer				):
                     if self.output_hidden_states:
                            lowercase		:Any       							=				all_hidden_states + (hidden_states,)

                     lowercase		:Union[str, Any]       							=				layer_module(
                         snake_case__  ,     snake_case__  ,     head_mask[i]  ,     snake_case__  ,     snake_case__				)
                     lowercase		:Tuple       							=				layer_outputs[0]

                     if self.output_attentions:
                            lowercase		:Optional[Any]       							=				all_attentions + (layer_outputs[1],)

                     lowercase		:str       							=				(hidden_states,)
                     if self.output_hidden_states:
                            lowercase		:Union[str, Any]       							=				current_outputs + (all_hidden_states,)
                     if self.output_attentions:
                            lowercase		:Tuple       							=				current_outputs + (all_attentions,)

                     lowercase		:Any       							=				self.highway[i](snake_case__				)
                     # logits, pooled_output

                     if not self.training:
                            lowercase		:Optional[Any]       							=				highway_exit[0]
                            lowercase		:Tuple       							=				entropy(snake_case__				)
                            lowercase		:Union[str, Any]       							=				highway_exit + (highway_entropy,)  # logits, hidden_states(?), entropy
                            lowercase		:Optional[Any]       							=				all_highway_exits + (highway_exit,)

                            if highway_entropy < self.early_exit_entropy[i]:
                                   lowercase		:Union[str, Any]       							=				(highway_logits,) + current_outputs[1:] + (all_highway_exits,)
                                   raise HighwayException(snake_case__  ,     i + 1				)
                     else:
                            lowercase		:List[Any]       							=				all_highway_exits + (highway_exit,)

        # Add last layer
              if self.output_hidden_states:
                     lowercase		:List[Any]       							=				all_hidden_states + (hidden_states,)

              lowercase		:int       							=				(hidden_states,)
              if self.output_hidden_states:
                     lowercase		:Optional[Any]       							=				outputs + (all_hidden_states,)
              if self.output_attentions:
                     lowercase		:str       							=				outputs + (all_attentions,)

              lowercase		:Union[str, Any]       							=				outputs + (all_highway_exits,)
              return outputs  # last-layer hidden state, (all hidden states), (all attentions), all highway exits







@add_start_docstrings(
    "The Bert Model transformer with early exiting (DeeBERT). "			,						__UpperCAmelCase			,						)
class 					__magic_name__		(			__UpperCAmelCase     ):






       def __init__(   self  :    Dict  ,     snake_case__  :    Optional[int]				):

              '''simple docstring'''






              super().__init__(snake_case__				)
              lowercase		:int       							=				config

              lowercase		:Optional[Any]       							=				BertEmbeddings(snake_case__				)
              lowercase		:Optional[Any]       							=				DeeBertEncoder(snake_case__				)
              lowercase		:List[str]       							=				BertPooler(snake_case__				)

              self.init_weights()






       def  __snake_case						(   self  :    List[str]				):

              '''simple docstring'''






              self.encoder.init_highway_pooler(self.pooler				)






       def  __snake_case						(   self  :    Optional[Any]				):

              '''simple docstring'''






              return self.embeddings.word_embeddings






       def  __snake_case						(   self  :    Optional[Any]  ,     snake_case__  :    List[Any]				):

              '''simple docstring'''






              lowercase		:List[str]       							=				value






       def  __snake_case						(   self  :    int  ,     snake_case__  :    Dict				):

              '''simple docstring'''






              for layer, heads in heads_to_prune.items():
                     self.encoder.layer[layer].attention.prune_heads(snake_case__				)







       @add_start_docstrings_to_model_forward(snake_case__				)
       def  __snake_case						(   self  :    Optional[Any]  ,     snake_case__  :    List[Any]=None  ,     snake_case__  :    Optional[Any]=None  ,     snake_case__  :    str=None  ,     snake_case__  :    Any=None  ,     snake_case__  :    Tuple=None  ,     snake_case__  :    Union[str, Any]=None  ,     snake_case__  :    Optional[int]=None  ,     snake_case__  :    int=None  ,     ):

              '''simple docstring'''






              if input_ids is not None and inputs_embeds is not None:
                     raise ValueError('''You cannot specify both input_ids and inputs_embeds at the same time'''				)
              elif input_ids is not None:
                     lowercase		:Tuple       							=				input_ids.size()
              elif inputs_embeds is not None:
                     lowercase		:str       							=				inputs_embeds.size()[:-1]
              else:
                     raise ValueError('''You have to specify either input_ids or inputs_embeds'''				)

              lowercase		:List[Any]       							=				input_ids.device if input_ids is not None else inputs_embeds.device

              if attention_mask is None:
                     lowercase		:Optional[int]       							=				torch.ones(snake_case__  ,     device=snake_case__				)
              if encoder_attention_mask is None:
                     lowercase		:Optional[int]       							=				torch.ones(snake_case__  ,     device=snake_case__				)
              if token_type_ids is None:
                     lowercase		:Tuple       							=				torch.zeros(snake_case__  ,     dtype=torch.long  ,     device=snake_case__				)

              # We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
              # ourselves in which case we just need to make it broadcastable to all heads.
              lowercase		:torch.Tensor       							=				self.get_extended_attention_mask(snake_case__  ,     snake_case__  ,     snake_case__				)

              # If a 2D ou 3D attention mask is provided for the cross-attention
              # we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
              if encoder_attention_mask.dim() == 3:
                     lowercase		:Tuple       							=				encoder_attention_mask[:, None, :, :]
              if encoder_attention_mask.dim() == 2:
                     lowercase		:int       							=				encoder_attention_mask[:, None, None, :]

              lowercase		:Optional[int]       							=				encoder_extended_attention_mask.to(
                  dtype=next(self.parameters()				).dtype				)  # fp16 compatibility
              lowercase		:str       							=				(1.0 - encoder_extended_attention_mask) * -1_00_00.0

              # Prepare head mask if needed
              # 1.0 in head_mask indicate we keep the head
              # attention_probs has shape bsz x n_heads x N x N
              # input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
              # and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
              lowercase		:str       							=				self.get_head_mask(snake_case__  ,     self.config.num_hidden_layers				)

              lowercase		:int       							=				self.embeddings(
                  input_ids=snake_case__  ,     position_ids=snake_case__  ,     token_type_ids=snake_case__  ,     inputs_embeds=snake_case__				)
              lowercase		:List[Any]       							=				self.encoder(
                  snake_case__  ,     attention_mask=snake_case__  ,     head_mask=snake_case__  ,     encoder_hidden_states=snake_case__  ,     encoder_attention_mask=snake_case__  ,     )
              lowercase		:Optional[Any]       							=				encoder_outputs[0]
              lowercase		:List[str]       							=				self.pooler(snake_case__				)

              lowercase		:int       							=				(
                  sequence_output,
                  pooled_output,
              ) + encoder_outputs[
                  1:
              ]  # add hidden_states and attentions if they are here
              return outputs  # sequence_output, pooled_output, (hidden_states), (attentions), highway exits







class 					__magic_name__		(			__UpperCAmelCase     ):






       def __init__(   self  :    Any  ,     snake_case__  :    Optional[Any]  ,     snake_case__  :    str				):

              '''simple docstring'''






              lowercase		:Any       							=				message
              lowercase		:Any       							=				exit_layer  # start from 1!







class 					__magic_name__		(			nn.Module     ):






       def __init__(   self  :    Dict  ,     snake_case__  :    Dict				):

              '''simple docstring'''






              super().__init__()
              lowercase		:Optional[int]       							=				BertPooler(snake_case__				)
              lowercase		:List[Any]       							=				nn.Dropout(config.hidden_dropout_prob				)
              lowercase		:Optional[int]       							=				nn.Linear(config.hidden_size  ,     config.num_labels				)







       def  __snake_case						(   self  :    Optional[int]  ,     snake_case__  :    Optional[int]				):

              '''simple docstring'''






              lowercase		:Tuple       							=				encoder_outputs[0]
              lowercase		:Any       							=				self.pooler(snake_case__				)
              # "return" pooler_output

              # BertModel
              lowercase		:Optional[Any]       							=				(pooler_input, pooler_output) + encoder_outputs[1:]
              # "return" bmodel_output

              # Dropout and classification
              lowercase		:str       							=				bmodel_output[1]

              lowercase		:str       							=				self.dropout(snake_case__				)
              lowercase		:Tuple       							=				self.classifier(snake_case__				)

              return logits, pooled_output







@add_start_docstrings(
    "Bert Model (with early exiting - DeeBERT) with a classifier on top,\n    also takes care of multi-layer training. "			,						__UpperCAmelCase			,						)
class 					__magic_name__		(			__UpperCAmelCase     ):






       def __init__(   self  :    Tuple  ,     snake_case__  :    List[Any]				):

              '''simple docstring'''






              super().__init__(snake_case__				)
              lowercase		:List[Any]       							=				config.num_labels
              lowercase		:Dict       							=				config.num_hidden_layers

              lowercase		:int       							=				DeeBertModel(snake_case__				)
              lowercase		:Union[str, Any]       							=				nn.Dropout(config.hidden_dropout_prob				)
              lowercase		:Optional[Any]       							=				nn.Linear(config.hidden_size  ,     self.config.num_labels				)

              self.init_weights()







       @add_start_docstrings_to_model_forward(snake_case__				)
       def  __snake_case						(   self  :    List[str]  ,     snake_case__  :    int=None  ,     snake_case__  :    List[Any]=None  ,     snake_case__  :    Dict=None  ,     snake_case__  :    Optional[int]=None  ,     snake_case__  :    List[str]=None  ,     snake_case__  :    Any=None  ,     snake_case__  :    List[str]=None  ,     snake_case__  :    str=-1  ,     snake_case__  :    int=False  ,     ):

              '''simple docstring'''






              lowercase		:Union[str, Any]       							=				self.num_layers
              try:
                     lowercase		:Optional[Any]       							=				self.bert(
                         snake_case__  ,     attention_mask=snake_case__  ,     token_type_ids=snake_case__  ,     position_ids=snake_case__  ,     head_mask=snake_case__  ,     inputs_embeds=snake_case__  ,     )
                     # sequence_output, pooled_output, (hidden_states), (attentions), highway exits

                     lowercase		:str       							=				outputs[1]

                     lowercase		:Dict       							=				self.dropout(snake_case__				)
                     lowercase		:Dict       							=				self.classifier(snake_case__				)
                     lowercase		:List[str]       							=				(logits,) + outputs[2:]  # add hidden states and attention if they are here
              except HighwayException as e:
                     lowercase		:Optional[int]       							=				e.message
                     lowercase		:int       							=				e.exit_layer
                     lowercase		:List[str]       							=				outputs[0]

              if not self.training:
                     lowercase		:int       							=				entropy(snake_case__				)
                     lowercase		:Any       							=				[]
                     lowercase		:str       							=				[]
              if labels is not None:
                     if self.num_labels == 1:
                            #  We are doing regression
                            lowercase		:str       							=				MSELoss()
                            lowercase		:Dict       							=				loss_fct(logits.view(-1				)  ,     labels.view(-1				)				)
                     else:
                            lowercase		:Optional[int]       							=				CrossEntropyLoss()
                            lowercase		:Dict       							=				loss_fct(logits.view(-1  ,     self.num_labels				)  ,     labels.view(-1				)				)

                     # work with highway exits
                     lowercase		:Optional[int]       							=				[]
                     for highway_exit in outputs[-1]:
                            lowercase		:Tuple       							=				highway_exit[0]
                            if not self.training:
                                   highway_logits_all.append(snake_case__				)
                                   highway_entropy.append(highway_exit[2]				)
                            if self.num_labels == 1:
                                   #  We are doing regression
                                   lowercase		:Any       							=				MSELoss()
                                   lowercase		:Optional[Any]       							=				loss_fct(highway_logits.view(-1				)  ,     labels.view(-1				)				)
                            else:
                                   lowercase		:List[Any]       							=				CrossEntropyLoss()
                                   lowercase		:Optional[Any]       							=				loss_fct(highway_logits.view(-1  ,     self.num_labels				)  ,     labels.view(-1				)				)
                            highway_losses.append(snake_case__				)

                     if train_highway:
                            lowercase		:int       							=				(sum(highway_losses[:-1]				),) + outputs
                            # exclude the final highway, of course
                     else:
                            lowercase		:List[str]       							=				(loss,) + outputs
              if not self.training:
                     lowercase		:Optional[int]       							=				outputs + ((original_entropy, highway_entropy), exit_layer)
                     if output_layer >= 0:
                            lowercase		:List[str]       							=				(
                                (outputs[0],) + (highway_logits_all[output_layer],) + outputs[2:]
                            )  # use the highway of the last layer

              return outputs  # (loss), logits, (hidden_states), (attentions), (highway_exits)





"""simple docstring"""



import unittest

import numpy as np






def       lowerCamelCase					(a_						:np.ndarray				,				a_						:np.ndarray				,				a_						:np.ndarray				,				a_						:np.ndarray | None = None				,				)     ->							np.ndarray:
       lowercase		:str       							=				np.shape(a_)
       lowercase		:Dict       							=				np.shape(a_)
       lowercase		:Optional[int]       							=				np.shape(a_)

       if shape_a[0] != shape_b[0]:
              lowercase		:List[str]       							=				(
                  '''Expected the same number of rows for A and B. '''
                  F"""Instead found A of size {shape_a} and B of size {shape_b}"""
              )
              raise ValueError(a_)

       if shape_b[1] != shape_c[1]:
              lowercase		:List[str]       							=				(
                  '''Expected the same number of columns for B and C. '''
                  F"""Instead found B of size {shape_b} and C of size {shape_c}"""
              )
              raise ValueError(a_)

       lowercase		:Optional[Any]       							=				pseudo_inv
       if a_inv is None:
              try:
                     lowercase		:int       							=				np.linalg.inv(a_)
              except np.linalg.LinAlgError:
                     raise ValueError(
                         '''Input matrix A is not invertible. Cannot compute Schur complement.''')

       return mat_c - mat_b.T @ a_inv @ mat_b
class 					__magic_name__		(			unittest.TestCase     ):






       def  __snake_case						(   self  :    int				):

              '''simple docstring'''






              lowercase		:Dict       							=				np.array([[1, 2, 1], [2, 1, 2], [3, 2, 4]]				)
              lowercase		:List[str]       							=				np.array([[0, 3], [3, 0], [2, 3]]				)
              lowercase		:List[str]       							=				np.array([[2, 1], [6, 3]]				)

              lowercase		:Any       							=				schur_complement(snake_case__  ,     snake_case__  ,     snake_case__				)

              lowercase		:int       							=				np.block([[a, b], [b.T, c]]				)

              lowercase		:Tuple       							=				np.linalg.det(snake_case__				)
              lowercase		:Any       							=				np.linalg.det(snake_case__				)
              lowercase		:List[Any]       							=				np.linalg.det(snake_case__				)

              self.assertAlmostEqual(snake_case__  ,     det_a * det_s				)






       def  __snake_case						(   self  :    Optional[int]				):

              '''simple docstring'''






              lowercase		:Union[str, Any]       							=				np.array([[1, 2, 1], [2, 1, 2], [3, 2, 4]]				)
              lowercase		:Union[str, Any]       							=				np.array([[0, 3], [3, 0], [2, 3]]				)
              lowercase		:List[str]       							=				np.array([[2, 1], [6, 3]]				)

              with self.assertRaises(snake_case__				):
                     schur_complement(snake_case__  ,     snake_case__  ,     snake_case__				)







       def  __snake_case						(   self  :    List[Any]				):

              '''simple docstring'''






              lowercase		:List[str]       							=				np.array([[1, 2, 1], [2, 1, 2], [3, 2, 4]]				)
              lowercase		:int       							=				np.array([[0, 3], [3, 0], [2, 3]]				)
              lowercase		:List[Any]       							=				np.array([[2, 1, 3], [6, 3, 5]]				)

              with self.assertRaises(snake_case__				):
                     schur_complement(snake_case__  ,     snake_case__  ,     snake_case__				)


if __name__ == "__main__":
 import doctest

 doctest.testmod()
 unittest.main()

