code: def _progressive_validation(
      dataset: base.typing.Dataset,
      model,
      metric: metrics.base.Metric,
      checkpoints: typing.Iterator[int],
      moment: str | typing.Callable[[dict], dt.datetime] | None = None,
      delay: str | int | dt.timedelta | typing.Callable | None = None,
      measure_time=False,
      measure_memory=False,
      yield_predictions=False,
  ):
      # Check that the model and the metric are in accordance
      if not metric.works_with(model):
          raise ValueError(
              f"{metric.__class__.__name__} metric is not compatible with {model}"
          )
  
      # Determine if predict_one or predict_proba_one should be used in case of a classifier
      if utils.inspect.isanomalydetector(
          model
      ) or utils.inspect.isanomalyfilter(model):
          pred_func = model.score_one
      elif utils.inspect.isclassifier(model) and not metric.requires_labels:  # type: ignore
          pred_func = model.predict_proba_one
      else:
          pred_func = model.predict_one
  
      preds = {}
  
      # If we are dealing with an active learner, we need to check whether or not a label should be
      # used for training or not. We'll also record how many times labels were used.
      active_learning = utils.inspect.isactivelearner(model)
      n_samples_learned = 0
  
      prev_checkpoint = None
      next_checkpoint = next(checkpoints, None)
      n_total_answers = 0
      if measure_time:
          start = time.perf_counter()
  
      def report(y_pred):
          if isinstance(metric, metrics.base.Metrics):
              state = {m.__class__.__name__: m for m in metric}
          else:
              state = {metric.__class__.__name__: metric}
          state["Step"] = n_total_answers
          if active_learning:
              state["Samples used"] = n_samples_learned
          if measure_time:
              now = time.perf_counter()
              state["Time"] = dt.timedelta(seconds=now - start)
          if measure_memory:
              state["Memory"] = model._raw_memory_usage
          if yield_predictions:
              state["Prediction"] = y_pred
  
          return state
  
      for i, x, y, *kwargs in stream.simulate_qa(
          dataset, moment, delay, copy=True
      ):
          kwargs = kwargs[0] if kwargs else {}
  
          # Case 1: no ground truth, just make a prediction
          if y is None:
              y_pred = pred_func(x, **kwargs)
              y_pred, ask_for_label = (
                  y_pred if active_learning else (y_pred, True)
              )
              if utils.inspect.isanomalyfilter(model):
                  y_pred = model.classify(y_pred)
              preds[i] = y_pred, ask_for_label
              continue
  
          # Case 2: there's a ground truth, model and metric can be updated
          y_pred, use_label = preds.pop(i)
  
          # Update the metric
          if y_pred != {} and y_pred is not None:
              metric.update(y_true=y, y_pred=y_pred)
  
          # Update the model
          if use_label:
              n_samples_learned += 1
              if model._supervised:
                  model.learn_one(x, y, **kwargs)
              else:
                  model.learn_one(x, **kwargs)
  
          # Yield current results
          n_total_answers += 1
          if n_total_answers == next_checkpoint:
              yield report(y_pred=y_pred)
              prev_checkpoint = next_checkpoint
              next_checkpoint = next(checkpoints, None)
      else:
          # If the dataset was exhausted, we need to make sure that we yield the final results
          if prev_checkpoint and n_total_answers != prev_checkpoint:
              yield report(y_pred=None)
  
file_path: river/river/evaluate/progressive_validation.py
imports: from __future__ import annotations
  import datetime
  import itertools
  import time
  import typing
  from river import base, metrics, stream, utils
label_desc: Use image node_id 1 for calling a global function with example usage: _progressive_validation(dataset, model, metric, checkpoints, moment, delay, measure_time, measure_memory, yield_predictions) and returns: state