




import math
from dataclasses import dataclass
from typing import Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin
@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->UnCLIP
class 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





						lowerCAmelCase_							=  42
						lowerCAmelCase_							=  None







def 		lowerCamelCase__       (						_A	,      _A=0.9_99	,      _A="cosine"	,      ):







	'''simple docstring'''





	if alpha_transform_type == "cosine":

		def alpha_bar_fn(_A				):
			return math.cos((t + 0.0_08) / 1.0_08 * math.pi / 2				) ** 2

	elif alpha_transform_type == "exp":

		def alpha_bar_fn(_A				):
			return math.exp(t * -12.0				)

	else:
		raise ValueError(f"Unsupported alpha_tranform_type: {alpha_transform_type}"				)

	snake_case_						       =	[]
	for i in range(_A				):
		snake_case_						       =	i / num_diffusion_timesteps
		snake_case_						       =	(i + 1) / num_diffusion_timesteps
		betas.append(min(1 - alpha_bar_fn(_A				) / alpha_bar_fn(_A				)	,      _A				)				)
	return torch.tensor(_A	,      dtype=torch.floataa				)






class 						UpperCAmelCase  (							UpperCAmelCase__			,  UpperCAmelCase__     ):







						'''simple docstring'''





						@register_to_config
						def __init__(		self				:	str			,      __lowercase				:	int = 10_00			,      __lowercase				:	str = "fixed_small_log"			,      __lowercase				:	bool = True			,      __lowercase				:	Optional[float] = 1.0			,      __lowercase				:	str = "epsilon"			,      __lowercase				:	str = "squaredcos_cap_v2"			,      ):






							"""simple docstring"""






							if beta_schedule != "squaredcos_cap_v2":
								raise ValueError("UnCLIPScheduler only supports `beta_schedule`: 'squaredcos_cap_v2'"		)

							snake_case_						       =	betas_for_alpha_bar(__lowercase		)

							snake_case_						       =	1.0 - self.betas
							snake_case_						       =	torch.cumprod(self.alphas			,      dim=0		)
							snake_case_						       =	torch.tensor(1.0		)

							# standard deviation of the initial noise distribution
							snake_case_						       =	1.0

							# setable values
							snake_case_						       =	None
							snake_case_						       =	torch.from_numpy(np.arange(0			,      __lowercase		)[::-1].copy()		)

							snake_case_						       =	variance_type






						def 			snake_case__		(		self				:	Optional[Any]			,      __lowercase				:	torch.FloatTensor			,      __lowercase				:	Optional[int] = None		):






							"""simple docstring"""






							return sample






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	int			,      __lowercase				:	Union[str, torch.device] = None		):






							"""simple docstring"""






							snake_case_						       =	num_inference_steps
							snake_case_						       =	(self.config.num_train_timesteps - 1) / (self.num_inference_steps - 1)
							snake_case_						       =	(np.arange(0			,      __lowercase		) * step_ratio).round()[::-1].copy().astype(np.intaa		)
							snake_case_						       =	torch.from_numpy(__lowercase		).to(__lowercase		)






						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	str			,      __lowercase				:	Optional[int]=None			,      __lowercase				:	Tuple=None			,      __lowercase				:	str=None		):






							"""simple docstring"""






							if prev_timestep is None:
								snake_case_						       =	t - 1

							snake_case_						       =	self.alphas_cumprod[t]
							snake_case_						       =	self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
							snake_case_						       =	1 - alpha_prod_t
							snake_case_						       =	1 - alpha_prod_t_prev

							if prev_timestep == t - 1:
								snake_case_						       =	self.betas[t]
							else:
								snake_case_						       =	1 - alpha_prod_t / alpha_prod_t_prev

							# For t > 0, compute predicted variance βt (see formula (6) and (7) from https://arxiv.org/pdf/2006.11239.pdf)
							# and sample from it to get previous sample
							# x_{t-1} ~ N(pred_prev_sample, variance) == add variance to pred_sample
							snake_case_						       =	beta_prod_t_prev / beta_prod_t * beta

							if variance_type is None:
								snake_case_						       =	self.config.variance_type

							# hacks - were probably added for training stability
							if variance_type == "fixed_small_log":
								snake_case_						       =	torch.log(torch.clamp(__lowercase			,      min=1E-20		)		)
								snake_case_						       =	torch.exp(0.5 * variance		)
							elif variance_type == "learned_range":
								# NOTE difference with DDPM scheduler
								snake_case_						       =	variance.log()
								snake_case_						       =	beta.log()

								snake_case_						       =	(predicted_variance + 1) / 2
								snake_case_						       =	frac * max_log + (1 - frac) * min_log

							return variance






						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	torch.FloatTensor			,      __lowercase				:	int			,      __lowercase				:	torch.FloatTensor			,      __lowercase				:	Optional[int] = None			,      __lowercase				:	Tuple=None			,      __lowercase				:	bool = True			,      ):






							"""simple docstring"""






							snake_case_						       =	timestep

							if model_output.shape[1] == sample.shape[1] * 2 and self.variance_type == "learned_range":
								snake_case_       , snake_case_						       =	torch.split(__lowercase			,      sample.shape[1]			,      dim=1		)
							else:
								snake_case_						       =	None

							# 1. compute alphas, betas
							if prev_timestep is None:
								snake_case_						       =	t - 1

							snake_case_						       =	self.alphas_cumprod[t]
							snake_case_						       =	self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
							snake_case_						       =	1 - alpha_prod_t
							snake_case_						       =	1 - alpha_prod_t_prev

							if prev_timestep == t - 1:
								snake_case_						       =	self.betas[t]
								snake_case_						       =	self.alphas[t]
							else:
								snake_case_						       =	1 - alpha_prod_t / alpha_prod_t_prev
								snake_case_						       =	1 - beta

							# 2. compute predicted original sample from predicted noise also called
							# "predicted x_0" of formula (15) from https://arxiv.org/pdf/2006.11239.pdf
							if self.config.prediction_type == "epsilon":
								snake_case_						       =	(sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
							elif self.config.prediction_type == "sample":
								snake_case_						       =	model_output
							else:
								raise ValueError(
								    f"prediction_type given as {self.config.prediction_type} must be one of `epsilon` or `sample`"
								    " for the UnCLIPScheduler."		)

							# 3. Clip "predicted x_0"
							if self.config.clip_sample:
								snake_case_						       =	torch.clamp(
								    __lowercase			,      -self.config.clip_sample_range			,      self.config.clip_sample_range		)

							# 4. Compute coefficients for pred_original_sample x_0 and current sample x_t
							# See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
							snake_case_						       =	(alpha_prod_t_prev ** 0.5 * beta) / beta_prod_t
							snake_case_						       =	alpha ** 0.5 * beta_prod_t_prev / beta_prod_t

							# 5. Compute predicted previous sample µ_t
							# See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
							snake_case_						       =	pred_original_sample_coeff * pred_original_sample + current_sample_coeff * sample

							# 6. Add noise
							snake_case_						       =	0
							if t > 0:
								snake_case_						       =	randn_tensor(
								    model_output.shape			,      dtype=model_output.dtype			,      generator=__lowercase			,      device=model_output.device		)

								snake_case_						       =	self._get_variance(
								    __lowercase			,      predicted_variance=__lowercase			,      prev_timestep=__lowercase			,      )

								if self.variance_type == "fixed_small_log":
									snake_case_						       =	variance
								elif self.variance_type == "learned_range":
									snake_case_						       =	(0.5 * variance).exp()
								else:
									raise ValueError(
									    f"variance_type given as {self.variance_type} must be one of `fixed_small_log` or `learned_range`"
									    " for the UnCLIPScheduler."		)

								snake_case_						       =	variance * variance_noise

							snake_case_						       =	pred_prev_sample + variance

							if not return_dict:
								return (pred_prev_sample,)

							return UnCLIPSchedulerOutput(prev_sample=__lowercase			,      pred_original_sample=__lowercase		)






						def 			snake_case__		(		self				:	List[Any]			,      __lowercase				:	torch.FloatTensor			,      __lowercase				:	torch.FloatTensor			,      __lowercase				:	torch.IntTensor			,      ):






							"""simple docstring"""






							snake_case_						       =	self.alphas_cumprod.to(device=original_samples.device			,      dtype=original_samples.dtype		)
							snake_case_						       =	timesteps.to(original_samples.device		)

							snake_case_						       =	alphas_cumprod[timesteps] ** 0.5
							snake_case_						       =	sqrt_alpha_prod.flatten()
							while len(sqrt_alpha_prod.shape		) < len(original_samples.shape		):
								snake_case_						       =	sqrt_alpha_prod.unsqueeze(-1		)

							snake_case_						       =	(1 - alphas_cumprod[timesteps]) ** 0.5
							snake_case_						       =	sqrt_one_minus_alpha_prod.flatten()
							while len(sqrt_one_minus_alpha_prod.shape		) < len(original_samples.shape		):
								snake_case_						       =	sqrt_one_minus_alpha_prod.unsqueeze(-1		)

							snake_case_						       =	sqrt_alpha_prod * original_samples + sqrt_one_minus_alpha_prod * noise
							return noisy_samples






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				):







	'''simple docstring'''





	snake_case_						       =	torch.exp(_A				)
	snake_case_						       =	torch.sum(_A	,      dim=1				)  # sum of exp(x_i)
	snake_case_						       =	torch.sum(x * exp_x	,      dim=1				)  # sum of x_i * exp(x_i)
	return torch.log(_A				) - B / A
