docstr: Replica of ``Worker.digests_total`` and
  ``Scheduler.cumulative_worker_metrics``, but only for the metrics that can be
  attributed to the current span tree. The span id has been removed from the key.
  
  At the moment of writing, all keys are
  ``("execute", <task prefix>, <activity>, <unit>)``
  or
  ``("p2p", <where>, <activity>, <unit>)``
  but more may be added in the future with a different format; please test e.g.
  for ``k[0] == "execute"``.
docstr_tok: ["Replica","of","``","Worker.digests_total","``","and","``","Scheduler.cumulative_worker_metrics","``",",","but","only","for","the","metrics","that","can","be","attributed","to","the","current","span","tree",".","The","span","id","has","been","removed","from","the","key",".","At","the","moment","of","writing",",","all","keys","are","``","(","``","execute","''",",","<","task","prefix",">",",","<","activity",">",",","<","unit",">",")","``","or","``","(","``","p2p","''",",","<","where",">",",","<","activity",">",",","<","unit",">",")","``","but","more","may","be","added","in","the","future","with","a","different","format",";","please","test","e.g",".","for","``","k","[","0","]","==","``","execute","''","``","."]
code: def cumulative_worker_metrics(
      self,
  ) -> dict[tuple[Hashable, ...], float]:
      """Replica of ``Worker.digests_total`` and
      ``Scheduler.cumulative_worker_metrics``, but only for the metrics that can be
      attributed to the current span tree. The span id has been removed from the key.
  
      At the moment of writing, all keys are
      ``("execute", <task prefix>, <activity>, <unit>)``
      or
      ``("p2p", <where>, <activity>, <unit>)``
      but more may be added in the future with a different format; please test e.g.
      for ``k[0] == "execute"``.
      """
      out = sum_mappings(
          child._cumulative_worker_metrics
          for child in self.traverse_spans()
      )
      known_seconds = sum(
          v
          for k, v in out.items()
          if k[0] == "execute" and k[-1] == "seconds"
      )
      # Besides rounding errors, you may get negative unknown seconds if a user
      # manually invokes `context_meter.digest_metric`.
      unknown_seconds = max(
          0.0, self.active_cpu_seconds - known_seconds
      )
  
      out[
          "execute", "N/A", "idle or other spans", "seconds"
      ] = unknown_seconds
      return out
  
imports: from __future__ import annotations
  import uuid
  import weakref
  from collections import defaultdict
  from collections.abc import Hashable, Iterable, Iterator
  from contextlib import contextmanager
  from itertools import islice
  from typing import TYPE_CHECKING, Any
  import dask.config
  from dask.typing import Key
  from distributed.collections import sum_mappings
  from distributed.itertools import ffill
  from distributed.metrics import time
label_desc: Use image node_id 15 for calling the Span obj's underlying member method code with example usage: obj.cumulative_worker_metrics() and returns: out