class_docstr: Continual Prototype Evolution plugin.
  
  Each class has a prototype for nearest-neighbor classification.
  The prototypes are updated continually with an exponentially moving average,
  using class-balanced replay to keep the prototypes up-to-date.
  The embedding space is optimized using the PseudoPrototypicalProxy-loss,
  exploiting both prototypes and batch information.
  
  This plugin doesn't use task identities in training or eval
  (data incremental) and is designed for online learning (1 epoch per task).
class_docstr_tok: ["Continual","Prototype","Evolution","plugin",".","Each","class","has","a","prototype","for","nearest-neighbor","classification",".","The","prototypes","are","updated","continually","with","an","exponentially","moving","average",",","using","class-balanced","replay","to","keep","the","prototypes","up-to-date",".","The","embedding","space","is","optimized","using","the","PseudoPrototypicalProxy-loss",",","exploiting","both","prototypes","and","batch","information",".","This","plugin","does","n't","use","task","identities","in","training","or","eval","(","data","incremental",")","and","is","designed","for","online","learning","(","1","epoch","per","task",")","."]
docstr: Find closest prototype for output samples in batch x.
  :param x: Batch of network logits.
  :return: one-hot representation of the predicted class.
docstr_tok: ["Find","closest","prototype","for","output","samples","in","batch","x.",":","param","x",":","Batch","of","network","logits",".",":","return",":","one-hot","representation","of","the","predicted","class","."]
code: def _get_nearest_neigbor_distr(self, x: Tensor) -> Tensor:
      """
      Find closest prototype for output samples in batch x.
      :param x: Batch of network logits.
      :return: one-hot representation of the predicted class.
      """
      ns = x.size(0)
      nd = x.view(ns, -1).shape[-1]
  
      # Get prototypes
      seen_c = len(self.p_mem.keys())
      if (
          seen_c == 0
      ):  # no prototypes yet, output uniform distr. all classes
          return (
              torch.Tensor(ns, self.n_classes)
              .fill_(1.0 / self.n_classes)
              .to(x.device)
          )
      means = torch.ones(seen_c, nd).to(x.device) * float("inf")
      for c, c_proto in self.p_mem.items():
          means[c] = c_proto  # Class idx gets allocated its prototype
  
      # Predict nearest mean
      classpred = torch.LongTensor(ns)
      for s_idx in range(ns):  # Per sample
          dist = -torch.mm(means, x[s_idx].unsqueeze(-1))  # Dot product
          _, ii = dist.min(0)  # Min dist (no proto = inf)
          ii = ii.squeeze()
          classpred[s_idx] = ii.item()  # Allocate class idx
  
      # Convert to 1-hot
      out = torch.zeros(ns, self.n_classes).to(x.device)
      for s_idx in range(ns):
          out[s_idx, classpred[s_idx]] = 1
      return out
  
imports: from typing import Dict
  import torch
  from torch import Tensor
  from torch.nn.functional import normalize
  from torch.nn.modules import Module
  from avalanche.training.utils import get_last_fc_layer, swap_last_fc_layer
  from avalanche.training.plugins.strategy_plugin import SupervisedPlugin
  from avalanche.training.storage_policy import ClassBalancedBuffer
  from avalanche.benchmarks.utils.data_loader import ReplayDataLoader
label_desc: Use image node_id 11 for calling the CoPEPlugin obj's underlying member method code with example usage: obj._get_nearest_neigbor_distr(x) and returns: out, torch