class_docstr: Keep track of high-level timing information for task group progress
class_docstr_tok: ["Keep","track","of","high-level","timing","information","for","task","group","progress"]
code: def transition(self, key, start, finish, *args, **kwargs):
      # We are mostly interested in when tasks complete for now, so just look
      # for when processing transitions to memory. Later we could also extend
      # this if we can come up with useful visual channels to show it in.
      if start == "processing" and finish == "memory":
          startstops = kwargs.get("startstops")
          if not startstops:
              logger.warning(
                  f"Task {key} finished processing, but timing information seems to "
                  "be missing"
              )
              return
  
          # Possibly extend the timeseries if another dt has passed
          now = time()
          self.time[-1] = now
          while self.time[-1] - self.time[-2] > self.dt:
              self.time[-1] = self.time[-2] + self.dt
              self.time.append(now)
              self.nthreads.append(self.scheduler.total_nthreads)
              for g in self.compute.values():
                  g.append(0.0)
  
          # Get the task
          task = self.scheduler.tasks[key]
          group = task.group
  
          # If the group is new, add it to the timeseries as if it has been
          # here the whole time
          if group.name not in self.compute:
              self.compute[group.name] = [0.0] * len(self.time)
  
          for startstop in startstops:
              if startstop["action"] != "compute":
                  continue
              stop = startstop["stop"]
              start = startstop["start"]
              idx = len(self.time) - 1
              # If the stop time is after the most recent bin,
              # roll back the current index. Not clear how often this happens.
              while idx > 0 and self.time[idx - 1] > stop:
                  idx -= 1
              # Allocate the timing information of the task to the time bins.
              while idx > 0 and stop > start:
                  delta = stop - max(self.time[idx - 1], start)
                  self.compute[group.name][idx] += delta
  
                  stop -= delta
                  idx -= 1
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import warnings
  from collections import defaultdict
  from timeit import default_timer
  from typing import ClassVar
  from tlz import groupby, valmap
  from dask.base import tokenize
  from dask.utils import key_split
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.metrics import time
label_desc: Use image node_id 3 for calling the GroupTiming obj's underlying member method code with example usage: obj.transition(key, start, finish) without return types