
import math
from collections import defaultdict
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import randn_tensor
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput



def        lowerCamelCase_  (   __UpperCamelCase ,						__UpperCamelCase=0.999 ,						__UpperCamelCase="cosine" ,						):
				if alpha_transform_type == "cosine":

								def alpha_bar_fn(__UpperCamelCase     ):
												return math.cos((t + 0.008) / 1.008 * math.pi / 2     ) ** 2

				elif alpha_transform_type == "exp":

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

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

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






class 					__lowercase		(					A							,       A       ):
					__magic_name__						:							List[str]     	=	[e.name for e in KarrasDiffusionSchedulers]
					__magic_name__						:							Optional[int]     	=	2


					@register_to_config
					def __init__(	self       ,       a__ = 1_0_0_0       ,       a__ = 0.0_00_85       ,       a__ = 0.0_12       ,       a__ = "linear"       ,       a__ = None       ,       a__ = "epsilon"       ,       a__ = "linspace"       ,       a__ = 0       ,       )		->							Any:





									'''simple docstring'''
									if trained_betas is not None:
													A_       	=  torch.tensor(a__       ,       dtype=torch.floataa						)
									elif beta_schedule == "linear":
													A_       	=  torch.linspace(a__       ,       a__       ,       a__       ,       dtype=torch.floataa						)
									elif beta_schedule == "scaled_linear":
													# this schedule is very specific to the latent diffusion model.
													A_       	=  (
													    torch.linspace(beta_start**0.5       ,       beta_end**0.5       ,       a__       ,       dtype=torch.floataa						) ** 2
													)
									elif beta_schedule == "squaredcos_cap_v2":
													# Glide cosine schedule
													A_       	=  betas_for_alpha_bar(a__						)
									else:
													raise NotImplementedError(F"{beta_schedule} does is not implemented for {self.__class__}"						)

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

									#  set all values
									self.set_timesteps(a__       ,       a__       ,       a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__=None						)		->							str:





									'''simple docstring'''
									if schedule_timesteps is None:
													A_       	=  self.timesteps

									A_       	=  (schedule_timesteps == timestep).nonzero()

									# The sigma index that is taken for the **very** first `step`
									# is always the second index (or the last index if there is only 1)
									# This way we can ensure we don't accidentally skip a sigma in
									# case we start in the middle of the denoising schedule (e.g. for image-to-image)
									if len(self._index_counter						) == 0:
													A_       	=  1 if len(a__						) > 1 else 0
									else:
													A_       	=  timestep.cpu().item() if torch.is_tensor(a__						) else timestep
													A_       	=  self._index_counter[timestep_int]

									return indices[pos].item()


					@property
					def        lowerCAmelCase_    (	self						)		->							List[Any]:





									'''simple docstring'''
									# standard deviation of the initial noise distribution
									if self.config.timestep_spacing in ["linspace", "trailing"]:
													return self.sigmas.max()

									return (self.sigmas.max() ** 2 + 1) ** 0.5


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       )		->							torch.FloatTensor:





									'''simple docstring'''
									A_       	=  self.index_for_timestep(a__						)

									if self.state_in_first_order:
													A_       	=  self.sigmas[step_index]
									else:
													A_       	=  self.sigmas_interpol[step_index - 1]

									A_       	=  sample / ((sigma**2 + 1) ** 0.5)
									return sample


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__ = None       ,       a__ = None       ,       )		->							int:





									'''simple docstring'''
									A_       	=  num_inference_steps

									A_       	=  num_train_timesteps or self.config.num_train_timesteps

									# "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
									if self.config.timestep_spacing == "linspace":
													A_       	=  np.linspace(0       ,       num_train_timesteps - 1       ,       a__       ,       dtype=a__						)[::-1].copy()
									elif self.config.timestep_spacing == "leading":
													A_       	=  num_train_timesteps // self.num_inference_steps
													# creates integer timesteps by multiplying by ratio
													# casting to int to avoid issues when num_inference_step is power of 3
													A_       	=  (np.arange(0       ,       a__						) * step_ratio).round()[::-1].copy().astype(a__						)
													timesteps += self.config.steps_offset
									elif self.config.timestep_spacing == "trailing":
													A_       	=  num_train_timesteps / self.num_inference_steps
													# creates integer timesteps by multiplying by ratio
													# casting to int to avoid issues when num_inference_step is power of 3
													A_       	=  (np.arange(a__       ,       0       ,       -step_ratio						)).round().copy().astype(a__						)
													timesteps -= 1
									else:
													raise ValueError(
													    F"{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."						)

									A_       	=  np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5						)
									A_       	=  torch.from_numpy(np.log(a__						)						).to(a__						)

									A_       	=  np.interp(a__       ,       np.arange(0       ,       len(a__						)						)       ,       a__						)
									A_       	=  np.concatenate([sigmas, [0.0]]						).astype(np.floataa						)
									A_       	=  torch.from_numpy(a__						).to(device=a__						)

									# compute up and down sigmas
									A_       	=  sigmas.roll(-1						)
									A_       	=  0.0
									A_       	=  (sigmas_next**2 * (sigmas**2 - sigmas_next**2) / sigmas**2) ** 0.5
									A_       	=  (sigmas_next**2 - sigmas_up**2) ** 0.5
									A_       	=  0.0

									# compute interpolated sigmas
									A_       	=  sigmas.log().lerp(sigmas_down.log()       ,       0.5						).exp()
									A_       	=  0.0

									# set sigmas
									A_       	=  torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2						), sigmas[-1:]]						)
									A_       	=  torch.cat(
									    [sigmas_interpol[:1], sigmas_interpol[1:].repeat_interleave(2						), sigmas_interpol[-1:]]						)
									A_       	=  torch.cat([sigmas_up[:1], sigmas_up[1:].repeat_interleave(2						), sigmas_up[-1:]]						)
									A_       	=  torch.cat([sigmas_down[:1], sigmas_down[1:].repeat_interleave(2						), sigmas_down[-1:]]						)

									if str(a__						).startswith('''mps'''						):
													# mps does not support float64
													A_       	=  torch.from_numpy(a__						).to(a__       ,       dtype=torch.floataa						)
									else:
													A_       	=  torch.from_numpy(a__						).to(a__						)

									A_       	=  self.sigma_to_t(a__						).to(a__       ,       dtype=timesteps.dtype						)
									A_       	=  torch.stack((timesteps_interpol[:-2, None], timesteps[1:, None])       ,       dim=-1						).flatten()

									A_       	=  torch.cat([timesteps[:1], interleaved_timesteps]						)

									A_       	=  None

									# for exp beta schedules, such as the one for `pipeline_shap_e.py`
									# we need an index counter
									A_       	=  defaultdict(a__						)


					def        lowerCAmelCase_    (	self       ,       a__						)		->							Dict:





									'''simple docstring'''
									# get log sigma
									A_       	=  sigma.log()

									# get distribution
									A_       	=  log_sigma - self.log_sigmas[:, None]

									# get sigmas range
									A_       	=  dists.ge(0						).cumsum(dim=0						).argmax(dim=0						).clamp(max=self.log_sigmas.shape[0] - 2						)
									A_       	=  low_idx + 1

									A_       	=  self.log_sigmas[low_idx]
									A_       	=  self.log_sigmas[high_idx]

									# interpolate sigmas
									A_       	=  (low - log_sigma) / (low - high)
									A_       	=  w.clamp(0       ,       1						)

									# transform interpolation to time range
									A_       	=  (1 - w) * low_idx + w * high_idx
									A_       	=  t.view(sigma.shape						)
									return t


					@property
					def        lowerCAmelCase_    (	self						)		->							int:





									'''simple docstring'''
									return self.sample is None


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__       ,       a__ = None       ,       a__ = True       ,       )		->							Union[SchedulerOutput, Tuple]:





									'''simple docstring'''
									A_       	=  self.index_for_timestep(a__						)

									# advance index counter by 1
									A_       	=  timestep.cpu().item() if torch.is_tensor(a__						) else timestep
									self._index_counter[timestep_int] += 1

									if self.state_in_first_order:
													A_       	=  self.sigmas[step_index]
													A_       	=  self.sigmas_interpol[step_index]
													A_       	=  self.sigmas_up[step_index]
													A_       	=  self.sigmas_down[step_index - 1]
									else:
													# 2nd order / KPDM2's method
													A_       	=  self.sigmas[step_index - 1]
													A_       	=  self.sigmas_interpol[step_index - 1]
													A_       	=  self.sigmas_up[step_index - 1]
													A_       	=  self.sigmas_down[step_index - 1]

									# currently only gamma=0 is supported. This usually works best anyways.
									# We can support gamma in the future but then need to scale the timestep before
									# passing it to the model which requires a change in API
									A_       	=  0
									A_       	=  sigma * (gamma + 1)  # Note: sigma_hat == sigma for now

									A_       	=  model_output.device
									A_       	=  randn_tensor(model_output.shape       ,       dtype=model_output.dtype       ,       device=a__       ,       generator=a__						)

									# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
									if self.config.prediction_type == "epsilon":
													A_       	=  sigma_hat if self.state_in_first_order else sigma_interpol
													A_       	=  sample - sigma_input * model_output
									elif self.config.prediction_type == "v_prediction":
													A_       	=  sigma_hat if self.state_in_first_order else sigma_interpol
													A_       	=  model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
													    sample / (sigma_input**2 + 1)
													)
									elif self.config.prediction_type == "sample":
													raise NotImplementedError('''prediction_type not implemented yet: sample'''						)
									else:
													raise ValueError(
													    F"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"						)

									if self.state_in_first_order:
													# 2. Convert to an ODE derivative for 1st order
													A_       	=  (sample - pred_original_sample) / sigma_hat
													# 3. delta timestep
													A_       	=  sigma_interpol - sigma_hat

													# store for 2nd order step
													A_       	=  sample
													A_       	=  dt
													A_       	=  sample + derivative * dt
									else:
													# DPM-Solver-2
													# 2. Convert to an ODE derivative for 2nd order
													A_       	=  (sample - pred_original_sample) / sigma_interpol
													# 3. delta timestep
													A_       	=  sigma_down - sigma_hat

													A_       	=  self.sample
													A_       	=  None

													A_       	=  sample + derivative * dt
													A_       	=  prev_sample + noise * sigma_up

									if not return_dict:
													return (prev_sample,)

									return SchedulerOutput(prev_sample=a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__       ,       )		->							torch.FloatTensor:





									'''simple docstring'''
									# Make sure sigmas and timesteps have the same device and dtype as original_samples
									A_       	=  self.sigmas.to(device=original_samples.device       ,       dtype=original_samples.dtype						)
									if original_samples.device.type == "mps" and torch.is_floating_point(a__						):
													# mps does not support float64
													A_       	=  self.timesteps.to(original_samples.device       ,       dtype=torch.floataa						)
													A_       	=  timesteps.to(original_samples.device       ,       dtype=torch.floataa						)
									else:
													A_       	=  self.timesteps.to(original_samples.device						)
													A_       	=  timesteps.to(original_samples.device						)

									A_       	=  [self.index_for_timestep(a__       ,       a__						) for t in timesteps]

									A_       	=  sigmas[step_indices].flatten()
									while len(sigma.shape						) < len(original_samples.shape						):
													A_       	=  sigma.unsqueeze(-1						)

									A_       	=  original_samples + noise * sigma
									return noisy_samples


					def __len__(	self						)		->							Union[str, Any]:





									'''simple docstring'''
									return self.config.num_train_timesteps

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


