code: def counts(scheduler, allprogress):
      return merge(
          {
              "all": valmap(len, allprogress.all),
              "nbytes": allprogress.nbytes,
          },
          {
              state: valmap(len, allprogress.state[state])
              for state in [
                  "memory",
                  "erred",
                  "released",
                  "processing",
                  "queued",
              ]
          },
      )
  
file_path: distributed/distributed/diagnostics/progress_stream.py
imports: from __future__ import annotations
  import logging
  from tlz import merge, valmap
  from dask.utils import key_split
  from distributed.core import coerce_to_address, connect
  from distributed.diagnostics.progress import AllProgress
  from distributed.utils import color_of
  from distributed.worker import dumps_function
label_desc: Use image node_id 1 for calling a global function with example usage: counts(scheduler, allprogress) and returns: merge