class_docstr: Controls the setup and execution of a pipeline of processes.
class_docstr_tok: ["Controls","the","setup","and","execution","of","a","pipeline","of","processes","."]
docstr: Create a dot file with connection info
docstr_tok: ["Create","a","dot","file","with","connection","info"]
code: def _get_dot(
      self,
      prefix=None,
      hierarchy=None,
      colored=False,
      simple_form=True,
      level=0,
  ):
      """Create a dot file with connection info"""
      import networkx as nx
  
      if prefix is None:
          prefix = "  "
      if hierarchy is None:
          hierarchy = []
      colorset = [
          "#FFFFC8",  # Y
          "#0000FF",
          "#B4B4FF",
          "#E6E6FF",  # B
          "#FF0000",
          "#FFB4B4",
          "#FFE6E6",  # R
          "#00A300",
          "#B4FFB4",
          "#E6FFE6",  # G
          "#0000FF",
          "#B4B4FF",
      ]  # loop B
      if level > len(colorset) - 2:
          level = 3  # Loop back to blue
  
      dotlist = [f'{prefix}label="{self.name}";']
      for node in nx.topological_sort(self._graph):
          fullname = ".".join(hierarchy + [node.fullname])
          nodename = fullname.replace(".", "_")
          if not isinstance(node, Workflow):
              node_class_name = get_print_name(
                  node, simple_form=simple_form
              )
              if not simple_form:
                  node_class_name = ".".join(
                      node_class_name.split(".")[1:]
                  )
              if hasattr(node, "iterables") and node.iterables:
                  dotlist.append(
                      (
                          '%s[label="%s", shape=box3d,'
                          "style=filled, color=black, colorscheme"
                          "=greys7 fillcolor=2];"
                      )
                      % (nodename, node_class_name)
                  )
              else:
                  if colored:
                      dotlist.append(
                          (
                              '%s[label="%s", style=filled,'
                              ' fillcolor="%s"];'
                          )
                          % (nodename, node_class_name, colorset[level])
                      )
                  else:
                      dotlist.append(
                          ('%s[label="%s"];')
                          % (nodename, node_class_name)
                      )
  
      for node in nx.topological_sort(self._graph):
          if isinstance(node, Workflow):
              fullname = ".".join(hierarchy + [node.fullname])
              nodename = fullname.replace(".", "_")
              dotlist.append("subgraph cluster_%s {" % nodename)
              if colored:
                  dotlist.append(
                      prefix
                      + prefix
                      + 'edge [color="%s"];' % (colorset[level + 1])
                  )
                  dotlist.append(prefix + prefix + "style=filled;")
                  dotlist.append(
                      prefix
                      + prefix
                      + 'fillcolor="%s";' % (colorset[level + 2])
                  )
              dotlist.append(
                  node._get_dot(
                      prefix=prefix + prefix,
                      hierarchy=hierarchy + [self.name],
                      colored=colored,
                      simple_form=simple_form,
                      level=level + 3,
                  )
              )
              dotlist.append("}")
          else:
              for subnode in self._graph.successors(node):
                  if node._hierarchy != subnode._hierarchy:
                      continue
                  if not isinstance(subnode, Workflow):
                      nodefullname = ".".join(
                          hierarchy + [node.fullname]
                      )
                      subnodefullname = ".".join(
                          hierarchy + [subnode.fullname]
                      )
                      nodename = nodefullname.replace(".", "_")
                      subnodename = subnodefullname.replace(".", "_")
                      for _ in self._graph.get_edge_data(node, subnode)[
                          "connect"
                      ]:
                          dotlist.append(
                              f"{nodename} -> {subnodename};"
                          )
                      logger.debug("connection: %s", dotlist[-1])
      # add between workflow connections
      for u, v, d in self._graph.edges(data=True):
          uname = ".".join(hierarchy + [u.fullname])
          vname = ".".join(hierarchy + [v.fullname])
          for src, dest in d["connect"]:
              uname1 = uname
              vname1 = vname
              if isinstance(src, tuple):
                  srcname = src[0]
              else:
                  srcname = src
              if "." in srcname:
                  uname1 += "." + ".".join(srcname.split(".")[:-1])
              if "." in dest and "@" not in dest:
                  if not isinstance(v, Workflow):
                      if (
                          "datasink"
                          not in str(v._interface.__class__).lower()
                      ):
                          vname1 += "." + ".".join(dest.split(".")[:-1])
                  else:
                      vname1 += "." + ".".join(dest.split(".")[:-1])
              if uname1.split(".")[:-1] != vname1.split(".")[:-1]:
                  dotlist.append(
                      "%s -> %s;"
                      % (
                          uname1.replace(".", "_"),
                          vname1.replace(".", "_"),
                      )
                  )
                  logger.debug("cross connection: %s", dotlist[-1])
      return ("\n" + prefix).join(dotlist)
  
imports: import os
  import os.path
  import sys
  from datetime import datetime
  from copy import deepcopy
  import pickle
  import shutil
  import numpy
  from ...None import config, logging
  from ...utils.misc import str2bool
  from ...utils.functions import getsource, create_function_from_source
  from ...interfaces.base import traits, TraitedSpec, TraitDictObject, TraitListObject
  from ...utils.filemanip import save_json
  from .utils import generate_expanded_graph, export_graph, write_workflow_prov, write_workflow_resources, format_dot, topological_sort, get_print_name, merge_dict, format_node
  from .base import EngineBase
  from .nodes import MapNode
label_desc: Use image node_id 32 for calling the Workflow obj's underlying member method code with example usage: obj._get_dot(prefix, hierarchy, colored, simple_form, level) and returns: unknown