class_docstr: A harness for early stopping.
docstr: Evaluate on a metric and compare to past evaluations to decide if training should stop.
docstr_tok: ["Evaluate","on","a","metric","and","compare","to","past","evaluations","to","decide","if","training","should","stop","."]
code: def should_stop(self, epoch: int) -> bool:
      """Evaluate on a metric and compare to past evaluations to decide if training should stop."""
      # for mypy
      assert self.best_model_path is not None
      # Evaluate
      metric_results = self.evaluator.evaluate(
          model=self.model,
          additional_filter_triples=self.training_triples_factory.mapped_triples,
          mapped_triples=self.evaluation_triples_factory.mapped_triples,
          use_tqdm=False,
          batch_size=self.evaluation_batch_size,
          slice_size=self.evaluation_slice_size,
          # Only perform time-consuming checks for the first call.
          do_time_consuming_checks=self.evaluation_batch_size is None,
      )
      # After the first evaluation pass the optimal batch and slice size is obtained and saved for re-use
      self.evaluation_batch_size = self.evaluator.batch_size
      self.evaluation_slice_size = self.evaluator.slice_size
  
      if self.result_tracker is not None:
          self.result_tracker.log_metrics(
              metrics=metric_results.to_flat_dict(),
              step=epoch,
              prefix="validation",
          )
      result = metric_results.get_metric(self.metric)
  
      # Append to history
      self.results.append(result)
  
      for result_callback in self.result_callbacks:
          result_callback(self, result, epoch)
  
      self.stopped = self._stopper.report_result(
          metric=result, epoch=epoch
      )
      if self.stopped:
          logger.info(
              f"Stopping early at epoch {epoch}. The best result {self.best_metric} occurred at "
              f"epoch {self.best_epoch}.",
          )
          for stopped_callback in self.stopped_callbacks:
              stopped_callback(self, result, epoch)
          logger.info(
              f"Re-loading weights from best epoch from {self.best_model_path}"
          )
          self.model.load_state_dict(torch.load(self.best_model_path))
          if self.clean_up_checkpoint:
              self.best_model_path.unlink()
              logger.debug(
                  f"Clean up checkpoint with best weights: {self.best_model_path}"
              )
          return True
  
      if self._stopper.is_best:
          torch.save(self.model.state_dict(), self.best_model_path)
          logger.info(
              f"New best result at epoch {epoch}: {self.best_metric}. Saved model weights to {self.best_model_path}",
          )
  
      for continue_callback in self.continue_callbacks:
          continue_callback(self, result, epoch)
      return False
  
imports: import dataclasses
  import logging
  import math
  import pathlib
  from dataclasses import dataclass
  from typing import Any, Callable, List, Mapping, Optional, Union
  from uuid import uuid4
  import torch
  from .stopper import Stopper
  from ..constants import PYKEEN_CHECKPOINTS
  from ..evaluation import Evaluator
  from ..models import Model
  from ..trackers import ResultTracker
  from ..triples import CoreTriplesFactory
  from ..utils import fix_dataclass_init_docs
label_desc: Use image node_id 7 for calling the EarlyStopper obj's underlying member method code with example usage: obj.should_stop(epoch) and returns: False, True