Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
Chen Wei
csst_shao_interface_templet
Commits
0a0fde28
Commit
0a0fde28
authored
Nov 09, 2023
by
chenwei@shao.ac.cn
Browse files
update
parent
7041526a
Changes
9
Hide whitespace changes
Inline
Side-by-side
csst_shao_common/__init__.py
deleted
100644 → 0
View file @
7041526a
csst_shao_common/common.py
deleted
100644 → 0
View file @
7041526a
"""
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
csst_shao_common/config.py
deleted
100644 → 0
View file @
7041526a
"""
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
csst_shao_common/exit_code_list.py
deleted
100644 → 0
View file @
7041526a
"""
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
csst_shao_common/ifs_status.py
deleted
100644 → 0
View file @
7041526a
"""
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
csst_shao_common/interface.py
deleted
100644 → 0
View file @
7041526a
# -*- 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
csst_shao_common/logger.py
deleted
100644 → 0
View file @
7041526a
"""
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 ...")
csst_shao_common/top_level_interface.py
deleted
100644 → 0
View file @
7041526a
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"
]
csst_shao_templet.py
View file @
0a0fde28
...
...
@@ -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
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment