code: def _gen_chunk_stage_info(
      cls,
      op: "BaseDataFrameExpandingAgg",
      chunk_cols=None,
      min_periods=1,
  ):
      map_groups = OrderedDict()
      map_sources = OrderedDict()
      combine_sources = OrderedDict()
      combine_columns = OrderedDict()
      combine_funcs = OrderedDict()
      key_to_funcs = OrderedDict()
      valid_columns = []
      min_periods_func_name = None
  
      def _clean_dict(d):
          return OrderedDict(
              (k, sorted(v) if v != [None] else None)
              for k, v in d.items()
          )
  
      def _fun_to_str(fun):
          if isinstance(fun, str):
              return fun
          fun_str = tokenize(fun)
          key_to_funcs[fun_str] = fun
          return fun if isinstance(fun, str) else tokenize(fun)
  
      def _add_column_to_functions(col, fun_name, mappers, aggregator):
          sources = []
          for mapper in mappers:
              mapper_str = _fun_to_str(mapper)
              cls._safe_append(map_groups, mapper_str, col)
              sources.append(mapper_str)
  
          combine_sources[fun_name] = sources
          cls._safe_append(combine_columns, fun_name, col)
          combine_funcs[fun_name] = _fun_to_str(aggregator)
  
      chunk_cols = set(chunk_cols) if chunk_cols is not None else None
      if isinstance(op.func, list):
          op_func = {None: op.func}
      elif isinstance(op.func, str):
          op_func = {None: [op.func]}
      else:
          op_func = op.func
  
      for col, funcs in op_func.items():
          if col is not None:
              if chunk_cols is not None and col not in chunk_cols:
                  continue
              valid_columns.append(col)
  
          if min_periods > 1:
              min_periods_func_name = tokenize(
                  chunk_cols, "min_periods"
              )
              _add_column_to_functions(
                  col,
                  min_periods_func_name,
                  *cls._get_stage_functions(op, "_data_count"),
              )
  
          for func in funcs:
              mapper_funcs, combine_func = cls._get_stage_functions(
                  op, func
              )
              _add_column_to_functions(
                  col, func, mapper_funcs, combine_func
              )
  
      return _stage_info(
          map_groups=_clean_dict(map_groups),
          map_sources=map_sources,
          combine_sources=combine_sources,
          combine_columns=_clean_dict(combine_columns),
          combine_funcs=combine_funcs,
          key_to_funcs=key_to_funcs,
          valid_columns=valid_columns or None,
          min_periods_func_name=min_periods_func_name,
      )
  
imports: from collections import namedtuple, OrderedDict
  from collections.abc import Iterable
  from typing import Dict
  import numpy
  import pandas
  from ...core.operand import OperandStage
  from ...serialization.serializables import AnyField, BoolField, Int32Field, Int64Field, DictField, StringField
  from ...utils import tokenize
  from ..core import DATAFRAME_TYPE
  from ..merge import DataFrameConcat
  from ..operands import DataFrameOperand, DataFrameOperandMixin
  from ..utils import build_df, parse_index, build_empty_series, filter_dtypes_by_index
label_desc: Use image node_id 20 for calling the BaseDataFrameExpandingAgg obj's underlying member method code with example usage: obj._gen_chunk_stage_info(cls, op, chunk_cols, min_periods) and returns: _stage_info, OrderedDict, unknown, fun