class_docstr: Abstract class for trainable models, functions, etc.
  
  A call to ``train()`` on a trainable will execute one logical iteration of
  training. As a rule of thumb, the execution time of one train call should
  be large enough to avoid overheads (i.e. more than a few seconds), but
  short enough to report progress periodically (i.e. at most a few minutes).
  
  Calling ``save()`` should save the training state of a trainable to disk,
  and ``restore(path)`` should restore a trainable to the given state.
  
  Generally you only need to implement ``setup``, ``step``,
  ``save_checkpoint``, and ``load_checkpoint`` when subclassing Trainable.
  
  Other implementation methods that may be helpful to override are
  ``log_result``, ``reset_config``, ``cleanup``, and ``_export_model``.
  
  Tune will convert this class into a Ray actor, which runs on a separate process.
  By default, Tune will also change the current working directory of this process to
  its corresponding trial-level log directory ``self.logdir``.
  This is designed so that different trials that run on the same physical node won't
  accidentally write to the same location and overstep each other.
  
  The behavior of changing the working directory can be disabled by setting the
  `RAY_CHDIR_TO_TRIAL_DIR=0` environment variable. This allows access to files
  in the original working directory, but relative paths should be used for read only
  purposes, and you must make sure that the directory is synced on all nodes if
  running on multiple machines.
  
  The `TUNE_ORIG_WORKING_DIR` environment variable was the original workaround for
  accessing paths relative to the original working directory. This environment
  variable is deprecated, and the `RAY_CHDIR_TO_TRIAL_DIR` environment variable
  described above should be used instead.
  
  This class supports checkpointing to and restoring from remote storage.
docstr: Runs one logical iteration of training.
  
  Calls ``step()`` internally. Subclasses should override ``step()``
  instead to return results.
  This method automatically fills the following fields in the result:
  
      `done` (bool): training is terminated. Filled only if not provided.
  
      `time_this_iter_s` (float): Time in seconds this iteration
      took to run. This may be overridden in order to override the
      system-computed time difference.
  
      `time_total_s` (float): Accumulated time in seconds for this
      entire experiment.
  
      `training_iteration` (int): The index of this
      training iteration, e.g. call to train(). This is incremented
      after `step()` is called.
  
      `pid` (str): The pid of the training process.
  
      `date` (str): A formatted date of when the result was processed.
  
      `timestamp` (str): A UNIX timestamp of when the result
      was processed. This may be overridden.
  
      `hostname` (str): Hostname of the machine hosting the training
      process.
  
      `node_ip` (str): Node ip of the machine hosting the training
      process.
  
  Returns:
      A dict that describes training progress.
