class_docstr: Utility class for inspecting the state of a cluster dump
  
  .. code-block:: python
  
      dump = DumpArtefact.from_url("dump.msgpack.gz")
      memory_tasks = dump.scheduler_tasks("memory")
      executing_tasks = dump.worker_tasks("executing")
class_docstr_tok: ["Utility","class","for","inspecting","the","state","of","a","cluster","dump","..","code-block",":",":","python","dump","=","DumpArtefact.from_url","(","``","dump.msgpack.gz","''",")","memory_tasks","=","dump.scheduler_tasks","(","``","memory","''",")","executing_tasks","=","dump.worker_tasks","(","``","executing","''",")"]
docstr: Splits the Dump Artefact into a tree of yaml files with
  ``root_dir`` as it's base.
  
  The root level of the tree contains a directory for the scheduler
  and directories for each individual worker.
  Each directory contains yaml files describing the state of the scheduler
  or worker when the artefact was created.
  
  In general, keys associated with the state are compacted into a ``general.yaml``
  file, unless they are in ``scheduler_expand_keys`` and ``worker_expand_keys``.
  
  Parameters
  ----------
  root_dir : str or Path
      The root directory into which the tree is written.
      Defaults to the current working directory if ``None``.
  worker_expand_keys : iterable of str
      An iterable of artefact worker keys that will be expanded
      into separate yaml files.
      Keys that are not in this iterable are compacted into a
      `general.yaml` file.
  scheduler_expand_keys : iterable of str
      An iterable of artefact scheduler keys that will be expanded
      into separate yaml files.
      Keys that are not in this iterable are compacted into a
      ``general.yaml`` file.
docstr_tok: ["Splits","the","Dump","Artefact","into","a","tree","of","yaml","files","with","``","root_dir","``","as","it","'s","base",".","The","root","level","of","the","tree","contains","a","directory","for","the","scheduler","and","directories","for","each","individual","worker",".","Each","directory","contains","yaml","files","describing","the","state","of","the","scheduler","or","worker","when","the","artefact","was","created",".","In","general",",","keys","associated","with","the","state","are","compacted","into","a","``","general.yaml","``","file",",","unless","they","are","in","``","scheduler_expand_keys","``","and","``","worker_expand_keys","``",".","Parameters","--","--","--","--","--","root_dir",":","str","or","Path","The","root","directory","into","which","the","tree","is","written",".","Defaults","to","the","current","working","directory","if","``","None","``",".","worker_expand_keys",":","iterable","of","str","An","iterable","of","artefact","worker","keys","that","will","be","expanded","into","separate","yaml","files",".","Keys","that","are","not","in","this","iterable","are","compacted","into","a","`","general.yaml","`","file",".","scheduler_expand_keys",":","iterable","of","str","An","iterable","of","artefact","scheduler","keys","that","will","be","expanded","into","separate","yaml","files",".","Keys","that","are","not","in","this","iterable","are","compacted","into","a","``","general.yaml","``","file","."]
code: def to_yamls(
      self,
      root_dir: str | Path | None = None,
      worker_expand_keys: Collection[str] = (
          "config",
          "log",
          "logs",
          "tasks",
      ),
      scheduler_expand_keys: Collection[str] = (
          "events",
          "extensions",
          "log",
          "task_groups",
          "tasks",
          "transition_log",
          "workers",
      ),
  ) -> None:
      """
      Splits the Dump Artefact into a tree of yaml files with
      ``root_dir`` as it's base.
  
      The root level of the tree contains a directory for the scheduler
      and directories for each individual worker.
      Each directory contains yaml files describing the state of the scheduler
      or worker when the artefact was created.
  
      In general, keys associated with the state are compacted into a ``general.yaml``
      file, unless they are in ``scheduler_expand_keys`` and ``worker_expand_keys``.
  
      Parameters
      ----------
      root_dir : str or Path
          The root directory into which the tree is written.
          Defaults to the current working directory if ``None``.
      worker_expand_keys : iterable of str
          An iterable of artefact worker keys that will be expanded
          into separate yaml files.
          Keys that are not in this iterable are compacted into a
          `general.yaml` file.
      scheduler_expand_keys : iterable of str
          An iterable of artefact scheduler keys that will be expanded
          into separate yaml files.
          Keys that are not in this iterable are compacted into a
          ``general.yaml`` file.
      """
      import yaml
  
      root_dir = Path(root_dir) if root_dir else Path.cwd()
      dumper = yaml.CSafeDumper
      scheduler_expand_keys = set(scheduler_expand_keys)
      worker_expand_keys = set(worker_expand_keys)
  
      workers = self.dump["workers"]
      for info in workers.values():
          try:
              worker_id = info["id"]
          except KeyError:
              continue
  
          worker_state = self._compact_state(info, worker_expand_keys)
  
          log_dir = root_dir / worker_id
          log_dir.mkdir(parents=True, exist_ok=True)
  
          for name, _logs in worker_state.items():
              filename = str(log_dir / f"{name}.yaml")
              with open(filename, "w") as fd:
                  yaml.dump(_logs, fd, Dumper=dumper)
  
      context = "scheduler"
      scheduler_state = self._compact_state(
          self.dump[context], scheduler_expand_keys
      )
  
      log_dir = root_dir / context
      log_dir.mkdir(parents=True, exist_ok=True)
      # Compact smaller keys into a general dict
  
      for name, _logs in scheduler_state.items():
          filename = str(log_dir / f"{name}.yaml")
  
          with open(filename, "w") as fd:
              yaml.dump(_logs, fd, Dumper=dumper)
  
imports: from __future__ import annotations
  import asyncio
  from collections import defaultdict
  from collections.abc import Awaitable, Callable, Collection, Mapping
  from pathlib import Path
  from typing import IO, Any, Literal
  import msgpack
  from dask.typing import Key
  from distributed._stories import scheduler_story
  from distributed._stories import worker_story
label_desc: Use image node_id 13 for calling the DumpArtefact obj's underlying member method code with example usage: obj.to_yamls(root_dir, worker_expand_keys, scheduler_expand_keys) without return types