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 __init__(
      self,
      core_context: core.Context,
      env: det.EnvContext,
      global_batch_size: int,
  ) -> None:
      self.core_context = core_context
      self.env = env
      self._run_id = env.trial_run_id
      self._trial_id = int(env.det_trial_id)
      self._allocation_id = env.allocation_id
      self._exp_id = int(env.det_experiment_id)
  
      unit = self.core_context.searcher.get_configured_units()
      if unit is None:
          logger.warning(
              "The searcher configuration provided was configured without units, but the "
              "training loop you are using (one of the Trial APIs) requires a searcher "
              "configured with units.  Proceeding anyway, and assuming that the lengths "
              "configured in the searcher are in terms of epochs."
          )
          unit = core.Unit.EPOCHS
      self._unit = unit
  
      self.val_from_previous_run = (
          self.core_context.train._get_last_validation()
      )
  
      self.want_initial_val = self.env.experiment_config.get(
          "perform_initial_validation", False
      )
  
      self.ckpt_policy = self.env.experiment_config.get(
          "checkpoint_policy", "best"
      )
  
      self.state = self.SavableState(trial_id=self._trial_id)
  
      # precalculated periods, in batches
      self.records_per_epoch = (
          env.experiment_config.get_records_per_epoch()
      )
      self.global_batch_size = global_batch_size
      if (
          env.experiment_config.get_min_checkpoint_period().get(
              "epochs"
          )
          is not None
      ):
          if self.records_per_epoch is None:
              raise ValueError(
                  "records_per_epoch must be specified if min_checkpoint_period is configured "
                  "in epoch units."
              )
  
      if (
          env.experiment_config.get_min_validation_period().get(
              "epochs"
          )
          is not None
      ):
          if self.records_per_epoch is None:
              raise ValueError(
                  "records_per_epoch must be specified if min_validation_period is configured "
                  "in epoch units."
              )
      self.min_val_period_batches = self.as_batches(
          **env.experiment_config.get_min_validation_period()
      )
      self.min_ckpt_period_batches = self.as_batches(
          **env.experiment_config.get_min_checkpoint_period()
      )
      if self.min_val_period_batches < 1:
          self.min_val_period_batches = sys.maxsize
      if self.min_ckpt_period_batches < 1:
          self.min_ckpt_period_batches = sys.maxsize
  
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 1 to create a new WorkloadSequencer object from inherited base classes: workload with example: obj = WorkloadSequencer(core_context, env, global_batch_size)