split level0data

.gitignore

@@ -41,3 +41,4 @@ test_reports.xml
 
 # tmp test
 .tmp/*
+.hypothesis
\ No newline at end of file

csst_dfs_api_local/facility/__init__.py

-from .calmerge import CalMergeApi
-from .level0 import Level0DataApi
 from .detector import DetectorApi
-from .level0prc import Level0PrcApi
 from .observation import ObservationApi
\ No newline at end of file

csst_dfs_api_local/ifs/__init__.py

-from .level1 import Level1DataApi
+from .calmerge import CalMergeApi
\ No newline at end of file
+from .level0 import Level0DataApi
+from .level0prc import Level0PrcApi
+from .level1 import Level1DataApi
+from .level1prc import Level1PrcApi
\ No newline at end of file

csst_dfs_api_local/ifs/calmerge.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.ifs import CalMergeRecord
+from csst_dfs_commons.models.common import from_dict_list
+from .level0 import Level0DataApi
+
+log = logging.getLogger('csst')
+
+class CalMergeApi(object):
+    def __init__(self, sub_system = "ifs"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+        self.level0Api = Level0DataApi()
+
+    def get_latest_by_l0(self, **kwargs):
+        ''' retrieve calibration merge records from database by level0 data
+        
+        :param kwargs: Parameter dictionary, key items support:
+            level0_id: [str],
+            ref_type: [str]
+
+        :returns: csst_dfs_common.models.Result
+        '''
+        try:
+            level0_id = get_parameter(kwargs, "level0_id")
+            ref_type = get_parameter(kwargs, "ref_type")
+
+            level0_data = self.level0Api.get_by_level0_id(level0_id)
+            if level0_data is None:
+                return Result.error(message = "level0 data [%s]not found"%(level0_id))
+            
+            sql_data = f"select * from ifs_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time >= '{level0_data.data.obs_time}' order by obs_time ASC limit 1"
+
+            r = self.db.select_one(sql_data)
+            if r:
+                rec = CalMergeRecord().from_dict(r)
+                return Result.ok_data(data=rec)
+            
+            sql_data = f"select * from ifs_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time <= '{level0_data.data.obs_time}' order by obs_time DESC limit 1"
+
+            r = self.db.select_one(sql_data)
+            if r:
+                rec = CalMergeRecord().from_dict(r)
+                return Result.ok_data(data=rec)            
+
+            return Result.error(message = "not found")
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def find(self, **kwargs):
+        ''' retrieve calibration merge records from database
+
+        parameter kwargs:
+            detector_no: [str]
+            ref_type: [str]
+            obs_time: (start,end)
+            qc1_status : [int]
+            prc_status : [int]
+            file_name: [str]
+            limit: limits returns the number of records,default 0:no-limit
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            detector_no = get_parameter(kwargs, "detector_no")
+            ref_type = get_parameter(kwargs, "ref_type")
+            exp_time_start = get_parameter(kwargs, "obs_time", [None, None])[0]
+            exp_time_end = get_parameter(kwargs, "obs_time", [None, None])[1]              
+            qc1_status = get_parameter(kwargs, "qc1_status")
+            prc_status = get_parameter(kwargs, "prc_status")
+            file_name = get_parameter(kwargs, "file_name")
+            limit = get_parameter(kwargs, "limit", 0)
+
+            sql_count = "select count(*) as c from ifs_cal_merge where 1=1"
+            sql_data = f"select * from ifs_cal_merge where 1=1"
+
+            sql_condition = "" 
+            if detector_no:
+                sql_condition = f"{sql_condition} and detector_no='{detector_no}'"
+            if ref_type:
+                sql_condition = f"{sql_condition} and ref_type='{ref_type}'"
+            if exp_time_start:
+                sql_condition = f"{sql_condition} and obs_time >='{exp_time_start}'"
+            if exp_time_end:
+                sql_condition = f"{sql_condition} and obs_time <='{exp_time_end}'"
+            if qc1_status:
+                sql_condition = f"{sql_condition} and qc1_status={qc1_status}"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+
+            if file_name:
+                sql_condition = f" and filename={file_name}"  
+
+            sql_count = f"{sql_count} {sql_condition}"
+            sql_data = f"{sql_data} {sql_condition}"
+
+            log.info(sql_count)
+            log.info(sql_data)
+
+            if limit > 0:
+                sql_data = f"{sql_data} limit {limit}"   
+
+            totalCount = self.db.select_one(sql_count)
+            _, records = self.db.select_many(sql_data)
+
+            return Result.ok_data(data=from_dict_list(CalMergeRecord, records)).append("totalCount", totalCount['c'])
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def get(self, **kwargs):
+        '''  fetch a record from database
+
+        parameter kwargs:
+            id : [int],
+            cal_id : [str]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        cal_id = get_parameter(kwargs, "cal_id", "")
+
+        if id == 0 and cal_id == "":
+            return Result.error(message="at least define id or cal_id") 
+
+        if id != 0: 
+            return self.get_by_id(id)
+        if cal_id != "": 
+            return self.get_by_cal_id(cal_id)
+
+    def get_by_id(self, iid: int):
+        try:
+            r = self.db.select_one(
+                "select * from ifs_cal_merge where id=?", (iid,))
+            if r:
+
+                sql_get_level0_id = f"select level0_id from ifs_cal2level0 where merge_id={r['id']}" 
+                _, records = self.db.select_many(sql_get_level0_id)
+                level0_ids = [r["level0_id"] for r in records]
+
+                rec = CalMergeRecord().from_dict(r)
+                rec.level0_ids = level0_ids
+                return Result.ok_data(data=rec)
+            else:
+                return Result.error(message=f"id:{iid} not found")
+                    
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def get_by_cal_id(self, cal_id: str):
+        try:
+            r = self.db.select_one(
+                "select * from ifs_cal_merge where cal_id=?", (cal_id,))
+            if r:
+
+                sql_get_level0_id = f"select level0_id from ifs_cal2level0 where merge_id={r['id']}" 
+                _, records = self.db.select_many(sql_get_level0_id)
+                level0_ids = [r["level0_id"] for r in records]
+
+                rec = CalMergeRecord().from_dict(r)
+                rec.level0_ids = level0_ids
+                return Result.ok_data(data=rec)
+            else:
+                return Result.error(message=f"id:{id} not found")
+                    
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def update_qc1_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            cal_id : [str],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        cal_id = get_parameter(kwargs, "cal_id", "")
+        result = self.get(id = id, cal_id = cal_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+
+        try:
+            self.db.execute(
+                'update ifs_cal_merge set qc1_status=?, qc1_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            cal_id : [str],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        cal_id = get_parameter(kwargs, "cal_id", "")
+        result = self.get(id = id, cal_id = cal_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+
+        try:
+            existed = self.db.exists(
+                "select * from ifs_cal_merge where id=?",
+                (id,)
+            )
+            if not existed:
+                log.warning('%s not found' %(id, ))
+                return Result.error(message ='%s not found' %(id, ))
+            self.db.execute(
+                'update ifs_cal_merge set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a calibration merge record into database
+ 
+        parameter kwargs:
+            cal_id : [str]
+            detector_no : [str]
+            ref_type : [str]
+            obs_time : [str]
+            exp_time : [float]
+            prc_status : [int]
+            prc_time : [str]
+            filename : [str]
+            file_path : [str]
+            level0_ids : [list]
+        return csst_dfs_common.models.Result
+        '''   
+
+        rec = CalMergeRecord(
+            id = 0,
+            cal_id =  get_parameter(kwargs, "cal_id"),
+            detector_no = get_parameter(kwargs, "detector_no"),
+            ref_type = get_parameter(kwargs, "ref_type"),
+            obs_time = get_parameter(kwargs, "obs_time"),
+            exp_time = get_parameter(kwargs, "exp_time"),
+            filename = get_parameter(kwargs, "filename"),
+            file_path = get_parameter(kwargs, "file_path"),
+            prc_status = get_parameter(kwargs, "prc_status", -1),
+            prc_time = get_parameter(kwargs, "prc_time"),
+            level0_ids = get_parameter(kwargs, "level0_ids", [])
+        )
+        try:
+            self.db.execute(
+                'INSERT INTO ifs_cal_merge (cal_id,detector_no,ref_type,obs_time,exp_time,filename,file_path,prc_status,prc_time,create_time) \
+                    VALUES(?,?,?,?,?,?,?,?,?,?)',
+                (rec.cal_id, rec.detector_no, rec.ref_type, rec.obs_time, rec.exp_time, rec.filename, rec.file_path,rec.prc_status,rec.prc_time,format_time_ms(time.time()))
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            sql_level0_ids = "insert into ifs_cal2level0 (merge_id,level0_id) values "
+            values = ["(%s,%s)"%(rec.id,i) for i in rec.level0_ids]
+            _ = self.db.execute(sql_level0_ids + ",".join(values))            
+
+            self.db.end()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
\ No newline at end of file

csst_dfs_api_local/ifs/ingest.py

+import os, sys
+import argparse
+import logging
+from astropy.io import fits
+import datetime
+import shutil
+
+from csst_dfs_api_local.common.db import DBClient
+log = logging.getLogger('csst-dfs-api-local')
+
+def ingest():
+    
+    parser = argparse.ArgumentParser(prog=f"{sys.argv[0]}", description="ingest the local files")
+    parser.add_argument('-i','--infile', dest="infile", help="a file or a directory")
+    parser.add_argument('-m', '--copyfiles', dest="copyfiles", action='store_true', default=False, help="whether copy files after import")
+    args = parser.parse_args(sys.argv[1:])
+
+    import_root_dir = args.infile
+    if import_root_dir is None or (not os.path.isfile(import_root_dir) and not os.path.isdir(import_root_dir)):
+        parser.print_help()
+        sys.exit(0)
+
+    db = DBClient()
+    if os.path.isfile(import_root_dir):
+        log.info(f"prepare import {import_root_dir}")
+        ingest_one(import_root_dir, db, args.copyfiles)
+    if os.path.isdir(import_root_dir):
+        for (path, _, file_names) in os.walk(import_root_dir):
+            for filename in file_names:
+                if filename.find(".fits") > 0:
+                    file_full_path = os.path.join(path, filename)
+                    log.info(f"prepare import {file_full_path}")
+                    try:
+                        ingest_one(file_full_path, db, args.copyfiles)
+                    except Exception as e:
+                        print(f"{file_full_path} import error!!!")
+                        log.error(e)
+
+    db.close()
+
+def ingest_one(file_path, db, copyfiles):
+    dest_root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+
+    hdul = fits.open(file_path)
+    header = hdul[0].header
+    header1 = hdul[1].header
+
+    obs_id = header["OBSID"]
+    exp_start_time = f"{header['DATE-OBS']}"
+    exp_time = header['EXPTIME']
+    
+    module_id = header["INSTRUME"]
+    obs_type = header["OBSTYPE"]
+    qc0_status = -1
+    prc_status = -1
+    time_now = datetime.datetime.now()
+    create_time = time_now.strftime('%Y-%m-%d %H:%M:%S')
+
+    facility_status_id = 0
+    module_status_id = 0
+
+    existed = db.exists("select * from t_observation where obs_id=?", (obs_id,))
+    if not existed:
+        db.execute("insert into t_observation \
+            (obs_id,obs_time,exp_time,module_id,obs_type,facility_status_id, module_status_id, qc0_status, prc_status,create_time) \
+            values (?,?,?,?,?,?,?,?,?,?)",
+        (obs_id,exp_start_time,exp_time,module_id,obs_type,facility_status_id,module_status_id,qc0_status, prc_status,create_time))
+        db.end()
+    #level0
+    detector = header1["DETNAM"]
+    filename = header["FILENAME"]
+    
+    existed = db.exists(
+            "select * from ifs_level0_data where filename=?",
+            (filename,)
+        )
+    if existed:
+        log.warning('%s has already been imported' %(file_path, ))
+        db.end()
+        return    
+
+    detector_status_id = 0
+
+    file_full_path = file_path
+
+    if copyfiles:
+        file_dir = f"{dest_root_dir}/{module_id}/{obs_type.upper()}/{header['EXPSTART']}/{obs_id}/MS"
+        if not os.path.exists(file_dir):
+            os.makedirs(file_dir)
+        file_full_path = f"{file_dir}/{filename}.fits"
+
+    level0_id = f"{obs_id}{detector}"  
+
+    c = db.execute("insert into ifs_level0_data \
+        (level0_id, obs_id, detector_no, obs_type, obs_time, exp_time,detector_status_id, filename, file_path,qc0_status, prc_status,create_time) \
+        values (?,?,?,?,?,?,?,?,?,?,?,?)",
+        (level0_id, obs_id, detector, obs_type, exp_start_time, exp_time, detector_status_id, filename, file_full_path, qc0_status, prc_status,create_time))
+    db.end()
+    level0_id_id = db.last_row_id()
+    #level0-header
+    ra_obj = header["OBJ_RA"]
+    dec_obj = header["OBJ_DEC"]
+    db.execute("delete from ifs_level0_header where id=?",(level0_id_id,))    
+    db.execute("insert into ifs_level0_header \
+        (id, obs_time, exp_time, ra, `dec`, create_time) \
+        values (?,?,?,?,?,?)",
+        (level0_id_id, exp_start_time, exp_time, ra_obj, dec_obj, create_time))
+    
+    if copyfiles:
+        #copy files
+        shutil.copyfile(file_path, file_full_path)
+
+    db.end()
+
+    print(f"{file_path} imported")
+
+if __name__ == "__main__":
+    ingest()
\ No newline at end of file

csst_dfs_api_local/facility/level0.py → csst_dfs_api_local/ifs/level0.py

@@ -6,13 +6,14 @@ import shutil
 from ..common.db import DBClient
 from ..common.utils import *
 from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.facility import Level0Record
+from csst_dfs_commons.models.ifs import Level0Record
 from csst_dfs_commons.models.common import from_dict_list
 
 log = logging.getLogger('csst')
 
 class Level0DataApi(object):
-    def __init__(self):
+    def __init__(self, sub_system = "ifs"):
+        self.sub_system = sub_system
         self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
         self.db = DBClient()
 
@@ -42,8 +43,8 @@ class Level0DataApi(object):
             file_name = get_parameter(kwargs, "file_name")
             limit = get_parameter(kwargs, "limit", 0)
 
-            sql_count = "select count(*) as c from t_level0_data where 1=1"
-            sql_data = f"select * from t_level0_data where 1=1"
+            sql_count = "select count(*) as c from ifs_level0_data where 1=1"
+            sql_data = f"select * from ifs_level0_data where 1=1"
 
             sql_condition = "" 
             if obs_id:
@@ -100,7 +101,7 @@ class Level0DataApi(object):
     def get_by_id(self, id: int):
         try:
             r = self.db.select_one(
-                "select * from t_level0_data where id=?", (id,))
+                "select * from ifs_level0_data where id=?", (id,))
             if r:
                 return Result.ok_data(data=Level0Record().from_dict(r))
             else:
@@ -112,7 +113,7 @@ class Level0DataApi(object):
     def get_by_level0_id(self, level0_id: str):
         try:
             r = self.db.select_one(
-                "select * from t_level0_data where level0_id=?", (level0_id,))
+                "select * from ifs_level0_data where level0_id=?", (level0_id,))
             if r:
                 return Result.ok_data(data=Level0Record().from_dict(r))
             else:
@@ -142,7 +143,7 @@ class Level0DataApi(object):
         status = get_parameter(kwargs, "status")
         try:
             self.db.execute(
-                'update t_level0_data set prc_status=?, prc_time=? where id=?',
+                'update ifs_level0_data set prc_status=?, prc_time=? where id=?',
                 (status, format_time_ms(time.time()), id)
             )  
             self.db.end() 
@@ -171,7 +172,7 @@ class Level0DataApi(object):
         status = get_parameter(kwargs, "status")
         try:
             self.db.execute(
-                'update t_level0_data set qc0_status=?, qc0_time=? where id=?',
+                'update ifs_level0_data set qc0_status=?, qc0_time=? where id=?',
                 (status, format_time_ms(time.time()), id)
             )  
             self.db.end() 
@@ -208,7 +209,7 @@ class Level0DataApi(object):
         rec.level0_id = f"{rec.obs_id}{rec.detector_no}"
         try:
             existed = self.db.exists(
-                    "select * from t_level0_data where filename=?",
+                    "select * from ifs_level0_data where filename=?",
                     (rec.filename,)
                 )
             if existed:
@@ -216,7 +217,7 @@ class Level0DataApi(object):
                 return Result.error(message ='%s existed' %(rec.filename, ))
 
             self.db.execute(
-                'INSERT INTO t_level0_data (level0_id, obs_id, detector_no, obs_type, obs_time, exp_time,detector_status_id, filename, file_path,qc0_status, prc_status,create_time) \
+                'INSERT INTO ifs_level0_data (level0_id, obs_id, detector_no, obs_type, obs_time, exp_time,detector_status_id, filename, file_path,qc0_status, prc_status,create_time) \
                     VALUES(?,?,?,?,?,?,?,?,?,?,?,?)',
                 (rec.level0_id, rec.obs_id, rec.detector_no, rec.obs_type, rec.obs_time, rec.exp_time, rec.detector_status_id, rec.filename, rec.file_path,-1,-1,format_time_ms(time.time()))
             )

csst_dfs_api_local/facility/level0prc.py → csst_dfs_api_local/ifs/level0prc.py

@@ -6,13 +6,14 @@ import shutil
 from ..common.db import DBClient
 from ..common.utils import *
 from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.facility import Level0PrcRecord
+from csst_dfs_commons.models.ifs import Level0PrcRecord
 from csst_dfs_commons.models.common import from_dict_list
 
 log = logging.getLogger('csst')
 
 class Level0PrcApi(object):
-    def __init__(self):
+    def __init__(self, sub_system = "ifs"):
+        self.sub_system = sub_system
         self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
         self.db = DBClient()
 
@@ -33,7 +34,7 @@ class Level0PrcApi(object):
             prc_module = get_parameter(kwargs, "prc_module")
             prc_status = get_parameter(kwargs, "prc_status")
 
-            sql_data = f"select * from t_level0_prc"
+            sql_data = f"select * from ifs_level0_prc"
 
             sql_condition = f"where level0_id='{level0_id}'"
             if pipeline_id:
@@ -65,14 +66,14 @@ class Level0PrcApi(object):
 
         try:
             existed = self.db.exists(
-                "select * from t_level0_prc where id=?",
+                "select * from ifs_level0_prc where id=?",
                 (id,)
             )
             if not existed:
                 log.warning('%s not found' %(id, ))
                 return Result.error(message ='%s not found' %(id, ))
             self.db.execute(
-                'update t_level0_prc set prc_status=?, prc_time=? where id=?',
+                'update ifs_level0_prc set prc_status=?, prc_time=? where id=?',
                 (status, format_time_ms(time.time()), id)
             )  
             self.db.end() 
@@ -108,7 +109,7 @@ class Level0PrcApi(object):
         )
         try:
             self.db.execute(
-                'INSERT INTO t_level0_prc (level0_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
+                'INSERT INTO ifs_level0_prc (level0_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
                     VALUES(?,?,?,?,?,?,?)',
                 (rec.level0_id, rec.pipeline_id, rec.prc_module, rec.params_file_path, rec.prc_status, rec.prc_time, rec.result_file_path)
             )

csst_dfs_api_local/facility/level1prc.py → csst_dfs_api_local/ifs/level1prc.py

@@ -6,7 +6,7 @@ import shutil
 from ..common.db import DBClient
 from ..common.utils import *
 from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.facility import Level1PrcRecord
+from csst_dfs_commons.models.ifs import Level1PrcRecord
 from csst_dfs_commons.models.common import from_dict_list
 
 log = logging.getLogger('csst')
@@ -33,7 +33,7 @@ class Level1PrcApi(object):
             prc_module = get_parameter(kwargs, "prc_module")
             prc_status = get_parameter(kwargs, "prc_status")
 
-            sql_data = f"select * from t_level1_prc"
+            sql_data = f"select * from ifs_level1_prc"
 
             sql_condition = f"where level1_id={level1_id}"
             if pipeline_id:
@@ -65,14 +65,14 @@ class Level1PrcApi(object):
 
         try:
             existed = self.db.exists(
-                "select * from t_level1_prc where id=?",
+                "select * from ifs_level1_prc where id=?",
                 (id,)
             )
             if not existed:
                 log.warning('%s not found' %(id, ))
                 return Result.error(message ='%s not found' %(id, ))
             self.db.execute(
-                'update t_level1_prc set prc_status=?, prc_time=? where id=?',
+                'update ifs_level1_prc set prc_status=?, prc_time=? where id=?',
                 (status, format_time_ms(time.time()), id)
             )  
             self.db.end() 
@@ -108,7 +108,7 @@ class Level1PrcApi(object):
         )
         try:
             self.db.execute(
-                'INSERT INTO t_level1_prc (level1_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
+                'INSERT INTO ifs_level1_prc (level1_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
                     VALUES(?,?,?,?,?,?,?)',
                 (rec.level1_id, rec.pipeline_id, rec.prc_module, rec.params_file_path, rec.prc_status, rec.prc_time, rec.result_file_path)
             )

csst_dfs_api_local/msc/__init__.py

-from .level1 import Level1DataApi
+from .calmerge import CalMergeApi
\ No newline at end of file
+from .level0 import Level0DataApi
+from .level0prc import Level0PrcApi
+from .level1 import Level1DataApi
+from .level1prc import Level1PrcApi
+from .level2catalog import Level2CatalogApi
\ No newline at end of file

csst_dfs_api_local/facility/calmerge.py → csst_dfs_api_local/msc/calmerge.py

@@ -6,14 +6,15 @@ import shutil
 from ..common.db import DBClient
 from ..common.utils import *
 from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.facility import CalMergeRecord
+from csst_dfs_commons.models.msc import CalMergeRecord
 from csst_dfs_commons.models.common import from_dict_list
 from .level0 import Level0DataApi
 
 log = logging.getLogger('csst')
 
 class CalMergeApi(object):
-    def __init__(self):
+    def __init__(self, sub_system = "msc"):
+        self.sub_system = sub_system
         self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
         self.db = DBClient()
         self.level0Api = Level0DataApi()
@@ -35,14 +36,14 @@ class CalMergeApi(object):
             if level0_data is None:
                 return Result.error(message = "level0 data [%s]not found"%(level0_id))
             
-            sql_data = f"select * from t_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time >= '{level0_data.data.obs_time}' order by obs_time ASC limit 1"
+            sql_data = f"select * from msc_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time >= '{level0_data.data.obs_time}' order by obs_time ASC limit 1"
 
             r = self.db.select_one(sql_data)
             if r:
                 rec = CalMergeRecord().from_dict(r)
                 return Result.ok_data(data=rec)
             
-            sql_data = f"select * from t_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time <= '{level0_data.data.obs_time}' order by obs_time DESC limit 1"
+            sql_data = f"select * from msc_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time <= '{level0_data.data.obs_time}' order by obs_time DESC limit 1"
 
             r = self.db.select_one(sql_data)
             if r:
@@ -78,8 +79,8 @@ class CalMergeApi(object):
             file_name = get_parameter(kwargs, "file_name")
             limit = get_parameter(kwargs, "limit", 0)
 
-            sql_count = "select count(*) as c from t_cal_merge where 1=1"
-            sql_data = f"select * from t_cal_merge where 1=1"
+            sql_count = "select count(*) as c from msc_cal_merge where 1=1"
+            sql_data = f"select * from msc_cal_merge where 1=1"
 
             sql_condition = "" 
             if detector_no:
@@ -138,10 +139,10 @@ class CalMergeApi(object):
     def get_by_id(self, id: str):
         try:
             r = self.db.select_one(
-                "select * from t_cal_merge where id=?", (id,))
+                "select * from msc_cal_merge where id=?", (id,))
             if r:
 
-                sql_get_level0_id = f"select level0_id from t_cal2level0 where merge_id={r['id']}" 
+                sql_get_level0_id = f"select level0_id from msc_cal2level0 where merge_id={r['id']}" 
                 _, records = self.db.select_many(sql_get_level0_id)
                 level0_ids = [r["level0_id"] for r in records]
 
@@ -158,10 +159,10 @@ class CalMergeApi(object):
     def get_by_cal_id(self, cal_id: str):
         try:
             r = self.db.select_one(
-                "select * from t_cal_merge where cal_id=?", (cal_id,))
+                "select * from msc_cal_merge where cal_id=?", (cal_id,))
             if r:
 
-                sql_get_level0_id = f"select level0_id from t_cal2level0 where merge_id={r['id']}" 
+                sql_get_level0_id = f"select level0_id from msc_cal2level0 where merge_id={r['id']}" 
                 _, records = self.db.select_many(sql_get_level0_id)
                 level0_ids = [r["level0_id"] for r in records]
 
@@ -197,7 +198,7 @@ class CalMergeApi(object):
 
         try:
             self.db.execute(
-                'update t_cal_merge set qc1_status=?, qc1_time=? where id=?',
+                'update msc_cal_merge set qc1_status=?, qc1_time=? where id=?',
                 (status, format_time_ms(time.time()), id)
             )  
             self.db.end() 
@@ -229,14 +230,14 @@ class CalMergeApi(object):
 
         try:
             existed = self.db.exists(
-                "select * from t_cal_merge where id=?",
+                "select * from msc_cal_merge where id=?",
                 (id,)
             )
             if not existed:
                 log.warning('%s not found' %(id, ))
                 return Result.error(message ='%s not found' %(id, ))
             self.db.execute(
-                'update t_cal_merge set prc_status=?, prc_time=? where id=?',
+                'update msc_cal_merge set prc_status=?, prc_time=? where id=?',
                 (status, format_time_ms(time.time()), id)
             )
             self.db.end() 
@@ -278,14 +279,14 @@ class CalMergeApi(object):
         )
         try:
             self.db.execute(
-                'INSERT INTO t_cal_merge (cal_id,detector_no,ref_type,obs_time,exp_time,filename,file_path,prc_status,prc_time,create_time) \
+                'INSERT INTO msc_cal_merge (cal_id,detector_no,ref_type,obs_time,exp_time,filename,file_path,prc_status,prc_time,create_time) \
                     VALUES(?,?,?,?,?,?,?,?,?,?)',
                 (rec.cal_id, rec.detector_no, rec.ref_type, rec.obs_time, rec.exp_time, rec.filename, rec.file_path,rec.prc_status,rec.prc_time,format_time_ms(time.time()))
             )
             self.db.end()
             rec.id = self.db.last_row_id()
 
-            sql_level0_ids = "insert into t_cal2level0 (merge_id,level0_id) values "
+            sql_level0_ids = "insert into msc_cal2level0 (merge_id,level0_id) values "
             values = ["(%s,%s)"%(rec.id,i) for i in rec.level0_ids]
             _ = self.db.execute(sql_level0_ids + ",".join(values))            
 

csst_dfs_api_local/msc/ingest.py

+import os, sys
+import argparse
+import logging
+from astropy.io import fits
+import datetime
+import shutil
+
+from csst_dfs_api_local.common.db import DBClient
+log = logging.getLogger('csst-dfs-api-local')
+
+def ingest():
+    
+    parser = argparse.ArgumentParser(prog=f"{sys.argv[0]}", description="ingest the local files")
+    parser.add_argument('-i','--infile', dest="infile", help="a file or a directory")
+    parser.add_argument('-m', '--copyfiles', dest="copyfiles", action='store_true', default=False, help="whether copy files after import")
+    args = parser.parse_args(sys.argv[1:])
+    
+    import_root_dir = args.infile
+    if import_root_dir is None or (not os.path.isfile(import_root_dir) and not os.path.isdir(import_root_dir)):
+        parser.print_help()
+        sys.exit(0)
+
+    db = DBClient()
+    if os.path.isfile(import_root_dir):
+        log.info(f"prepare import {import_root_dir}")
+        ingest_one(import_root_dir, db, args.copyfiles)
+    if os.path.isdir(import_root_dir):
+        for (path, _, file_names) in os.walk(import_root_dir):
+            for filename in file_names:
+                if filename.find(".fits") > 0:
+                    file_full_path = os.path.join(path, filename)
+                    log.info(f"prepare import {file_full_path}")
+                    ingest_one(file_full_path, db, args.copyfiles)
+
+    db.close()
+
+def ingest_one(file_path, db, copyfiles):
+    dest_root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+
+    hdul = fits.open(file_path)
+    header = hdul[0].header
+
+    obs_id = header["OBSID"]
+    exp_start_time = f"{header['DATE-OBS']} {header['TIME-OBS']}"
+    exp_time = header['EXPTIME']
+    
+    module_id = header["INSTRUME"]
+    obs_type = header["FILETYPE"]
+    qc0_status = -1
+    prc_status = -1
+    time_now = datetime.datetime.now()
+    create_time = time_now.strftime('%Y-%m-%d %H:%M:%S')
+
+    facility_status_id = 0
+    module_status_id = 0
+
+    existed = db.exists("select * from t_observation where obs_id=?", (obs_id,))
+    if not existed:
+        db.execute("insert into t_observation \
+            (obs_id,obs_time,exp_time,module_id,obs_type,facility_status_id, module_status_id, qc0_status, prc_status,create_time) \
+            values (?,?,?,?,?,?,?,?,?,?)",
+        (obs_id,exp_start_time,exp_time,module_id,obs_type,facility_status_id,module_status_id,qc0_status, prc_status,create_time))
+        db.end()
+    #level0
+    detector = header["DETECTOR"]
+    filename = header["FILENAME"]
+    
+    existed = db.exists(
+            "select * from t_level0_data where filename=?",
+            (filename,)
+        )
+    if existed:
+        log.warning('%s has already been imported' %(file_path, ))
+        db.end()
+        return    
+
+    detector_status_id = 0
+
+    file_full_path = file_path
+
+    if copyfiles:
+        file_dir = f"{dest_root_dir}/{module_id}/{obs_type.upper()}/{header['EXPSTART']}/{obs_id}/MS"
+        if not os.path.exists(file_dir):
+            os.makedirs(file_dir)
+        file_full_path = f"{file_dir}/{filename}.fits"
+
+    level0_id = f"{obs_id}{detector}"  
+
+    c = db.execute("insert into msc_level0_data \
+        (level0_id, obs_id, detector_no, obs_type, obs_time, exp_time,detector_status_id, filename, file_path,qc0_status, prc_status,create_time) \
+        values (?,?,?,?,?,?,?,?,?,?,?,?)",
+        (level0_id, obs_id, detector, obs_type, exp_start_time, exp_time, detector_status_id, filename, file_full_path, qc0_status, prc_status,create_time))
+    db.end()
+    level0_id_id = db.last_row_id()
+    #level0-header
+    ra_obj = header["RA_OBJ"]
+    dec_obj = header["DEC_OBJ"]
+    db.execute("delete from msc_level0_header where id=?",(level0_id_id,))    
+    db.execute("insert into msc_level0_header \
+        (id, obs_time, exp_time, ra, `dec`, create_time) \
+        values (?,?,?,?,?,?)",
+        (level0_id_id, exp_start_time, exp_time, ra_obj, dec_obj, create_time))
+    
+    if copyfiles:
+        #copy files
+        shutil.copyfile(file_path, file_full_path)
+
+    db.end()
+
+    print(f"{file_path} imported")
+
+if __name__ == "__main__":
+    ingest()
\ No newline at end of file

csst_dfs_api_local/msc/level0.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.msc import Level0Record
+from csst_dfs_commons.models.common import from_dict_list
+
+log = logging.getLogger('csst')
+
+class Level0DataApi(object):
+    def __init__(self, sub_system = "msc"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+
+    def find(self, **kwargs):
+        ''' retrieve level0 records from database
+
+        parameter kwargs:
+            obs_id: [str]
+            detector_no: [str]
+            obs_type: [str]
+            obs_time : (start, end),
+            qc0_status : [int],
+            prc_status : [int],
+            file_name: [str]
+            limit: limits returns the number of records,default 0:no-limit
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            obs_id = get_parameter(kwargs, "obs_id")
+            detector_no = get_parameter(kwargs, "detector_no")
+            obs_type = get_parameter(kwargs, "obs_type")
+            exp_time_start = get_parameter(kwargs, "obs_time", [None, None])[0]
+            exp_time_end = get_parameter(kwargs, "obs_time", [None, None])[1]
+            qc0_status = get_parameter(kwargs, "qc0_status")
+            prc_status = get_parameter(kwargs, "prc_status")
+            file_name = get_parameter(kwargs, "file_name")
+            limit = get_parameter(kwargs, "limit", 0)
+
+            sql_count = "select count(*) as c from msc_level0_data where 1=1"
+            sql_data = f"select * from msc_level0_data where 1=1"
+
+            sql_condition = "" 
+            if obs_id:
+                sql_condition = f"{sql_condition} and obs_id='{obs_id}'"              
+            if detector_no:
+                sql_condition = f"{sql_condition} and detector_no='{detector_no}'"
+            if obs_type:
+                sql_condition = f"{sql_condition} and obs_type='{obs_type}'"
+            if exp_time_start:
+                sql_condition = f"{sql_condition} and obs_time >='{exp_time_start}'"
+            if exp_time_end:
+                sql_condition = f"{sql_condition} and obs_time <='{exp_time_end}'"
+            if qc0_status:
+                sql_condition = f"{sql_condition} and qc0_status={qc0_status}"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+            if file_name:
+                sql_condition = f" and filename='{file_name}'"  
+
+            sql_count = f"{sql_count} {sql_condition}"
+            sql_data = f"{sql_data} {sql_condition}"
+
+            if limit > 0:
+                sql_data = f"{sql_data} limit {limit}"   
+
+            totalCount = self.db.select_one(sql_count)
+            _, records = self.db.select_many(sql_data)
+
+            return Result.ok_data(data=from_dict_list(Level0Record, records)).append("totalCount", totalCount['c'])
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def get(self, **kwargs):
+        '''  fetch a record from database
+
+        parameter kwargs:
+            id : [int],
+            level0_id : [str]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        level0_id = get_parameter(kwargs, "level0_id", "")
+
+        if id == 0 and level0_id == "":
+            return Result.error(message="at least define id or level0_id") 
+
+        if id != 0: 
+            return self.get_by_id(id)
+        if level0_id != "": 
+            return self.get_by_level0_id(level0_id)
+
+    def get_by_id(self, id: int):
+        try:
+            r = self.db.select_one(
+                "select * from msc_level0_data where id=?", (id,))
+            if r:
+                return Result.ok_data(data=Level0Record().from_dict(r))
+            else:
+                return Result.error(message=f"id:{id} not found")  
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))  
+
+    def get_by_level0_id(self, level0_id: str):
+        try:
+            r = self.db.select_one(
+                "select * from msc_level0_data where level0_id=?", (level0_id,))
+            if r:
+                return Result.ok_data(data=Level0Record().from_dict(r))
+            else:
+                return Result.error(message=f"level0_id:{level0_id} not found")  
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))  
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            level0_id : [str],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id")
+        level0_id = get_parameter(kwargs, "level0_id")
+        result = self.get(id = id, level0_id = level0_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+        try:
+            self.db.execute(
+                'update msc_level0_data set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def update_qc0_status(self, **kwargs):
+        ''' update the status of QC0
+        
+        parameter kwargs:
+            id : [int],
+            level0_id : [str],
+            status : [int]
+        '''        
+        id = get_parameter(kwargs, "id")
+        level0_id = get_parameter(kwargs, "level0_id")
+        result = self.get(id = id, level0_id = level0_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+        try:
+            self.db.execute(
+                'update msc_level0_data set qc0_status=?, qc0_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a level0 data record into database
+ 
+        parameter kwargs:
+            obs_id = [str]
+            detector_no = [str]
+            obs_type = [str]        
+            obs_time = [str]
+            exp_time = [int]
+            detector_status_id = [int]
+            filename = [str]
+            file_path = [str]
+        return: csst_dfs_common.models.Result
+        '''          
+        rec = Level0Record(
+            obs_id = get_parameter(kwargs, "obs_id"),
+            detector_no = get_parameter(kwargs, "detector_no"),
+            obs_type = get_parameter(kwargs, "obs_type"),
+            obs_time = get_parameter(kwargs, "obs_time"),
+            exp_time = get_parameter(kwargs, "exp_time"),
+            detector_status_id = get_parameter(kwargs, "detector_status_id"),
+            filename = get_parameter(kwargs, "filename"),
+            file_path = get_parameter(kwargs, "file_path")
+        )
+        rec.level0_id = f"{rec.obs_id}{rec.detector_no}"
+        try:
+            existed = self.db.exists(
+                    "select * from msc_level0_data where filename=?",
+                    (rec.filename,)
+                )
+            if existed:
+                log.warning('%s existed' %(rec.filename, ))
+                return Result.error(message ='%s existed' %(rec.filename, ))
+
+            self.db.execute(
+                'INSERT INTO msc_level0_data (level0_id, obs_id, detector_no, obs_type, obs_time, exp_time,detector_status_id, filename, file_path,qc0_status, prc_status,create_time) \
+                    VALUES(?,?,?,?,?,?,?,?,?,?,?,?)',
+                (rec.level0_id, rec.obs_id, rec.detector_no, rec.obs_type, rec.obs_time, rec.exp_time, rec.detector_status_id, rec.filename, rec.file_path,-1,-1,format_time_ms(time.time()))
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
+
+

csst_dfs_api_local/msc/level0prc.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.msc import Level0PrcRecord
+from csst_dfs_commons.models.common import from_dict_list
+
+log = logging.getLogger('csst')
+
+class Level0PrcApi(object):
+    def __init__(self, sub_system = "msc"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+
+    def find(self, **kwargs):
+        ''' retrieve level0 procedure records from database
+
+        parameter kwargs:
+            level0_id: [str]
+            pipeline_id: [str]
+            prc_module: [str]
+            prc_status : [int]
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            level0_id = get_parameter(kwargs, "level0_id")
+            pipeline_id = get_parameter(kwargs, "pipeline_id")
+            prc_module = get_parameter(kwargs, "prc_module")
+            prc_status = get_parameter(kwargs, "prc_status")
+
+            sql_data = f"select * from msc_level0_prc"
+
+            sql_condition = f"where level0_id='{level0_id}'"
+            if pipeline_id:
+                sql_condition = sql_condition + " and pipeline_id='" + pipeline_id + "'"
+            if prc_module:
+                sql_condition = sql_condition + " and prc_module ='" + prc_module + "'"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+
+            sql_data = f"{sql_data} {sql_condition}"
+
+            _, records = self.db.select_many(sql_data)
+            return Result.ok_data(data=from_dict_list(Level0PrcRecord, records)).append("totalCount", len(records))
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id")
+        status = get_parameter(kwargs, "status")
+
+        try:
+            existed = self.db.exists(
+                "select * from msc_level0_prc where id=?",
+                (id,)
+            )
+            if not existed:
+                log.warning('%s not found' %(id, ))
+                return Result.error(message ='%s not found' %(id, ))
+            self.db.execute(
+                'update msc_level0_prc set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a level0 procedure record into database
+ 
+        parameter kwargs:
+            level0_id : [str]
+            pipeline_id : [str]
+            prc_module : [str]
+            params_file_path : [str]
+            prc_status : [int]
+            prc_time : [str]
+            result_file_path : [str]
+        return csst_dfs_common.models.Result
+        '''   
+
+        rec = Level0PrcRecord(
+            id = 0,
+            level0_id = get_parameter(kwargs, "level0_id"),
+            pipeline_id = get_parameter(kwargs, "pipeline_id"),
+            prc_module = get_parameter(kwargs, "prc_module"),
+            params_file_path = get_parameter(kwargs, "params_file_path"),
+            prc_status = get_parameter(kwargs, "prc_status", -1),
+            prc_time = get_parameter(kwargs, "prc_time"),
+            result_file_path = get_parameter(kwargs, "result_file_path")
+        )
+        try:
+            self.db.execute(
+                'INSERT INTO msc_level0_prc (level0_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
+                    VALUES(?,?,?,?,?,?,?)',
+                (rec.level0_id, rec.pipeline_id, rec.prc_module, rec.params_file_path, rec.prc_status, rec.prc_time, rec.result_file_path)
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
+

csst_dfs_api_local/msc/level1prc.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.msc import Level1PrcRecord
+from csst_dfs_commons.models.common import from_dict_list
+
+log = logging.getLogger('csst')
+
+class Level1PrcApi(object):
+    def __init__(self, sub_system = "msc"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+
+    def find(self, **kwargs):
+        ''' retrieve level1 procedure records from database
+
+        parameter kwargs:
+            level1_id: [int]
+            pipeline_id: [str]
+            prc_module: [str]
+            prc_status : [int]
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            level1_id = get_parameter(kwargs, "level1_id", 0)
+            pipeline_id = get_parameter(kwargs, "pipeline_id")
+            prc_module = get_parameter(kwargs, "prc_module")
+            prc_status = get_parameter(kwargs, "prc_status")
+
+            sql_data = f"select * from msc_level1_prc"
+
+            sql_condition = f"where level1_id={level1_id}"
+            if pipeline_id:
+                sql_condition = sql_condition + " and pipeline_id='" + pipeline_id + "'"
+            if prc_module:
+                sql_condition = sql_condition + " and prc_module ='" + prc_module + "'"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+
+            sql_data = f"{sql_data} {sql_condition}"
+
+            _, records = self.db.select_many(sql_data)
+            return Result.ok_data(data=from_dict_list(Level1PrcRecord, records)).append("totalCount", len(records))
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id")
+        status = get_parameter(kwargs, "status")
+
+        try:
+            existed = self.db.exists(
+                "select * from msc_level1_prc where id=?",
+                (id,)
+            )
+            if not existed:
+                log.warning('%s not found' %(id, ))
+                return Result.error(message ='%s not found' %(id, ))
+            self.db.execute(
+                'update msc_level1_prc set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a level1 procedure record into database
+ 
+        parameter kwargs:
+            level1_id : [int]
+            pipeline_id : [str]
+            prc_module : [str]
+            params_file_path : [str]
+            prc_status : [int]
+            prc_time : [str]
+            result_file_path : [str]
+        return csst_dfs_common.models.Result
+        '''   
+
+        rec = Level1PrcRecord(
+            id = 0,
+            level1_id = get_parameter(kwargs, "level1_id", 0),
+            pipeline_id = get_parameter(kwargs, "pipeline_id"),
+            prc_module = get_parameter(kwargs, "prc_module"),
+            params_file_path = get_parameter(kwargs, "params_file_path"),
+            prc_status = get_parameter(kwargs, "prc_status", -1),
+            prc_time = get_parameter(kwargs, "prc_time"),
+            result_file_path = get_parameter(kwargs, "result_file_path")
+        )
+        try:
+            self.db.execute(
+                'INSERT INTO msc_level1_prc (level1_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
+                    VALUES(?,?,?,?,?,?,?)',
+                (rec.level1_id, rec.pipeline_id, rec.prc_module, rec.params_file_path, rec.prc_status, rec.prc_time, rec.result_file_path)
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
+

csst_dfs_api_local/msc/level2catalog.py

+import os
+import logging
+import time, datetime
+import shutil
+from traceback import print_stack
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.common import from_dict_list
+from csst_dfs_commons.models.msc import MSCLevel2CatalogRecord
+
+log = logging.getLogger('csst')
+
+class Level2CatalogApi(object):
+    def __init__(self, sub_system = "msc"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+
+    def find(self, **kwargs):
+        ''' retrieve level1 records from database
+
+        parameter kwargs:
+            obs_id: [str]
+            detector_no: [str]
+            obs_time : (start, end),
+            filename: [str]
+            limit: limits returns the number of records,default 0:no-limit
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            obs_id = get_parameter(kwargs, "obs_id")
+            detector_no = get_parameter(kwargs, "detector_no")
+            obs_time_start = get_parameter(kwargs, "obs_time", [None, None])[0]
+            obs_time_end = get_parameter(kwargs, "obs_time", [None, None])[1]
+            limit = get_parameter(kwargs, "limit", 0)
+
+            sql_count = "select count(*) as c from msc_level2_catalog where 1=1"
+            sql_data = f"select * from msc_level2_catalog where 1=1"
+
+            sql_condition = "" 
+            if obs_id:
+                sql_condition = f"{sql_condition} and obs_id='{obs_id}'"
+            if detector_no:
+                sql_condition = f"{sql_condition} and detector_no='{detector_no}'"
+            if obs_time_start:
+                sql_condition = f"{sql_condition} and obs_time >='{obs_time_start}'"
+            if obs_time_end:
+                sql_condition = f"{sql_condition} and obs_time <='{obs_time_end}'"
+
+            sql_count = f"{sql_count} {sql_condition}"
+            sql_data = f"{sql_data} {sql_condition}"
+
+            if limit > 0:
+                sql_data = f"{sql_data} limit {limit}"  
+
+            totalCount = self.db.select_one(sql_count)
+            _, recs = self.db.select_many(sql_data)
+            return Result.ok_data(data=from_dict_list(MSCLevel2CatalogRecord, recs)).append("totalCount", totalCount['c'])
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def write(self, records):
+       
+        try:
+            recordStrs = []
+            for recordStr in records:
+                recordStrs.append(f"({recordStr})")
+            if recordStrs:
+                self.db.execute(
+                f"insert into msc_level2_catalog values {','.join(recordStrs)}" 
+                )
+                self.db.end()
+
+            return Result.ok_data()
+        except Exception as e:
+            print_stack()
+            log.error(e)
+            return Result.error(message=str(e))            
\ No newline at end of file

csst_dfs_api_local/sls/calmerge.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.sls import CalMergeRecord
+from csst_dfs_commons.models.common import from_dict_list
+from .level0 import Level0DataApi
+
+log = logging.getLogger('csst')
+
+class CalMergeApi(object):
+    def __init__(self, sub_system = "sls"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+        self.level0Api = Level0DataApi()
+
+    def get_latest_by_l0(self, **kwargs):
+        ''' retrieve calibration merge records from database by level0 data
+        
+        :param kwargs: Parameter dictionary, key items support:
+            level0_id: [str],
+            ref_type: [str]
+
+        :returns: csst_dfs_common.models.Result
+        '''
+        try:
+            level0_id = get_parameter(kwargs, "level0_id")
+            ref_type = get_parameter(kwargs, "ref_type")
+
+            level0_data = self.level0Api.get_by_level0_id(level0_id)
+            if level0_data is None:
+                return Result.error(message = "level0 data [%s]not found"%(level0_id))
+            
+            sql_data = f"select * from sls_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time >= '{level0_data.data.obs_time}' order by obs_time ASC limit 1"
+
+            r = self.db.select_one(sql_data)
+            if r:
+                rec = CalMergeRecord().from_dict(r)
+                return Result.ok_data(data=rec)
+            
+            sql_data = f"select * from sls_cal_merge where detector_no='{level0_data.data.detector_no}' and ref_type='{ref_type}' and obs_time <= '{level0_data.data.obs_time}' order by obs_time DESC limit 1"
+
+            r = self.db.select_one(sql_data)
+            if r:
+                rec = CalMergeRecord().from_dict(r)
+                return Result.ok_data(data=rec)            
+
+            return Result.error(message = "not found")
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def find(self, **kwargs):
+        ''' retrieve calibration merge records from database
+
+        parameter kwargs:
+            detector_no: [str]
+            ref_type: [str]
+            obs_time: (start,end)
+            qc1_status : [int]
+            prc_status : [int]
+            file_name: [str]
+            limit: limits returns the number of records,default 0:no-limit
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            detector_no = get_parameter(kwargs, "detector_no")
+            ref_type = get_parameter(kwargs, "ref_type")
+            exp_time_start = get_parameter(kwargs, "obs_time", [None, None])[0]
+            exp_time_end = get_parameter(kwargs, "obs_time", [None, None])[1]              
+            qc1_status = get_parameter(kwargs, "qc1_status")
+            prc_status = get_parameter(kwargs, "prc_status")
+            file_name = get_parameter(kwargs, "file_name")
+            limit = get_parameter(kwargs, "limit", 0)
+
+            sql_count = "select count(*) as c from sls_cal_merge where 1=1"
+            sql_data = f"select * from sls_cal_merge where 1=1"
+
+            sql_condition = "" 
+            if detector_no:
+                sql_condition = f"{sql_condition} and detector_no='{detector_no}'"
+            if ref_type:
+                sql_condition = f"{sql_condition} and ref_type='{ref_type}'"
+            if exp_time_start:
+                sql_condition = f"{sql_condition} and obs_time >='{exp_time_start}'"
+            if exp_time_end:
+                sql_condition = f"{sql_condition} and obs_time <='{exp_time_end}'"
+            if qc1_status:
+                sql_condition = f"{sql_condition} and qc1_status={qc1_status}"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+
+            if file_name:
+                sql_condition = f" and filename={file_name}"  
+
+            sql_count = f"{sql_count} {sql_condition}"
+            sql_data = f"{sql_data} {sql_condition}"
+
+            log.info(sql_count)
+            log.info(sql_data)
+
+            if limit > 0:
+                sql_data = f"{sql_data} limit {limit}"   
+
+            totalCount = self.db.select_one(sql_count)
+            _, records = self.db.select_many(sql_data)
+
+            return Result.ok_data(data=from_dict_list(CalMergeRecord, records)).append("totalCount", totalCount['c'])
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def get(self, **kwargs):
+        '''  fetch a record from database
+
+        parameter kwargs:
+            id : [int],
+            cal_id : [str]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        cal_id = get_parameter(kwargs, "cal_id", "")
+
+        if id == 0 and cal_id == "":
+            return Result.error(message="at least define id or cal_id") 
+
+        if id != 0: 
+            return self.get_by_id(id)
+        if cal_id != "": 
+            return self.get_by_cal_id(cal_id)
+
+    def get_by_id(self, id: str):
+        try:
+            r = self.db.select_one(
+                "select * from sls_cal_merge where id=?", (id,))
+            if r:
+
+                sql_get_level0_id = f"select level0_id from sls_cal2level0 where merge_id={r['id']}" 
+                _, records = self.db.select_many(sql_get_level0_id)
+                level0_ids = [r["level0_id"] for r in records]
+
+                rec = CalMergeRecord().from_dict(r)
+                rec.level0_ids = level0_ids
+                return Result.ok_data(data=rec)
+            else:
+                return Result.error(message=f"id:{id} not found")
+                    
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def get_by_cal_id(self, cal_id: str):
+        try:
+            r = self.db.select_one(
+                "select * from sls_cal_merge where cal_id=?", (cal_id,))
+            if r:
+
+                sql_get_level0_id = f"select level0_id from sls_cal2level0 where merge_id={r['id']}" 
+                _, records = self.db.select_many(sql_get_level0_id)
+                level0_ids = [r["level0_id"] for r in records]
+
+                rec = CalMergeRecord().from_dict(r)
+                rec.level0_ids = level0_ids
+                return Result.ok_data(data=rec)
+            else:
+                return Result.error(message=f"id:{id} not found")
+                    
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def update_qc1_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            cal_id : [str],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        cal_id = get_parameter(kwargs, "cal_id", "")
+        result = self.get(id = id, cal_id = cal_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+
+        try:
+            self.db.execute(
+                'update sls_cal_merge set qc1_status=?, qc1_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            cal_id : [str],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        cal_id = get_parameter(kwargs, "cal_id", "")
+        result = self.get(id = id, cal_id = cal_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+
+        try:
+            existed = self.db.exists(
+                "select * from sls_cal_merge where id=?",
+                (id,)
+            )
+            if not existed:
+                log.warning('%s not found' %(id, ))
+                return Result.error(message ='%s not found' %(id, ))
+            self.db.execute(
+                'update sls_cal_merge set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a calibration merge record into database
+ 
+        parameter kwargs:
+            cal_id : [str]
+            detector_no : [str]
+            ref_type : [str]
+            obs_time : [str]
+            exp_time : [float]
+            prc_status : [int]
+            prc_time : [str]
+            filename : [str]
+            file_path : [str]
+            level0_ids : [list]
+        return csst_dfs_common.models.Result
+        '''   
+
+        rec = CalMergeRecord(
+            id = 0,
+            cal_id =  get_parameter(kwargs, "cal_id"),
+            detector_no = get_parameter(kwargs, "detector_no"),
+            ref_type = get_parameter(kwargs, "ref_type"),
+            obs_time = get_parameter(kwargs, "obs_time"),
+            exp_time = get_parameter(kwargs, "exp_time"),
+            filename = get_parameter(kwargs, "filename"),
+            file_path = get_parameter(kwargs, "file_path"),
+            prc_status = get_parameter(kwargs, "prc_status", -1),
+            prc_time = get_parameter(kwargs, "prc_time"),
+            level0_ids = get_parameter(kwargs, "level0_ids", [])
+        )
+        try:
+            self.db.execute(
+                'INSERT INTO sls_cal_merge (cal_id,detector_no,ref_type,obs_time,exp_time,filename,file_path,prc_status,prc_time,create_time) \
+                    VALUES(?,?,?,?,?,?,?,?,?,?)',
+                (rec.cal_id, rec.detector_no, rec.ref_type, rec.obs_time, rec.exp_time, rec.filename, rec.file_path,rec.prc_status,rec.prc_time,format_time_ms(time.time()))
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            sql_level0_ids = "insert into sls_cal2level0 (merge_id,level0_id) values "
+            values = ["(%s,%s)"%(rec.id,i) for i in rec.level0_ids]
+            _ = self.db.execute(sql_level0_ids + ",".join(values))            
+
+            self.db.end()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
\ No newline at end of file

csst_dfs_api_local/sls/level0.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.sls import Level0Record
+from csst_dfs_commons.models.common import from_dict_list
+
+log = logging.getLogger('csst')
+
+class Level0DataApi(object):
+    def __init__(self, sub_system = "sls"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+
+    def find(self, **kwargs):
+        ''' retrieve level0 records from database
+
+        parameter kwargs:
+            obs_id: [str]
+            detector_no: [str]
+            obs_type: [str]
+            obs_time : (start, end),
+            qc0_status : [int],
+            prc_status : [int],
+            file_name: [str]
+            limit: limits returns the number of records,default 0:no-limit
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            obs_id = get_parameter(kwargs, "obs_id")
+            detector_no = get_parameter(kwargs, "detector_no")
+            obs_type = get_parameter(kwargs, "obs_type")
+            exp_time_start = get_parameter(kwargs, "obs_time", [None, None])[0]
+            exp_time_end = get_parameter(kwargs, "obs_time", [None, None])[1]
+            qc0_status = get_parameter(kwargs, "qc0_status")
+            prc_status = get_parameter(kwargs, "prc_status")
+            file_name = get_parameter(kwargs, "file_name")
+            limit = get_parameter(kwargs, "limit", 0)
+
+            sql_count = "select count(*) as c from sls_level0_data where 1=1"
+            sql_data = f"select * from sls_level0_data where 1=1"
+
+            sql_condition = "" 
+            if obs_id:
+                sql_condition = f"{sql_condition} and obs_id='{obs_id}'"              
+            if detector_no:
+                sql_condition = f"{sql_condition} and detector_no='{detector_no}'"
+            if obs_type:
+                sql_condition = f"{sql_condition} and obs_type='{obs_type}'"
+            if exp_time_start:
+                sql_condition = f"{sql_condition} and obs_time >='{exp_time_start}'"
+            if exp_time_end:
+                sql_condition = f"{sql_condition} and obs_time <='{exp_time_end}'"
+            if qc0_status:
+                sql_condition = f"{sql_condition} and qc0_status={qc0_status}"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+            if file_name:
+                sql_condition = f" and filename='{file_name}'"  
+
+            sql_count = f"{sql_count} {sql_condition}"
+            sql_data = f"{sql_data} {sql_condition}"
+
+            if limit > 0:
+                sql_data = f"{sql_data} limit {limit}"   
+
+            totalCount = self.db.select_one(sql_count)
+            _, records = self.db.select_many(sql_data)
+
+            return Result.ok_data(data=from_dict_list(Level0Record, records)).append("totalCount", totalCount['c'])
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def get(self, **kwargs):
+        '''  fetch a record from database
+
+        parameter kwargs:
+            id : [int],
+            level0_id : [str]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id", 0)
+        level0_id = get_parameter(kwargs, "level0_id", "")
+
+        if id == 0 and level0_id == "":
+            return Result.error(message="at least define id or level0_id") 
+
+        if id != 0: 
+            return self.get_by_id(id)
+        if level0_id != "": 
+            return self.get_by_level0_id(level0_id)
+
+    def get_by_id(self, id: int):
+        try:
+            r = self.db.select_one(
+                "select * from sls_level0_data where id=?", (id,))
+            if r:
+                return Result.ok_data(data=Level0Record().from_dict(r))
+            else:
+                return Result.error(message=f"id:{id} not found")  
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))  
+
+    def get_by_level0_id(self, level0_id: str):
+        try:
+            r = self.db.select_one(
+                "select * from sls_level0_data where level0_id=?", (level0_id,))
+            if r:
+                return Result.ok_data(data=Level0Record().from_dict(r))
+            else:
+                return Result.error(message=f"level0_id:{level0_id} not found")  
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))  
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            level0_id : [str],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id")
+        level0_id = get_parameter(kwargs, "level0_id")
+        result = self.get(id = id, level0_id = level0_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+        try:
+            self.db.execute(
+                'update sls_level0_data set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def update_qc0_status(self, **kwargs):
+        ''' update the status of QC0
+        
+        parameter kwargs:
+            id : [int],
+            level0_id : [str],
+            status : [int]
+        '''        
+        id = get_parameter(kwargs, "id")
+        level0_id = get_parameter(kwargs, "level0_id")
+        result = self.get(id = id, level0_id = level0_id)
+
+        if not result.success:
+            return Result.error(message="not found")
+
+        id = result.data.id
+        status = get_parameter(kwargs, "status")
+        try:
+            self.db.execute(
+                'update sls_level0_data set qc0_status=?, qc0_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a level0 data record into database
+ 
+        parameter kwargs:
+            obs_id = [str]
+            detector_no = [str]
+            obs_type = [str]        
+            obs_time = [str]
+            exp_time = [int]
+            detector_status_id = [int]
+            filename = [str]
+            file_path = [str]
+        return: csst_dfs_common.models.Result
+        '''          
+        rec = Level0Record(
+            obs_id = get_parameter(kwargs, "obs_id"),
+            detector_no = get_parameter(kwargs, "detector_no"),
+            obs_type = get_parameter(kwargs, "obs_type"),
+            obs_time = get_parameter(kwargs, "obs_time"),
+            exp_time = get_parameter(kwargs, "exp_time"),
+            detector_status_id = get_parameter(kwargs, "detector_status_id"),
+            filename = get_parameter(kwargs, "filename"),
+            file_path = get_parameter(kwargs, "file_path")
+        )
+        rec.level0_id = f"{rec.obs_id}{rec.detector_no}"
+        try:
+            existed = self.db.exists(
+                    "select * from sls_level0_data where filename=?",
+                    (rec.filename,)
+                )
+            if existed:
+                log.warning('%s existed' %(rec.filename, ))
+                return Result.error(message ='%s existed' %(rec.filename, ))
+
+            self.db.execute(
+                'INSERT INTO sls_level0_data (level0_id, obs_id, detector_no, obs_type, obs_time, exp_time,detector_status_id, filename, file_path,qc0_status, prc_status,create_time) \
+                    VALUES(?,?,?,?,?,?,?,?,?,?,?,?)',
+                (rec.level0_id, rec.obs_id, rec.detector_no, rec.obs_type, rec.obs_time, rec.exp_time, rec.detector_status_id, rec.filename, rec.file_path,-1,-1,format_time_ms(time.time()))
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
+
+

csst_dfs_api_local/sls/level0prc.py

+import os
+import logging
+import time, datetime
+import shutil
+
+from ..common.db import DBClient
+from ..common.utils import *
+from csst_dfs_commons.models import Result
+from csst_dfs_commons.models.sls import Level0PrcRecord
+from csst_dfs_commons.models.common import from_dict_list
+
+log = logging.getLogger('csst')
+
+class Level0PrcApi(object):
+    def __init__(self, sub_system = "sls"):
+        self.sub_system = sub_system
+        self.root_dir = os.getenv("CSST_LOCAL_FILE_ROOT", "/opt/temp/csst")
+        self.db = DBClient()
+
+    def find(self, **kwargs):
+        ''' retrieve level0 procedure records from database
+
+        parameter kwargs:
+            level0_id: [str]
+            pipeline_id: [str]
+            prc_module: [str]
+            prc_status : [int]
+
+        return: csst_dfs_common.models.Result
+        '''
+        try:
+            level0_id = get_parameter(kwargs, "level0_id")
+            pipeline_id = get_parameter(kwargs, "pipeline_id")
+            prc_module = get_parameter(kwargs, "prc_module")
+            prc_status = get_parameter(kwargs, "prc_status")
+
+            sql_data = f"select * from sls_level0_prc"
+
+            sql_condition = f"where level0_id='{level0_id}'"
+            if pipeline_id:
+                sql_condition = sql_condition + " and pipeline_id='" + pipeline_id + "'"
+            if prc_module:
+                sql_condition = sql_condition + " and prc_module ='" + prc_module + "'"
+            if prc_status:
+                sql_condition = f"{sql_condition} and prc_status={prc_status}"   
+
+            sql_data = f"{sql_data} {sql_condition}"
+
+            _, records = self.db.select_many(sql_data)
+            return Result.ok_data(data=from_dict_list(Level0PrcRecord, records)).append("totalCount", len(records))
+
+        except Exception as e:
+            return Result.error(message=str(e))
+
+    def update_proc_status(self, **kwargs):
+        ''' update the status of reduction
+
+        parameter kwargs:
+            id : [int],
+            status : [int]
+
+        return csst_dfs_common.models.Result
+        '''
+        id = get_parameter(kwargs, "id")
+        status = get_parameter(kwargs, "status")
+
+        try:
+            existed = self.db.exists(
+                "select * from sls_level0_prc where id=?",
+                (id,)
+            )
+            if not existed:
+                log.warning('%s not found' %(id, ))
+                return Result.error(message ='%s not found' %(id, ))
+            self.db.execute(
+                'update sls_level0_prc set prc_status=?, prc_time=? where id=?',
+                (status, format_time_ms(time.time()), id)
+            )  
+            self.db.end() 
+            return Result.ok_data()
+           
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e))
+
+    def write(self, **kwargs):
+        ''' insert a level0 procedure record into database
+ 
+        parameter kwargs:
+            level0_id : [str]
+            pipeline_id : [str]
+            prc_module : [str]
+            params_file_path : [str]
+            prc_status : [int]
+            prc_time : [str]
+            result_file_path : [str]
+        return csst_dfs_common.models.Result
+        '''   
+
+        rec = Level0PrcRecord(
+            id = 0,
+            level0_id = get_parameter(kwargs, "level0_id"),
+            pipeline_id = get_parameter(kwargs, "pipeline_id"),
+            prc_module = get_parameter(kwargs, "prc_module"),
+            params_file_path = get_parameter(kwargs, "params_file_path"),
+            prc_status = get_parameter(kwargs, "prc_status", -1),
+            prc_time = get_parameter(kwargs, "prc_time"),
+            result_file_path = get_parameter(kwargs, "result_file_path")
+        )
+        try:
+            self.db.execute(
+                'INSERT INTO sls_level0_prc (level0_id,pipeline_id,prc_module, params_file_path, prc_status,prc_time,result_file_path) \
+                    VALUES(?,?,?,?,?,?,?)',
+                (rec.level0_id, rec.pipeline_id, rec.prc_module, rec.params_file_path, rec.prc_status, rec.prc_time, rec.result_file_path)
+            )
+            self.db.end()
+            rec.id = self.db.last_row_id()
+
+            return Result.ok_data(data=rec)
+
+        except Exception as e:
+            log.error(e)
+            return Result.error(message=str(e)) 
+