
import tensorflow as tf

from ...tf_utils import shape_list



class 		UpperCamelCase__  (	tf.keras.layers.Layer   ):
   def __init__(self			: List[str] ,    snake_case_			: Optional[Any] ,    snake_case_			: int ,    snake_case_			: Optional[Any] ,    snake_case_			: Optional[Any] ,    snake_case_			: str=1 ,    snake_case_			: str=False ,    **snake_case_			: Dict       ):
    super().__init__(**snake_case_       )

    __a			:  Tuple		    =  vocab_size
    __a			:  Optional[Any]		    =  d_embed
    __a			:  Union[str, Any]		    =  d_proj

    __a			:  List[Any]		    =  cutoffs + [vocab_size]
    __a			:  str		    =  [0] + self.cutoffs
    __a			:  List[str]		    =  div_val

    __a			:  str		    =  self.cutoffs[0]
    __a			:  Dict		    =  len(self.cutoffs       ) - 1
    __a			:  Dict		    =  self.shortlist_size + self.n_clusters
    __a			:  Optional[Any]		    =  keep_order

    __a			:  List[Any]		    =  []
    __a			:  Any		    =  []
   def 				lowerCAmelCase      (self			: Any ,    snake_case_			: Optional[Any]       ):
    if self.n_clusters > 0:
     __a			:  Optional[Any]		    =  self.add_weight(
         shape=(self.n_clusters, self.d_embed) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name='''cluster_weight'''       )
     __a			:  Tuple		    =  self.add_weight(
         shape=(self.n_clusters,) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name='''cluster_bias'''       )

    if self.div_val == 1:
     for i in range(len(self.cutoffs       )       ):
      if self.d_proj != self.d_embed:
       __a			:  List[str]		    =  self.add_weight(
           shape=(self.d_embed, self.d_proj) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name=f"out_projs_._{i}" ,    )
       self.out_projs.append(snake_case_       )
      else:
       self.out_projs.append(snake_case_       )
      __a			:  Tuple		    =  self.add_weight(
          shape=(self.vocab_size, self.d_embed) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name=f"out_layers_._{i}_._weight" ,    )
      __a			:  Optional[Any]		    =  self.add_weight(
          shape=(self.vocab_size,) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name=f"out_layers_._{i}_._bias" ,    )
      self.out_layers.append((weight, bias)       )
    else:
     for i in range(len(self.cutoffs       )       ):
      __a    ,	__a			:  Tuple		    =  self.cutoff_ends[i], self.cutoff_ends[i + 1]
      __a			:  Union[str, Any]		    =  self.d_embed // (self.div_val**i)

      __a			:  Optional[Any]		    =  self.add_weight(
          shape=(d_emb_i, self.d_proj) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name=f"out_projs_._{i}"       )
      self.out_projs.append(snake_case_       )
      __a			:  List[str]		    =  self.add_weight(
          shape=(r_idx - l_idx, d_emb_i) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name=f"out_layers_._{i}_._weight" ,    )
      __a			:  Tuple		    =  self.add_weight(
          shape=(r_idx - l_idx,) ,    initializer='''zeros''' ,    trainable=snake_case_ ,    name=f"out_layers_._{i}_._bias" ,    )
      self.out_layers.append((weight, bias)       )
    super().build(snake_case_       )
   @staticmethod
   def 				lowerCAmelCase      (snake_case_			: Any ,    snake_case_			: Dict ,    snake_case_			: Union[str, Any] ,    snake_case_			: int=None       ):
    __a			:  Dict		    =  x
    if proj is not None:
     __a			:  Dict		    =  tf.einsum('''ibd,ed->ibe''' ,    snake_case_ ,    snake_case_       )
    return tf.einsum('''ibd,nd->ibn''' ,    snake_case_ ,    snake_case_       ) + b
   @staticmethod
   def 				lowerCAmelCase      (snake_case_			: Any ,    snake_case_			: Any       ):
    __a			:  Union[str, Any]		    =  shape_list(snake_case_       )
    __a			:  List[str]		    =  tf.range(lp_size[0] ,    dtype=target.dtype       )
    __a			:  Optional[Any]		    =  tf.stack([r, target] ,    1       )
    return tf.gather_nd(snake_case_ ,    snake_case_       )







   def 				lowerCAmelCase      (self			: Optional[Any] ,    snake_case_			: List[str] ,    snake_case_			: List[Any] ,    snake_case_			: Optional[Any]=True ,    snake_case_			: Tuple=False       ):
    __a			:  int		    =  0
    if self.n_clusters == 0:
     __a			:  Dict		    =  self._logit(snake_case_ ,    self.out_layers[0][0] ,    self.out_layers[0][1] ,    self.out_projs[0]       )
     if target is not None:
      __a			:  str		    =  tf.nn.sparse_softmax_cross_entropy_with_logits(labels=snake_case_ ,    logits=snake_case_       )
     __a			:  int		    =  tf.nn.log_softmax(snake_case_ ,    axis=-1       )
    else:
     __a			:  str		    =  shape_list(snake_case_       )
     __a			:  Union[str, Any]		    =  []
     __a			:  List[Any]		    =  tf.zeros(hidden_sizes[:2]       )
     for i in range(len(self.cutoffs       )       ):
      __a    ,	__a			:  Tuple		    =  self.cutoff_ends[i], self.cutoff_ends[i + 1]
      if target is not None:
       __a			:  str		    =  (target >= l_idx) & (target < r_idx)
       __a			:  str		    =  tf.where(snake_case_       )
       __a			:  Dict		    =  tf.boolean_mask(snake_case_ ,    snake_case_       ) - l_idx

      if self.div_val == 1:
       __a			:  Union[str, Any]		    =  self.out_layers[0][0][l_idx:r_idx]
       __a			:  List[Any]		    =  self.out_layers[0][1][l_idx:r_idx]
      else:
       __a			:  str		    =  self.out_layers[i][0]
       __a			:  Optional[int]		    =  self.out_layers[i][1]

      if i == 0:
       __a			:  Dict		    =  tf.concat([cur_W, self.cluster_weight] ,    0       )
       __a			:  int		    =  tf.concat([cur_b, self.cluster_bias] ,    0       )

       __a			:  Union[str, Any]		    =  self._logit(snake_case_ ,    snake_case_ ,    snake_case_ ,    self.out_projs[0]       )
       __a			:  Any		    =  tf.nn.log_softmax(snake_case_       )
       out.append(head_logprob[..., : self.cutoffs[0]]       )
       if target is not None:
        __a			:  Dict		    =  tf.boolean_mask(snake_case_ ,    snake_case_       )
        __a			:  List[str]		    =  self._gather_logprob(snake_case_ ,    snake_case_       )
      else:
       __a			:  Union[str, Any]		    =  self._logit(snake_case_ ,    snake_case_ ,    snake_case_ ,    self.out_projs[i]       )
       __a			:  List[Any]		    =  tf.nn.log_softmax(snake_case_       )
       __a			:  str		    =  self.cutoffs[0] + i - 1  # No probability for the head cluster
       __a			:  List[Any]		    =  head_logprob[..., cluster_prob_idx, None] + tail_logprob
       out.append(snake_case_       )
       if target is not None:
        __a			:  List[str]		    =  tf.boolean_mask(snake_case_ ,    snake_case_       )
        __a			:  Union[str, Any]		    =  tf.boolean_mask(snake_case_ ,    snake_case_       )
        __a			:  Optional[Any]		    =  self._gather_logprob(snake_case_ ,    snake_case_       )
        cur_logprob += cur_head_logprob[:, self.cutoff_ends[1] + i - 1]
      if target is not None:
       loss += tf.scatter_nd(snake_case_ ,    -cur_logprob ,    shape_list(snake_case_       )       )
     __a			:  Union[str, Any]		    =  tf.concat(snake_case_ ,    axis=-1       )

    if target is not None:
     if return_mean:
      __a			:  Optional[Any]		    =  tf.reduce_mean(snake_case_       )
     # Add the training-time loss value to the layer using `self.add_loss()`.
     self.add_loss(snake_case_       )

     # Log the loss as a metric (we could log arbitrary metrics,
     # including different metrics for training and inference.
     self.add_metric(snake_case_ ,    name=self.name ,    aggregation='''mean''' if return_mean else ''''''       )

    return out