SCREAMING_SNAKE_CASE   :		str					 =       logging.get_logger(__name__)

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






class 					__lowercase		(					A       ):
					__magic_name__						:							Any     	=	RetriBertConfig
					__magic_name__						:							Optional[Any]     	=	None
					__magic_name__						:							str     	=	'''retribert'''


					def        lowerCAmelCase_    (	self       ,       a__						)		->							int:





									'''simple docstring'''
									if isinstance(a__       ,       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(a__       ,       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(a__       ,       nn.LayerNorm						):
													module.bias.data.zero_()
													module.weight.data.fill_(1.0						)
SCREAMING_SNAKE_CASE   :		Optional[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 					__lowercase		(					A       ):


					def __init__(	self       ,       a__						)		->							None:





									'''simple docstring'''
									super().__init__(a__						)
									A_       	=  config.projection_dim

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

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

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


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__       ,       a__=-1       ,       )		->							int:





									'''simple docstring'''
									# reproduces BERT forward pass with checkpointing
									if checkpoint_batch_size < 0 or input_ids.shape[0] < checkpoint_batch_size:
													return sent_encoder(a__       ,       attention_mask=a__						)[1]
									else:
													# prepare implicit variables
													A_       	=  input_ids.device
													A_       	=  input_ids.size()
													A_       	=  torch.zeros(a__       ,       dtype=torch.long       ,       device=a__						)
													A_       	=  [None] * sent_encoder.config.num_hidden_layers
													A_       	=  sent_encoder.get_extended_attention_mask(
													    a__       ,       a__						)

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

													# run embedding layer on everything at once
													A_       	=  sent_encoder.embeddings(
													    input_ids=a__       ,       position_ids=a__       ,       token_type_ids=a__       ,       inputs_embeds=a__						)
													# 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(a__       ,       a__       ,       a__						)
																	pooled_output_list.append(a__						)
													return torch.cat(a__       ,       dim=0						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__=None       ,       a__=-1       ,       )		->							int:





									'''simple docstring'''
									A_       	=  self.embed_sentences_checkpointed(
									    a__       ,       a__       ,       self.bert_query       ,       a__       ,       )
									return self.project_query(a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__=None       ,       a__=-1       ,       )		->							Optional[Any]:





									'''simple docstring'''
									A_       	=  self.embed_sentences_checkpointed(
									    a__       ,       a__       ,       self.bert_query if self.bert_doc is None else self.bert_doc       ,       a__       ,       )
									return self.project_doc(a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__       ,       a__       ,       a__ = -1       ,       )		->							torch.FloatTensor:





									'''simple docstring'''
									A_       	=  input_ids_query.device
									A_       	=  self.embed_questions(a__       ,       a__       ,       a__						)
									A_       	=  self.embed_answers(a__       ,       a__       ,       a__						)
									A_       	=  torch.mm(a__       ,       a_reps.t()						)
									A_       	=  self.ce_loss(a__       ,       torch.arange(compare_scores.shape[1]						).to(a__						)						)
									A_       	=  self.ce_loss(compare_scores.t()       ,       torch.arange(compare_scores.shape[0]						).to(a__						)						)
									A_       	=  (loss_qa + loss_aq) / 2
									return loss