



"""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 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  List[str]					):



					"""simple docstring"""





					snake_case_					:		str 					=					torch.exp(__a					)
					snake_case_					:		List[Any] 					=					torch.sum(__a				,    dim=1					)  # sum of exp(x_i)
					snake_case_					:		int 					=					torch.sum(x * exp_x				,    dim=1					)  # sum of x_i * exp(x_i)
					return torch.log(__a					) - B / A







class    __lowercase     (					nn.Module):







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											super().__init__()
											snake_case_					:		List[Any] 					=					config.output_attentions
											snake_case_					:		Optional[Any] 					=					config.output_hidden_states
											snake_case_					:		Union[str, Any] 					=					nn.ModuleList([BertLayer(SCREAMING_SNAKE_CASE_       ) for _ in range(config.num_hidden_layers       )]       )
											snake_case_					:		List[str] 					=					nn.ModuleList([BertHighway(SCREAMING_SNAKE_CASE_       ) for _ in range(config.num_hidden_layers       )]       )

											snake_case_					:		Tuple 					=					[-1 for _ in range(config.num_hidden_layers       )]







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											if (type(SCREAMING_SNAKE_CASE_       ) is float) or (type(SCREAMING_SNAKE_CASE_       ) is int):
																for i in range(len(self.early_exit_entropy       )       ):
																					snake_case_					:		List[str] 					=					x
											else:
																snake_case_					:		Optional[int] 					=					x







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		List[Any] 					=					pooler.state_dict()
											for highway in self.highway:
																for name, param in highway.pooler.state_dict().items():
																					param.copy_(loaded_model[name]       )







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		):
											snake_case_					:		Optional[int] 					=					()
											snake_case_					:		List[str] 					=					()
											snake_case_					:		str 					=					()
											for i, layer_module in enumerate(self.layer       ):
																if self.output_hidden_states:
																					snake_case_					:		Tuple 					=					all_hidden_states + (hidden_states,)

																snake_case_					:		Tuple 					=					layer_module(
																    SCREAMING_SNAKE_CASE_   ,		SCREAMING_SNAKE_CASE_   ,		head_mask[i]   ,		SCREAMING_SNAKE_CASE_   ,		SCREAMING_SNAKE_CASE_       )
																snake_case_					:		List[Any] 					=					layer_outputs[0]

																if self.output_attentions:
																					snake_case_					:		Union[str, Any] 					=					all_attentions + (layer_outputs[1],)

																snake_case_					:		str 					=					(hidden_states,)
																if self.output_hidden_states:
																					snake_case_					:		Optional[Any] 					=					current_outputs + (all_hidden_states,)
																if self.output_attentions:
																					snake_case_					:		Any 					=					current_outputs + (all_attentions,)

																snake_case_					:		Tuple 					=					self.highway[i](SCREAMING_SNAKE_CASE_       )
																# logits, pooled_output

																if not self.training:
																					snake_case_					:		List[str] 					=					highway_exit[0]
																					snake_case_					:		str 					=					entropy(SCREAMING_SNAKE_CASE_       )
																					snake_case_					:		Any 					=					highway_exit + (highway_entropy,)  # logits, hidden_states(?), entropy
																					snake_case_					:		List[Any] 					=					all_highway_exits + (highway_exit,)

																					if highway_entropy < self.early_exit_entropy[i]:
																										snake_case_					:		str 					=					(highway_logits,) + current_outputs[1:] + (all_highway_exits,)
																										raise HighwayException(SCREAMING_SNAKE_CASE_   ,		i + 1       )
																else:
																					snake_case_					:		str 					=					all_highway_exits + (highway_exit,)

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

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

											snake_case_					:		Optional[int] 					=					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). """					,    __lowerCamelCase					,    )
