add observation.py, but this is useless

csst_dag/_csst/observation.py

 from .dict import DotDict
 from .instrument import csst
-from astropy.table import Table
+from astropy.table import Table, vstack
 import numpy as np
 
 # from csst_dag.dict import DotDict
@@ -40,6 +40,36 @@ class CsstPlanObsid(DotDict):
     def from_plan(plan_data: dict) -> "CsstPlanObsid":
         return CsstPlanObsid(**plan_data)
 
+    @property
+    def expected_data_table(self):
+        contents = []
+        for detector in self.detectors:
+            n_frame = self.params.num_epec_frame if self.instrument == "HSTDM" else 1
+            for _ in range(n_frame):
+                contents.append(
+                    dict(
+                        dataset=self.dataset,
+                        instrument=self.instrument,
+                        obs_type=self.obs_type,
+                        obs_group=self.obs_group,
+                        obs_id=self.obs_id,
+                        detector=detector.name,
+                    )
+                )
+        # post processing
+        t = Table(contents)
+        t.sort(
+            keys=[
+                "dataset",
+                "instrument",
+                "obs_type",
+                "obs_group",
+                "obs_id",
+                "detector",
+            ]
+        )
+        return t
+
 
 class CsstPlanObsgroup(DotDict):
     def __init__(self, *args, **kwargs):
@@ -103,3 +133,45 @@ class CsstPlanObsgroup(DotDict):
         return CsstPlanObsgroup(
             **{_["obs_id"]: CsstPlanObsid.from_plan(_) for _ in plan_data}
         )
+
+    @property
+    def expected_data_table(self):
+        t = vstack([self[k].expected_data_table for k in self])
+        t.sort(
+            keys=[
+                "dataset",
+                "instrument",
+                "obs_type",
+                "obs_group",
+                "obs_id",
+                "detector",
+            ]
+        )
+        return t
+
+
+class CsstLevel0(DotDict):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    @property
+    def n_file_expected(self):
+        return sum([self[_].n_file_expected for _ in self.obs_id_list])
+
+    @staticmethod
+    def extract_data_table(level0_data: list[dict]) -> Table:
+        d_list = [
+            dict(
+                dataset=d["dataset"],
+                instrument=d["instrument"],
+                obs_type=d["obs_type"],
+                obs_group=d["obs_group"],
+                obs_id=d["obs_id"],
+                detector=d["detector"],
+            )
+            for d in level0_data
+        ]
+
+        return CsstPlanObsgroup(
+            **{_["obs_id"]: CsstPlanObsid.from_plan(_) for _ in plan_data}
+        )

tests/test_observation.py

+import joblib
+import numpy as np
+
+from csst_dag import CsstPlanObsid, CsstPlanObsgroup, csst
+
+plan_data = joblib.load("/Users/cham/CsstProjects/csst-dag/tests/plan.dump")
+
+csst_obsid1 = CsstPlanObsid(plan_data[0])
+csst_obsid = CsstPlanObsid.from_plan(plan_data[0])
+csst_obsid.params
+t = csst_obsid.expected_data_table
+tp = copy(t)
+
+csst_obsgroup = CsstPlanObsgroup.from_plan(plan_data)
+csst_obsgroup.instrument
+csst_obsgroup.obs_id_list
+csst_obsgroup.keys()
+csst_obsgroup.values()
+csst_obsgroup.n_file_expected