def _validate()

in tensorflow_fold/blocks/blocks.py [0:0]

• 47 lines of code
• 30 McCabe index (conditional complexity)

  def _validate(self, compiler_ctx):
    # pylint: disable=protected-access
    # Sort children in topological order, and detect cycles.
    topo_children = []
    visited = {}

    def visit(child):
      """Depth-first traversal."""
      if child is self._input_ph:
        return
      visited[child] = 1    # mark as visiting for cycle detection
      if child in self._child_input_wire_dict:
        for (b, _) in self._child_input_wire_dict[child]:
          if b in visited:
            if visited[b] == 1:
              raise ValueError('Composition cannot have cycles.')
          else:
            visit(b)
      topo_children.append(child)
      visited[child] = 2    # mark as visited

    sinks = set(self.children)
    for input_wires in six.itervalues(self._child_input_wire_dict):
      for wire in input_wires:
        sinks.discard(wire.block)
    if self._output_wires:
      for source, _ in self._output_wires:
        if source not in visited: visit(source)

    # Reorder sinks to match children to avoid nondeterminism.
    sinks = [c for c in self.children if c in sinks]

    # Find any children that were not reachable from the output.
    for child in sinks:
      assert child not in visited
      visit(child)

    self._children = topo_children

    # TODO(delesley): write test cases for error conditions
    self._child_input_wires = [[]]
    self._child_to_index[self.input] = 0
    for (i, b) in enumerate(self.children, 1):
      # Find the list of input wires for child b
      self._child_to_index[b] = i
      in_wires = self._child_input_wire_dict.get(b, [])
      # Set _child_input_wires for fast lookup during evaluation
      self._child_input_wires.append(in_wires)
      # Validate child, which will ensure it has an output type.
      b._validate(compiler_ctx)

    # Check that all child outputs are used.
    unused = [b for b in sinks if not isinstance(b.output_type, tdt.VoidType)]
    if unused:
      raise TypeError('children have unused outputs: %s' %
                      ', '.join(str(u) for u in unused))

    # Corner case: set input type to Void if all children have Void input.
    if self.input_type is None:
      if all([isinstance(child.input_type, tdt.VoidType)
              for child in self._children]):
        self.set_input_type(tdt.VoidType())

    # Check that composition has an input type.
    self._check_input_type()

    # Infer the composition output type.
    if not (self._output_wires or isinstance(self.output_type, tdt.VoidType)):
      # We could infer void output type here but more likely the user made a
      # mistake, so we throw unless the VoidType was set explicitly.
      raise TypeError('Composition block has no output: %s' % self)
    self._check_output_type()