add CRDS and DFS

csst_common/crds.py

+import functools
+import os
+from crds import client
+
+
+class CRDS:
+
+    def __init__(self, crdsroot="/crdsroot", crdscache="/pipeline/crds_cache"):
+        print("Setting CRDS root path ... ", end="")
+        self.crdsroot = crdsroot
+        print(self.crdsroot)
+
+        print("Setting CRDS cache path ... ", end="")
+        self.crdscache = crdscache
+        print(self.crdscache)
+
+        print("Setting CRDS environment variables ... ", end="")
+        os.environ['CRDS_MODE'] = 'remote'
+        os.environ['CRDS_PATH'] = crdscache
+        os.environ['CRDS_OBSERVATORY'] = 'csst'
+        print("Done")
+
+        print("Query for observatory ... ", end="")
+        self.observatory = client.get_default_observatory()
+        print(self.observatory)
+        assert self.observatory == "csst", f"observatory [{self.observatory}] is not csst!"
+
+        print("Query for operational_context ... ", end="")
+        try:
+            self.operational_context = client.get_default_context(self.observatory)
+            self.is_available = True
+        except BaseException:
+            self.operational_context = ""
+            self.is_available = False
+        print(self.operational_context)
+        # assert self.operational_context != "", f"operational_context [{self.operational_context}] is empty!"
+
+    def __repr__(self):
+        return f"< CRDS url='{os.getenv('CRDS_SERVER_URL', default='')}' >\n" \
+               f" - is_available={self.is_available}\n" \
+               f" - observatory='{self.observatory}'\n" \
+               f" - operational_context='{self.operational_context}'\n" \
+               f" - crdsroot='{self.crdsroot}'\n" \
+               f" - crdscache='{self.crdscache}'\n"
+
+    def get_refs(self,
+                 file_path="/dfsroot/L0/MSC/SCIE/62030/10160000105/MS/CSST_MSC_MS_SCIE_20280916072059_20280916072329_10160000105_10_L0_V01.fits"):
+        # print("get min header")
+        hdrs = client.api.get_minimum_header(self.operational_context, file_path, ignore_cache=False)
+        # {'CAMERA': 'MS',
+        #  'CHIPID': '10',
+        #  'DATE-OBS': '2028-09-16T07:20:59',
+        #  'FILTER': 'GI',
+        #  'INSTRUME': 'MSC',
+        #  'REFTYPE': 'UNDEFINED'}
+        # print("get best ref")
+        refs = client.api.get_best_references(self.operational_context, hdrs)
+        # {'bias': 'csst_msc_ms_bias_10_000001.fits',
+        #  'dark': 'csst_msc_ms_dark_10_000001.fits',
+        #  'ledflat': 'csst_msc_ms_ledflat_10_000001.fits',
+        #  'shutter': "NOT FOUND  parameter='CHIPID' value='10' is not in ['06', '07', '08', '09', '11', '12', '13',
+        #  '14', '15', '16', '17', '18', '19', '20', '22', '23', '24', '25']"}
+        # assert ref_type in refs.keys(), f"ref_type [{ref_type}] not in {refs.keys()}"
+        # print("get file info")
+        refs_fp = dict()
+        for ref_type in refs.keys():
+            d = client.api.get_file_info(self.operational_context, refs[ref_type])
+            if d:
+                fp = os.path.join(self.crdsroot, "references", d["file_path"])
+                assert os.path.exists, f"File path [{fp}] does not exist!"
+                refs_fp[ref_type] = fp
+            else:
+                print(f"Failed to get file info for [{ref_type}:{refs[ref_type]}]")
+        # {'bias': '/crdsroot/references/msc/csst_msc_ms_bias_10_000001.fits',
+        #  'dark': '/crdsroot/references/msc/csst_msc_ms_dark_10_000001.fits',
+        #  'ledflat': '/crdsroot/references/msc/csst_msc_ms_ledflat_10_000001.fits'}
+        return refs_fp
+
+    @staticmethod
+    def retry(func, n, *args, **kwargs):
+        for attempt in range(1, n + 1):
+            try:
+                res = func(*args, **kwargs)
+                # assert res["code"] == 0, res
+                return res
+            except BaseException as e:
+                print(f"Error occurs: {e.__repr__()}")
+        raise RuntimeError("All attempts failed.")
+
+
+def retry(func, n=3):
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        for attempt in range(1, n + 1):
+            try:
+                result = func(*args, **kwargs)
+                return result
+            except Exception as e:
+                print(f"尝试 {attempt}/{n} 失败: {e}")
+        raise RuntimeError("All attempts failed.")
+
+
+"""
+file_path="/dfsroot/L0/MSC/SCIE/62030/10160000105/MS/CSST_MSC_MS_SCIE_20280916072059_20280916072329_10160000105_10_L0_V01.fits"
+hdrs = client.api.get_minimum_header(c.operational_context, file_path, ignore_cache=True)
+refs = client.api.get_best_references(c.operational_context, hdrs)
+refs = crds.getrecommendations(hdrs, reftypes=None, context=c.operational_context, ignore_cache=True, observatory=c.observatory, fast=False)
+refs = crds.getreferences(hdrs, reftypes=['bias'], context=c.operational_context, ignore_cache=False, observatory=c.observatory)
+
+
+docker run --rm -it \
+    -v /share/crdsdata/data:/crdsroot \
+    -e CRDS_SERVER_URL=http://172.24.27.2:29000 \
+    -v /share/dfs:/dfsroot \
+    csst/csst-msc-l1-mbi bash
+"""
+
+if __name__ == "__main__":
+    c = CRDS()
+    refs = c.get_refs()
+    print(refs)
+
+    refs = c.retry(
+        c.get_refs, 3,
+        file_path="/dfsroot/L0/MSC/SCIE/62030/10160000105/MS/CSST_MSC_MS_SCIE_20280916072059_20280916072329_10160000105_11_L0_V01.fits"
+    )
+    print(refs)
+    # {'bias': '/crdsroot/references/msc/csst_msc_ms_bias_11_000001.fits',
+    #  'dark': '/crdsroot/references/msc/csst_msc_ms_dark_11_000001.fits',
+    #  'ledflat': '/crdsroot/references/msc/csst_msc_ms_ledflat_11_000001.fits',
+    #  'shutter': '/crdsroot/references/msc/csst_msc_ms_shutter_11_000001.fits'}

