(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.
Constructs a DAGPipeline.
Constructs a DAGPipeline .
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
def __init__ ( self , dag , pipeline_name = 'DAGPipeline' ) : # 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