code: def _get_execution_specs(
      graph_def: tf.compat.v1.GraphDef,
      graph_output_names: List[str],
      graph: tf.Graph,
      node_name_to_node_def: Mapping[str, tf.compat.v1.NodeDef],
      remote_op_to_immediate_dep: Mapping[str, List[str]],
  ) -> List[execution_spec.ExecutionSpec]:
      """Generates the ExecutionSpecs for a graph.
  
      A "layer" contains one or more nodes inside a graph. There are two types of
      layers: subgraph layer and remote op layer. A subgraph layer doesn't
      contain remote ops, whereas a remote op layer only contains remote ops.
      Remote ops inside a remote op layer don't depend on each other's output,
      so the order of execution between those remote ops doesn't matter.
  
      We first identify the remote op layers of a graph. Then, based on the
      remote op layers, we can derive the subgraph layers. For example, after
      identifying the first remote op layer, we can equate the inputs of the
      remote op layer to the outputs of the previous subgraph layer. We can
      then traverse and construct the previous subgraph layer.
  
      Each subgraph layer can be captured into one ExecutionSpec, but each
      remote op layer need to be stored into N ExecutionSpecs, where N equals
      to the number of remote ops inside a remote op layer. This happens
      because each remote op essentially represents a graph.
  
      Args:
        graph_def: A `GraphDef` proto.
        graph_output_names: A list of graph output node names.
        graph: A tf.Graph representing the same graph as graph_def.
        node_name_to_node_def: A mapping from node names to `NodeDef` protos.
        remote_op_to_immediate_dep: A mapping from remote op name to a list of
          remote op immediate dependencies' names.
  
      Returns:
        A list of ExecutionSpecs, where the order of the list represents the
        order of execution.
      """
      execution_specs = []  # type: List[execution_spec.ExecutionSpec]
      previously_visited = set()  # type: Set[str]
  
      for remote_op_layer in _RemoteOpLayers(
          remote_op_to_immediate_dep
      ):
          # Get one subgraph layer
          output_node_names = (
              _get_previous_subgraph_layer_output_node_names(
                  remote_op_layer, node_name_to_node_def
              )
          )
  
          if output_node_names:
              spec = _get_execution_spec_for_subgraph_layer(
                  graph_def,
                  graph,
                  node_name_to_node_def,
                  previously_visited,
                  output_node_names,
              )
              execution_specs.append(spec)
              previously_visited |= _get_non_input_names(spec.subgraph)
  
          # Get one remote op layer
          specs = _get_execution_specs_for_remote_op_layer(
              remote_op_layer, node_name_to_node_def
          )
          execution_specs.extend(specs)
  
      # Get the last subgraph layer
      output_node_names = set(graph_output_names)
      spec = _get_execution_spec_for_subgraph_layer(
          graph_def,
          graph,
          node_name_to_node_def,
          previously_visited,
          output_node_names,
      )
      execution_specs.append(spec)
  
      return execution_specs
  
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 11 for calling a global function with example usage: _get_execution_specs(graph_def, graph_output_names, graph, node_name_to_node_def, remote_op_to_immediate_dep) and returns: execution_specs