add data_id_set

csst_dag/cli/csst_cpic_l1.py

@@ -82,11 +82,12 @@ dag_group_run = BaseDAG.gen_dag_group_run(
 print("\n"*2)
 print(">>> Matching DAGs ...")
 dag_run_list = []
+data_id_set = set()
 for dag in [
     "csst-cpic-l1",
     "csst-cpic-l1-qc0",
 ]:
-    this_dag_run_list = CSST_DAGS[dag].schedule(
+    this_dag_run_list, this_data_id_set = CSST_DAGS[dag].schedule(
         dag_group_run=dag_group_run,
         plan_basis=plan_basis,
         data_basis=data_basis,
@@ -94,6 +95,7 @@ for dag in [
     )
     print(f"- [{dag}] : {len(this_dag_run_list)} dag_runs")
     dag_run_list.extend(this_dag_run_list)
+    data_id_set.union(this_data_id_set)
 
 # print dag_group_run and dag_run_list
 print("\n")

csst_dag/cli/csst_msc_l1.py

@@ -123,13 +123,14 @@ dag_group_run = BaseDAG.gen_dag_group_run(
 print("\n"*2)
 print(">>> Matching DAGs ...")
 dag_run_list = []
+data_id_set = set()
 for dag in [
     "csst-msc-l1-mbi",
     "csst-msc-l1-ast",
     "csst-msc-l1-sls",
     "csst-msc-l1-qc0",
 ]:
-    this_dag_run_list = CSST_DAGS[dag].schedule(
+    this_dag_run_list, this_data_id_set = CSST_DAGS[dag].schedule(
         dag_group_run=dag_group_run,
         plan_basis=plan_basis,
         data_basis=data_basis,
@@ -138,6 +139,7 @@ for dag in [
     )
     print(f"- [{dag}] : {len(this_dag_run_list)} dag_runs")
     dag_run_list.extend(this_dag_run_list)
+    data_id_set.union(this_data_id_set)
 
 # print dag_group_run and dag_run_list
 print("\n")

csst_dag/dag/_base_dag.py

@@ -103,22 +103,29 @@ class BaseDAG:
         dag_group_run: dict,
         data_basis: table.Table,
         plan_basis: table.Table,
+        force_success: bool = False,
         **kwargs,
-    ) -> list[dict]:
+    ) -> tuple[list, set]:
+        # filter plan and data basis
         filtered_plan_basis, filtered_data_basis = self.filter_basis(
             plan_basis, data_basis
         )
+        # dispatch tasks
         task_list = self.dispatcher(filtered_plan_basis, filtered_data_basis)
-        dag_run_list = []
+        # convert tasks to dag_run_list
+        dag_run_list = []  # each element is a dag_run (dict)
+        data_id_set = set()  # each element is a data_id (str)
         for this_task in task_list:
-            dag_run = self.gen_dag_run(
-                **dag_group_run,
-                **this_task["task"],
-                **kwargs,
-            )
-            dag_run_list.append(dag_run)
-
-        return dag_run_list
+            # only convert success tasks
+            if force_success or this_task["success"]:
+                dag_run = self.gen_dag_run(
+                    **dag_group_run,
+                    **this_task["task"],
+                    **kwargs,
+                )
+                dag_run_list.append(dag_run)
+                data_id_set.union(this_task["relevant_data"]["_id"])
+        return dag_run_list, data_id_set
 
     @staticmethod
     def generate_sha1():