def analyze_tensor()

in tensorflow_transform/graph_tools.py [0:0]

• 26 lines of code
• 12 McCabe index (conditional complexity)

  def analyze_tensor(self, tensor_or_op):
    """Analyzes the `tensor_or_op` for its dependencies and readiness.

    Computes the transitive dependencies of a tensor or operation and decides
    whether it is ready to run using iterative DFS. `source_info_dict` are used
    as terminal nodes.  An error is thrown if a table or placeholder is reached:
    they must be set using source_info_dict. This function is memoized using the
    _memoized_analyze_tensor_result cache. Cycles are ignored (so a cycle is
    considered ready to run).

    Args:
      tensor_or_op: A `Tensor` or `Operation`.

    Returns:
      An _AnalysisResult which includes whether this op or tensor is ready to
      run, a path from it to its sources and its dependent sources from
      `source_info_dict`.

    Raises:
      _UnexpectedTableError: If an initializable table op is encountered.
      _UnexpectedPlaceholderError: If a placeholder is encountered.
    """
    stack = collections.deque()
    # Note that because tensors are no longer hashable, we need to convert to
    # their reference in order to use them in sets or dicts.
    stack.append(tf_utils.hashable_tensor_or_op(tensor_or_op))
    # Contains the nodes of the path starting from tensor_or_op to current
    # visiting node, used for loop detection. We assume that any loop is a
    # valid while loop and so it will be able to run as long as all the other
    # parents are ready.
    path = set()
    while stack:
      current = stack[-1]
      if current in self._memoized_analyze_tensor_result:
        stack.pop()
        continue
      path.add(current)
      parents = self._get_parents(tf_utils.deref_tensor_or_op(current))
      parents = [parent for parent in map(tf_utils.hashable_tensor_or_op,
                                          parents) if parent not in path]
      if all(
          parent in self._memoized_analyze_tensor_result for parent in parents):
        parent_results = [
            self._memoized_analyze_tensor_result[parent] for parent in parents
        ]
        current_result = self._compute_analysis_result(
            tf_utils.deref_tensor_or_op(current), parent_results)
        self._memoized_analyze_tensor_result[current] = current_result
        path.discard(stack.pop())
      else:
        stack.extend(parents)
    return self._memoized_analyze_tensor_result[tf_utils.hashable_tensor_or_op(
        tensor_or_op)]