'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:
InvalidDAGException: If each key value pair in the `dag` dictionary is
not of the form
(Pipeline or DagOutput): (Pipeline, PipelineKey, or DagInput).
TypeMismatchException: The type signature of each key and value in `dag`
must match, otherwise this will be thrown.
DuplicateNameException: If two `Pipeline` instances in `dag` have the
same string name.
BadInputOrOutputException: If there are no `DagOutput` instaces in `dag`
or not exactly one `DagInput` plus type combination in `dag`.
InvalidDictionaryOutput: If `DagOutput()` is not connected to a
dictionary, or `DagOutput(name)` is not connected to a Pipeline,
PipelineKey, or DagInput instance.
NotConnectedException: If a `Pipeline` used in a dependency has nothing
feeding into it, or a `Pipeline` used as a destination does not feed
anywhere.
BadTopologyException: If there there is a directed cycle in `dag`.
Exception: Misc. exceptions.'

def __init__(self, dag, pipeline_name='DAGPipeline'):

    self.dag = dict(self._expand_dag_shorthands(dag))
    for (unit, dependency) in self.dag.items():
        if (not isinstance(unit, (pipeline.Pipeline, DagOutput))):
            raise InvalidDAGException(('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, basestring) for name in dependency])):
                raise InvalidDAGException(('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) or (isinstance(subordinate, pipeline.PipelineKey) and isinstance(subordinate.unit, pipeline.Pipeline)) or isinstance(subordinate, DagInput))):
                raise InvalidDAGException(('Dependency    {%s:    %s}    is    invalid.    Right    hand    side    subordinate    %s    must    be    either    a    Pipeline,    PipelineKey,    or    DagInput    object' % (unit, dependency, subordinate)))
        if isinstance(unit, DagOutput):
            continue
        if (unit.input_type != self._get_type_signature_for_dependency(dependency)):
            raise TypeMismatchException(('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))))
    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 DuplicateNameException(('Pipelines    %s    and    %s    both    have    name    "%s".    Each    Pipeline    must    have    a    unique    name.' % (unit, other_unit, name)))
    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 BadInputOrOutputException('No    DagInput    object    found.    DagInput    is    the    start    of    the    pipeline.')
        else:
            raise BadInputOrOutputException('Multiple    DagInput    objects    found.    Only    one    input    is    supported.')
    if (not self.outputs):
        raise BadInputOrOutputException('No    DagOutput    objects    found.    DagOutput    is    the    end    of    the    pipeline.')
    self.input = inputs.pop()
    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)
    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 NotConnectedException(('%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 NotConnectedException(('%s    is    given    as    a    destination    in    the    DAG    but    does    not    output    anywhere.    It    is    a    deadend.' % units_with_no_output.pop()))
    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 = []
    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)))
    for unit in graph:
        if (graph[unit][1] != 0):
            raise BadTopologyException(('Dependency    loop    found    on    %s' % unit))
    if (set(call_list) != set(((list(all_subordinates) + self.outputs) + [self.input]))):
        raise BadTopologyException('Not    all    pipelines    feed    into    an    output    or    there    is    a    dependency    loop.')
    call_list.reverse()
    assert (call_list[0] == self.input)
