code: def to_graphviz(
      hg,
      data_attributes=None,
      function_attributes=None,
      rankdir="BT",
      graph_attr=None,
      node_attr=None,
      edge_attr=None,
      **kwargs,
  ):
      from dask.dot import label, name
  
      graphviz = import_required(
          "graphviz",
          "Drawing dask graphs with the graphviz visualization engine requires the `graphviz` "
          "python library and the `graphviz` system library.\n\n"
          "Please either conda or pip install as follows:\n\n"
          "  conda install python-graphviz     # either conda install\n"
          "  python -m pip install graphviz    # or pip install and follow installation instructions",
      )
  
      data_attributes = data_attributes or {}
      function_attributes = function_attributes or {}
      graph_attr = graph_attr or {}
      node_attr = node_attr or {}
      edge_attr = edge_attr or {}
  
      graph_attr["rankdir"] = rankdir
      node_attr["shape"] = "box"
      node_attr["fontname"] = "helvetica"
  
      graph_attr.update(kwargs)
      g = graphviz.Digraph(
          graph_attr=graph_attr,
          node_attr=node_attr,
          edge_attr=edge_attr,
      )
  
      n_tasks = {}
      for layer in hg.dependencies:
          n_tasks[layer] = len(hg.layers[layer])
  
      min_tasks = min(n_tasks.values())
      max_tasks = max(n_tasks.values())
  
      cache = {}
  
      color = kwargs.get("color")
      if color == "layer_type":
          layer_colors = {
              "DataFrameIOLayer": ["#CCC7F9", False],  # purple
              "ShuffleLayer": ["#F9CCC7", False],  # rose
              "SimpleShuffleLayer": ["#F9CCC7", False],  # rose
              "ArrayOverlayLayer": ["#FFD9F2", False],  # pink
              "BroadcastJoinLayer": ["#D9F2FF", False],  # blue
              "Blockwise": ["#D9FFE6", False],  # green
              "BlockwiseLayer": ["#D9FFE6", False],  # green
              "MaterializedLayer": ["#DBDEE5", False],  # gray
          }
  
      for layer in hg.dependencies:
          layer_name = name(layer)
          attrs = data_attributes.get(layer, {})
  
          node_label = label(layer, cache=cache)
          node_size = (
              20
              if max_tasks == min_tasks
              else int(
                  20
                  + (
                      (n_tasks[layer] - min_tasks)
                      / (max_tasks - min_tasks)
                  )
                  * 20
              )
          )
  
          layer_type = str(type(hg.layers[layer]).__name__)
          node_tooltips = f"A {layer_type.replace('Layer', '')} Layer with {n_tasks[layer]} Tasks.\n"
  
          layer_ca = hg.layers[layer].collection_annotations
          if layer_ca:
              if layer_ca.get("type") == "dask.array.core.Array":
                  node_tooltips += (
                      f"Array Shape: {layer_ca.get('shape')}\n"
                      f"Data Type: {layer_ca.get('dtype')}\n"
                      f"Chunk Size: {layer_ca.get('chunksize')}\n"
                      f"Chunk Type: {layer_ca.get('chunk_type')}\n"
                  )
  
              if (
                  layer_ca.get("type")
                  == "dask.dataframe.core.DataFrame"
              ):
                  dftype = {"pandas.core.frame.DataFrame": "pandas"}
                  cols = layer_ca.get("columns")
  
                  node_tooltips += (
                      f"Number of Partitions: {layer_ca.get('npartitions')}\n"
                      f"DataFrame Type: {dftype.get(layer_ca.get('dataframe_type'))}\n"
                      f"{len(cols)} DataFrame Columns: {str(cols) if len(str(cols)) <= 40 else '[...]'}\n"
                  )
  
          attrs.setdefault("label", str(node_label))
          attrs.setdefault("fontsize", str(node_size))
          attrs.setdefault("tooltip", str(node_tooltips))
  
          if color == "layer_type":
              node_color = layer_colors.get(layer_type)[0]
              layer_colors.get(layer_type)[1] = True
  
              attrs.setdefault("fillcolor", str(node_color))
              attrs.setdefault("style", "filled")
  
          g.node(layer_name, **attrs)
  
      for layer, deps in hg.dependencies.items():
          layer_name = name(layer)
          for dep in deps:
              dep_name = name(dep)
              g.edge(dep_name, layer_name)
  
      if color == "layer_type":
          legend_title = "Key"
  
          legend_label = (
              '<<TABLE BORDER="0" CELLBORDER="1" CELLSPACING="0" CELLPADDING="5">'
              "<TR><TD><B>Legend: Layer types</B></TD></TR>"
          )
  
          for layer_type, color in layer_colors.items():
              if color[1]:
                  legend_label += f'<TR><TD BGCOLOR="{color[0]}">{layer_type}</TD></TR>'
  
          legend_label += "</TABLE>>"
  
          attrs = data_attributes.get(legend_title, {})
          attrs.setdefault("label", str(legend_label))
          attrs.setdefault("fontsize", "20")
          attrs.setdefault("margin", "0")
  
          g.node(legend_title, **attrs)
  
      return g
  
imports: from __future__ import annotations
  import abc
  import copy
  import html
  from collections.abc import Collection, Hashable, ItemsView, Iterable, Iterator, KeysView, Mapping, Sequence, Set, ValuesView
  from typing import Any
  import tlz
  import dask
  from dask import config
  from dask.base import clone_key, flatten, is_dask_collection
  from dask.core import keys_in_tasks, reverse_dict
  from dask.typing import DaskCollection, Graph, Key
  from dask.utils import ensure_dict, import_required, key_split
  from dask.widgets import get_template
label_desc: Use image node_id 2 for calling a global function with example usage: to_graphviz(hg, data_attributes, function_attributes, rankdir, graph_attr, node_attr, edge_attr) and returns: g