class_docstr: An implementation of the Bernoulli negative sampling approach proposed by [wang2014]_.
  
  The probability of corrupting the head $h$ or tail $t$ in a relation $(h,r,t) \in \mathcal{K}$
  is determined by global properties of the relation $r$:
  
  - $r$ is *one-to-many* (e.g. *motherOf*): a higher probability is assigned to replace $h$
  - $r$ is *many-to-one* (e.g. *bornIn*): a higher probability is assigned to replace $t$.
  
  More precisely, for each relation $r \in \mathcal{R}$, the average number of tails per head
  (``tph``) and heads per tail (``hpt``) are first computed.
  
  Then, the head corruption probability $p_r$ is defined as $p_r = \frac{tph}{tph + hpt}$.
  The tail corruption probability is defined as $1 - p_r = \frac{hpt}{tph + hpt}$.
  
  For each triple $(h,r,t) \in \mathcal{K}$, the head is corrupted with probability $p_r$ and the tail is
  corrupted with probability $1 - p_r$.
  
  If ``filtered`` is set to ``True``, all proposed corrupted triples that also exist as
  actual positive triples $(h,r,t) \in \mathcal{K}$ will be removed.
code: def corrupt_batch(
      self, positive_batch: torch.LongTensor
  ) -> torch.LongTensor:  # noqa: D102
      batch_shape = positive_batch.shape[:-1]
  
      # Decide whether to corrupt head or tail
      head_corruption_probability = self.corrupt_head_probability[
          positive_batch[..., 1]
      ].unsqueeze(dim=-1)
      head_mask = torch.rand(
          *batch_shape,
          self.num_negs_per_pos,
          device=positive_batch.device,
      ) < head_corruption_probability.to(device=positive_batch.device)
  
      # clone positive batch for corruption (.repeat_interleave creates a copy)
      negative_batch = positive_batch.view(-1, 3).repeat_interleave(
          self.num_negs_per_pos, dim=0
      )
      # flatten mask
      head_mask = head_mask.view(-1)
  
      for index, mask in (
          (COLUMN_HEAD, head_mask),
          # Tails are corrupted if heads are not corrupted
          (COLUMN_TAIL, ~head_mask),
      ):
          random_replacement_(
              batch=negative_batch,
              index=index,
              selection=mask,
              size=mask.sum(),
              max_index=self.num_entities,
          )
  
      return negative_batch.view(*batch_shape, self.num_negs_per_pos, 3)
  
file_path: pykeen/src/pykeen/sampling/bernoulli_negative_sampler.py
imports: import torch
  from .basic_negative_sampler import random_replacement_
  from .negative_sampler import NegativeSampler
  from ..typing import COLUMN_HEAD, COLUMN_TAIL, MappedTriples
label_desc: Use image node_id 2 for calling the BernoulliNegativeSampler obj's underlying member method code with example usage: obj.corrupt_batch(positive_batch) and returns: negative_batch