add CsstMsFile

csst_common/data_manager.py

@@ -149,6 +149,7 @@ class CsstMsDataManager:
     >>> # define an L1 file (non-detector-specified)
     >>> dm.l1_file("flipped_image.fits")
     """
+
     # TODO: update docstring
     def __init__(self,
                  ver_sim: str = "C6.2",
@@ -1004,3 +1005,80 @@ class CsstMsDataManager:
 
 # temporarily compatible with old interface
 CsstMbiDataManager = CsstMsDataManager
+
+
+class CsstMsFile(dict):
+    def __init__(self, filepath, ver_sim="C6.2", dir_l0=".", dir_l1=".", dfs_root=None, **kwargs):
+        super(CsstMsFile, self).__init__()
+        self.ver_sim = ver_sim
+        self.dir_l0 = dir_l0
+        self.dir_l1 = dir_l1
+        self.dfs_root = dfs_root
+
+        self.filename = os.path.basename(filepath)
+        self.dir_l0 = os.path.dirname(filepath)
+        # parse filename
+        pattern = re.compile(
+            r"(?P<telescope>[A-Z]+)_"
+            r"(?P<instrument>[A-Z]+)_"
+            r"(?P<project>[A-Z]+)_"
+            r"(?P<obs_type>[A-Z]+)_"
+            r"(?P<exp_start>[0-9]{14})_"
+            r"(?P<exp_stop>[0-9]{14})_"
+            r"(?P<obs_id>[0-9]{11})_"
+            r"(?P<detector>[0-9]{2})_"
+            r"L(?P<level>[0-9]{1})_"
+            r"V(?P<version>[0-9]{2})"
+            r".(?P<ext>[a-z]{4})"
+        )
+        mo = re.fullmatch(pattern, self.filename)
+        assert mo is not None
+        mogd = mo.groupdict()
+        # mogd.pop("detector")
+
+        for k, v in mogd.items():
+            self.__setattr__(k, v)
+
+        for k, v in kwargs.items():
+            self.__setattr__(k, v)
+
+    @property
+    def dir_l0(self):
+        if self.dfs_root is not None:
+            return os.path.join(self.dfs_root, self._dir_l0)
+        else:
+            return self._dir_l0
+
+    @dir_l0.setter
+    def dir_l0(self, dir_l0):
+        self._dir_l0 = dir_l0
+
+    def fpo(self, post=".fits"):
+        if post.startswith("."):
+            fn = f"{self.telescope}_{self.instrument}_{self.project}_{self.obs_type}_{self.exp_start}_{self.exp_stop}_" \
+                 f"{self.obs_id}_{self.detector}_L1_V{self.version}{post}"
+        else:
+            fn = f"{self.telescope}_{self.instrument}_{self.project}_{self.obs_type}_{self.exp_start}_{self.exp_stop}_" \
+                 f"{self.obs_id}_{self.detector}_L1_V{self.version}_{post}"
+        return os.path.join(self.dir_l1, fn)
+
+    def fpi(self):
+        return os.path.join(self.dir_l0, self.filename)
+
+    @staticmethod
+    def from_l1id(id=17796, ver_sim="C6.2", dir_l0=".", dir_l1=".", dfs_root=None):
+        L1DataApi = Level1DataApi()
+        rec = L1DataApi.get(id=id)
+        try:
+            assert rec["code"] == 0
+        except AssertionError as ae:
+            print(rec)
+            raise ae
+        header = rec["data"].header
+        return CsstMsFile(
+            filepath=rec["data"].file_path, ver_sim=ver_sim, dir_l0=dir_l0, dir_l1=dir_l1, dfs_root=dfs_root, **header)
+
+# file = CsstMsFile(
+#     "L1/MSC/SCI/62173/10160000108/CSST_MSC_MS_SCI_20290206174352_20290206174622_10160000108_21_L1_V01.fits")
+#
+# file = CsstMsFile.from_l1id(id=17796, dfs_root="/share/dfs")