class    __lowercase     (					__lowerCamelCase):







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											super().__init__(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		List[str] 					=					config

											snake_case_					:		Optional[Any] 					=					BertEmbeddings(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		Any 					=					DeeBertEncoder(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		Dict 					=					BertPooler(SCREAMING_SNAKE_CASE_       )

											self.init_weights()







						def 						__UpperCamelCase							(self       ):
											self.encoder.init_highway_pooler(self.pooler       )







						def 						__UpperCamelCase							(self       ):
											return self.embeddings.word_embeddings







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Optional[Any] 					=					value







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											for layer, heads in heads_to_prune.items():
																self.encoder.layer[layer].attention.prune_heads(SCREAMING_SNAKE_CASE_       )







						@add_start_docstrings_to_model_forward(SCREAMING_SNAKE_CASE_       )
						def 						__UpperCamelCase							(self   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		):
											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:
																snake_case_					:		int 					=					input_ids.size()
											elif inputs_embeds is not None:
																snake_case_					:		List[Any] 					=					inputs_embeds.size()[:-1]
											else:
																raise ValueError("""You have to specify either input_ids or inputs_embeds"""       )

											snake_case_					:		Union[str, Any] 					=					input_ids.device if input_ids is not None else inputs_embeds.device

											if attention_mask is None:
																snake_case_					:		List[str] 					=					torch.ones(SCREAMING_SNAKE_CASE_   ,		device=SCREAMING_SNAKE_CASE_       )
											if encoder_attention_mask is None:
																snake_case_					:		Tuple 					=					torch.ones(SCREAMING_SNAKE_CASE_   ,		device=SCREAMING_SNAKE_CASE_       )
											if token_type_ids is None:
																snake_case_					:		Any 					=					torch.zeros(SCREAMING_SNAKE_CASE_   ,		dtype=torch.long   ,		device=SCREAMING_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.
											snake_case_					:		List[str] 					=					self.get_extended_attention_mask(SCREAMING_SNAKE_CASE_   ,		SCREAMING_SNAKE_CASE_   ,		SCREAMING_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:
																snake_case_					:		Optional[Any] 					=					encoder_attention_mask[:, None, :, :]
											if encoder_attention_mask.dim() == 2:
																snake_case_					:		Optional[int] 					=					encoder_attention_mask[:, None, None, :]

											snake_case_					:		List[str] 					=					encoder_extended_attention_mask.to(
											    dtype=next(self.parameters()       ).dtype       )  # fp16 compatibility
											snake_case_					:		Tuple 					=					(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]
											snake_case_					:		Union[str, Any] 					=					self.get_head_mask(SCREAMING_SNAKE_CASE_   ,		self.config.num_hidden_layers       )

											snake_case_					:		Dict 					=					self.embeddings(
											    input_ids=SCREAMING_SNAKE_CASE_   ,		position_ids=SCREAMING_SNAKE_CASE_   ,		token_type_ids=SCREAMING_SNAKE_CASE_   ,		inputs_embeds=SCREAMING_SNAKE_CASE_       )
											snake_case_					:		Optional[int] 					=					self.encoder(
											    SCREAMING_SNAKE_CASE_   ,		attention_mask=SCREAMING_SNAKE_CASE_   ,		head_mask=SCREAMING_SNAKE_CASE_   ,		encoder_hidden_states=SCREAMING_SNAKE_CASE_   ,		encoder_attention_mask=SCREAMING_SNAKE_CASE_   ,		)
											snake_case_					:		Tuple 					=					encoder_outputs[0]
											snake_case_					:		str 					=					self.pooler(SCREAMING_SNAKE_CASE_       )

											snake_case_					:		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    __lowercase     (					__lowerCamelCase):







						"""simple docstring"""







						def __init__(self   ,		lowercase__   ,		lowercase__       ):
											snake_case_					:		Optional[Any] 					=					message
											snake_case_					:		str 					=					exit_layer  # start from 1!







class    __lowercase     (					nn.Module):







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											super().__init__()
											snake_case_					:		Any 					=					BertPooler(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		List[str] 					=					nn.Dropout(config.hidden_dropout_prob       )
											snake_case_					:		List[Any] 					=					nn.Linear(config.hidden_size   ,		config.num_labels       )







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											# Pooler
											snake_case_					:		int 					=					encoder_outputs[0]
											snake_case_					:		Optional[int] 					=					self.pooler(SCREAMING_SNAKE_CASE_       )
											# "return" pooler_output

											# BertModel
											snake_case_					:		str 					=					(pooler_input, pooler_output) + encoder_outputs[1:]
											# "return" bmodel_output

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

											snake_case_					:		Tuple 					=					self.dropout(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		Optional[Any] 					=					self.classifier(SCREAMING_SNAKE_CASE_       )

											return logits, pooled_output







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







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											super().__init__(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		Union[str, Any] 					=					config.num_labels
											snake_case_					:		Any 					=					config.num_hidden_layers

											snake_case_					:		Optional[Any] 					=					DeeBertModel(SCREAMING_SNAKE_CASE_       )
											snake_case_					:		Any 					=					nn.Dropout(config.hidden_dropout_prob       )
											snake_case_					:		Optional[int] 					=					nn.Linear(config.hidden_size   ,		self.config.num_labels       )

											self.init_weights()







						@add_start_docstrings_to_model_forward(SCREAMING_SNAKE_CASE_       )
						def 						__UpperCamelCase							(self   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=-1   ,		lowercase__=False   ,		):
											snake_case_					:		Optional[Any] 					=					self.num_layers
											try:
																snake_case_					:		Union[str, Any] 					=					self.bert(
																    SCREAMING_SNAKE_CASE_   ,		attention_mask=SCREAMING_SNAKE_CASE_   ,		token_type_ids=SCREAMING_SNAKE_CASE_   ,		position_ids=SCREAMING_SNAKE_CASE_   ,		head_mask=SCREAMING_SNAKE_CASE_   ,		inputs_embeds=SCREAMING_SNAKE_CASE_   ,		)
																# sequence_output, pooled_output, (hidden_states), (attentions), highway exits

																snake_case_					:		str 					=					outputs[1]

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

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

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

																if train_highway:
																					snake_case_					:		Tuple 					=					(sum(highway_losses[:-1]       ),) + outputs
																					# exclude the final highway, of course
																else:
																					snake_case_					:		Tuple 					=					(loss,) + outputs
											if not self.training:
																snake_case_					:		List[Any] 					=					outputs + ((original_entropy, highway_entropy), exit_layer)
																if output_layer >= 0:
																					snake_case_					:		int 					=					(
																					    (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"""

from manim import *







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Union[str, Any] 					=					Rectangle(height=0.5   ,		width=0.5       )
											snake_case_					:		str 					=					Rectangle(height=0.46   ,		width=0.46       ).set_stroke(width=0       )

											snake_case_					:		Optional[Any] 					=					[mem.copy() for i in range(6       )]
											snake_case_					:		str 					=					[mem.copy() for i in range(6       )]
											snake_case_					:		str 					=					VGroup(*lowercase__       ).arrange(lowercase__   ,		buff=0       )
											snake_case_					:		Any 					=					VGroup(*lowercase__       ).arrange(lowercase__   ,		buff=0       )
											snake_case_					:		List[str] 					=					VGroup(lowercase__   ,		lowercase__       ).arrange(lowercase__   ,		buff=0       )
											snake_case_					:		List[Any] 					=					Text("""CPU"""   ,		font_size=24       )
											snake_case_					:		Tuple 					=					Group(lowercase__   ,		lowercase__       ).arrange(lowercase__   ,		buff=0.5   ,		aligned_edge=lowercase__       )
											cpu.move_to([-2.5, -0.5, 0]       )
											self.add(lowercase__       )

											snake_case_					:		List[Any] 					=					[mem.copy() for i in range(4       )]
											snake_case_					:		Tuple 					=					VGroup(*lowercase__       ).arrange(lowercase__   ,		buff=0       )
											snake_case_					:		List[str] 					=					Text("""GPU"""   ,		font_size=24       )
											snake_case_					:		Any 					=					Group(lowercase__   ,		lowercase__       ).arrange(lowercase__   ,		buff=0.5   ,		aligned_edge=lowercase__       )
											gpu.move_to([-1, -1, 0]       )
											self.add(lowercase__       )

											snake_case_					:		Optional[Any] 					=					[mem.copy() for i in range(6       )]
											snake_case_					:		List[Any] 					=					VGroup(*lowercase__       ).arrange(lowercase__   ,		buff=0       )

											snake_case_					:		Dict 					=					Text("""Model"""   ,		font_size=24       )
											snake_case_					:		int 					=					Group(lowercase__   ,		lowercase__       ).arrange(lowercase__   ,		buff=0.5   ,		aligned_edge=lowercase__       )
											model.move_to([3, -1.0, 0]       )
											self.add(lowercase__       )

											snake_case_					:		Dict 					=					[]
											for i, rect in enumerate(lowercase__       ):
																rect.set_stroke(lowercase__       )
																# target = fill.copy().set_fill(YELLOW, opacity=0.7)
																# target.move_to(rect)
																# self.add(target)

																snake_case_					:		List[str] 					=					Rectangle(height=0.46 / 4   ,		width=0.46 / 3       ).set_stroke(width=0.0       ).set_fill(lowercase__   ,		opacity=0.7       )

																if i == 0:
																					cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT       )   ,		buff=0.02   ,		direction=lowercase__       )
																					cpu_target.set_x(cpu_target.get_x() + 0.1       )
																elif i == 3:
																					cpu_target.next_to(cpu_targs[0]   ,		direction=lowercase__   ,		buff=0.0       )
																else:
																					cpu_target.next_to(cpu_targs[i - 1]   ,		direction=lowercase__   ,		buff=0.0       )
																self.add(lowercase__       )
																cpu_targs.append(lowercase__       )

											snake_case_					:		List[str] 					=					[mem.copy() for i in range(6       )]
											snake_case_					:		List[str] 					=					VGroup(*lowercase__       ).arrange(lowercase__   ,		buff=0       )

											snake_case_					:		str 					=					Text("""Loaded Checkpoint"""   ,		font_size=24       )
											snake_case_					:		Any 					=					Group(lowercase__   ,		lowercase__       ).arrange(lowercase__   ,		aligned_edge=lowercase__   ,		buff=0.4       )
											checkpoint.move_to([3, 0.5, 0]       )

											snake_case_					:		Optional[Any] 					=					Square(side_length=2.2       )
											key.move_to([-5, 2, 0]       )

											snake_case_					:		Union[str, Any] 					=					MarkupText(
											    f'<b>Key:</b>\n\n<span fgcolor=\'{YELLOW}\'>●</span> Empty Model'   ,		font_size=18   ,		)

											key_text.move_to([-5, 2.4, 0]       )

											self.add(lowercase__   ,		lowercase__       )

											snake_case_					:		List[Any] 					=					MarkupText(
											    f'<span fgcolor=\'{BLUE}\'>●</span> Checkpoint'   ,		font_size=18   ,		)

											blue_text.next_to(lowercase__   ,		DOWN * 2.4   ,		aligned_edge=key_text.get_left()       )

											snake_case_					:		List[Any] 					=					MarkupText(
											    f'Next, a <i><span fgcolor="{BLUE}">second</span></i> model is loaded into memory,\nwith the weights of a <span fgcolor="{BLUE}">single shard</span>.'   ,		font_size=24   ,		)
											step_a.move_to([2, 2, 0]       )
											self.play(Write(lowercase__       )   ,		Write(lowercase__       )       )

											self.play(Write(lowercase__   ,		run_time=1       )   ,		Create(lowercase__   ,		run_time=1       )       )

											snake_case_					:		Optional[int] 					=					[]
											snake_case_					:		List[str] 					=					[]
											for i, rect in enumerate(lowercase__       ):
																snake_case_					:		Optional[Any] 					=					fill.copy().set_fill(lowercase__   ,		opacity=0.7       )
																target.move_to(lowercase__       )
																first_animations.append(GrowFromCenter(lowercase__   ,		run_time=1       )       )

																snake_case_					:		List[Any] 					=					target.copy()
																cpu_target.generate_target()
																if i < 5:
																					cpu_target.target.move_to(cpu_left_col_base[i + 1]       )
																else:
																					cpu_target.target.move_to(cpu_right_col_base[i - 5]       )
																second_animations.append(MoveToTarget(lowercase__   ,		run_time=1.5       )       )

											self.play(*lowercase__       )
											self.play(*lowercase__       )
											self.wait()




