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: Connect nodes in the pipeline.
  
  This routine also checks if inputs and outputs are actually provided by
  the nodes that are being connected.
  
  Creates edges in the directed graph using the nodes and edges specified
  in the `connection_list`.  Uses the NetworkX method
  DiGraph.add_edges_from.
  
  Parameters
  ----------
  
  args : list or a set of four positional arguments
  
      Four positional arguments of the form::
  
        connect(source, sourceoutput, dest, destinput)
  
      source : nodewrapper node
      sourceoutput : string (must be in source.outputs)
      dest : nodewrapper node
      destinput : string (must be in dest.inputs)
  
      A list of 3-tuples of the following form::
  
       [(source, target,
           [('sourceoutput/attribute', 'targetinput'),
           ...]),
       ...]
  
      Or::
  
       [(source, target, [(('sourceoutput1', func, arg2, ...),
                                   'targetinput'), ...]),
       ...]
       sourceoutput1 will always be the first argument to func
       and func will be evaluated and the results sent to targetinput
  
       currently func needs to define all its needed imports within the
       function as we use the inspect module to get at the source code
       and execute it remotely
docstr_tok: ["Connect","nodes","in","the","pipeline",".","This","routine","also","checks","if","inputs","and","outputs","are","actually","provided","by","the","nodes","that","are","being","connected",".","Creates","edges","in","the","directed","graph","using","the","nodes","and","edges","specified","in","the","`","connection_list","`",".","Uses","the","NetworkX","method","DiGraph.add_edges_from",".","Parameters","--","--","--","--","--","args",":","list","or","a","set","of","four","positional","arguments","Four","positional","arguments","of","the","form",":",":","connect","(","source",",","sourceoutput",",","dest",",","destinput",")","source",":","nodewrapper","node","sourceoutput",":","string","(","must","be","in","source.outputs",")","dest",":","nodewrapper","node","destinput",":","string","(","must","be","in","dest.inputs",")","A","list","of","3-tuples","of","the","following","form",":",":","[","(","source",",","target",",","[","(","'sourceoutput\/attribute","'",",","'targetinput","'",")",",","...","]",")",",","...","]","Or",":",":","[","(","source",",","target",",","[","(","(","'sourceoutput1","'",",","func",",","arg2",",","...",")",",","'targetinput","'",")",",","...","]",")",",","...","]","sourceoutput1","will","always","be","the","first","argument","to","func","and","func","will","be","evaluated","and","the","results","sent","to","targetinput","currently","func","needs","to","define","all","its","needed","imports","within","the","function","as","we","use","the","inspect","module","to","get","at","the","source","code","and","execute","it","remotely"]
code: def connect(self, *args, **kwargs):
      """Connect nodes in the pipeline.
  
      This routine also checks if inputs and outputs are actually provided by
      the nodes that are being connected.
  
      Creates edges in the directed graph using the nodes and edges specified
      in the `connection_list`.  Uses the NetworkX method
      DiGraph.add_edges_from.
  
      Parameters
      ----------
  
      args : list or a set of four positional arguments
  
          Four positional arguments of the form::
  
            connect(source, sourceoutput, dest, destinput)
  
          source : nodewrapper node
          sourceoutput : string (must be in source.outputs)
          dest : nodewrapper node
          destinput : string (must be in dest.inputs)
  
          A list of 3-tuples of the following form::
  
           [(source, target,
               [('sourceoutput/attribute', 'targetinput'),
               ...]),
           ...]
  
          Or::
  
           [(source, target, [(('sourceoutput1', func, arg2, ...),
                                       'targetinput'), ...]),
           ...]
           sourceoutput1 will always be the first argument to func
           and func will be evaluated and the results sent to targetinput
  
           currently func needs to define all its needed imports within the
           function as we use the inspect module to get at the source code
           and execute it remotely
      """
      if len(args) == 1:
          connection_list = args[0]
      elif len(args) == 4:
          connection_list = [(args[0], args[2], [(args[1], args[3])])]
      else:
          raise TypeError(
              "connect() takes either 4 arguments, or 1 list of"
              " connection tuples (%d args given)" % len(args)
          )
  
      disconnect = False
      if kwargs:
          disconnect = kwargs.get("disconnect", False)
  
      if disconnect:
          self.disconnect(connection_list)
          return
  
      newnodes = set()
      for srcnode, destnode, _ in connection_list:
          if self in [srcnode, destnode]:
              msg = (
                  "Workflow connect cannot contain itself as node:"
                  " src[%s] dest[%s] workflow[%s]"
              ) % (srcnode, destnode, self.name)
  
              raise OSError(msg)
          if (srcnode not in newnodes) and not self._has_node(srcnode):
              newnodes.add(srcnode)
          if (destnode not in newnodes) and not self._has_node(
              destnode
          ):
              newnodes.add(destnode)
      if newnodes:
          self._check_nodes(newnodes)
          for node in newnodes:
              if node._hierarchy is None:
                  node._hierarchy = self.name
      not_found = []
      connected_ports = {}
      for srcnode, destnode, connects in connection_list:
          if destnode not in connected_ports:
              connected_ports[destnode] = set()
          # check to see which ports of destnode are already
          # connected.
          if not disconnect and (destnode in self._graph.nodes()):
              for edge in self._graph.in_edges(destnode):
                  data = self._graph.get_edge_data(*edge)
                  connected_ports[destnode].update(
                      destname for _, destname in data["connect"]
                  )
          for source, dest in connects:
              # Currently datasource/sink/grabber.io modules
              # determine their inputs/outputs depending on
              # connection settings.  Skip these modules in the check
              if dest in connected_ports[destnode]:
                  raise Exception(
                      """\
  Trying to connect %s:%s to %s:%s but input '%s' of node '%s' is already
  connected.
  """
                      % (
                          srcnode,
                          source,
                          destnode,
                          dest,
                          dest,
                          destnode,
                      )
                  )
              if not (
                  hasattr(destnode, "_interface")
                  and (
                      ".io" in str(destnode._interface.__class__)
                      or any(
                          [
                              ".io" in str(val)
                              for val in destnode._interface.__class__.__bases__
                          ]
                      )
                  )
              ):
                  if not destnode._check_inputs(dest):
                      not_found.append(["in", destnode.name, dest])
              if not (
                  hasattr(srcnode, "_interface")
                  and (
                      ".io" in str(srcnode._interface.__class__)
                      or any(
                          [
                              ".io" in str(val)
                              for val in srcnode._interface.__class__.__bases__
                          ]
                      )
                  )
              ):
                  if isinstance(source, tuple):
                      # handles the case that source is specified
                      # with a function
                      sourcename = source[0]
                  elif isinstance(source, (str, bytes)):
                      sourcename = source
                  else:
                      raise Exception(
                          (
                              "Unknown source specification in "
                              "connection from output of %s"
                          )
                          % srcnode.name
                      )
                  if sourcename and not srcnode._check_outputs(
                      sourcename
                  ):
                      not_found.append(
                          ["out", srcnode.name, sourcename]
                      )
              connected_ports[destnode].add(dest)
      infostr = []
      for info in not_found:
          infostr += [
              f"Module {info[1]} has no {info[0]}put called {info[2]}\n"
          ]
      if not_found:
          raise Exception(
              "\n".join(["Some connections were not found"] + infostr)
          )
  
      # turn functions into strings
      for srcnode, destnode, connects in connection_list:
          for idx, (src, dest) in enumerate(connects):
              if isinstance(src, tuple) and not isinstance(
                  src[1], (str, bytes)
              ):
                  function_source = getsource(src[1])
                  connects[idx] = (
                      (src[0], function_source, src[2:]),
                      dest,
                  )
  
      # add connections
      for srcnode, destnode, connects in connection_list:
          edge_data = self._graph.get_edge_data(srcnode, destnode, None)
          if edge_data:
              logger.debug(
                  "(%s, %s): Edge data exists: %s",
                  srcnode,
                  destnode,
                  str(edge_data),
              )
              for data in connects:
                  if data not in edge_data["connect"]:
                      edge_data["connect"].append(data)
                  if disconnect:
                      logger.debug("Removing connection: %s", str(data))
                      edge_data["connect"].remove(data)
              if edge_data["connect"]:
                  self._graph.add_edges_from(
                      [(srcnode, destnode, edge_data)]
                  )
              else:
                  # pass
                  logger.debug(
                      "Removing connection: %s->%s", srcnode, destnode
                  )
                  self._graph.remove_edges_from([(srcnode, destnode)])
          elif not disconnect:
              logger.debug("(%s, %s): No edge data", srcnode, destnode)
              self._graph.add_edges_from(
                  [(srcnode, destnode, {"connect": connects})]
              )
          edge_data = self._graph.get_edge_data(srcnode, destnode)
          logger.debug(
              "(%s, %s): new edge data: %s",
              srcnode,
              destnode,
              str(edge_data),
          )
  
      if newnodes:
          self._update_node_cache()
  
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 3 for calling the Workflow obj's underlying member method code with example usage: obj.connect() without return types