class_docstr: A directed acyclic graph pipeline.
  
  This Pipeline can be given an arbitrary graph composed of Pipeline instances
  and will run all of those pipelines feeding outputs to inputs. See README.md
  for details.
  
  Use DAGPipeline to compose multiple smaller pipelines together.
class_docstr_tok: ["A","directed","acyclic","graph","pipeline",".","This","Pipeline","can","be","given","an","arbitrary","graph","composed","of","Pipeline","instances","and","will","run","all","of","those","pipelines","feeding","outputs","to","inputs",".","See","README.md","for","details",".","Use","DAGPipeline","to","compose","multiple","smaller","pipelines","together","."]
docstr: Constructs a DAGPipeline.
  
  A DAG (direct acyclic graph) is given which fully specifies what the
  DAGPipeline runs.
  
  Args:
    dag: A dictionary mapping `Pipeline` or `DagOutput` instances to any of
       `Pipeline`, `PipelineKey`, `DagInput`. `dag` defines a directed acyclic
       graph.
    pipeline_name: String name of this Pipeline object.
  
  Raises:
    InvalidDAGError: If each key value pair in the `dag` dictionary is
        not of the form
        (Pipeline or DagOutput): (Pipeline, PipelineKey, or DagInput).
    TypeMismatchError: The type signature of each key and value in `dag`
        must match, otherwise this will be thrown.
    DuplicateNameError: If two `Pipeline` instances in `dag` have the
        same string name.
    BadInputOrOutputError: If there are no `DagOutput` instaces in `dag`
        or not exactly one `DagInput` plus type combination in `dag`.
    InvalidDictionaryOutputError: If `DagOutput()` is not connected to a
        dictionary, or `DagOutput(name)` is not connected to a Pipeline,
        PipelineKey, or DagInput instance.
    NotConnectedError: If a `Pipeline` used in a dependency has nothing
        feeding into it, or a `Pipeline` used as a destination does not feed
        anywhere.
    BadTopologyError: If there there is a directed cycle in `dag`.
    Exception: Misc. exceptions.
