class_docstr: Spike Detection Callback.
  
  Terminates training with a ``TrainingSpikeException`` when a loss-spike was detected and
  saves the batches to skip when resuming to a file.
  
  We skip the current and the previous batch since it is unclear whether the previous batch
  altered the weights in a way that it causes the spike or just the current batch is corrupted somehow.
  
  Args:
      mode: Whether to minimize or maximize the tracked metric
      window: A running mean of metrics with ``window`` size. Serves as reference value for spikes.
      warmup: After how many batches spike-tracking should start
      atol: An absolute tolerance.  Every diff between the running mean and the current value,
          that's not an improvement and above ``atol`` will be considered a spike
      rtol: A relative tolerance. Every diff between the running mean and the current value,
          that's higher than ``rtol * running_mean`` is considered a spike
      exclude_batches_path: Where to save the file that contains the batches to exclude.
          Will default to current directory.
      finite_only: If set to ``False``, consider non-finite values like NaN, inf and -inf a spike as well.
docstr: Checks if we currently have a loss-spike.
docstr_tok: ["Checks","if","we","currently","have","a","loss-spike","."]
code: def on_train_batch_end(
      self,
      fabric: "Fabric",
      loss: torch.Tensor,
      batch: Any,
      batch_idx: int,
  ) -> None:
      """Checks if we currently have a loss-spike."""
      if batch_idx == 0:
          self.running_mean.to(fabric.strategy.root_device)
  
      if self.exclude_batches_path is None:
          self.exclude_batches_path = os.getcwd()
  
      if not str(self.exclude_batches_path).endswith(".json"):
          self.exclude_batches_path = os.path.join(
              self.exclude_batches_path, "skip_batches.json"
          )
  
      is_spike = bool(batch_idx >= self.warmup and self._is_spike(loss))
      fabric.strategy.barrier()
  
      # While spike-detection happens on a per-rank level, we need to fail all ranks if any rank detected a spike
      is_spike_global = fabric.strategy.reduce_boolean_decision(
          is_spike, all=False
      )
  
      if is_spike_global:
          self._handle_spike(fabric, batch_idx)
      else:
          is_finite_all = (
              self.finite_only
              or fabric.strategy.reduce_boolean_decision(
                  bool(torch.isfinite(loss).all()), all=True
              )
          )
          if is_finite_all:
              self._update_stats(loss)
  
imports: import json
  import operator
  import os
  import warnings
  from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union
  import torch
  from lightning_utilities.core.imports import compare_version
  from lightning.fabric.utilities.types import _PATH
label_desc: Use image node_id 2 for calling the SpikeDetection obj's underlying member method code with example usage: obj.on_train_batch_end(fabric, loss, batch, batch_idx) without return types