class 						UpperCAmelCase  (							nn.Module     ):







						'''simple docstring'''





						def __init__(		self				:	Optional[int]			,      __lowercase				:	List[Any]		):






							"""simple docstring"""






							super().__init__()
							snake_case_						       =	config.output_attentions
							snake_case_						       =	config.output_hidden_states
							snake_case_						       =	nn.ModuleList([BertLayer(__lowercase		) for _ in range(config.num_hidden_layers		)]		)
							snake_case_						       =	nn.ModuleList([BertHighway(__lowercase		) for _ in range(config.num_hidden_layers		)]		)

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






						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	Optional[int]		):






							"""simple docstring"""






							if (type(__lowercase		) is float) or (type(__lowercase		) is int):
								for i in range(len(self.early_exit_entropy		)		):
									snake_case_						       =	x
							else:
								snake_case_						       =	x






						def 			snake_case__		(		self				:	int			,      __lowercase				:	Tuple		):






							"""simple docstring"""






							snake_case_						       =	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				:	List[Any]			,      __lowercase				:	List[str]			,      __lowercase				:	List[Any]=None			,      __lowercase				:	Optional[int]=None			,      __lowercase				:	Optional[Any]=None			,      __lowercase				:	List[str]=None			,      ):






							"""simple docstring"""






							snake_case_						       =	()
							snake_case_						       =	()
							snake_case_						       =	()
							for i, layer_module in enumerate(self.layer		):
								if self.output_hidden_states:
									snake_case_						       =	all_hidden_states + (hidden_states,)

								snake_case_						       =	layer_module(
								    __lowercase			,      __lowercase			,      head_mask[i]			,      __lowercase			,      __lowercase		)
								snake_case_						       =	layer_outputs[0]

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

								snake_case_						       =	(hidden_states,)
								if self.output_hidden_states:
									snake_case_						       =	current_outputs + (all_hidden_states,)
								if self.output_attentions:
									snake_case_						       =	current_outputs + (all_attentions,)

								snake_case_						       =	self.highway[i](__lowercase		)
								# logits, pooled_output

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

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

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

							snake_case_						       =	(hidden_states,)
							if self.output_hidden_states:
								snake_case_						       =	outputs + (all_hidden_states,)
							if self.output_attentions:
								snake_case_						       =	outputs + (all_attentions,)

							snake_case_						       =	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 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





						def __init__(		self				:	Union[str, Any]			,      __lowercase				:	List[Any]		):






							"""simple docstring"""






							super().__init__(__lowercase		)
							snake_case_						       =	config

							snake_case_						       =	BertEmbeddings(__lowercase		)
							snake_case_						       =	DeeBertEncoder(__lowercase		)
							snake_case_						       =	BertPooler(__lowercase		)

							self.init_weights()






						def 			snake_case__		(		self				:	Tuple		):






							"""simple docstring"""






							self.encoder.init_highway_pooler(self.pooler		)






						def 			snake_case__		(		self				:	Tuple		):






							"""simple docstring"""






							return self.embeddings.word_embeddings






						def 			snake_case__		(		self				:	Optional[int]			,      __lowercase				:	Any		):






							"""simple docstring"""






							snake_case_						       =	value






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






							"""simple docstring"""






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






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

							snake_case_						       =	input_ids.device if input_ids is not None else inputs_embeds.device

							if attention_mask is None:
								snake_case_						       =	torch.ones(__lowercase			,      device=__lowercase		)
							if encoder_attention_mask is None:
								snake_case_						       =	torch.ones(__lowercase			,      device=__lowercase		)
							if token_type_ids is None:
								snake_case_						       =	torch.zeros(__lowercase			,      dtype=torch.long			,      device=__lowercase		)

							# 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_						       =	self.get_extended_attention_mask(__lowercase			,      __lowercase			,      __lowercase		)

							# 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_						       =	encoder_attention_mask[:, None, :, :]
							if encoder_attention_mask.dim() == 2:
								snake_case_						       =	encoder_attention_mask[:, None, None, :]

							snake_case_						       =	encoder_extended_attention_mask.to(
							    dtype=next(self.parameters()		).dtype		)  # fp16 compatibility
							snake_case_						       =	(1.0 - encoder_extended_attention_mask) * -1_0000.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_						       =	self.get_head_mask(__lowercase			,      self.config.num_hidden_layers		)

							snake_case_						       =	self.embeddings(
							    input_ids=__lowercase			,      position_ids=__lowercase			,      token_type_ids=__lowercase			,      inputs_embeds=__lowercase		)
							snake_case_						       =	self.encoder(
							    __lowercase			,      attention_mask=__lowercase			,      head_mask=__lowercase			,      encoder_hidden_states=__lowercase			,      encoder_attention_mask=__lowercase			,      )
							snake_case_						       =	encoder_outputs[0]
							snake_case_						       =	self.pooler(__lowercase		)

							snake_case_						       =	(
							    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 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





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






							"""simple docstring"""






							snake_case_						       =	message
							snake_case_						       =	exit_layer  # start from 1!







class 						UpperCAmelCase  (							nn.Module     ):







						'''simple docstring'''





						def __init__(		self				:	int			,      __lowercase				:	Optional[int]		):






							"""simple docstring"""






							super().__init__()
							snake_case_						       =	BertPooler(__lowercase		)
							snake_case_						       =	nn.Dropout(config.hidden_dropout_prob		)
							snake_case_						       =	nn.Linear(config.hidden_size			,      config.num_labels		)






						def 			snake_case__		(		self				:	List[str]			,      __lowercase				:	str		):






							"""simple docstring"""






							snake_case_						       =	encoder_outputs[0]
							snake_case_						       =	self.pooler(__lowercase		)
							# "return" pooler_output

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

							# Dropout and classification
							snake_case_						       =	bmodel_output[1]

							snake_case_						       =	self.dropout(__lowercase		)
							snake_case_						       =	self.classifier(__lowercase		)

							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. '''			,  UpperCAmelCase__			,  )
class 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





						def __init__(		self				:	Union[str, Any]			,      __lowercase				:	Union[str, Any]		):






							"""simple docstring"""






							super().__init__(__lowercase		)
							snake_case_						       =	config.num_labels
							snake_case_						       =	config.num_hidden_layers

							snake_case_						       =	DeeBertModel(__lowercase		)
							snake_case_						       =	nn.Dropout(config.hidden_dropout_prob		)
							snake_case_						       =	nn.Linear(config.hidden_size			,      self.config.num_labels		)

							self.init_weights()






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






							"""simple docstring"""






							snake_case_						       =	self.num_layers
							try:
								snake_case_						       =	self.bert(
								    __lowercase			,      attention_mask=__lowercase			,      token_type_ids=__lowercase			,      position_ids=__lowercase			,      head_mask=__lowercase			,      inputs_embeds=__lowercase			,      )
								# sequence_output, pooled_output, (hidden_states), (attentions), highway exits

								snake_case_						       =	outputs[1]

								snake_case_						       =	self.dropout(__lowercase		)
								snake_case_						       =	self.classifier(__lowercase		)
								snake_case_						       =	(logits,) + outputs[2:]  # add hidden states and attention if they are here
							except HighwayException as e:
								snake_case_						       =	e.message
								snake_case_						       =	e.exit_layer
								snake_case_						       =	outputs[0]

							if not self.training:
								snake_case_						       =	entropy(__lowercase		)
								snake_case_						       =	[]
								snake_case_						       =	[]
							if labels is not None:
								if self.num_labels == 1:
									#  We are doing regression
									snake_case_						       =	MSELoss()
									snake_case_						       =	loss_fct(logits.view(-1		)			,      labels.view(-1		)		)
								else:
									snake_case_						       =	CrossEntropyLoss()
									snake_case_						       =	loss_fct(logits.view(-1			,      self.num_labels		)			,      labels.view(-1		)		)

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

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