diff --git a/csst_shao_common/__init__.py b/csst_shao_common/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/csst_shao_common/common.py b/csst_shao_common/common.py deleted file mode 100644 index d2a28ea92edd2ef3f3c5066e87091e1a7c7a8de4..0000000000000000000000000000000000000000 --- a/csst_shao_common/common.py +++ /dev/null @@ -1,190 +0,0 @@ -""" -Identifier: KSC-SJ6-csst_ifs_common/common.py -Name: Common.py -Description: CSST IFS COMMON -Author: Wei Chen -""" - -from .ifs_status import CsstIFSStatus -from .exit_code_list import IFSExitCode -from .logger import get_logger -# from .ifs_status import IFSError -from .config import IFSConfig -import os - - -class CsstIFSInit: - - def __init__(self): - self.exit_code = "100" - self.status = CsstIFSStatus - self.fun_exit_code = IFSExitCode() - self._env_init() - self._dm_init() - self.config_ifs_rss_env() - self.config_ifs_cube_env() - self.logger = get_logger() - - def _env_init(self): - # code 100: "l1 normal exit" - exit_code = "100" - # self.logger.info("IFS Env Setting...") - try: - self.config = IFSConfig() - self.config.set_ifs_config("dir_log", self.config.get_ifs_config("dir_local")) - path_log = os.path.join(self.config.get_ifs_config("dir_log"), "./log.txt") - self.config.set_ifs_config("path_log", path_log) - self.logger = get_logger("CSST IFS log", self.config.get_ifs_config("path_log")) - # self.logger.info("IFS Env Set Finish!") - except Exception as e: - # code 101: "l1 Env Set Error", - exit_code = "101" - print("{}".format(e)) - self.logger.error(self.fun_exit_code.get_ifs_exit_code('101')) - self.logger.error(e) - status = 2 - assert status in [CsstIFSStatus.PERFECT, CsstIFSStatus.WARNING] - finally: - return exit_code - - def _dm_init(self): - # code 100: "l1 normal exit" - exit_code = "100" - try: - from .data_manager import CsstIFSDataManager - self.dm = CsstIFSDataManager() - except Exception as e: - # code 102: "Not found DFS system", - # exit_code = "102" - exit_code = "103" - print(self.fun_exit_code.get_ifs_exit_code('103')) - # print("{}".format(e)) - # self.logger.error(self.fun_exit_code.get_ifs_exit_code('102')) - # self.logger.error(e) - # status = 2 - # assert status in [CsstIFSStatus.PERFECT, CsstIFSStatus.WARNING] - finally: - return exit_code - - def config_ifs_rss_env(self): - # /L1Pipeline/test_shao/test_data/ this dir is cdrs dir - # self.config.set_ifs_config("path_RSSlog", - # "/L1Pipeline/integration/ifs/csst_ifs_rss/proc_data/path_RSSlog.txt") - # self.config.set_ifs_config("dir_rss_proc", "/L1Pipeline/integration/ifs/csst_ifs_rss/proc_data/") - # self.config.set_ifs_config("dir_aux", "/L1Pipeline/integration/ifs/csst_ifs_rss/cal_data/") - # self._config_ifs_rss_env_crds() - # # self.config.set_ifs_config("path_l0_rss", self.dm.get_data_path_l0_sci(self.config.get_ifs_config("obsid"))) - # path_list = self.get_all_file_path("/L1Pipeline/integration/ifs/csst_ifs_rss/raw_data/") - # self.config.set_ifs_config("path_l0_rss", path_list) - # - # self.config.set_ifs_config("path_l1_wcs_output", - # "/L1Pipeline/test_shao/test_data/ifs/csst_ifs_wcs/temp_for_cube") - - self.config.set_ifs_config( - "path_RSSlog", - "/CSST_test/l1_all/workspace/test_data/ifs/sim/L0/IFS_simData_2023-06-26/proc_data/path_RSSlog.txt") - # self.config.set_ifs_config( - # "dir_rss_proc", "/CSST_test/l1_all/workspace/test_data/ifs/sim/L0/IFS_simData_2023-06-26/proc_data/") - self.config.set_ifs_config( - "dir_rss_proc", "/datapool/CSST_PIPELINE/workspace/temp_data/ifs/sim/L1/sim_ver230725/proc_l1_rss/") - self.config.set_ifs_config( - "dir_aux", - "/CSST_test/l1_all/workspace/test_data/ifs/sim/L0/IFS_CRDS_20230725/") - self._config_ifs_rss_env_crds() - # self.config.set_ifs_config("path_l0_rss", self.dm.get_data_path_l0_sci(self.config.get_ifs_config("obsid"))) - path_list = self.get_all_file_path( - "/CSST_test/l1_all/workspace/test_data/ifs/sim/L0/IFS_simData_2023-07-25/sky_Data_temp/") - self.config.set_ifs_config("path_l0_rss", path_list) - - self.config.set_ifs_config( - "path_l1_wcs_output", - "/CSST_test/l1_all/workspace/test_data/ifs/sim/L0/IFS_simData_2023-07-25/temp_for_cube") - - return - - @staticmethod - def get_all_file_path(dir_temp): - path_list = list() - for root, dirs, names in os.walk(dir_temp): - for filename in names: - path_list.append(os.path.join(root, filename)) - return path_list - - @staticmethod - def get_file_path_include_keywords(dir_temp, keywords): - path_list = list() - for root, dirs, names in os.walk(dir_temp): - for filename in names: - if keywords in filename: - path_list.append(os.path.join(root, filename)) - return path_list - - def _config_ifs_rss_env_crds(self): - - path = os.path.join(self.config.get_ifs_config("dir_aux"), self.dm.get_data_name_rss_bias_b()) - self.config.set_ifs_config("path_rss_biasm_b", path) - path = os.path.join(self.config.get_ifs_config("dir_aux"), self.dm.get_data_name_rss_bias_r()) - self.config.set_ifs_config("path_rss_biasm_r", path) - - path = os.path.join(self.config.get_ifs_config("dir_aux"), self.dm.get_data_name_rss_flat_b()) - self.config.set_ifs_config("path_rss_flatm_b", path) - path = os.path.join(self.config.get_ifs_config("dir_aux"), self.dm.get_data_name_rss_flat_r()) - self.config.set_ifs_config("path_rss_flatm_r", path) - - self.config.set_ifs_config("path_rss_dark_b", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_dark_b()) - self.config.set_ifs_config("path_rss_dark_r", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_dark_r()) - - self.config.set_ifs_config("path_rss_flatnorm_b", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_flatnorm_b()) - self.config.set_ifs_config("path_rss_flatnorm_r", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_flatnorm_r()) - - self.config.set_ifs_config("path_rss_trace_b", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_trace_b()) - self.config.set_ifs_config("path_rss_trace_r", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_trace_r()) - - self.config.set_ifs_config("path_rss_ptow_b", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_ptow_b()) - self.config.set_ifs_config("path_rss_ptow_r", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_ptow_r()) - - self.config.set_ifs_config("path_rss_starfile_b", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_starfile_b()) - self.config.set_ifs_config("path_rss_starfile_r", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_starfile_r()) - - self.config.set_ifs_config("path_rss_arclab_b", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_arclab_b()) - self.config.set_ifs_config("path_rss_arclab_r", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_arclab_r()) - - self.config.set_ifs_config("path_rss_starlab", self.config.get_ifs_config( - "dir_aux") + self.dm.get_data_name_rss_starlab()) - return - - def config_ifs_cube_env(self): - # self.config.set_ifs_config("dir_tmp", - # "/nfsdata/users/csstpipeline/L1Pipeline/integration/ifs/csst_ifs_cube/aimdir") - self.config.set_ifs_config("objid", "NGC6397") - # self.config.set_ifs_config("path_l0_cube", self.dm.get_data_path_l1_sci(self.config.get_ifs_config("obsid"))) - - # hdir = '/nfsdata/users/csstpipeline/L1Pipeline/integration/ifs/csst_ifs_cube/test_file/' - # aimdir = '/nfsdata/users/csstpipeline/L1Pipeline/integration/ifs/csst_ifs_cube/aimdir/' - target = 'NGC6397' - hdir = '/nfsdata/users/csstpipeline/L1Pipeline/integration/ifs/csst_ifs_cube/test_file/' - aimdir = '/nfsdata/users/csstpipeline/L1Pipeline/integration/ifs/csst_ifs_cube/aimdir/' - softver = 'R7201' - ver = 1 - lamr = hdir + "CSST_IFS_LAMP_MASTER_20220610_R_L1_R7201_bew_____.fits" - lamb = hdir + "CSST_IFS_LAMP_MASTER_20220610_B_L1_R7201_bew_____.fits" - self.config.set_ifs_config('objid', target) - self.config.set_ifs_config('hdir', hdir) - self.config.set_ifs_config('aimdir', aimdir) - self.config.set_ifs_config('softver', softver) - self.config.set_ifs_config('ver', ver) - self.config.set_ifs_config('lamr', lamr) - self.config.set_ifs_config('lamb', lamb) - return diff --git a/csst_shao_common/config.py b/csst_shao_common/config.py deleted file mode 100644 index 9193d2ce3abc28130a5c8f7a63c082de65ae56a1..0000000000000000000000000000000000000000 --- a/csst_shao_common/config.py +++ /dev/null @@ -1,87 +0,0 @@ -""" -Identifier: KSC-SJ6-csst_ifs_common/config.py -Name: config.py -Description: CSST IFS CONFIG -Author: Wei Chen -""" - -import os -import threading -from .ifs_status import CsstIFSStatus - - -class IFSConfig(object): - """ - IFS Config as public parameters. - """ - _instance_lock = threading.Lock() - - def __init__(self): - self._set_ifs_config_init() - - def __new__(cls, *args, **kwargs): - if not hasattr(IFSConfig, "_instance"): - with IFSConfig._instance_lock: - if not hasattr(IFSConfig, "_instance"): - IFSConfig._instance = object.__new__(cls) - cls.config_dict = dict() - cls.public_variable_dict = dict() - return IFSConfig._instance - - def _set_ifs_config_init(self): - local_path = os.getcwd() - config_dict = { - "dir_l0": ".", - "dir_l1": ".", - "dir_log": ".", - "dir_rss": ".", - "dir_cube": ".", - "dir_local": local_path, - "dir_rss_proc": "/L1Pipeline/integration/ifs/csst_ifs_rss/proc_data/" - } - if self.config_dict == {}: - self.config_dict = config_dict - - def get_ifs_config(self, ifs_config_key): - """ - get ifs config. - - Parameters - ---------- - ifs_config_key : str - The string of dict key - - Returns - ------- - dict_value : string - The string of dict_value. - """ - if ifs_config_key in self.config_dict: - ifs_config_value = self.config_dict[ifs_config_key] - return ifs_config_value - else: - print("Error: key "+ifs_config_key+" not exist") - status = 2 - assert status in [CsstIFSStatus.PERFECT, CsstIFSStatus.WARNING] - - def set_ifs_config(self, ifs_config_key, ifs_config_value): - """ - get ifs config. - - Parameters - ---------- - ifs_config_key : str - The string of dict key - - ifs_config_value : anytype - The string of dict value - - Returns - ------- - None - """ - # if ifs_config_key in self.config_dict: - # print("Warning: key "+ifs_config_key+" has exist, overwrite!") - # status = 1 - # assert status in [CsstIFSStatus.PERFECT, CsstIFSStatus.WARNING] - self.config_dict[ifs_config_key] = ifs_config_value diff --git a/csst_shao_common/exit_code_list.py b/csst_shao_common/exit_code_list.py deleted file mode 100644 index b397ffca01182eba40858de445fcb93bedb17e38..0000000000000000000000000000000000000000 --- a/csst_shao_common/exit_code_list.py +++ /dev/null @@ -1,84 +0,0 @@ -""" -Identifier: KSC-SJ6-csst_ifs_common/exit_code_list.py -Name: exit_code_list.py -Description: CSST IFS Exit Code List -Author: Wei Chen -""" - -import threading - - -class IFSExitCode(object): - """ - IFS Exit Code as public. - """ - _instance_lock = threading.Lock() - - def __init__(self): - self.config_dict = {} - self._set_ifs_exit_code_init() - - def __new__(cls, *args, **kwargs): - if not hasattr(IFSExitCode, "_instance"): - with IFSExitCode._instance_lock: - if not hasattr(IFSExitCode, "_instance"): - IFSExitCode._instance = object.__new__(cls) - return IFSExitCode._instance - - # modify exit code here! - def _set_ifs_exit_code_init(self): - config_dict = { - "100": "l1 normal exit", - "101": "l1 Env Set Error", - "102": "Not found DFS system", - "103": "CSST Dummy DFS system Initialize", - "199": "l1 error exit", - "200": "rss normal exit", - "201": "rss error exit", - "300": "cube normal exit", - "301": "cube error exit", - "400": "qc normal exit", - "401": "qc error exit", - } - self.config_dict = config_dict - - def get_ifs_exit_code(self, ifs_exit_code_key): - """ - get ifs config. - - Parameters - ---------- - ifs_exit_code_key : str - The string of dict key - - Returns - ------- - dict_value : string - The exit_code of program status. - """ - if ifs_exit_code_key in self.config_dict: - ifs_exit_code_value = self.config_dict[ifs_exit_code_key] - return ifs_exit_code_value - else: - print("Error: key "+ifs_exit_code_key+" not exist") - - def set_ifs_exit_code(self, ifs_exit_code_key, ifs_exit_code_value): - """ - get ifs config. - - Parameters - ---------- - ifs_exit_code_key : str - The string of dict key - - ifs_exit_code_value : str - The string of dict value - - Returns - ------- - None - """ - if ifs_exit_code_key in self.config_dict: - print("Warning: key "+ifs_exit_code_key+" has existed, overwrite!") - - self.config_dict[ifs_exit_code_key] = ifs_exit_code_value diff --git a/csst_shao_common/ifs_status.py b/csst_shao_common/ifs_status.py deleted file mode 100644 index 546f311b12d9386a69f5b329cef39b6cb9aec06d..0000000000000000000000000000000000000000 --- a/csst_shao_common/ifs_status.py +++ /dev/null @@ -1,36 +0,0 @@ -""" -Identifier: KSC-SJ6-csst_ifs_common/ifs_status.py -Name: ifs_status.py -Description: CSST IFS Status -Author: Wei Chen -""" - -from enum import IntEnum - - -class CsstIFSStatus(IntEnum): - """ - The CSST IFS Status class. - - This class provides a set of status which should be returned - from each CSST L1 IFS functional module. - - Examples - -------- - >>> def f(num): - >>> if num > 0: - >>> return CsstIFSStatus.PERFECT - >>> return CsstIFSStatus.ERROR - """ - PERFECT = 0 - WARNING = 1 - ERROR = 2 - # status list to be completed - -# -# class IFSError(Exception): -# def __init__(self, msg): -# self.msg = msg -# -# def __str__(self): -# return self.msg diff --git a/csst_shao_common/interface.py b/csst_shao_common/interface.py deleted file mode 100644 index 05ff5cc07ffc46dea965f61fa9641a0381083aa5..0000000000000000000000000000000000000000 --- a/csst_shao_common/interface.py +++ /dev/null @@ -1,44 +0,0 @@ -# -*- coding: utf-8 -*- - -from abc import abstractmethod, ABCMeta -from csst_ifs_common.common import CsstIFSInit -from csst_ifs_common.ifs_status import CsstIFSStatus -from csst_ifs_common.ifs_result import CsstIFSResult - -csst_ifs = CsstIFSInit() - - -class BasePipelineInterface(metaclass=ABCMeta): - @abstractmethod - def run(self): - """run the pipeline""" - pass - - @abstractmethod - def para_set(self, *args): - pass - - @abstractmethod - def result_check(self): - pass - - -# 数据处理软件编写主调用类范例 -class L1Example(BasePipelineInterface): - def __init__(self): - self.result = CsstIFSStatus - self.logger = csst_ifs.logger - self.status = CsstIFSResult - - # 在此处编写代码，使得用户通过执行此方法运行程序 - def run(self): - pass - - # 在此处编写代码，使得用户通过执行此方法配置参数，所有参数应当含有默认值，输入参数非必须。此处可以通过指定参数，给与不同情况的初始化参数配置 - def para_set(self, list_para): - pass - - # 在此处编写代码，使得用户通过执行此方法进行结果检查 - def result_check(self): - pass - diff --git a/csst_shao_common/logger.py b/csst_shao_common/logger.py deleted file mode 100644 index c10a6c37e3f363b40e13444d4d20e141275a3f62..0000000000000000000000000000000000000000 --- a/csst_shao_common/logger.py +++ /dev/null @@ -1,64 +0,0 @@ -""" -Identifier: KSC-SJ6-csst_ifs_common/logger.py -Name: logger.py -Description: CSST IFS Logger -Author: Wei Chen -""" - -import logging - - -def get_logger(name="CSST L1 IFS Pipeline", filename=""): - """ - Get a logger for CSST L1 Pipeline. - - Just copy form bo zhang. - - need mod for ifs - """ - - # initialize logger - logger = logging.getLogger(name=name) - logger.handlers.clear() - logger.setLevel(logging.DEBUG) - - # logging formatter - formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(module)s.py:%(lineno)d - %(message)s") - - # stream handler - stream_handler = logging.StreamHandler() - stream_handler.setFormatter(formatter) - stream_handler.setLevel(logging.DEBUG) - logger.addHandler(stream_handler) - - # file handler - if filename != "": - file_handler = logging.FileHandler(filename=filename, encoding="utf-8") - file_handler.setFormatter(formatter) - file_handler.setLevel(logging.INFO) - logger.addHandler(file_handler) - - return logger - -# -# def upload_logger(path_file=""): -# # logger upload code here -# print(path_file) -# return -# -# -# def check_logger(path_file=""): -# # check upload code here -# print(path_file) -# return -# -# -# if __name__ == "__main__": -# import os -# print(os.getcwd()) -# logger = get_logger(filename="test_logger.log") -# print(type(logger), logger.name) -# logger.debug("Well ..") -# logger.info("Hello ...") -# logger.warning("There ...") -# logger.error("Good morning ...") diff --git a/csst_shao_common/top_level_interface.py b/csst_shao_common/top_level_interface.py deleted file mode 100644 index 37692c047f66533d9e3941f4fc89ad513a0d9d69..0000000000000000000000000000000000000000 --- a/csst_shao_common/top_level_interface.py +++ /dev/null @@ -1,8 +0,0 @@ -from .logger import get_logger -from .ifs_status import CsstIFSStatus -from .exit_code_list import IFSExitCode -from .data_manager import CsstIFSDataManager -from .config import IFSConfig -from .common import CsstIFSInit - -__all__ = ["get_logger", "CsstIFSStatus", "IFSExitCode", "CsstIFSDataManager", "IFSConfig", "CsstIFSInit"] diff --git a/csst_shao_templet.py b/csst_shao_templet.py index 8589848168d42a3aa898c469883fd0e265b98adf..3597f1835149aaec06dcce25aa09e727b7766f6e 100644 --- a/csst_shao_templet.py +++ b/csst_shao_templet.py @@ -4,25 +4,244 @@ from abc import abstractmethod, ABCMeta from csst_ifs_common.common import CsstIFSInit from csst_ifs_common.ifs_status import CsstIFSStatus from csst_ifs_common.ifs_result import CsstIFSResult +from csst_shao_common.interface import BasePipelineInterface csst_ifs = CsstIFSInit() -# 数据处理软件编写主调用类范例 -class L1Example(BasePipelineInterface): - def __init__(self): - self.result = CsstIFSStatus - self.logger = csst_ifs.logger - self.status = CsstIFSResult - # 在此处编写代码，使得用户通过执行此方法运行程序 +class BasePipelineInterface(metaclass=ABCMeta): + @abstractmethod def run(self): + """run the pipeline""" pass - # 在此处编写代码，使得用户通过执行此方法配置参数，所有参数应当含有默认值，输入参数非必须。此处可以通过指定参数，给与不同情况的初始化参数配置 - def para_set(self, list_para): + @abstractmethod + def para_set(self, *args): pass - # 在此处编写代码，使得用户通过执行此方法进行结果检查 + @abstractmethod def result_check(self): pass + def loger(func) -> typeing.Callable: + def inner(*args, **kwargs): + try: + result = func(*args, **kwargs) + csst_mci.logger.info(f"log: {func.__name__}{args}->{result}") + return result + except Exception as e: + csst_mci.logger.error(f"error: {func.__name__}->{str(e)}") + + return inner + + +# 数据处理软件编写主调用类范例 +class PipelineL1MCIInstrument(BasePipelineInterface): + # 此处对程序进行概要介绍, 对类的属性(Attributes)、方法(Methods)等进行一一介绍。 + """ + Instrument is xxxx + + ... + + Attributes + ---------- + obsid : str + The obsid of this item. + + logger : + Inheritance the Logging package for Python. Get a logger for CSST L1 Pipeline. + + self.status = CsstResult + + Methods + ------- + para_set(image_path: str = "/path/to/image", + weight_path: str = "/path/to/weight", + flag_path: str = "/path/to/flag", + cat_ref_path: str = "/path/to/cat_ref", + deg: int = 2, + ) -> CsstResult + + set para + + run() + run the pipeline + + result_check() + check the result + + """ + + def __init__(self): + self.result = CsstStatus + self.logger = csst_mci.logger + self.status = None + self.obsid = None + + # 在此处编写代码，使得用户通过执行此方法运行程序 + def run(self): + """Method, run the pipeline. + + Parameters + ---------- + # None + + Returns + ------- + CsstResult + Result containing `status`, `file_list`, and `output`. + """ + + from csst_mci_instrument.top_level_interface import inst_corr + + # 获取在para_set中配置的参数，此处用于整理，也可以不写，直接喂给需要运行的程序。 + bias_ref = self._bias_ref + dark_ref = self._dark_ref + flat_ref = self._flat_ref + shutter_ref = self._shutter_ref + sci_path = self._sci_path + output_dir = self._output_dir + + # 核心调用代码，用于调用开发者各自开发的程序，其中self.status是个人自定义的返回码，可以以任意形式返回 + self.logger.info("%s running..." % self.__name__) + self.status = inst_corr( + sci_path, bias_ref, dark_ref, flat_ref, shutter_ref, output_dir, logger=None, os_bias=False, cr_flag=False) + + # 如果生成结果是内存形式存在的fits头文件的情况。 + fits_header = "xxxxx" + + # 根据程序运行结果，返回成功、警告、或者报错，此处为一种方法，也可以用其他方式返回错误。 + def case_perfect_1(): + status = CsstStatus.PERFECT + self.logger.info("Pipeline instrument Done!") + return status + + def case_warning_1(): + status = CsstStatus.WARNING + self.logger.info("Pipeline instrument something warning, xxxx") + return status + + def case_error_1(): + status = CsstStatus.ERROR + self.logger.info("Pipeline instrument something wrong, xxxx") + return status + + def case_error_2(): + status = CsstStatus.ERROR + self.logger.info("Pipeline instrument something wrong, xxxx") + return status + + switch_dict = { + 20: case_perfect_1, + 31: case_warning_1, + 41: case_error_1, + 42: case_error_2, + "default": case_perfect_1 + } + + status = switch_dict.get(self.status) + + # 最终返回结果 + result = CsstResult( + status=status, + files=[ + "/path/to/file.fits", + "/path/to/file.txt", + ], + para_01=1, + fits_header=fits_header + ) + return result + + # 在此处编写代码，使得用户通过执行此方法配置流水线运行参数，所有参数应当含有默认值，输入参数非必须。此处可以通过指定参数，给与不同情况的初始化参数配置 + def para_set(self, + image_path: str = "/path/to/image", + weight_path: str = "/path/to/weight", + flag_path: str = "/path/to/flag", + cat_ref_path: str = "/path/to/cat_ref", + deg: int = 2, + ) -> CsstResult: + """Method, estimate image distortion for a single MSC-MBI image. + + Parameters + ---------- + image_path: str + Path of image, defaulting to "/path/to/image". + weight_path: str + Path of weight, defaulting to "/path/to/weight". + flag_path: str + Path of flag, defaulting to "/path/to/flag". + cat_ref_path: str + Path of reference catalog, defaulting to "/path/to/cat_ref". + deg + + + Returns + ------- + CsstResult + Result containing `status`, `file_list`, and `output`. + """ + from csst_mci_common.config_crds import env_crds_set_ins_csst1 + from csst_mci_common.config_dfs import env_dfs_set_ins_csst1 + from csst_mci_common.config_dfs import env_dfs_set_l0_csst1 + + env_crds_set_ins_csst1() + self.obsid = csst_mci.config.get_mci_config("obsid") + env_dfs_set_l0_csst1(obsid=self.obsid) + env_dfs_set_ins_csst1(target_detector) + + self._bias_ref = csst_mci.config.get_mci_config("bias_ref") + self._dark_ref = csst_mci.config.get_mci_config("dark_ref") + self._flat_ref = csst_mci.config.get_mci_config("flat_ref") + self._shutter_ref = csst_mci.config.get_mci_config("shutter_ref") + self._sci_path = csst_mci.config.get_mci_config("sci_path") + self._output_dir = csst_mci.config.get_mci_config("ins_output") + + return + + # 在此处编写代码，使得用户通过执行此方法配置本地运行所需要的参数，此处方法可用于单元测试或实际单机运行。 + def para_set_local( + self, + image_path: str = "/path/to/image", + bias_ref_path: str = "/path/to/bias", + weight_path: str = "/path/to/weight", + flag_path: str = "/path/to/flag", + cat_ref_path: str = "/path/to/cat_ref", + deg: int = 2, + ) -> CsstResult: + """Method, estimate image distortion for a single MSC-MBI image. + + Parameters + ---------- + image_path: str + Path of image, defaulting to "/path/to/image". + weight_path: str + Path of weight, defaulting to "/path/to/weight". + flag_path: str + Path of flag, defaulting to "/path/to/flag". + cat_ref_path: str + Path of reference catalog, defaulting to "/path/to/cat_ref". + deg + + + Returns + ------- + CsstResult + Result containing `status`, `file_list`, and `output`. + """ + self.image_path = image_path + self.bias_ref = bias_ref_path + self.dark_ref = csst_mci.config.get_mci_config("dark_ref") + self.flat_ref = csst_mci.config.get_mci_config("flat_ref") + self.shutter_ref = csst_mci.config.get_mci_config("shutter_ref") + self.sci_path = csst_mci.config.get_mci_config("sci_path") + self.output_dir = csst_mci.config.get_mci_config("ins_output") + return + + + def para_set_custom_1(self): + return + + # 在此处编写代码，使得用户通过执行此方法进行结果检查 + def result_check(self): + pass