docstr_tok: ["Constructs","a","DAGPipeline",".","A","DAG","(","direct","acyclic","graph",")","is","given","which","fully","specifies","what","the","DAGPipeline","runs",".","Args",":","dag",":","A","dictionary","mapping","`","Pipeline","`","or","`","DagOutput","`","instances","to","any","of","`","Pipeline","`",",","`","PipelineKey","`",",","`","DagInput","`",".","`","dag","`","defines","a","directed","acyclic","graph",".","pipeline_name",":","String","name","of","this","Pipeline","object",".","Raises",":","InvalidDAGError",":","If","each","key","value","pair","in","the","`","dag","`","dictionary","is","not","of","the","form","(","Pipeline","or","DagOutput",")",":","(","Pipeline",",","PipelineKey",",","or","DagInput",")",".","TypeMismatchError",":","The","type","signature","of","each","key","and","value","in","`","dag","`","must","match",",","otherwise","this","will","be","thrown",".","DuplicateNameError",":","If","two","`","Pipeline","`","instances","in","`","dag","`","have","the","same","string","name",".","BadInputOrOutputError",":","If","there","are","no","`","DagOutput","`","instaces","in","`","dag","`","or","not","exactly","one","`","DagInput","`","plus","type","combination","in","`","dag","`",".","InvalidDictionaryOutputError",":","If","`","DagOutput","(",")","`","is","not","connected","to","a","dictionary",",","or","`","DagOutput","(","name",")","`","is","not","connected","to","a","Pipeline",",","PipelineKey",",","or","DagInput","instance",".","NotConnectedError",":","If","a","`","Pipeline","`","used","in","a","dependency","has","nothing","feeding","into","it",",","or","a","`","Pipeline","`","used","as","a","destination","does","not","feed","anywhere",".","BadTopologyError",":","If","there","there","is","a","directed","cycle","in","`","dag","`",".","Exception",":","Misc",".","exceptions","."]
code: def __init__(self, dag, pipeline_name="DAGPipeline"):
      """Constructs a DAGPipeline.
  
      A DAG (direct acyclic graph) is given which fully specifies what the
      DAGPipeline runs.
  
      Args:
        dag: A dictionary mapping `Pipeline` or `DagOutput` instances to any of
           `Pipeline`, `PipelineKey`, `DagInput`. `dag` defines a directed acyclic
           graph.
        pipeline_name: String name of this Pipeline object.
  
      Raises:
        InvalidDAGError: If each key value pair in the `dag` dictionary is
            not of the form
            (Pipeline or DagOutput): (Pipeline, PipelineKey, or DagInput).
        TypeMismatchError: The type signature of each key and value in `dag`
            must match, otherwise this will be thrown.
        DuplicateNameError: If two `Pipeline` instances in `dag` have the
            same string name.
        BadInputOrOutputError: If there are no `DagOutput` instaces in `dag`
            or not exactly one `DagInput` plus type combination in `dag`.
        InvalidDictionaryOutputError: If `DagOutput()` is not connected to a
            dictionary, or `DagOutput(name)` is not connected to a Pipeline,
            PipelineKey, or DagInput instance.
        NotConnectedError: If a `Pipeline` used in a dependency has nothing
            feeding into it, or a `Pipeline` used as a destination does not feed
            anywhere.
        BadTopologyError: If there there is a directed cycle in `dag`.
        Exception: Misc. exceptions.
      """
      # Expand DAG shorthand.
      self.dag = dict(self._expand_dag_shorthands(dag))
  
      # Make sure DAG is valid.
      # DagInput types match output types. Nothing depends on outputs.
      # Things that require input get input. DAG is composed of correct types.
      for unit, dependency in self.dag.items():
          if not isinstance(unit, (pipeline.Pipeline, DagOutput)):
              raise InvalidDAGError(
                  "Dependency {%s: %s} is invalid. Left hand side value %s must "
                  "either be a Pipeline or DagOutput object"
                  % (unit, dependency, unit)
              )
          if isinstance(dependency, dict):
              if not all(
                  isinstance(name, six.string_types)
                  for name in dependency
              ):
                  raise InvalidDAGError(
                      "Dependency {%s: %s} is invalid. Right hand side keys %s must be "
                      "strings" % (unit, dependency, dependency.keys())
                  )
              values = dependency.values()
          else:
              values = [dependency]
          for subordinate in values:
              if not (
                  isinstance(subordinate, (pipeline.Pipeline, DagInput))
                  or (
                      isinstance(subordinate, pipeline.PipelineKey)
                      and isinstance(
                          subordinate.unit, pipeline.Pipeline
                      )
                  )
              ):
                  raise InvalidDAGError(
                      "Dependency {%s: %s} is invalid. Right hand side subordinate %s "
                      "must be either a Pipeline, PipelineKey, or DagInput object"
                      % (unit, dependency, subordinate)
                  )
  
          # Check that all input types match output types.
          if isinstance(unit, DagOutput):
              # DagOutput objects don't know their types.
              continue
          if (
              unit.input_type
              != self._get_type_signature_for_dependency(dependency)
          ):
              raise TypeMismatchError(
                  "Invalid dependency {%s: %s}. Required `input_type` of left hand "
                  "side is %s. DagOutput type of right hand side is %s."
                  % (
                      unit,
                      dependency,
                      unit.input_type,
                      self._get_type_signature_for_dependency(
                          dependency
                      ),
                  )
              )
  
      # Make sure all Pipeline names are unique, so that Statistic objects don't
      # clash.
      sorted_unit_names = sorted(
          [(unit, unit.name) for unit in self.dag], key=lambda t: t[1]
      )
      for index, (unit, name) in enumerate(sorted_unit_names[:-1]):
          if name == sorted_unit_names[index + 1][1]:
              other_unit = sorted_unit_names[index + 1][0]
              raise DuplicateNameError(
                  'Pipelines %s and %s both have name "%s". Each Pipeline must have '
                  "a unique name." % (unit, other_unit, name)
              )
  
      # Find DagInput and DagOutput objects and make sure they are being used
      # correctly.
      self.outputs = [
          unit for unit in self.dag if isinstance(unit, DagOutput)
      ]
      self.output_names = dict(
          (output.name, output) for output in self.outputs
      )
      for output in self.outputs:
          output.input_type = (
              output.output_type
          ) = self._get_type_signature_for_dependency(self.dag[output])
      inputs = set()
      for deps in self.dag.values():
          units = self._get_units(deps)
          for unit in units:
              if isinstance(unit, DagInput):
                  inputs.add(unit)
      if len(inputs) != 1:
          if not inputs:
              raise BadInputOrOutputError(
                  "No DagInput object found. DagInput is the start of the pipeline."
              )
          else:
              raise BadInputOrOutputError(
                  "Multiple DagInput objects found. Only one input is supported."
              )
      if not self.outputs:
          raise BadInputOrOutputError(
              "No DagOutput objects found. DagOutput is the end of the pipeline."
          )
      self.input = inputs.pop()
  
      # Compute output_type for self and call super constructor.
      output_signature = dict(
          (output.name, output.output_type) for output in self.outputs
      )
      super(DAGPipeline, self).__init__(
          input_type=self.input.output_type,
          output_type=output_signature,
          name=pipeline_name,
      )
  
      # Make sure all Pipeline objects have DAG vertices that feed into them,
      # and feed their output into other DAG vertices.
      all_subordinates = set(
          dep_unit
          for unit in self.dag
          for dep_unit in self._get_units(self.dag[unit])
      ).difference(set([self.input]))
      all_destinations = set(self.dag.keys()).difference(
          set(self.outputs)
      )
      if all_subordinates != all_destinations:
          units_with_no_input = all_subordinates.difference(
              all_destinations
          )
          units_with_no_output = all_destinations.difference(
              all_subordinates
          )
          if units_with_no_input:
              raise NotConnectedError(
                  "%s is given as a dependency in the DAG but has nothing connected "
                  "to it. Nothing in the DAG feeds into it."
                  % units_with_no_input.pop()
              )
          else:
              raise NotConnectedError(
                  "%s is given as a destination in the DAG but does not output "
                  "anywhere. It is a deadend."
                  % units_with_no_output.pop()
              )
  
      # Construct topological ordering to determine the execution order of the
      # pipelines.
      # https://en.wikipedia.org/wiki/Topological_sorting#Kahn.27s_algorithm
  
      # `graph` maps a pipeline to the pipelines it depends on. Each dict value
      # is a list with the dependency pipelines in the 0th position, and a count
      # of forward connections to the key pipeline (how many pipelines use this
      # pipeline as a dependency).
      graph = dict(
          (unit, [self._get_units(self.dag[unit]), 0])
          for unit in self.dag
      )
      graph[self.input] = [[], 0]
      for unit, (forward_connections, _) in graph.items():
          for to_unit in forward_connections:
              graph[to_unit][1] += 1
      self.call_list = (
          call_list
      ) = []  # Topologically sorted elements go here.
      nodes = set(self.outputs)
      while nodes:
          n = nodes.pop()
          call_list.append(n)
          for m in graph[n][0]:
              graph[m][1] -= 1
              if graph[m][1] == 0:
                  nodes.add(m)
              elif graph[m][1] < 0:
                  raise Exception(
                      "Congratulations, you found a bug! Please report this issue at "
                      "https://github.com/tensorflow/magenta/issues and copy/paste the "
                      "following: dag=%s, graph=%s, call_list=%s"
                      % (self.dag, graph, call_list)
                  )
      # Check for cycles by checking if any edges remain.
      for unit in graph:
          if graph[unit][1] != 0:
              raise BadTopologyError(
                  "Dependency loop found on %s" % unit
              )
  
      # Note: this exception should never be raised. Disconnected graphs will be
      # caught where NotConnectedError is raised. If this exception goes off
      # there is likely a bug.
      if set(call_list) != set(
          list(all_subordinates) + self.outputs + [self.input]
      ):
          raise BadTopologyError(
              "Not all pipelines feed into an output or "
              "there is a dependency loop."
          )
  
      call_list.reverse()
      assert call_list[0] == self.input
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import itertools
  from magenta.pipelines import pipeline
  import six
label_desc: Use image node_id 1 to create a new DAGPipeline object from inherited base classes: pipeline with example: obj = DAGPipeline(dag, pipeline_name)