code: def _get_execution_spec_for_subgraph_layer(
      graph_def: tf.compat.v1.GraphDef,
      graph: tf.Graph,
      node_name_to_node_def: Mapping[str, tf.compat.v1.NodeDef],
      previously_visited: Set[str],
      output_node_names: Set[str],
  ) -> execution_spec.ExecutionSpec:
      """Constructs one subgraph layer.
  
      As discussed in _get_execution_specs(), a subgraph layer contains one or
      more nodes excluding remote ops. Based on a set of output node names, we
      traverse toward the ancestors (upward) until encountering a "special" node.
  
      Here, we traverse upward because each node's node_def contains input names
      but not output names.
  
      A "special" node could be either a placeholder node, a remote op, or a node
      visited by a previous layer. Since it is computed/stored prior to the
      current subgraph layer, we can treat it as an input of the current subgraph
      layer.
  
      Args:
        graph_def: A `GraphDef` proto for the original graph.
        graph: A tf.Graph instance for the original graph.
        node_name_to_node_def: A mapping from node names to `NodeDef` protos.
        previously_visited: A set of node names from previous subgraph layers.
        output_node_names: A set of output node names for the current subgraph.
  
      Returns:
        An ExecutionSpec representing a subgraph layer.
      """
      subgraph = tf.compat.v1.GraphDef()
      subgraph.versions.CopyFrom(graph_def.versions)
      subgraph.library.CopyFrom(graph_def.library)
  
      queue = collections.deque(output_node_names)
      visited = set()
  
      while queue:
          current_node_name = queue.popleft()
          current_node = node_name_to_node_def[current_node_name]
  
          if current_node_name not in visited:
              visited.add(current_node_name)
  
              if (
                  _is_remote_op(current_node)
                  or _is_placeholder_op(current_node)
                  or current_node_name in previously_visited
              ):
                  # These ops must be computed before this subgraph layer. Hence,
                  # we treat them as placeholder inputs.
                  placeholder_node = (
                      _create_placeholder_node_from_existing_node(
                          current_node, graph
                      )
                  )
                  subgraph.node.append(placeholder_node)
  
              else:
                  subgraph.node.append(current_node)
                  queue.extend(
                      node_name_to_node_def[current_node_name].input
                  )
  
      return execution_spec.ExecutionSpec(
          subgraph=subgraph,
          input_names=_get_input_names(subgraph),
          output_names=set(output_node_names),
          is_remote_op=False,
      )
  
imports: import collections
  from typing import Dict, List, Mapping, Set
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.experimental.distributed_inference.graphdef_experiments.subgraph_partitioning import execution_spec
label_desc: Use image node_id 13 for calling a global function with example usage: _get_execution_spec_for_subgraph_layer(graph_def, graph, node_name_to_node_def, previously_visited, output_node_names) and returns: execution_spec