csst_common/data_manager.py

@@ -33,7 +33,9 @@ from csst_dfs_api.sls.level2spectra import Level2SpectraApi as SlsLevel2DataApi
 from .logger import get_logger
 from .params import CSST_PARAMS as CP
 from .params import DFS_CONF
-from .time import now
+from .time import now, TimeStamp
+from .crds import CRDS
+from .dfs import DFS
 
 
 class CsstMsDataManager:
@@ -295,6 +297,9 @@ class CsstMsDataManager:
             self.l1_whitelist.append(self.l1_file(self.stamps))
         self.write_stamp()
 
+        self.crds = CRDS()
+        self.dfs = DFS()
+
     def write_stamp(self):
         if self.stamps is not None and self.stamps != "":
             with open(self.l1_file(self.stamps), "a+") as f:

csst_common/dfs.py

+import os
+
+
+class DFS:
+    def __init__(self, n_try=5):
+        self.n_try = n_try
+        pass
+
+    @staticmethod
+    def retry(func, n, *args, **kwargs):
+        for attempt in range(1, n + 1):
+            try:
+                res = func(*args, **kwargs)
+                assert res["code"] == 0, res
+                return res
+            except BaseException as e:
+                print(f"Error occurs: {e.__repr__()}")
+        raise RuntimeError("All attempts failed.")
+
+
+
+
+