




'''simple docstring'''







import tensorflow as tf

from ...tf_utils import shape_list







class 					_a (tf.keras.layers.Layer	):





     '''simple docstring'''



     def __init__(				self							,__a							,__a							,__a							,__a							,__a=1							,__a=False							,**__a							)      ->				Tuple:
          super().__init__(**__a							)

          snake_case			:      List[Any]        =			vocab_size
          snake_case			:      Any        =			d_embed
          snake_case			:      int        =			d_proj

          snake_case			:      Dict        =			cutoffs + [vocab_size]
          snake_case			:      Any        =			[0] + self.cutoffs
          snake_case			:      Dict        =			div_val

          snake_case			:      Tuple        =			self.cutoffs[0]
          snake_case			:      int        =			len(self.cutoffs							) - 1
          snake_case			:      Optional[Any]        =			self.shortlist_size + self.n_clusters
          snake_case			:      Any        =			keep_order

          snake_case			:      Union[str, Any]        =			[]
          snake_case			:      Union[str, Any]        =			[]







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

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

                    snake_case			:      List[str]        =			self.add_weight(
                        shape=(d_emb_i, self.d_proj)							,initializer="""zeros"""							,trainable=__a							,name=F'''out_projs_._{i}'''							)
                    self.out_projs.append(__a							)
                    snake_case			:      Tuple        =			self.add_weight(
                        shape=(r_idx - l_idx, d_emb_i)							,initializer="""zeros"""							,trainable=__a							,name=F'''out_layers_._{i}_._weight'''							,)
                    snake_case			:      int        =			self.add_weight(
                        shape=(r_idx - l_idx,)							,initializer="""zeros"""							,trainable=__a							,name=F'''out_layers_._{i}_._bias'''							,)
                    self.out_layers.append((weight, bias)							)
          super().build(__a							)







     @staticmethod
     def 			snake_case_      (				__a							,__a							,__a							,__a=None							)      ->				Union[str, Any]:
          snake_case			:      List[Any]        =			x
          if proj is not None:
               snake_case			:      Optional[Any]        =			tf.einsum("""ibd,ed->ibe"""							,__a							,__a							)
          return tf.einsum("""ibd,nd->ibn"""							,__a							,__a							) + b







     @staticmethod
     def 			snake_case_      (				__a							,__a							)      ->				Dict:
          snake_case			:      List[Any]        =			shape_list(__a							)
          snake_case			:      List[str]        =			tf.range(lp_size[0]							,dtype=target.dtype							)
          snake_case			:      int        =			tf.stack([r, target]							,1							)
          return tf.gather_nd(__a							,__a							)







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

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

                    if i == 0:
                         snake_case			:      Optional[int]        =			tf.concat([cur_W, self.cluster_weight]							,0							)
                         snake_case			:      Union[str, Any]        =			tf.concat([cur_b, self.cluster_bias]							,0							)

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

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

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

          return out









'''simple docstring'''







import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
lowercase      :   List[str]      =       """."""


if __name__ == "__main__":
      lowercase      :   List[str]      =       os.path.join(REPO_PATH, """utils/documentation_tests.txt""")
      lowercase      :   Union[str, Any]      =       []
      lowercase      :   List[str]      =       []
      with open(doctest_file_path) as fp:
            for line in fp:
                  lowercase      :   List[str]      =       line.strip()
                  lowercase      :   Optional[Any]      =       os.path.join(REPO_PATH, line)
                  if not (os.path.isfile(path) or os.path.isdir(path)):
                        non_existent_paths.append(line)
                  all_paths.append(path)
      if len(non_existent_paths) > 0:
            lowercase      :   int      =       """\n""".join(non_existent_paths)
            raise ValueError(F"""`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}""")
      if all_paths != sorted(all_paths):
            raise ValueError("""Files in `utils/documentation_tests.txt` are not in alphabetical order.""")



