class_docstr: A `DAGRepr` represents a complete DAG.
  
  Args:
      nodes (`Dict[str, NodeRepr]`): All nodes in the dag, which start with _input and end with _output node.
      edges (`Dict[int, List]`): The edges about data queue schema, such as:
                          { 0: {'data': [(a, ColumnType.SCALAR), (b, ColumnType.SCALAR)], 'schema': {'a', SchemaRepr, 'b', SchemaRepr}}
                            1: {'data': [(b, ColumnType.SCALAR), (c, ColumnType.SCALAR)], 'schema': {'b', SchemaRepr, 'c', SchemaRepr}}
                            2: {'data': [(a, ColumnType.SCALAR), (c, ColumnType.SCALAR)], 'schema': {'a', SchemaRepr, 'c', SchemaRepr}}
                          }
      dag_dict(`Dict`): The dag dict.
class_docstr_tok: ["A","`","DAGRepr","`","represents","a","complete","DAG",".","Args",":","nodes","(","`","Dict","[","str",",","NodeRepr","]","`",")",":","All","nodes","in","the","dag",",","which","start","with","_input","and","end","with","_output","node",".","edges","(","`","Dict","[","int",",","List","]","`",")",":","The","edges","about","data","queue","schema",",","such","as",":","{","0",":","{","'data","'",":","[","(","a",",","ColumnType.SCALAR",")",",","(","b",",","ColumnType.SCALAR",")","]",",","'schema","'",":","{","'","a","'",",","SchemaRepr",",","'","b","'",",","SchemaRepr","}","}","1",":","{","'data","'",":","[","(","b",",","ColumnType.SCALAR",")",",","(","c",",","ColumnType.SCALAR",")","]",",","'schema","'",":","{","'","b","'",",","SchemaRepr",",","'","c","'",",","SchemaRepr","}","}","2",":","{","'data","'",":","[","(","a",",","ColumnType.SCALAR",")",",","(","c",",","ColumnType.SCALAR",")","]",",","'schema","'",":","{","'","a","'",",","SchemaRepr",",","'","c","'",",","SchemaRepr","}","}","}","dag_dict","(","`","Dict","`",")",":","The","dag","dict","."]
code: def rebuild_dag(
      dag, sub_uid, op_info, iter_info, input_schema, output_schema
  ):
      sub_dag = op_info["dag"]
      sub_top_sort = op_info["top_sort"]
      iter_type = iter_info["type"]
      param = iter_info["param"]
  
      used_schemas = ()
      for _, node in dag.items():
          used_schemas += node["inputs"] + node["outputs"]
      if iter_type == FilterConst.name:
          filter_cols = param[FilterConst.param.filter_by]
          new_dag, sub_uid_out = DAGRepr.rebuild_sub_dag(
              sub_dag,
              sub_uid,
              sub_top_sort,
              filter_cols,
              filter_cols,
              used_schemas,
          )
      else:
          new_dag, sub_uid_out = DAGRepr.rebuild_sub_dag(
              sub_dag,
              sub_uid,
              sub_top_sort,
              input_schema,
              output_schema,
              used_schemas,
          )
      new_dag[sub_uid_out]["next_nodes"] += dag[sub_uid]["next_nodes"]
  
      if iter_type == FlatMapConst.name:
          new_dag[sub_uid_out]["iter_info"] = {
              "type": FlatMapConst.name,
              "param": None,
          }
      elif iter_type == FilterConst.name:
          new_dag[sub_uid_out]["iter_info"] = {
              "type": FilterConst.name,
              "param": {
                  FilterConst.param.filter_by: new_dag[sub_uid_out][
                      "inputs"
                  ]
              },
          }
          new_dag[sub_uid_out]["inputs"] = input_schema
          new_dag[sub_uid_out]["outputs"] = output_schema
      elif iter_type == WindowConst.name:
          new_dag[sub_uid]["iter_info"] = {
              "type": WindowConst.name,
              "param": {
                  WindowConst.param.size: param[WindowConst.param.size],
                  WindowConst.param.step: param[WindowConst.param.step],
              },
          }
      elif iter_type == TimeWindowConst.name:
          new_dag[sub_uid]["iter_info"] = {
              "type": TimeWindowConst.name,
              "param": {
                  TimeWindowConst.param.time_range_sec: param[
                      TimeWindowConst.param.time_range_sec
                  ],
                  TimeWindowConst.param.time_step_sec: param[
                      TimeWindowConst.param.time_step_sec
                  ],
                  TimeWindowConst.param.timestamp_col: param[
                      TimeWindowConst.param.timestamp_col
                  ],
              },
          }
      elif iter_type == WindowAllConst.name:
          new_dag[sub_uid]["iter_info"] = {
              "type": WindowAllConst.name,
              "param": None,
          }
      elif iter_type == ReduceConst.name:
          new_dag[sub_uid]["iter_info"] = {
              "type": ReduceConst.name,
              "param": None,
          }
  
      for _, node in new_dag.items():
          if node["config"]["name"].startswith(InputConst.name) or node[
              "config"
          ]["name"].startswith(OutputConst.name):
              node["config"]["name"] = node["config"]["name"].split(
                  "-"
              )[0]
          else:
              node["config"].pop("name")
      dag.update(new_dag)
  
imports: import types
  import uuid
  import json
  from copy import deepcopy
  from typing import Dict, Any, Set, List, Tuple, Optional
  from pydantic import BaseModel
  from towhee.runtime.check_utils import check_set, check_node_iter
  from towhee.runtime.node_repr import NodeRepr
  from towhee.runtime.schema_repr import SchemaRepr
  from towhee.runtime.node_config import TowheeConfig
  from towhee.runtime.constants import WindowAllConst, WindowConst, ConcatConst, MapConst, ReduceConst, FilterConst, TimeWindowConst, FlatMapConst, InputConst, OutputConst, OPType
label_desc: Use image node_id 12 for calling the DAGRepr obj's underlying member method code with example usage: obj.rebuild_dag(dag, sub_uid, op_info, iter_info, input_schema, output_schema) without return types