rm module level1

csst_dfs_api_cluster/cpic/__init__.py

-from .level1 import Level1DataApi
-from .level1prc import Level1PrcApi
\ No newline at end of file

csst_dfs_api_cluster/cpic/level1.py

-import os
-import grpc
-import datetime
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.cpic import Level1Record
-from csst_dfs_commons.models.constants import UPLOAD_CHUNK_SIZE
-from csst_dfs_proto.cpic.level1 import level1_pb2, level1_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1DataApi(object):
-    """
-    Level1 Data Operation Class
-    """    
-    def __init__(self):
-        self.stub = level1_pb2_grpc.Level1SrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 records from database
-
-        parameter kwargs:
-            level0_id: [str]
-            data_type: [str]
-            create_time : (start, end),
-            qc1_status : [int],
-            prc_status : [int],
-            filename: [str]
-            limit: limits returns the number of records,default 0:no-limit
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1_pb2.FindLevel1Req(
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type"),
-                create_time_start = get_parameter(kwargs, "create_time", [None, None])[0],
-                create_time_end = get_parameter(kwargs, "create_time", [None, None])[1],
-                qc1_status = get_parameter(kwargs, "qc1_status"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                filename = get_parameter(kwargs, "filename"),
-                limit = get_parameter(kwargs, "limit", 0),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def get(self, **kwargs):
-        '''  fetch a record from database
-
-        parameter kwargs:
-            id : [int] 
-
-        return csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Get.with_call(level1_pb2.GetLevel1Req(
-                id = get_parameter(kwargs, "id"),
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type") 
-            ),metadata = get_auth_headers())
-
-            if resp.record is None or resp.record.id == 0:
-                return Result.error(message=f"data not found")  
-
-            return Result.ok_data(data = Level1Record().from_proto_model(resp.record))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))   
-
-    def update_proc_status(self, **kwargs):
-        ''' update the status of reduction
-
-        parameter kwargs:
-            id : [int],
-            status : [int]
-
-        return csst_dfs_common.models.Result
-        '''
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1_pb2.UpdateProcStatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def update_qc1_status(self, **kwargs):
-        ''' update the status of QC0
-        
-        parameter kwargs:
-            id : [int],
-            status : [int]
-        '''        
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateQc1Status.with_call(
-                level1_pb2.UpdateQc1StatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def write(self, **kwargs):
-        ''' insert a level1 record into database
- 
-        parameter kwargs:
-            level0_id : [str]
-            data_type : [str]
-            cor_sci_id : [int]
-            prc_params : [str]
-            filename : [str]
-            file_path : [str]            
-            prc_status : [int]
-            prc_time : [str]
-            pipeline_id : [str]
-            refs: [dict]
-
-        return csst_dfs_common.models.Result
-        '''   
-
-        rec = level1_pb2.Level1Record(
-            id = 0,
-            level0_id = get_parameter(kwargs, "level0_id"),
-            data_type = get_parameter(kwargs, "data_type"),
-            cor_sci_id = get_parameter(kwargs, "cor_sci_id"),
-            prc_params = get_parameter(kwargs, "prc_params"),
-            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", format_datetime(datetime.now())),
-            pipeline_id = get_parameter(kwargs, "pipeline_id"),
-            refs = get_parameter(kwargs, "refs", {})
-        )
-        def stream(rec):
-            with open(rec.file_path, 'rb') as f:
-                while True:
-                    data = f.read(UPLOAD_CHUNK_SIZE)
-                    if not data:
-                        break
-                    yield level1_pb2.WriteLevel1Req(record = rec, data = data)
-        try:
-            if not rec.file_path:
-                return Result.error(message="file_path is blank")
-            if not os.path.exists(rec.file_path):
-                return Result.error(message="the file [%s] not existed" % (rec.file_path, ))
-            if not rec.filename:
-                rec.filename = os.path.basename(rec.file_path)
-
-            resp,_ = self.stub.Write.with_call(stream(rec),metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))

csst_dfs_api_cluster/cpic/level1prc.py

-import grpc
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.cpic import Level1PrcRecord
-
-from csst_dfs_proto.cpic.level1prc import level1prc_pb2, level1prc_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1PrcApi(object):
-    def __init__(self):
-        self.stub = level1prc_pb2_grpc.Level1PrcSrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 procedure records from database
-
-        parameter kwargs:
-            level1_id: [str]
-            pipeline_id: [str]
-            prc_module: [str]
-            prc_status : [int]
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1prc_pb2.FindLevel1PrcReq(
-                level1_id = get_parameter(kwargs, "level1_id"),
-                pipeline_id = get_parameter(kwargs, "pipeline_id"),
-                prc_module = get_parameter(kwargs, "prc_module"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data = from_proto_model_list(Level1PrcRecord, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1prc_pb2.UpdateProcStatusReq(id=id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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 = level1prc_pb2.Level1PrcRecord(
-            id = 0,
-            level1_id = get_parameter(kwargs, "level1_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")
-        )
-        req = level1prc_pb2.WriteLevel1PrcReq(record = rec)
-        try:
-            resp,_ = self.stub.Write.with_call(req,metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data = Level1PrcRecord().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-   
-

csst_dfs_api_cluster/facility/__init__.py

@@ -2,6 +2,7 @@ from .brick import BrickApi
 from .detector import DetectorApi
 from .level2producer import Level2ProducerApi
 from .observation import ObservationApi
-from .calmerge import CalMergeApi
+from .level1 import Level1DataApi
 from .level0 import Level0DataApi
-from .level0prc import Level0PrcApi
\ No newline at end of file
+from .level0prc import Level0PrcApi
+from .level1prc import Level1PrcApi
\ No newline at end of file

csst_dfs_api_cluster/facility/calmerge.py

-import grpc
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.facility import CalMergeRecord
-
-from csst_dfs_proto.facility.calmerge import calmerge_pb2, calmerge_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class CalMergeApi(object):
-    def __init__(self):
-        self.stub = calmerge_pb2_grpc.CalMergeSrvStub(ServiceProxy().channel())
-
-    def get_latest_by_l0(self, **kwargs):
-        ''' retrieve calibration merge records from database by level0 data
-
-        parameter kwargs:
-            level0_id: [str]
-            ref_type: [str]
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.GetLatestByL0.with_call(calmerge_pb2.GetLatestByL0Req(
-                level0_id = get_parameter(kwargs, "level0_id"),
-                ref_type = get_parameter(kwargs, "ref_type")),
-            metadata = get_auth_headers())
-
-            if resp.record.id == 0:
-                return Result.error(message=f"not found")  
-
-            return Result.ok_data(data=CalMergeRecord().from_proto_model(resp.record))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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:
-            resp, _ =  self.stub.Find.with_call(calmerge_pb2.FindCalMergeReq(
-                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"),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(CalMergeRecord,resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def get(self, **kwargs):
-        '''  fetch a record from database
-
-        :param kwargs: Parameter dictionary, key items support:
-            id : [int]
-        
-        :returns: csst_dfs_common.models.Result
-        '''
-        try:
-            id = get_parameter(kwargs, "id", 0)
-            cal_id = get_parameter(kwargs, "cal_id", "")
-            resp, _ =  self.stub.Get.with_call(calmerge_pb2.GetCalMergeReq(
-                id = id,
-                cal_id = cal_id
-            ),metadata = get_auth_headers())
-
-            if resp.record.id == 0:
-                return Result.error(message=f"not found")  
-
-            return Result.ok_data(data=CalMergeRecord().from_proto_model(resp.record))
-           
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))   
-
-    def update_qc1_status(self, **kwargs):
-        ''' update the status of reduction
-
-        parameter kwargs:
-            id : [int],
-            cal_id = cal_id,
-            status : [int]
-
-        return csst_dfs_common.models.Result
-        '''
-        id = get_parameter(kwargs, "id", 0)
-        cal_id = get_parameter(kwargs, "cal_id", "")
-        status = get_parameter(kwargs, "status")
-
-        try:
-            resp,_ = self.stub.UpdateQc1Status.with_call(
-                calmerge_pb2.UpdateQc1StatusReq(
-                    id = id, 
-                    cal_id = cal_id,
-                    status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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", "")
-        status = get_parameter(kwargs, "status")
-
-        try:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                calmerge_pb2.UpdateProcStatusReq(
-                    id = id,
-                    cal_id = cal_id,
-                    status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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 = calmerge_pb2.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", [])
-        )
-        req = calmerge_pb2.WriteCalMergeReq(record = rec)
-        try:
-            resp,_ = self.stub.Write.with_call(req,metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data=CalMergeRecord().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-

csst_dfs_api_cluster/facility/level0.py

@@ -17,14 +17,20 @@ class Level0DataApi(object):
         ''' retrieve level0 records from database
 
         parameter kwargs:
-            obs_id: [str]
-            detector_no: [str]
-            obs_type: [str]
-            filter: [str]
-            obs_time : (start, end)
-            qc0_status : [int]
-            prc_status : [int]
-            file_name: [str]
+            obs_id: [str],
+            module_id: [str]
+            detector_no: [str],
+            obs_type: [str],
+            filter: [str],
+            obs_time : (start, end),
+            qc0_status : [int],
+            prc_status : [int],
+            file_name: [str],
+            ra_obj: [float],
+            dec_obj: [float],
+            radius: [float],
+            object_name: [str],
+            version: [str],
             limit: limits returns the number of records,default 0:no-limit
 
         return: csst_dfs_common.models.Result
@@ -33,6 +39,7 @@ class Level0DataApi(object):
             resp, _ =  self.stub.Find.with_call(level0_pb2.FindLevel0DataReq(
                 obs_id = get_parameter(kwargs, "obs_id"),
                 detector_no = get_parameter(kwargs, "detector_no"),
+                module_id = get_parameter(kwargs, "module_id"),
                 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],

csst_dfs_api_cluster/mbi/level1.py → csst_dfs_api_cluster/facility/level1.py

@@ -4,9 +4,9 @@ import datetime
 
 from csst_dfs_commons.models import Result
 from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.msc import Level1Record
+from csst_dfs_commons.models.facility import Level1Record
 from csst_dfs_commons.models.constants import UPLOAD_CHUNK_SIZE
-from csst_dfs_proto.msc.level1 import level1_pb2, level1_pb2_grpc
+from csst_dfs_proto.facility.level1 import level1_pb2, level1_pb2_grpc
 
 from ..common.service import ServiceProxy
 from ..common.utils import *
@@ -23,6 +23,7 @@ class Level1DataApi(object):
 
         parameter kwargs:
             level0_id: [str]
+            module_id: [str]
             data_type: [str]
             create_time : (start, end),
             qc1_status : [int],
@@ -35,6 +36,7 @@ class Level1DataApi(object):
         try:
             resp, _ =  self.stub.Find.with_call(level1_pb2.FindLevel1Req(
                 level0_id = get_parameter(kwargs, "level0_id"),
+                module_id = get_parameter(kwargs, "module_id"),
                 data_type = get_parameter(kwargs, "data_type"),
                 create_time_start = get_parameter(kwargs, "create_time", [None, None])[0],
                 create_time_end = get_parameter(kwargs, "create_time", [None, None])[1],
@@ -162,6 +164,7 @@ class Level1DataApi(object):
         rec = level1_pb2.Level1Record(
             id = 0,
             level0_id = get_parameter(kwargs, "level0_id"),
+            module_id = get_parameter(kwargs, "module_id", ''),
             data_type = get_parameter(kwargs, "data_type"),
             cor_sci_id = get_parameter(kwargs, "cor_sci_id"),
             prc_params = get_parameter(kwargs, "prc_params"),

csst_dfs_api_cluster/ifs/level1prc.py → csst_dfs_api_cluster/facility/level1prc.py

@@ -2,9 +2,9 @@ import grpc
 
 from csst_dfs_commons.models import Result
 from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.ifs import Level1PrcRecord
+from csst_dfs_commons.models.facility import Level1PrcRecord
 
-from csst_dfs_proto.ifs.level1prc import level1prc_pb2, level1prc_pb2_grpc
+from csst_dfs_proto.facility.level1prc import level1prc_pb2, level1prc_pb2_grpc
 
 from ..common.service import ServiceProxy
 from ..common.utils import *

csst_dfs_api_cluster/hstdm/__init__.py

-from .level1 import Level1DataApi
-from .level1prc import Level1PrcApi
\ No newline at end of file

csst_dfs_api_cluster/hstdm/level1.py

-import os
-import grpc
-import datetime
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.hstdm import Level1Record
-from csst_dfs_commons.models.constants import UPLOAD_CHUNK_SIZE
-from csst_dfs_proto.hstdm.level1 import level1_pb2, level1_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1DataApi(object):
-    """
-    Level1 Data Operation Class
-    """    
-    def __init__(self):
-        self.stub = level1_pb2_grpc.Level1SrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 records from database
-
-        parameter kwargs:
-            level0_id: [str]
-            data_type: [str]
-            create_time : (start, end),
-            qc1_status : [int],
-            prc_status : [int],
-            filename: [str]
-            limit: limits returns the number of records,default 0:no-limit
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1_pb2.FindLevel1Req(
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type"),
-                create_time_start = get_parameter(kwargs, "create_time", [None, None])[0],
-                create_time_end = get_parameter(kwargs, "create_time", [None, None])[1],
-                qc1_status = get_parameter(kwargs, "qc1_status"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                filename = get_parameter(kwargs, "filename"),
-                limit = get_parameter(kwargs, "limit", 0),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def get(self, **kwargs):
-        '''  fetch a record from database
-
-        parameter kwargs:
-            id : [int] 
-
-        return csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Get.with_call(level1_pb2.GetLevel1Req(
-                id = get_parameter(kwargs, "id"),
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type") 
-            ),metadata = get_auth_headers())
-
-            if resp.record is None or resp.record.id == 0:
-                return Result.error(message=f"data not found")  
-
-            return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-           
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))   
-
-    def update_proc_status(self, **kwargs):
-        ''' update the status of reduction
-
-        parameter kwargs:
-            id : [int],
-            status : [int]
-
-        return csst_dfs_common.models.Result
-        '''
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1_pb2.UpdateProcStatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def update_qc1_status(self, **kwargs):
-        ''' update the status of QC0
-        
-        parameter kwargs:
-            id : [int],
-            status : [int]
-        '''        
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateQc1Status.with_call(
-                level1_pb2.UpdateQc1StatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def write(self, **kwargs):
-        ''' insert a level1 record into database
- 
-        parameter kwargs:
-            level0_id : [str]
-            data_type : [str]
-            cor_sci_id : [int]
-            prc_params : [str]
-            filename : [str]
-            file_path : [str]            
-            prc_status : [int]
-            prc_time : [str]
-            pipeline_id : [str]
-            refs: [dict]
-
-        return csst_dfs_common.models.Result
-        '''   
-
-        rec = level1_pb2.Level1Record(
-            id = 0,
-            level0_id = get_parameter(kwargs, "level0_id"),
-            data_type = get_parameter(kwargs, "data_type"),
-            cor_sci_id = get_parameter(kwargs, "cor_sci_id"),
-            prc_params = get_parameter(kwargs, "prc_params"),
-            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", format_datetime(datetime.now())),
-            pipeline_id = get_parameter(kwargs, "pipeline_id"),
-            refs = get_parameter(kwargs, "refs", {})
-        )
-        def stream(rec):
-            with open(rec.file_path, 'rb') as f:
-                while True:
-                    data = f.read(UPLOAD_CHUNK_SIZE)
-                    if not data:
-                        break
-                    yield level1_pb2.WriteLevel1Req(record = rec, data = data)
-
-        try:
-            if not rec.file_path:
-                return Result.error(message="file_path is blank")
-            if not os.path.exists(rec.file_path):
-                return Result.error(message="the file [%s] not existed" % (rec.file_path, ))
-            if not rec.filename:
-                rec.filename = os.path.basename(rec.file_path)
-
-            resp,_ = self.stub.Write.with_call(stream(rec),metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))

csst_dfs_api_cluster/hstdm/level1prc.py

-import grpc
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.hstdm import Level1PrcRecord
-
-from csst_dfs_proto.hstdm.level1prc import level1prc_pb2, level1prc_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1PrcApi(object):
-    def __init__(self):
-        self.stub = level1prc_pb2_grpc.Level1PrcSrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 procedure records from database
-
-        parameter kwargs:
-            level1_id: [str]
-            pipeline_id: [str]
-            prc_module: [str]
-            prc_status : [int]
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1prc_pb2.FindLevel1PrcReq(
-                level1_id = get_parameter(kwargs, "level1_id"),
-                pipeline_id = get_parameter(kwargs, "pipeline_id"),
-                prc_module = get_parameter(kwargs, "prc_module"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data = from_proto_model_list(Level1PrcRecord, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1prc_pb2.UpdateProcStatusReq(id=id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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 = level1prc_pb2.Level1PrcRecord(
-            id = 0,
-            level1_id = get_parameter(kwargs, "level1_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")
-        )
-        req = level1prc_pb2.WriteLevel1PrcReq(record = rec)
-        try:
-            resp,_ = self.stub.Write.with_call(req,metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data = Level1PrcRecord().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-   
-

csst_dfs_api_cluster/ifs/__init__.py

-from .level1 import Level1DataApi
-from .level1prc import Level1PrcApi

csst_dfs_api_cluster/ifs/level1.py

-import os
-import grpc
-import datetime
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.ifs import Level1Record
-from csst_dfs_commons.models.constants import UPLOAD_CHUNK_SIZE
-from csst_dfs_proto.ifs.level1 import level1_pb2, level1_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1DataApi(object):
-    """
-    Level1 Data Operation Class
-    """    
-    def __init__(self):
-        self.stub = level1_pb2_grpc.Level1SrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 records from database
-
-        parameter kwargs:
-            level0_id: [str]
-            data_type: [str]
-            create_time : (start, end),
-            qc1_status : [int],
-            prc_status : [int],
-            filename: [str]
-            limit: limits returns the number of records,default 0:no-limit
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1_pb2.FindLevel1Req(
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type"),
-                create_time_start = get_parameter(kwargs, "create_time", [None, None])[0],
-                create_time_end = get_parameter(kwargs, "create_time", [None, None])[1],
-                qc1_status = get_parameter(kwargs, "qc1_status"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                filename = get_parameter(kwargs, "filename"),
-                limit = get_parameter(kwargs, "limit", 0),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def get(self, **kwargs):
-        '''  fetch a record from database
-
-        parameter kwargs:
-            id : [int] 
-
-        return csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Get.with_call(level1_pb2.GetLevel1Req(
-                id = get_parameter(kwargs, "id"),
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type") 
-            ),metadata = get_auth_headers())
-
-            if resp.record is None or resp.record.id == 0:
-                return Result.error(message=f"data not found")  
-
-            return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-           
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))   
-
-    def update_proc_status(self, **kwargs):
-        ''' update the status of reduction
-
-        parameter kwargs:
-            id : [int],
-            status : [int]
-
-        return csst_dfs_common.models.Result
-        '''
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1_pb2.UpdateProcStatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def update_qc1_status(self, **kwargs):
-        ''' update the status of QC0
-        
-        parameter kwargs:
-            id : [int],
-            status : [int]
-        '''        
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateQc1Status.with_call(
-                level1_pb2.UpdateQc1StatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def write(self, **kwargs):
-        ''' insert a level1 record into database
- 
-        parameter kwargs:
-            level0_id : [str]
-            data_type : [str]
-            cor_sci_id : [int]
-            prc_params : [str]
-            filename : [str]
-            file_path : [str]            
-            prc_status : [int]
-            prc_time : [str]
-            pipeline_id : [str]
-            refs: [dict]
-
-        return csst_dfs_common.models.Result
-        '''   
-
-        rec = level1_pb2.Level1Record(
-            id = 0,
-            level0_id = get_parameter(kwargs, "level0_id"),
-            data_type = get_parameter(kwargs, "data_type"),
-            cor_sci_id = get_parameter(kwargs, "cor_sci_id"),
-            prc_params = get_parameter(kwargs, "prc_params"),
-            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", format_datetime(datetime.now())),
-            pipeline_id = get_parameter(kwargs, "pipeline_id"),
-            refs = get_parameter(kwargs, "refs", {})
-        )
-        def stream(rec):
-            with open(rec.file_path, 'rb') as f:
-                while True:
-                    data = f.read(UPLOAD_CHUNK_SIZE)
-                    if not data:
-                        break
-                    yield level1_pb2.WriteLevel1Req(record = rec, data = data)
-
-        try:
-            if not rec.file_path:
-                return Result.error(message="file_path is blank")
-            if not os.path.exists(rec.file_path):
-                return Result.error(message="the file [%s] not existed" % (rec.file_path, ))
-            if not rec.filename:
-                rec.filename = os.path.basename(rec.file_path)
-
-            resp,_ = self.stub.Write.with_call(stream(rec),metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))

csst_dfs_api_cluster/mbi/__init__.py

-from .level1 import Level1DataApi
-from .level1prc import Level1PrcApi
 from .level2 import Level2DataApi
 from .level2co import Level2CoApi
\ No newline at end of file

csst_dfs_api_cluster/mbi/level1prc.py

-import grpc
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.msc import Level1PrcRecord
-
-from csst_dfs_proto.msc.level1prc import level1prc_pb2, level1prc_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1PrcApi(object):
-    def __init__(self):
-        self.stub = level1prc_pb2_grpc.Level1PrcSrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 procedure records from database
-
-        parameter kwargs:
-            level1_id: [str]
-            pipeline_id: [str]
-            prc_module: [str]
-            prc_status : [int]
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1prc_pb2.FindLevel1PrcReq(
-                level1_id = get_parameter(kwargs, "level1_id"),
-                pipeline_id = get_parameter(kwargs, "pipeline_id"),
-                prc_module = get_parameter(kwargs, "prc_module"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data = from_proto_model_list(Level1PrcRecord, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1prc_pb2.UpdateProcStatusReq(id=id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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 = level1prc_pb2.Level1PrcRecord(
-            id = 0,
-            level1_id = get_parameter(kwargs, "level1_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")
-        )
-        req = level1prc_pb2.WriteLevel1PrcReq(record = rec)
-        try:
-            resp,_ = self.stub.Write.with_call(req,metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data = Level1PrcRecord().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-   
-

csst_dfs_api_cluster/mci/__init__.py

-from .level1 import Level1DataApi
-from .level1prc import Level1PrcApi
\ No newline at end of file

csst_dfs_api_cluster/mci/level1.py

-import os
-import grpc
-import datetime
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.mci import Level1Record
-from csst_dfs_commons.models.constants import UPLOAD_CHUNK_SIZE
-from csst_dfs_proto.mci.level1 import level1_pb2, level1_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1DataApi(object):
-    """
-    Level1 Data Operation Class
-    """    
-    def __init__(self):
-        self.stub = level1_pb2_grpc.Level1SrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 records from database
-
-        parameter kwargs:
-            level0_id: [str]
-            data_type: [str]
-            create_time : (start, end),
-            qc1_status : [int],
-            prc_status : [int],
-            filename: [str]
-            limit: limits returns the number of records,default 0:no-limit
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1_pb2.FindLevel1Req(
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type"),
-                create_time_start = get_parameter(kwargs, "create_time", [None, None])[0],
-                create_time_end = get_parameter(kwargs, "create_time", [None, None])[1],
-                qc1_status = get_parameter(kwargs, "qc1_status"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                filename = get_parameter(kwargs, "filename"),
-                limit = get_parameter(kwargs, "limit", 0),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def get(self, **kwargs):
-        '''  fetch a record from database
-
-        parameter kwargs:
-            id : [int] 
-
-        return csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Get.with_call(level1_pb2.GetLevel1Req(
-                id = get_parameter(kwargs, "id"),
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type") 
-            ),metadata = get_auth_headers())
-
-            if resp.record is None or resp.record.id == 0:
-                return Result.error(message=f"data not found")  
-
-            return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-           
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))   
-
-    def update_proc_status(self, **kwargs):
-        ''' update the status of reduction
-
-        parameter kwargs:
-            id : [int],
-            status : [int]
-
-        return csst_dfs_common.models.Result
-        '''
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1_pb2.UpdateProcStatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def update_qc1_status(self, **kwargs):
-        ''' update the status of QC0
-        
-        parameter kwargs:
-            id : [int],
-            status : [int]
-        '''        
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateQc1Status.with_call(
-                level1_pb2.UpdateQc1StatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def write(self, **kwargs):
-        ''' insert a level1 record into database
- 
-        parameter kwargs:
-            level0_id : [str]
-            data_type : [str]
-            cor_sci_id : [int]
-            prc_params : [str]
-            filename : [str]
-            file_path : [str]            
-            prc_status : [int]
-            prc_time : [str]
-            pipeline_id : [str]
-            refs: [dict]
-
-        return csst_dfs_common.models.Result
-        '''   
-
-        rec = level1_pb2.Level1Record(
-            id = 0,
-            level0_id = get_parameter(kwargs, "level0_id"),
-            data_type = get_parameter(kwargs, "data_type"),
-            cor_sci_id = get_parameter(kwargs, "cor_sci_id"),
-            prc_params = get_parameter(kwargs, "prc_params"),
-            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", format_datetime(datetime.now())),
-            pipeline_id = get_parameter(kwargs, "pipeline_id"),
-            refs = get_parameter(kwargs, "refs", {})
-        )
-        def stream(rec):
-            with open(rec.file_path, 'rb') as f:
-                while True:
-                    data = f.read(UPLOAD_CHUNK_SIZE)
-                    if not data:
-                        break
-                    yield level1_pb2.WriteLevel1Req(record = rec, data = data)
-
-        try:
-            if not rec.file_path:
-                return Result.error(message="file_path is blank")
-            if not os.path.exists(rec.file_path):
-                return Result.error(message="the file [%s] not existed" % (rec.file_path, ))
-            if not rec.filename:
-                rec.filename = os.path.basename(rec.file_path)
-
-            resp,_ = self.stub.Write.with_call(stream(rec),metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))

csst_dfs_api_cluster/mci/level1prc.py

-import grpc
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.mci import Level1PrcRecord
-
-from csst_dfs_proto.mci.level1prc import level1prc_pb2, level1prc_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1PrcApi(object):
-    def __init__(self):
-        self.stub = level1prc_pb2_grpc.Level1PrcSrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 procedure records from database
-
-        parameter kwargs:
-            level1_id: [str]
-            pipeline_id: [str]
-            prc_module: [str]
-            prc_status : [int]
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1prc_pb2.FindLevel1PrcReq(
-                level1_id = get_parameter(kwargs, "level1_id"),
-                pipeline_id = get_parameter(kwargs, "pipeline_id"),
-                prc_module = get_parameter(kwargs, "prc_module"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                other_conditions = {"test":"cnlab.test"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data = from_proto_model_list(Level1PrcRecord, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1prc_pb2.UpdateProcStatusReq(id=id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    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 = level1prc_pb2.Level1PrcRecord(
-            id = 0,
-            level1_id = get_parameter(kwargs, "level1_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")
-        )
-        req = level1prc_pb2.WriteLevel1PrcReq(record = rec)
-        try:
-            resp,_ = self.stub.Write.with_call(req,metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data = Level1PrcRecord().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-   
-

csst_dfs_api_cluster/sls/__init__.py

-from .level1 import Level1DataApi
-from .level1prc import Level1PrcApi
 from .level2spectra import Level2SpectraApi
\ No newline at end of file

csst_dfs_api_cluster/sls/level1.py

-import os
-import grpc
-import datetime
-
-from csst_dfs_commons.models import Result
-from csst_dfs_commons.models.common import from_proto_model_list
-from csst_dfs_commons.models.sls import Level1Record
-from csst_dfs_commons.models.constants import UPLOAD_CHUNK_SIZE
-from csst_dfs_proto.sls.level1 import level1_pb2, level1_pb2_grpc
-
-from ..common.service import ServiceProxy
-from ..common.utils import *
-
-class Level1DataApi(object):
-    """
-    Level1 Data Operation Class
-    """    
-    def __init__(self):
-        self.stub = level1_pb2_grpc.Level1SrvStub(ServiceProxy().channel())
-
-    def find(self, **kwargs):
-        ''' retrieve level1 records from database
-
-        parameter kwargs:
-            level0_id: [str]
-            data_type: [str]
-            create_time : (start, end),
-            qc1_status : [int],
-            prc_status : [int],
-            filename: [str]
-            limit: limits returns the number of records,default 0:no-limit
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Find.with_call(level1_pb2.FindLevel1Req(
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type"),
-                create_time_start = get_parameter(kwargs, "create_time", [None, None])[0],
-                create_time_end = get_parameter(kwargs, "create_time", [None, None])[1],
-                qc1_status = get_parameter(kwargs, "qc1_status"),
-                prc_status = get_parameter(kwargs, "prc_status"),
-                filename = get_parameter(kwargs, "filename"),
-                limit = get_parameter(kwargs, "limit", 0),
-                other_conditions = {}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-            
-    def find_by_brick_ids(self, **kwargs):
-        ''' retrieve level1 records by brick_ids like [1,2,3,4]
-
-        :param kwargs: Parameter dictionary, key items support:
-            brick_ids: [list]
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.FindByBrickIds.with_call(level1_pb2.FindByBrickIdsReq(
-                brick_ids = get_parameter(kwargs, "brick_ids", [])
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records))
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def find_by_prc_status(self, **kwargs):
-        ''' retrieve level1 records from database
-
-        parameter kwargs:
-            level0_id: [str]
-            data_type: [str]
-            create_time : (start, end),
-            qc1_status : [int],
-            prc_status : [int],
-            filename: [str]
-            limit: limits returns the number of records,default 0:no-limit
-
-        return: csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.FindByPrcStatus.with_call(level1_pb2.FindLevel1Req(
-                level0_id = None,
-                data_type = None,
-                create_time_start = None,
-                create_time_end = None,
-                qc1_status = None,
-                prc_status = get_parameter(kwargs, "prc_status", -1),
-                limit = get_parameter(kwargs, "limit", 1),
-                other_conditions = {"orderBy":"create_time asc"}
-            ),metadata = get_auth_headers())
-
-            if resp.success:
-                return Result.ok_data(data=from_proto_model_list(Level1Record, resp.records)).append("totalCount", resp.totalCount)
-            else:
-                return Result.error(message = str(resp.error.detail))
-
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def get(self, **kwargs):
-        '''  fetch a record from database
-
-        parameter kwargs:
-            id : [int] 
-
-        return csst_dfs_common.models.Result
-        '''
-        try:
-            resp, _ =  self.stub.Get.with_call(level1_pb2.GetLevel1Req(
-                id = get_parameter(kwargs, "id"),
-                level0_id = get_parameter(kwargs, "level0_id"),
-                data_type = get_parameter(kwargs, "data_type")            
-            ),metadata = get_auth_headers())
-
-            if resp.record is None or resp.record.id == 0:
-                return Result.error(message=f"data not found")  
-
-            return Result.ok_data(data = Level1Record().from_proto_model(resp.record))
-           
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))   
-
-    def update_proc_status(self, **kwargs):
-        ''' update the status of reduction
-
-        parameter kwargs:
-            id : [int],
-            status : [int]
-
-        return csst_dfs_common.models.Result
-        '''
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateProcStatus.with_call(
-                level1_pb2.UpdateProcStatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def update_qc1_status(self, **kwargs):
-        ''' update the status of QC0
-        
-        parameter kwargs:
-            id : [int],
-            status : [int]
-        '''        
-        fits_id = get_parameter(kwargs, "id")
-        status = get_parameter(kwargs, "status")
-        try:
-            resp,_ = self.stub.UpdateQc1Status.with_call(
-                level1_pb2.UpdateQc1StatusReq(id=fits_id, status=status),
-                metadata = get_auth_headers()
-            )
-            if resp.success:
-                return Result.ok_data()
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))
-
-    def write(self, **kwargs):
-        ''' insert a level1 record into database
- 
-        parameter kwargs:
-            level0_id : [str]
-            data_type : [str]
-            prc_params : [str]
-            filename : [str]
-            file_path : [str]            
-            prc_status : [int]
-            prc_time : [str]
-            pipeline_id : [str]
-            refs: [dict]
-
-        return csst_dfs_common.models.Result
-        '''   
-
-        rec = level1_pb2.Level1Record(
-            id = 0,
-            level0_id = get_parameter(kwargs, "level0_id"),
-            data_type = get_parameter(kwargs, "data_type"),
-            prc_params = get_parameter(kwargs, "prc_params"),
-            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", format_datetime(datetime.now())),
-            pipeline_id = get_parameter(kwargs, "pipeline_id"),
-            refs = get_parameter(kwargs, "refs", {})
-        )
-        def stream(rec):
-            with open(rec.file_path, 'rb') as f:
-                while True:
-                    data = f.read(UPLOAD_CHUNK_SIZE)
-                    if not data:
-                        break
-                    yield level1_pb2.WriteLevel1Req(record = rec, data = data)
-        try:
-            if not rec.file_path:
-                return Result.error(message="file_path is blank")
-            if not os.path.exists(rec.file_path):
-                return Result.error(message="the file [%s] not existed" % (rec.file_path, ))
-            if not rec.filename:
-                rec.filename = os.path.basename(rec.file_path)
-
-            resp,_ = self.stub.Write.with_call(stream(rec),metadata = get_auth_headers())
-            if resp.success:
-                return Result.ok_data(data=Level1Record().from_proto_model(resp.record))
-            else:
-                return Result.error(message = str(resp.error.detail))
-        except grpc.RpcError as e:
-            return Result.error(message="%s:%s" % (e.code().value, e.details()))