import os
import sys
import unittest


lowercase__          =os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, 'utils'))

import get_test_info  # noqa: E402
from get_test_info import (  # noqa: E402
    get_model_to_test_mapping,
    get_model_to_tester_mapping,
    get_test_to_tester_mapping,
)


lowercase__          =os.path.join('tests', 'models', 'bert', 'test_modeling_bert.py')
lowercase__          =os.path.join('tests', 'models', 'blip', 'test_modeling_blip.py')



class 		UpperCamelCase__  (	unittest.TestCase   ):
   def 				lowerCAmelCase      (self			: List[Any]       ):
    __a			:  str		    =  get_test_to_tester_mapping(snake_case_       )
    __a			:  Tuple		    =  get_test_to_tester_mapping(snake_case_       )

    __a			:  Union[str, Any]		    =  {'''BertModelTest''': '''BertModelTester'''}

    __a			:  Tuple		    =  {
        '''BlipModelTest''': '''BlipModelTester''',
        '''BlipTextImageModelTest''': '''BlipTextImageModelsModelTester''',
        '''BlipTextModelTest''': '''BlipTextModelTester''',
        '''BlipTextRetrievalModelTest''': '''BlipTextRetrievalModelTester''',
        '''BlipVQAModelTest''': '''BlipVQAModelTester''',
        '''BlipVisionModelTest''': '''BlipVisionModelTester''',
    }

    self.assertEqual(get_test_info.to_json(snake_case_       ) ,    snake_case_       )
    self.assertEqual(get_test_info.to_json(snake_case_       ) ,    snake_case_       )
   def 				lowerCAmelCase      (self			: str       ):
    __a			:  Optional[int]		    =  get_model_to_test_mapping(snake_case_       )
    __a			:  Any		    =  get_model_to_test_mapping(snake_case_       )

    __a			:  List[Any]		    =  {
        '''BertForMaskedLM''': ['''BertModelTest'''],
        '''BertForMultipleChoice''': ['''BertModelTest'''],
        '''BertForNextSentencePrediction''': ['''BertModelTest'''],
        '''BertForPreTraining''': ['''BertModelTest'''],
        '''BertForQuestionAnswering''': ['''BertModelTest'''],
        '''BertForSequenceClassification''': ['''BertModelTest'''],
        '''BertForTokenClassification''': ['''BertModelTest'''],
        '''BertLMHeadModel''': ['''BertModelTest'''],
        '''BertModel''': ['''BertModelTest'''],
    }

    __a			:  Dict		    =  {
        '''BlipForConditionalGeneration''': ['''BlipTextImageModelTest'''],
        '''BlipForImageTextRetrieval''': ['''BlipTextRetrievalModelTest'''],
        '''BlipForQuestionAnswering''': ['''BlipVQAModelTest'''],
        '''BlipModel''': ['''BlipModelTest'''],
        '''BlipTextModel''': ['''BlipTextModelTest'''],
        '''BlipVisionModel''': ['''BlipVisionModelTest'''],
    }

    self.assertEqual(get_test_info.to_json(snake_case_       ) ,    snake_case_       )
    self.assertEqual(get_test_info.to_json(snake_case_       ) ,    snake_case_       )







   def 				lowerCAmelCase      (self			: int       ):
    __a			:  Any		    =  get_model_to_tester_mapping(snake_case_       )
    __a			:  List[str]		    =  get_model_to_tester_mapping(snake_case_       )

    __a			:  Any		    =  {
        '''BertForMaskedLM''': ['''BertModelTester'''],
        '''BertForMultipleChoice''': ['''BertModelTester'''],
        '''BertForNextSentencePrediction''': ['''BertModelTester'''],
        '''BertForPreTraining''': ['''BertModelTester'''],
        '''BertForQuestionAnswering''': ['''BertModelTester'''],
        '''BertForSequenceClassification''': ['''BertModelTester'''],
        '''BertForTokenClassification''': ['''BertModelTester'''],
        '''BertLMHeadModel''': ['''BertModelTester'''],
        '''BertModel''': ['''BertModelTester'''],
    }

    __a			:  int		    =  {
        '''BlipForConditionalGeneration''': ['''BlipTextImageModelsModelTester'''],
        '''BlipForImageTextRetrieval''': ['''BlipTextRetrievalModelTester'''],
        '''BlipForQuestionAnswering''': ['''BlipVQAModelTester'''],
        '''BlipModel''': ['''BlipModelTester'''],
        '''BlipTextModel''': ['''BlipTextModelTester'''],
        '''BlipVisionModel''': ['''BlipVisionModelTester'''],
    }

    self.assertEqual(get_test_info.to_json(snake_case_       ) ,    snake_case_       )
    self.assertEqual(get_test_info.to_json(snake_case_       ) ,    snake_case_       )