docstr_tok: ["Runs","one","logical","iteration","of","training",".","Calls","``","step","(",")","``","internally",".","Subclasses","should","override","``","step","(",")","``","instead","to","return","results",".","This","method","automatically","fills","the","following","fields","in","the","result",":","`","done","`","(","bool",")",":","training","is","terminated",".","Filled","only","if","not","provided",".","`","time_this_iter_s","`","(","float",")",":","Time","in","seconds","this","iteration","took","to","run",".","This","may","be","overridden","in","order","to","override","the","system-computed","time","difference",".","`","time_total_s","`","(","float",")",":","Accumulated","time","in","seconds","for","this","entire","experiment",".","`","training_iteration","`","(","int",")",":","The","index","of","this","training","iteration",",","e.g",".","call","to","train","(",")",".","This","is","incremented","after","`","step","(",")","`","is","called",".","`","pid","`","(","str",")",":","The","pid","of","the","training","process",".","`","date","`","(","str",")",":","A","formatted","date","of","when","the","result","was","processed",".","`","timestamp","`","(","str",")",":","A","UNIX","timestamp","of","when","the","result","was","processed",".","This","may","be","overridden",".","`","hostname","`","(","str",")",":","Hostname","of","the","machine","hosting","the","training","process",".","`","node_ip","`","(","str",")",":","Node","ip","of","the","machine","hosting","the","training","process",".","Returns",":","A","dict","that","describes","training","progress","."]
code: def train(self):
      """Runs one logical iteration of training.
  
      Calls ``step()`` internally. Subclasses should override ``step()``
      instead to return results.
      This method automatically fills the following fields in the result:
  
          `done` (bool): training is terminated. Filled only if not provided.
  
          `time_this_iter_s` (float): Time in seconds this iteration
          took to run. This may be overridden in order to override the
          system-computed time difference.
  
          `time_total_s` (float): Accumulated time in seconds for this
          entire experiment.
  
          `training_iteration` (int): The index of this
          training iteration, e.g. call to train(). This is incremented
          after `step()` is called.
  
          `pid` (str): The pid of the training process.
  
          `date` (str): A formatted date of when the result was processed.
  
          `timestamp` (str): A UNIX timestamp of when the result
          was processed. This may be overridden.
  
          `hostname` (str): Hostname of the machine hosting the training
          process.
  
          `node_ip` (str): Node ip of the machine hosting the training
          process.
  
      Returns:
          A dict that describes training progress.
      """
      start = time.time()
      try:
          result = self.step()
      except Exception as e:
          skipped = skip_exceptions(e)
          raise skipped from exception_cause(skipped)
  
      assert isinstance(result, dict), "step() needs to return a dict."
  
      # We do not modify internal state nor update this result if duplicate.
      if RESULT_DUPLICATE in result:
          return result
  
      result = result.copy()
  
      self._iteration += 1
      self._iterations_since_restore += 1
  
      if result.get(TIME_THIS_ITER_S) is not None:
          time_this_iter = result[TIME_THIS_ITER_S]
      else:
          time_this_iter = time.time() - start
      self._time_total += time_this_iter
      self._time_since_restore += time_this_iter
  
      result_timestamp = result.get(TIMESTAMP, None)
  
      result.setdefault(DONE, False)
  
      # self._timesteps_total should only be tracked if increments are provided
      if result.get(TIMESTEPS_THIS_ITER) is not None:
          if self._timesteps_total is None:
              self._timesteps_total = 0
          self._timesteps_total += result[TIMESTEPS_THIS_ITER]
          self._timesteps_since_restore += result[TIMESTEPS_THIS_ITER]
  
      # self._episodes_total should only be tracked if increments provided
      if result.get(EPISODES_THIS_ITER) is not None:
          if self._episodes_total is None:
              self._episodes_total = 0
          self._episodes_total += result[EPISODES_THIS_ITER]
  
      # self._timesteps_total should not override user-provided total
      if self._timesteps_total is not None:
          result.setdefault(TIMESTEPS_TOTAL, self._timesteps_total)
      if self._episodes_total is not None:
          result.setdefault(EPISODES_TOTAL, self._episodes_total)
      result.setdefault(TRAINING_ITERATION, self._iteration)
  
      now = datetime.today()
      result.update(
          self.get_auto_filled_metrics(
              now=now,
              time_this_iter=time_this_iter,
              timestamp=result_timestamp,
          )
      )
  
      monitor_data = self._monitor.get_data()
      if monitor_data:
          result.update(monitor_data)
  
      self.log_result(result)
  
      if self._stdout_context:
          self._stdout_stream.flush()
      if self._stderr_context:
          self._stderr_stream.flush()
  
      self._last_result = result
  
      if self._storage:
          # Launch background tasks to sync artifacts at some specified frequency.
          self._storage.persist_artifacts()
  
      return result
  
imports: import copy
  from datetime import datetime
  import logging
  import os
  from pathlib import Path
  import platform
  import sys
  import tempfile
  import time
  from contextlib import redirect_stderr, redirect_stdout
  from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
  import ray
  import ray.cloudpickle
  from ray.air._internal.util import skip_exceptions, exception_cause
  from ray.air.constants import TIMESTAMP, TIME_THIS_ITER_S, TRAINING_ITERATION
  from ray.train._internal.checkpoint_manager import _TrainingResult
  from ray.train._internal.storage import StorageContext, _exists_at_fs_path
  from ray.train import Checkpoint
  from ray.tune.result import DEBUG_METRICS, DEFAULT_RESULTS_DIR, DONE, EPISODES_THIS_ITER, EPISODES_TOTAL, HOSTNAME, NODE_IP, PID, RESULT_DUPLICATE, SHOULD_CHECKPOINT, STDERR_FILE, STDOUT_FILE, TIME_TOTAL_S, TIMESTEPS_THIS_ITER, TIMESTEPS_TOTAL, TRIAL_ID, TRIAL_INFO
  from ray.tune.utils import UtilMonitor
  from ray.tune.utils.log import disable_ipython
  from ray.tune.utils.util import Tee
  from ray.tune.execution.placement_groups import PlacementGroupFactory
  from ray.util.annotations import DeveloperAPI, PublicAPI
label_desc: Use image node_id 8 for calling the Trainable obj's underlying member method code with example usage: obj.train() and returns: result, result