code: def unpack_collections(expr):
      """Normalize a python object and merge all sub-graphs.
  
      - Replace ``Delayed`` with their keys
      - Convert literals to things the schedulers can handle
      - Extract dask graphs from all enclosed values
  
      Parameters
      ----------
      expr : object
          The object to be normalized. This function knows how to handle
          dask collections, as well as most builtin python types.
  
      Returns
      -------
      task : normalized task to be run
      collections : a tuple of collections
  
      Examples
      --------
      >>> import dask
      >>> a = delayed(1, 'a')
      >>> b = delayed(2, 'b')
      >>> task, collections = unpack_collections([a, b, 3])
      >>> task
      ['a', 'b', 3]
      >>> collections
      (Delayed('a'), Delayed('b'))
  
      >>> task, collections = unpack_collections({a: 1, b: 2})
      >>> task
      (<class 'dict'>, [['a', 1], ['b', 2]])
      >>> collections
      (Delayed('a'), Delayed('b'))
      """
      if isinstance(expr, Delayed):
          return expr._key, (expr,)
  
      if is_dask_collection(expr):
          finalized = finalize(expr)
          return finalized._key, (finalized,)
  
      if type(expr) is type(iter(list())):
          expr = list(expr)
      elif type(expr) is type(iter(tuple())):
          expr = tuple(expr)
      elif type(expr) is type(iter(set())):
          expr = set(expr)
  
      typ = type(expr)
  
      if typ in (list, tuple, set):
          args, collections = unzip(
              (unpack_collections(e) for e in expr), 2
          )
          args = list(args)
          collections = tuple(unique(concat(collections), key=id))
          # Ensure output type matches input type
          if typ is not list:
              args = (typ, args)
          return args, collections
  
      if typ is dict:
          args, collections = unpack_collections(
              [[k, v] for k, v in expr.items()]
          )
          return (dict, args), collections
  
      if typ is slice:
          args, collections = unpack_collections(
              [expr.start, expr.stop, expr.step]
          )
          return (slice, *args), collections
  
      if is_dataclass(expr):
          args, collections = unpack_collections(
              [
                  [f.name, getattr(expr, f.name)]
                  for f in fields(expr)
                  if hasattr(
                      expr, f.name
                  )  # if init=False, field might not exist
              ]
          )
          if not collections:
              return expr, ()
          try:
              _fields = {
                  f.name: getattr(expr, f.name)
                  for f in fields(expr)
                  if hasattr(expr, f.name)
              }
              replace(expr, **_fields)
          except TypeError as e:
              raise TypeError(
                  f"Failed to unpack {typ} instance. "
                  "Note that using a custom __init__ is not supported."
              ) from e
          except ValueError as e:
              raise ValueError(
                  f"Failed to unpack {typ} instance. "
                  "Note that using fields with `init=False` are not supported."
              ) from e
          return (apply, typ, (), (dict, args)), collections
  
      if is_namedtuple_instance(expr):
          args, collections = unpack_collections([v for v in expr])
          return (typ, *args), collections
  
      return expr, ()
  
imports: from __future__ import annotations
  import operator
  import types
  import uuid
  import warnings
  from collections.abc import Sequence
  from dataclasses import fields, is_dataclass, replace
  from functools import partial
  from tlz import concat, curry, merge, unique
  from dask import config
  from dask.base import DaskMethodsMixin, dont_optimize, is_dask_collection, named_schedulers, replace_name_in_key
  from dask.base import tokenize
  from dask.context import globalmethod
  from dask.core import flatten, quote
  from dask.highlevelgraph import HighLevelGraph
  from dask.typing import Graph, NestedKeys
  from dask.utils import OperatorMethodMixin, apply, funcname, is_namedtuple_instance, methodcaller
label_desc: Use image node_id 3 for calling a global function with example usage: unpack_collections(expr) and returns: expr,  , expr,  expr, finalized,  finalized, args,  collections, dict,  args,  collections, slice,  unknown,  collections, apply,  typ,  ,  dict,  args,  collections, typ,  unknown,  collections, expr,  