import itertools from astropy import table from ._base_dag import BaseDAG from ._dispatcher import Dispatcher from .._csst import csst def generate_permutations(**kwargs) -> table.Table: """ 生成关键字参数所有值的排列组合字典列表 参数: **kwargs: 关键字参数,值应为可迭代对象(如列表) 返回: list[dict]: 每个字典代表一种排列组合 """ # 验证输入值是否为可迭代对象 for key, values in kwargs.items(): if not isinstance(values, (list, tuple, set)): kwargs[key] = [values] # 如果不是可迭代对象,转换为列表 # 提取键和对应的值列表 keys = list(kwargs.keys()) value_lists = [kwargs[key] for key in keys] # 生成笛卡尔积(所有值组合) permutations = [] for combination in itertools.product(*value_lists): # 将每个组合转换为字典 {键: 值} perm_dict = dict(zip(keys, combination)) permutations.append(perm_dict) return table.Table(permutations) CSST_DAGS = { "csst-msc-l1-qc0": BaseDAG( dag="csst-msc-l1-qc0", pattern=generate_permutations( instrument=["MSC"], obs_type=["BIAS", "DARK", "FLAT"], ), dispatcher=Dispatcher.dispatch_file, ), "csst-msc-l1-mbi": BaseDAG( dag="csst-msc-l1-mbi", pattern=generate_permutations( instrument=["MSC"], obs_type=["WIDE", "DEEP"], detector=csst["MSC"]["MBI"].effective_detector_names, ), dispatcher=Dispatcher.dispatch_file, ), "csst-msc-l1-ast": BaseDAG( dag="csst-msc-l1-ast", pattern=generate_permutations( instrument=["MSC"], obs_type=["WIDE", "DEEP"], detector=csst["MSC"]["MBI"].effective_detector_names, ), dispatcher=Dispatcher.dispatch_file, ), "csst-msc-l1-sls": BaseDAG( dag="csst-msc-l1-sls", pattern=generate_permutations( instrument=["MSC"], obs_type=["WIDE", "DEEP"], detector=csst["MSC"]["SLS"].effective_detector_names, ), dispatcher=Dispatcher.dispatch_file, ), "csst-msc-l1-ooc": BaseDAG( dag="csst-msc-l1-ooc", pattern=generate_permutations( instrument=["MSC"], obs_type=["BIAS", "DARK", "FLAT"], detector=csst["MSC"].effective_detector_names, ), dispatcher=Dispatcher.dispatch_obsgroup_detector, ), "csst-cpic-l1": BaseDAG( dag="csst-cpic-l1", pattern=generate_permutations( instrument=["CPIC"], obs_type=["SCI", "DSF", "CALS"], detector=csst["CPIC"].effective_detector_names, ), dispatcher=Dispatcher.dispatch_file, ), "csst-cpic-l1-qc0": BaseDAG( dag="csst-cpic-l1-qc0", pattern=generate_permutations( instrument=["CPIC"], obs_type=["BIAS", "DARK", "FLAT", "BKG", "LASER"], detector=csst["CPIC"].effective_detector_names, ), dispatcher=Dispatcher.dispatch_file, ), # other DAGS # ... }