class_docstr: Test class for all metrics.
  
  Class used for efficiently run a lot of parametrized tests in DDP mode. Makes sure that DDP is only setup once and
  that pool of processes are used for all tests. All tests should subclass from this and implement a new method called
  ``test_metric_name`` where the method ``self.run_metric_test`` is called inside.
docstr: Core method that should be used for testing class. Call this inside testing methods.
  
  Args:
      ddp: bool, if running in ddp mode or not
      preds: torch tensor with predictions
      target: torch tensor with targets
      metric_class: metric class that should be tested
      reference_metric: callable function that is used for comparison
      dist_sync_on_step: bool, if true will synchronize metric state across processes at each ``forward()``
      metric_args: dict with additional arguments used for class initialization
      check_dist_sync_on_step: bool, if true will check if the metric is also correctly
          calculated per batch and per device (and not just at the end)
      check_batch: bool, if true will check if the metric is also correctly
          calculated across devices for each batch (and not just at the end)
      fragment_kwargs: whether tensors in kwargs should be divided as `preds` and `target` among processes
      check_scriptable: bool indicating if metric should also be tested if it can be scripted
      check_state_dict: bool indicating if metric should be tested that its state_dict by default is empty
      atol: absolute tolerance used for comparison of results, if None will use self.atol
      kwargs_update: Additional keyword arguments that will be passed with preds and
          target when running update on the metric.
docstr_tok: ["Core","method","that","should","be","used","for","testing","class",".","Call","this","inside","testing","methods",".","Args",":","ddp",":","bool",",","if","running","in","ddp","mode","or","not","preds",":","torch","tensor","with","predictions","target",":","torch","tensor","with","targets","metric_class",":","metric","class","that","should","be","tested","reference_metric",":","callable","function","that","is","used","for","comparison","dist_sync_on_step",":","bool",",","if","true","will","synchronize","metric","state","across","processes","at","each","``","forward","(",")","``","metric_args",":","dict","with","additional","arguments","used","for","class","initialization","check_dist_sync_on_step",":","bool",",","if","true","will","check","if","the","metric","is","also","correctly","calculated","per","batch","and","per","device","(","and","not","just","at","the","end",")","check_batch",":","bool",",","if","true","will","check","if","the","metric","is","also","correctly","calculated","across","devices","for","each","batch","(","and","not","just","at","the","end",")","fragment_kwargs",":","whether","tensors","in","kwargs","should","be","divided","as","`","preds","`","and","`","target","`","among","processes","check_scriptable",":","bool","indicating","if","metric","should","also","be","tested","if","it","can","be","scripted","check_state_dict",":","bool","indicating","if","metric","should","be","tested","that","its","state_dict","by","default","is","empty","atol",":","absolute","tolerance","used","for","comparison","of","results",",","if","None","will","use","self.atol","kwargs_update",":","Additional","keyword","arguments","that","will","be","passed","with","preds","and","target","when","running","update","on","the","metric","."]
code: def run_class_metric_test(
      self,
      ddp: bool,
      preds: Union[Tensor, List[Dict]],
      target: Union[Tensor, List[Dict]],
      metric_class: Metric,
      reference_metric: Callable,
      dist_sync_on_step: bool = False,
      metric_args: Optional[dict] = None,
      check_dist_sync_on_step: bool = True,
      check_batch: bool = True,
      fragment_kwargs: bool = False,
      check_scriptable: bool = True,
      check_state_dict: bool = True,
      atol: Optional[float] = None,
      **kwargs_update: Any,
  ):
      """Core method that should be used for testing class. Call this inside testing methods.
  
      Args:
          ddp: bool, if running in ddp mode or not
          preds: torch tensor with predictions
          target: torch tensor with targets
          metric_class: metric class that should be tested
          reference_metric: callable function that is used for comparison
          dist_sync_on_step: bool, if true will synchronize metric state across processes at each ``forward()``
          metric_args: dict with additional arguments used for class initialization
          check_dist_sync_on_step: bool, if true will check if the metric is also correctly
              calculated per batch and per device (and not just at the end)
          check_batch: bool, if true will check if the metric is also correctly
              calculated across devices for each batch (and not just at the end)
          fragment_kwargs: whether tensors in kwargs should be divided as `preds` and `target` among processes
          check_scriptable: bool indicating if metric should also be tested if it can be scripted
          check_state_dict: bool indicating if metric should be tested that its state_dict by default is empty
          atol: absolute tolerance used for comparison of results, if None will use self.atol
          kwargs_update: Additional keyword arguments that will be passed with preds and
              target when running update on the metric.
  
      """
      atol = atol or self.atol
      metric_args = metric_args or {}
      if ddp:
          if sys.platform == "win32":
              pytest.skip("DDP not supported on windows")
  
          pytest.pool.starmap(
              partial(
                  _class_test,
                  preds=preds,
                  target=target,
                  metric_class=metric_class,
                  reference_metric=reference_metric,
                  dist_sync_on_step=dist_sync_on_step,
                  metric_args=metric_args,
                  check_dist_sync_on_step=check_dist_sync_on_step,
                  check_batch=check_batch,
                  atol=atol,
                  device="cuda" if torch.cuda.is_available() else "cpu",
                  fragment_kwargs=fragment_kwargs,
                  check_scriptable=check_scriptable,
                  check_state_dict=check_state_dict,
                  **kwargs_update,
              ),
              [(rank, NUM_PROCESSES) for rank in range(NUM_PROCESSES)],
          )
      else:
          device = "cuda" if torch.cuda.is_available() else "cpu"
  
          _class_test(
              rank=0,
              world_size=1,
              preds=preds,
              target=target,
              metric_class=metric_class,
              reference_metric=reference_metric,
              dist_sync_on_step=dist_sync_on_step,
              metric_args=metric_args,
              check_dist_sync_on_step=check_dist_sync_on_step,
              check_batch=check_batch,
              atol=atol,
              device=device,
              fragment_kwargs=fragment_kwargs,
              check_scriptable=check_scriptable,
              check_state_dict=check_state_dict,
              **kwargs_update,
          )
  
imports: import pickle
  import sys
  from copy import deepcopy
  from functools import partial
  from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
  import numpy
  import pytest
  import torch
  from lightning_utilities import apply_to_collection
  from torch import Tensor, tensor
  from torchmetrics import Metric
  from torchmetrics.utilities.data import _flatten
  from unittests import NUM_PROCESSES
label_desc: Use image node_id 2 for calling the MetricTester obj's underlying member method code with example usage: obj.run_class_metric_test(ddp, preds, target, metric_class, reference_metric, dist_sync_on_step, metric_args, check_dist_sync_on_step, check_batch, fragment_kwargs, check_scriptable, check_state_dict, atol) without return types