code: def __init__(self):
      import ray
  
      @ray.remote
      class ProgressBarActor:
          def __init__(self):
              self._init()
  
          def submit(self, key, deps, now):
              for dep in deps.keys():
                  self.deps[key].add(dep)
              self.submitted[key] = now
              self.submission_queue.append((key, now))
  
          def task_scheduled(self, key, now):
              self.scheduled[key] = now
  
          def finish(self, key, now):
              self.finished[key] = now
  
          def result(self):
              return len(self.submitted), len(self.finished)
  
          def report(self):
              result = defaultdict(dict)
              for key, finished in self.finished.items():
                  submitted = self.submitted[key]
                  scheduled = self.scheduled[key]
                  # deps = self.deps[key]
                  result[key]["execution_time"] = (
                      finished - scheduled
                  ).total_seconds()
                  # Calculate the scheduling time.
                  # This is inaccurate.
                  # We should subtract scheduled - (last dep completed).
                  # But currently it is not easy because
                  # of how getitem is implemented in dask on ray sort.
                  result[key]["scheduling_time"] = (
                      scheduled - submitted
                  ).total_seconds()
              result["submission_order"] = self.submission_queue
              return result
  
          def ready(self):
              pass
  
          def reset(self):
              self._init()
  
          def _init(self):
              self.submission_queue = []
              self.submitted = defaultdict(None)
              self.scheduled = defaultdict(None)
              self.finished = defaultdict(None)
              self.deps = defaultdict(set)
  
      try:
          self.pb = ray.get_actor("_dask_on_ray_pb")
          ray.get(self.pb.reset.remote())
      except ValueError:
          self.pb = ProgressBarActor.options(
              name="_dask_on_ray_pb"
          ).remote()
          ray.get(self.pb.ready.remote())
  
imports: import contextlib
  from collections import namedtuple, defaultdict
  from datetime import datetime
  from dask.callbacks import Callback
label_desc: Use image node_id 1 to create a new ProgressBarCallback object from inherited base classes: RayDaskCallback with example: obj = ProgressBarCallback()