docstr: The compute to communication time ratio of a key
  
  Returns
  -------
  cost_multiplier: The increased cost from moving this task as a factor.
  For example a result of zero implies a task without dependencies.
  level: The location within a stealable list to place this value
docstr_tok: ["The","compute","to","communication","time","ratio","of","a","key","Returns","--","--","--","-","cost_multiplier",":","The","increased","cost","from","moving","this","task","as","a","factor",".","For","example","a","result","of","zero","implies","a","task","without","dependencies",".","level",":","The","location","within","a","stealable","list","to","place","this","value"]
code: def steal_time_ratio(
      self, ts: TaskState
  ) -> tuple[float, int] | tuple[None, None]:
      """The compute to communication time ratio of a key
  
      Returns
      -------
      cost_multiplier: The increased cost from moving this task as a factor.
      For example a result of zero implies a task without dependencies.
      level: The location within a stealable list to place this value
      """
      split = ts.prefix.name
      if split in fast_tasks:
          return None, None
  
      if not ts.dependencies:  # no dependencies fast path
          return 0, 0
  
      compute_time = self.scheduler.get_task_duration(ts)
  
      if not compute_time:
          # occupancy/ws.processing[ts] is only allowed to be zero for
          # long running tasks which cannot be stolen
          assert ts.processing_on
          assert ts in ts.processing_on.long_running
          return None, None
  
      nbytes = ts.get_nbytes_deps()
      transfer_time = nbytes / self.scheduler.bandwidth + LATENCY
      cost_multiplier = transfer_time / compute_time
  
      level = int(round(log2(cost_multiplier) + 6))
  
      if level < 1:
          level = 1
      elif level >= len(self.cost_multipliers):
          return None, None
  
      return cost_multiplier, level
  
imports: from __future__ import annotations
  import asyncio
  import logging
  from collections import defaultdict, deque
  from collections.abc import Container
  from functools import partial
  from math import log2
  from time import time
  from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast
  from tlz import topk
  import dask
  from dask.typing import Key
  from dask.utils import parse_timedelta
  from distributed.compatibility import PeriodicCallback
  from distributed.core import CommClosedError
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.utils import log_errors, recursive_to_dict
label_desc: Use image node_id 13 for calling the WorkStealing obj's underlying member method code with example usage: obj.steal_time_ratio(ts) and returns: cost_multiplier,  level, None,  None, int,  int, None,  None, None,  None