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 train(
      self, num_batches: int, op: core.SearcherOperation
  ) -> WorkloadGenerator:
      # Report a train step is starting.
      self.core_context.train.set_status("training")
  
      wkld = workload.Workload(
          kind=workload.Workload.Kind.RUN_STEP,
          e_id=self._exp_id,
          t_id=self._trial_id,
          s_id=self.state.step_id + 1,
          num_batches=num_batches,
          total_batches_processed=self.state.steps_completed,
      )
  
      response = yield from yield_and_await_response(wkld)
  
      # Train step is complete, process the result.
  
      if isinstance(response, workload.InvalidHP):
          # Exit before reporting metrics (which would be empty anyway).
          self.core_context.train.report_early_exit(
              core.EarlyExitReason.INVALID_HP
          )
          raise ShouldExit()
  
      metrics = response.get("metrics", {}).get("avg_metrics", {})
      batch_metrics = response.get("metrics", {}).get(
          "batch_metrics", []
      )
  
      self.state.steps_completed += num_batches
      self.state.step_id += 1
      self.core_context.train.report_training_metrics(
          steps_completed=self.state.steps_completed,
          metrics=metrics,
          batch_metrics=batch_metrics,
      )
  
      # Report progress to the searcher.  For historical reasons we only deal in batches.
      if self._unit == core.Unit.BATCHES:
          op.report_progress(self.state.steps_completed)
      elif self._unit == core.Unit.RECORDS:
          op.report_progress(
              self.global_batch_size * self.state.steps_completed
          )
      elif self._unit == core.Unit.EPOCHS:
          op.report_progress(
              self.state.steps_completed / self.as_batches(epochs=1)
          )
      else:
          raise ValueError(
              f"unrecognized searcher op unit: {self._unit}"
          )
  
      if response.get("stop_requested"):
          # Exit after reporting metrics.
          raise ShouldExit()
  
      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 6 for calling the WorkloadSequencer obj's underlying member method code with example usage: obj.train(num_batches, op) without return types