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: Returns
  -------
  stories : dict
      A dict of stories for the keys/stimulus ID's in ``*key_or_stimulus_id`.`
docstr_tok: ["Returns","--","--","--","-","stories",":","dict","A","dict","of","stories","for","the","keys\/stimulus","ID","'s","in","``","*","key_or_stimulus_id","`",".","`"]
code: def worker_story(self, *key_or_stimulus_id: str) -> dict:
      """
      Returns
      -------
      stories : dict
          A dict of stories for the keys/stimulus ID's in ``*key_or_stimulus_id`.`
      """
      keys = set(key_or_stimulus_id)
      stories = defaultdict(list)
  
      for worker_dump in self.dump["workers"].values():
          if isinstance(worker_dump, dict) and "log" in worker_dump:
              for story in _worker_story(keys, worker_dump["log"]):
                  stories[story[0]].append(tuple(story))
  
      return dict(stories)
  
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 10 for calling the DumpArtefact obj's underlying member method code with example usage: obj.worker_story() and returns: dict