class_docstr: WorkloadSequencer is the python rewrite of the old golang
  TrialWorkloadSequencer.
  
  Like the go version, it fuses the dual stream of searcher operations +
  descheduling decisions into a single stream of Workload events.
  
  When the sequencer was in the master, the resulting stream of Workloads was
  the basis for all master/harness communications, but now that the sequencer
  lives in the harness, all master/harness communications are over the new
  push APIs.
  
  This Workload stream (and the whole WorkloadSequencer) is only even needed
  for reverse-compatibility with the old TrialControllers that we don't care
  to update (TFKerasTrial and EstimatorTrial).
class_docstr_tok: ["WorkloadSequencer","is","the","python","rewrite","of","the","old","golang","TrialWorkloadSequencer",".","Like","the","go","version",",","it","fuses","the","dual","stream","of","searcher","operations","+","descheduling","decisions","into","a","single","stream","of","Workload","events",".","When","the","sequencer","was","in","the","master",",","the","resulting","stream","of","Workloads","was","the","basis","for","all","master\/harness","communications",",","but","now","that","the","sequencer","lives","in","the","harness",",","all","master\/harness","communications","are","over","the","new","push","APIs",".","This","Workload","stream","(","and","the","whole","WorkloadSequencer",")","is","only","even","needed","for","reverse-compatibility","with","the","old","TrialControllers","that","we","do","n't","care","to","update","(","TFKerasTrial","and","EstimatorTrial",")","."]
code: def validate(
      self, op: Optional[core.SearcherOperation]
  ) -> WorkloadGenerator:
      # Report a validation step is starting.
      self.core_context.train.set_status("validating")
  
      wkld = workload.Workload(
          kind=workload.Workload.Kind.COMPUTE_VALIDATION_METRICS,
          e_id=self._exp_id,
          t_id=self._trial_id,
          s_id=self.state.step_id,
          num_batches=0,
          total_batches_processed=self.state.steps_completed,
      )
  
      response = yield from yield_and_await_response(wkld)
  
      # Validation step is complete, process the result.
  
      if isinstance(response, workload.InvalidHP):
          self.core_context.train.report_early_exit(
              core.EarlyExitReason.INVALID_HP
          )
          raise ShouldExit()
  
      metrics = response["metrics"]["validation_metrics"]
  
      # Check that the validation metrics computed by the model code
      # includes the metric used by the search method.
      searcher_metric_name = self.env.experiment_config["searcher"][
          "metric"
      ]
      if searcher_metric_name not in metrics:
          raise RuntimeError(
              f"Search method is configured to use metric '{searcher_metric_name}' but model "
              f"definition returned validation metrics {list(metrics.keys())}. The metric "
              "used by the search method must be one of the validation "
              "metrics returned by the model definition."
          )
  
      # Check that the searcher metric has a scalar value so that it can be compared for
      # search purposes. Other metrics don't have to be scalars.
      searcher_metric = metrics[searcher_metric_name]
      if not util.is_numerical_scalar(searcher_metric):
          raise RuntimeError(
              f"Searcher validation metric '{searcher_metric_name}' returned "
              f"a non-scalar value: {searcher_metric}"
          )
  
      # Report to the searcher API first, so we don't end up in a situation where we die between
      # reporting to the metrics API and when we come back we refuse to repeat a validation, but
      # we also don't have any validation metrics to report the the searcher API.
      #
      # A simpler solution here would be to execute in the following order (which would be
      # suitable for most customers to implement on their own):
      #   - validation
      #   - report to metrics API
      #   - report to searcher API
      #   - checkpoint
      #
      # But we can't do that without breaking behavior.
      if op is not None and self.batches_until_op_complete(op) < 1:
          op.report_completed(searcher_metric)
  
      if (
          self.ckpt_policy == "best"
          and not self.checkpoint_is_current()
      ):
          # Before reporting our own validation metric, check what the best known validation is
          # without it.
          best_validation_before = (
              self.core_context.train.get_experiment_best_validation()
          )
  
      self.state.last_val = self.state.steps_completed
      self.core_context.train.report_validation_metrics(
          steps_completed=self.state.steps_completed,
          metrics=metrics,
      )
  
      if response.get("stop_requested"):
          raise ShouldExit()
  
      if not self.checkpoint_is_current():
          if self.ckpt_policy == "all" or (
              self.ckpt_policy == "best"
              and self.is_best_validation(
                  now=searcher_metric, before=best_validation_before
              )
          ):
              yield from self.checkpoint(already_exiting=False)
  
      self.check_for_preemption()
  
imports: import logging
  import sys
  from typing import Any, Generator, Optional, Tuple
  import determined
  from determined import core, util, workload
  from determined.common import check
label_desc: Use image node_id 8 for calling the WorkloadSequencer obj's underlying member method code with example usage: obj.validate(op) without return types