code: def combine_steps(
      g: Plottable,
      kind: str,
      steps: List[Tuple[ASTObject, Plottable]],
      engine: Engine,
  ) -> DataFrameT:
      """
      Collect nodes and edges, taking care to deduplicate and tag any names
      """
  
      id = getattr(g, "_node" if kind == "nodes" else "_edge")
      df_fld = "_nodes" if kind == "nodes" else "_edges"
      op_type = ASTNode if kind == "nodes" else ASTEdge
  
      if id is None:
          raise ValueError(
              f"Cannot combine steps with empty id for kind {kind}"
          )
  
      logger.debug(
          "combine_steps ops pre: %s", [op for (op, _) in steps]
      )
      if kind == "edges":
          logger.debug("EDGES << recompute forwards given reduced set")
          steps = [
              (
                  op,  # forward op
                  op(
                      g=g.edges(
                          g_step._edges
                      ),  # transition via any found edge
                      prev_node_wavefront=g_step._nodes,  # start from where backwards step says is reachable
                      # target_wave_front=steps[i+1][1]._nodes  # end at where next backwards step says is reachable
                      target_wave_front=None,  # ^^^ optimization: valid transitions already limit to known-good ones
                      engine=engine,
                  ),
              )
              for (op, g_step) in steps
          ]
  
      concat = df_concat(engine)
  
      # df[[id]]
      out_df = concat(
          [getattr(g_step, df_fld)[[id]] for (_, g_step) in steps]
      ).drop_duplicates(subset=[id])
  
      # df[[id, op_name1, ...]]
      logger.debug("combine_steps ops: %s", [op for (op, _) in steps])
      for op, g_step in steps:
          if op._name is not None and isinstance(op, op_type):
              logger.debug(
                  "tagging kind [%s] name %s", op_type, op._name
              )
              out_df = out_df.merge(
                  getattr(g_step, df_fld)[[id, op._name]],
                  on=id,
                  how="left",
              )
              out_df[op._name] = (
                  out_df[op._name].fillna(False).astype(bool)
              )
      out_df = out_df.merge(getattr(g, df_fld), on=id, how="left")
  
      logger.debug("COMBINED[%s] >> %s", kind, out_df)
  
      return out_df
  
file_path: pygraphistry/graphistry/compute/chain.py
imports: from typing import Any, Dict, Union, cast, List, Tuple, TYPE_CHECKING
  import pandas
  from graphistry.Engine import Engine, EngineAbstract, df_concat, resolve_engine
  from graphistry.Plottable import Plottable
  from graphistry.compute.ASTSerializable import ASTSerializable
  from graphistry.util import setup_logger
  from graphistry.utils.json import JSONVal
  from .ast import ASTObject, ASTNode, ASTEdge, from_json
  from .typing import DataFrameT
label_desc: Use image node_id 1 for calling a global function with example usage: combine_steps(g, kind, steps, engine) and returns: out_df