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


lowercase_  							=			logging.get_logger(__name__)

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







class      SCREAMING_SNAKE_CASE__					(  __UpperCamelCase    ):
       A							:	Any          =	RetriBertConfig
       A							:	str          =	None
       A							:	Optional[Any]          =	"retribert"

       def 			snake_case__    (							self					:     Tuple			,			_lowerCAmelCase					:     Dict			):
           if isinstance(_lowerCAmelCase			,			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(_lowerCAmelCase			,			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(_lowerCAmelCase			,			nn.LayerNorm			):
               module.bias.data.zero_()
               module.weight.data.fill_(1.0			)







lowercase_  							=			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."   ,							__UpperCamelCase   ,							)
class      SCREAMING_SNAKE_CASE__					(  __UpperCamelCase    ):

       def __init__(							self					:     Dict			,			_lowerCAmelCase					:     RetriBertConfig			):
           super().__init__(_lowerCAmelCase			)
           __snake_case       :				Any 			=    config.projection_dim

           __snake_case       :				int 			=    BertModel(_lowerCAmelCase			)
           __snake_case       :				Dict 			=    None if config.share_encoders else BertModel(_lowerCAmelCase			)
           __snake_case       :				Tuple 			=    nn.Dropout(config.hidden_dropout_prob			)
           __snake_case       :				Optional[int] 			=    nn.Linear(config.hidden_size			,			config.projection_dim			,			bias=_lowerCAmelCase			)
           __snake_case       :				Union[str, Any] 			=    nn.Linear(config.hidden_size			,			config.projection_dim			,			bias=_lowerCAmelCase			)

           __snake_case       :				Optional[Any] 			=    nn.CrossEntropyLoss(reduction="""mean"""			)

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

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

               # define function for checkpointing
               def partial_encode(*_lowerCAmelCase					:     Union[str, Any]			):
                   __snake_case       :				Optional[int] 			=    sent_encoder.encoder(
                       inputs[0]			,			attention_mask=inputs[1]			,			head_mask=_lowerCAmelCase			,			)
                   __snake_case       :				List[str] 			=    encoder_outputs[0]
                   __snake_case       :				Tuple 			=    sent_encoder.pooler(_lowerCAmelCase			)
                   return pooled_output

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

       def 			snake_case__    (							self					:     List[str]			,			_lowerCAmelCase					:     Tuple			,			_lowerCAmelCase					:     List[Any]=None			,			_lowerCAmelCase					:     str=-1			,			):
           __snake_case       :				int 			=    self.embed_sentences_checkpointed(
               _lowerCAmelCase			,			_lowerCAmelCase			,			self.bert_query			,			_lowerCAmelCase			,			)
           return self.project_query(_lowerCAmelCase			)

       def 			snake_case__    (							self					:     Dict			,			_lowerCAmelCase					:     Union[str, Any]			,			_lowerCAmelCase					:     Optional[int]=None			,			_lowerCAmelCase					:     Optional[int]=-1			,			):
           __snake_case       :				Optional[Any] 			=    self.embed_sentences_checkpointed(
               _lowerCAmelCase			,			_lowerCAmelCase			,			self.bert_query if self.bert_doc is None else self.bert_doc			,			_lowerCAmelCase			,			)
           return self.project_doc(_lowerCAmelCase			)





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




import tempfile
import unittest

from diffusers import (
    DDIMScheduler,
    DDPMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    PNDMScheduler,
    logging,
)
from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.utils.testing_utils import CaptureLogger







class      SCREAMING_SNAKE_CASE__					(  __UpperCamelCase    ):
       A							:	str          =	"config.json"

       @register_to_config
       def __init__(							self					:     Any			,			_lowerCAmelCase					:     Tuple=2			,			_lowerCAmelCase					:     List[Any]=5			,			_lowerCAmelCase					:     int=(2, 5)			,			_lowerCAmelCase					:     Dict="for diffusion"			,			_lowerCAmelCase					:     List[str]=[1, 3]			,			):
           pass







class      SCREAMING_SNAKE_CASE__					(  __UpperCamelCase    ):
       A							:	Optional[int]          =	"config.json"

       @register_to_config
       def __init__(							self					:     int			,			_lowerCAmelCase					:     Dict=2			,			_lowerCAmelCase					:     Union[str, Any]=5			,			_lowerCAmelCase					:     Any=(2, 5)			,			_lowerCAmelCase					:     Optional[int]="for diffusion"			,			_lowerCAmelCase					:     int=[1, 3]			,			):
           pass







class      SCREAMING_SNAKE_CASE__					(  __UpperCamelCase    ):
       A							:	str          =	"config.json"

       @register_to_config
       def __init__(							self					:     Any			,			_lowerCAmelCase					:     Optional[int]=2			,			_lowerCAmelCase					:     Optional[int]=5			,			_lowerCAmelCase					:     int=(2, 5)			,			_lowerCAmelCase					:     Tuple="for diffusion"			,			_lowerCAmelCase					:     Any=[1, 3]			,			_lowerCAmelCase					:     str=[1, 3]			,			):
           pass







class      SCREAMING_SNAKE_CASE__					(  __UpperCamelCase    ):
       A							:	Union[str, Any]          =	"config.json"

       @register_to_config
       def __init__(							self					:     Any			,			_lowerCAmelCase					:     Dict=2			,			_lowerCAmelCase					:     Optional[int]=5			,			_lowerCAmelCase					:     Optional[Any]=(2, 5)			,			_lowerCAmelCase					:     int="for diffusion"			,			_lowerCAmelCase					:     Dict=[1, 5]			,			_lowerCAmelCase					:     Optional[Any]=[5, 4]			,			):
           pass







class      SCREAMING_SNAKE_CASE__					(  unittest.TestCase    ):

       def 			snake_case__    (							self					:     List[str]			):
           with self.assertRaises(_lowerCAmelCase			):
               ConfigMixin.load_config("""dummy_path"""			)

       def 			snake_case__    (							self					:     List[Any]			):
           __snake_case       :				List[Any] 			=    SampleObject()
           __snake_case       :				Optional[Any] 			=    obj.config
           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == (2, 5)
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           # init ignore private arguments
           __snake_case       :				Optional[Any] 			=    SampleObject(_name_or_path="""lalala"""			)
           __snake_case       :				List[Any] 			=    obj.config
           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == (2, 5)
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           # can override default
           __snake_case       :				int 			=    SampleObject(c=6			)
           __snake_case       :				Union[str, Any] 			=    obj.config
           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == 6
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           # can use positional arguments.
           __snake_case       :				List[str] 			=    SampleObject(1			,			c=6			)
           __snake_case       :				str 			=    obj.config
           assert config["a"] == 1
           assert config["b"] == 5
           assert config["c"] == 6
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

       def 			snake_case__    (							self					:     Union[str, Any]			):
           __snake_case       :				Union[str, Any] 			=    SampleObject()
           __snake_case       :				Optional[Any] 			=    obj.config

           assert config["a"] == 2
           assert config["b"] == 5
           assert config["c"] == (2, 5)
           assert config["d"] == "for diffusion"
           assert config["e"] == [1, 3]

           with tempfile.TemporaryDirectory() as tmpdirname:
               obj.save_config(_lowerCAmelCase			)
               __snake_case       :				List[Any] 			=    SampleObject.from_config(SampleObject.load_config(_lowerCAmelCase			)			)
               __snake_case       :				Union[str, Any] 			=    new_obj.config

           # unfreeze configs
           __snake_case       :				Optional[Any] 			=    dict(_lowerCAmelCase			)
           __snake_case       :				Optional[Any] 			=    dict(_lowerCAmelCase			)

           assert config.pop("""c"""			) == (2, 5)  # instantiated as tuple
           assert new_config.pop("""c"""			) == [2, 5]  # saved & loaded as list because of json
           config.pop("""_use_default_values"""			)
           assert config == new_config

       def 			snake_case__    (							self					:     Tuple			):
           __snake_case       :				Optional[Any] 			=    logging.get_logger("""diffusers.configuration_utils"""			)
           # 30 for warning
           logger.setLevel(30			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger:
               __snake_case       :				List[Any] 			=    DDIMScheduler.from_pretrained(
                   """hf-internal-testing/tiny-stable-diffusion-torch"""			,			subfolder="""scheduler"""			)

           assert ddim.__class__ == DDIMScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 			snake_case__    (							self					:     Optional[int]			):
           __snake_case       :				int 			=    logging.get_logger("""diffusers.configuration_utils"""			)
           # 30 for warning
           logger.setLevel(30			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger:
               __snake_case       :				Optional[int] 			=    EulerDiscreteScheduler.from_pretrained(
                   """hf-internal-testing/tiny-stable-diffusion-torch"""			,			subfolder="""scheduler"""			)

           assert euler.__class__ == EulerDiscreteScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 			snake_case__    (							self					:     Tuple			):
           __snake_case       :				int 			=    logging.get_logger("""diffusers.configuration_utils"""			)
           # 30 for warning
           logger.setLevel(30			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger:
               __snake_case       :				Any 			=    EulerAncestralDiscreteScheduler.from_pretrained(
                   """hf-internal-testing/tiny-stable-diffusion-torch"""			,			subfolder="""scheduler"""			)

           assert euler.__class__ == EulerAncestralDiscreteScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 			snake_case__    (							self					:     List[Any]			):
           __snake_case       :				str 			=    logging.get_logger("""diffusers.configuration_utils"""			)
           # 30 for warning
           logger.setLevel(30			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger:
               __snake_case       :				Optional[int] 			=    PNDMScheduler.from_pretrained(
                   """hf-internal-testing/tiny-stable-diffusion-torch"""			,			subfolder="""scheduler"""			)

           assert pndm.__class__ == PNDMScheduler
           # no warning should be thrown
           assert cap_logger.out == ""

       def 			snake_case__    (							self					:     Tuple			):
           __snake_case       :				int 			=    logging.get_logger("""diffusers.configuration_utils"""			)
           # 30 for warning
           logger.setLevel(30			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger:
               __snake_case       :				List[str] 			=    DDPMScheduler.from_pretrained(
                   """hf-internal-testing/tiny-stable-diffusion-torch"""			,			subfolder="""scheduler"""			,			prediction_type="""sample"""			,			beta_end=8			,			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger_a:
               __snake_case       :				Optional[Any] 			=    DDPMScheduler.from_pretrained("""google/ddpm-celebahq-256"""			,			beta_start=88			)

           assert ddpm.__class__ == DDPMScheduler
           assert ddpm.config.prediction_type == "sample"
           assert ddpm.config.beta_end == 8
           assert ddpm_a.config.beta_start == 88

           # no warning should be thrown
           assert cap_logger.out == ""
           assert cap_logger_a.out == ""

       def 			snake_case__    (							self					:     List[Any]			):
           __snake_case       :				List[Any] 			=    logging.get_logger("""diffusers.configuration_utils"""			)
           # 30 for warning
           logger.setLevel(30			)

           with CaptureLogger(_lowerCAmelCase			) as cap_logger:
               __snake_case       :				str 			=    DPMSolverMultistepScheduler.from_pretrained(
                   """hf-internal-testing/tiny-stable-diffusion-torch"""			,			subfolder="""scheduler"""			)

           assert dpm.__class__ == DPMSolverMultistepScheduler
           # no warning should be thrown
           assert cap_logger.out == ""





       def 			snake_case__    (							self					:     List[Any]			):
           # let's first save a config that should be in the form
           #    a=2,
           #    b=5,
           #    c=(2, 5),
           #    d="for diffusion",
           #    e=[1, 3],

           __snake_case       :				Optional[Any] 			=    SampleObject()

           __snake_case       :				Union[str, Any] 			=    {k: v for k, v in config.config.items() if not k.startswith("""_"""			)}

           # make sure that default config has all keys in `_use_default_values`
           assert set(config_dict.keys()			) == set(config.config._use_default_values			)

           with tempfile.TemporaryDirectory() as tmpdirname:
               config.save_config(_lowerCAmelCase			)

               # now loading it with SampleObject2 should put f into `_use_default_values`
               __snake_case       :				Union[str, Any] 			=    SampleObjecta.from_config(_lowerCAmelCase			)

               assert "f" in config._use_default_values
               assert config.f == [1, 3]

           # now loading the config, should **NOT** use [1, 3] for `f`, but the default [1, 4] value
           # **BECAUSE** it is part of `config._use_default_values`
           __snake_case       :				Optional[int] 			=    SampleObjecta.from_config(config.config			)
           assert new_config.f == [5, 4]

           config.config._use_default_values.pop()
           __snake_case       :				Optional[Any] 			=    SampleObjecta.from_config(config.config			)
           assert new_config_a.f == [1, 3]

           # Nevertheless "e" should still be correctly loaded to [1, 3] from SampleObject2 instead of defaulting to [1, 5]
           assert new_config_a.e == [1, 3]



