class_docstr: Pipeline is a tool to create data transformation chains.
  
  Args:
      dag (`dict`): The dag for the pipeline.
      clo_node (`str`): The close node in the pipeline dag, defaults to '_input'.
docstr: Perform action on time windows.
  
  Args:
      input_schema (tuple): The input column/s of fn.
      output_schema (tuple): The output columns to fn.
      timestamp_col (str): Which column to use for creating windows.
      size (int): size of window.
      step (int): how far to progress window.
      fn (Operation | lambda | callable): The action to perform on the input_schema
                                          after window the date with timestamp_col.
      config (dict, optional): Config for the time window. Defaults to None.
  
  Returns:
      Pipeline: Pipeline with time_window action added.
  
  Examples:
      >>> from towhee import pipe
      >>> p = (pipe.input('d')
      ...         .flat_map('d', ('n1', 'n2', 't'), lambda x: ((a, b, c) for a, b, c in x))
      ...         .time_window(('n1', 'n2'), ('s1', 's2'), 't', 3, 3, lambda x, y: (sum(x), sum(y)))
      ...         .output('s1', 's2'))
      >>> data = [(i, i+1, i * 1000) for i in range(11) if i < 3 or i > 7] #[(0, 1), (1, 2), (2, 3), (8, 9), (9, 10), (10, 11)]
      >>> res = p(data)
      >>> res.get() #[(0, 1), (1, 2), (2, 3)]
      [3, 6]
      >>> res.get() #(8, 9)
      [8, 9]
      >>> res.get() #(9, 10), (10, 11)
      [19, 21]
docstr_tok: ["Perform","action","on","time","windows",".","Args",":","input_schema","(","tuple",")",":","The","input","column\/s","of","fn",".","output_schema","(","tuple",")",":","The","output","columns","to","fn",".","timestamp_col","(","str",")",":","Which","column","to","use","for","creating","windows",".","size","(","int",")",":","size","of","window",".","step","(","int",")",":","how","far","to","progress","window",".","fn","(","Operation","|","lambda","|","callable",")",":","The","action","to","perform","on","the","input_schema","after","window","the","date","with","timestamp_col",".","config","(","dict",",","optional",")",":","Config","for","the","time","window",".","Defaults","to","None",".","Returns",":","Pipeline",":","Pipeline","with","time_window","action","added",".","Examples",":",">",">",">","from","towhee","import","pipe",">",">",">","p","=","(","pipe.input","(","'d","'",")","...",".flat_map","(","'d","'",",","(","'n1","'",",","'n2","'",",","'t","'",")",",","lambda","x",":","(","(","a",",","b",",","c",")","for","a",",","b",",","c","in","x",")",")","...",".time_window","(","(","'n1","'",",","'n2","'",")",",","(","'s1","'",",","'s2","'",")",",","'t","'",",","3",",","3",",","lambda","x",",","y",":","(","sum","(","x",")",",","sum","(","y",")",")",")","...",".output","(","'s1","'",",","'s2","'",")",")",">",">",">","data","=","[","(","i",",","i+1",",","i","*","1000",")","for","i","in","range","(","11",")","if","i","<","3","or","i",">","7","]","#","[","(","0",",","1",")",",","(","1",",","2",")",",","(","2",",","3",")",",","(","8",",","9",")",",","(","9",",","10",")",",","(","10",",","11",")","]",">",">",">","res","=","p","(","data",")",">",">",">","res.get","(",")","#","[","(","0",",","1",")",",","(","1",",","2",")",",","(","2",",","3",")","]","[","3",",","6","]",">",">",">","res.get","(",")","#","(","8",",","9",")","[","8",",","9","]",">",">",">","res.get","(",")","#","(","9",",","10",")",",","(","10",",","11",")","[","19",",","21","]"]
code: def time_window(
      self,
      input_schema,
      output_schema,
      timestamp_col,
      size,
      step,
      fn,
      config=None,
  ) -> "Pipeline":
      """
      Perform action on time windows.
  
      Args:
          input_schema (tuple): The input column/s of fn.
          output_schema (tuple): The output columns to fn.
          timestamp_col (str): Which column to use for creating windows.
          size (int): size of window.
          step (int): how far to progress window.
          fn (Operation | lambda | callable): The action to perform on the input_schema
                                              after window the date with timestamp_col.
          config (dict, optional): Config for the time window. Defaults to None.
  
      Returns:
          Pipeline: Pipeline with time_window action added.
  
      Examples:
          >>> from towhee import pipe
          >>> p = (pipe.input('d')
          ...         .flat_map('d', ('n1', 'n2', 't'), lambda x: ((a, b, c) for a, b, c in x))
          ...         .time_window(('n1', 'n2'), ('s1', 's2'), 't', 3, 3, lambda x, y: (sum(x), sum(y)))
          ...         .output('s1', 's2'))
          >>> data = [(i, i+1, i * 1000) for i in range(11) if i < 3 or i > 7] #[(0, 1), (1, 2), (2, 3), (8, 9), (9, 10), (10, 11)]
          >>> res = p(data)
          >>> res.get() #[(0, 1), (1, 2), (2, 3)]
          [3, 6]
          >>> res.get() #(8, 9)
          [8, 9]
          >>> res.get() #(9, 10), (10, 11)
          [19, 21]
      """
      output_schema = self._check_schema(output_schema)
      input_schema = self._check_schema(input_schema)
  
      uid = uuid.uuid4().hex
      fn_action = self._to_action(fn)
      dag_dict = deepcopy(self._dag)
      dag_dict[uid] = {
          "inputs": input_schema,
          "outputs": output_schema,
          "op_info": fn_action.serialize(),
          "iter_info": {
              "type": TimeWindowConst.name,
              "param": {
                  TimeWindowConst.param.time_range_sec: size,
                  TimeWindowConst.param.time_step_sec: step,
                  TimeWindowConst.param.timestamp_col: timestamp_col,
              },
          },
          "config": config,
          "next_nodes": [],
      }
      dag_dict[self._clo_node]["next_nodes"].append(uid)
      return Pipeline(dag_dict, uid)
  
imports: import uuid
  from copy import deepcopy
  from towhee.runtime.check_utils import TupleForm
  from towhee.runtime.operator_manager import OperatorAction
  from towhee.runtime.factory import _OperatorWrapper
  from towhee.runtime.runtime_pipeline import RuntimePipeline
  from towhee.runtime.constants import MapConst, WindowAllConst, WindowConst, ReduceConst, FilterConst, TimeWindowConst, FlatMapConst, ConcatConst, InputConst, OutputConst, OPName
label_desc: Use image node_id 12 for calling the Pipeline obj's underlying member method code with example usage: obj.time_window(input_schema, output_schema, timestamp_col, size, step, fn, config) and returns: Pipeline