    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)
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]
towhee/runtime/pipeline.py