Skip to content
GitLab
Commit 1be038fd authored by Wei Shoulin's avatar Wei Shoulin
Browse files

un level0

parent 3b1aed9a
Show whitespace changes
Inline Side-by-side
csst_dfs_api/sls/level0prc.py deleted 100644 → 0
View file @ 3b1aed9a
from ..common.delegate import Delegate
class Level0PrcApi(object):
"""
Level 0 Data Operation API
"""
def __init__(self):
self.pymodule = Delegate().load(sub_module = "sls")
self.stub = getattr(self.pymodule, "Level0PrcApi")()
def find(self, **kwargs):
''' retrieve level0 procedure records from database
:param kwargs: Parameter dictionary, key items support:
level0_id: [str]
pipeline_id: [str]
prc_module: [str]
prc_status : [int]
:returns: csst_dfs_common.models.Result
'''
return self.stub.find(**kwargs)
def update_proc_status(self, **kwargs):
''' update the status of reduction
:param kwargs: Parameter dictionary, key items support:
id : [int],
status : [int]
:returns: csst_dfs_common.models.Result
'''
return self.stub.update_proc_status(**kwargs)
def write(self, **kwargs):
''' insert a level0 procedure record into database
:param kwargs: Parameter dictionary, key items support:
level0_id : [str]
pipeline_id : [str]
prc_module : [str]
params_file_path : [str]
prc_status : [int]
prc_time : [str]
result_file_path : [str]
:returns: csst_dfs_common.models.Result
'''
return self.stub.write(**kwargs)
csst_dfs_api/sls/level1.py
View file @ 1be038fd
......@@ -25,6 +25,17 @@ class Level1DataApi(object):
'''
return self.stub.find(**kwargs)
def find_by_prc_status(self, **kwargs):
''' retrieve level1 records from database
:param kwargs: Parameter dictionary, key items support:
prc_status : [int],
limit: limits returns the number of records,default 0:no-limit
:returns: csst_dfs_common.models.Result
'''
return self.stub.find_by_prc_status(**kwargs)
def get(self, **kwargs):
''' fetch a record from database
......
tests/common/test_common_catalog.py
View file @ 1be038fd
......@@ -13,18 +13,18 @@ class CommonCatalogTestCase(unittest.TestCase):
def test_catalog_query(self):
t= time.time()
result = self.api.catalog_query(
ra=193, # 15415
dec=26,
radius=0.1,
ra=193.299,
dec=19.08851,
radius=2,
catalog_name='gaia3',
min_mag=-1,
max_mag=-1,
obstime = -1,
limit = 1
limit = 0
)
print(result)
# dt = self.api.to_table(result)
# df = dt.to_pandas()
# print(df.head())
print('used:', time.time()-t)
print('return:', result['totalCount'])
# print('return:', result['totalCount'\])
tests/cpic/test_cpic_cal_merge.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.cpic.calmerge import CalMergeApi
class SLSCalMergeApiTestCase(unittest.TestCase):
def setUp(self):
self.api = CalMergeApi()
def test_find(self):
recs = self.api.find(detector_no='CCD01',
ref_type = "bias",
obs_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13"))
print('find:', recs)
def test_get_latest_by_l0(self):
rec = self.api.get_latest_by_l0(level0_id='000001102', ref_type = "bias")
print('get_latest_by_l0:', rec)
def test_get(self):
rec = self.api.get(id = 2)
print('get by id:', rec)
rec = self.api.get(cal_id = '2')
print('get by cal_id:', rec)
# def test_update_proc_status(self):
# rec = self.api.update_proc_status(id = 3, status = 1)
# print('update_proc_status:', rec)
# def test_update_qc1_status(self):
# rec = self.api.update_qc1_status(id = 3, status = 2)
# print('update_qc1_status:', rec)
# def test_write(self):
# rec = self.api.write(
# cal_id = '00002',
# detector_no = '01',
# ref_type = "bias",
# obs_time = "2021-06-04 11:12:13",
# exp_time = 150,
# filename = "/opt/dddasd1.params",
# file_path = "/opt/dddasd1.fits",
# prc_status = 3,
# prc_time = '2021-06-04 11:12:13',
# level0_ids = ['1','2','3','4'])
# print('write:', rec)
\ No newline at end of file
tests/cpic/test_cpic_level0.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.cpic.level0 import Level0DataApi
class CPICLevel0DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0DataApi()
def test_find(self):
recs = self.api.find(obs_id = '000009', obs_type = 'sci', limit = 0)
print('find:', recs)
def test_get(self):
rec = self.api.get(id = 100)
print('get:', rec)
rec = self.api.get(level0_id = '1000000102')
print('get:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(level0_id = '000001102', status = 6)
print('update_proc_status:', rec)
def test_update_qc0_status(self):
rec = self.api.update_qc0_status(level0_id = '000001102', status = 7)
print('update_qc0_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id = '000001101',
obs_id = '0000011',
detector_no = "01",
obs_type = "sci",
obs_time = "2021-06-06 11:12:13",
exp_time = 150,
detector_status_id = 3,
filename = "MSC_00001234",
file_path = "/opt/MSC_00001234.fits")
print('write:', rec)
\ No newline at end of file
tests/cpic/test_cpic_level0_prc.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.cpic.level0prc import Level0PrcApi
class CPICLevel0PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0PrcApi()
def test_find(self):
recs = self.api.find(level0_id='134')
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 8, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='134',
pipeline_id = "P1",
prc_module = "QC0",
params_file_path = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-06-04 11:12:13',
result_file_path = "/opt/dddasd.header")
print('write:', rec)
\ No newline at end of file
tests/cpic/test_cpic_level1.py
View file @ 1be038fd
......@@ -2,7 +2,7 @@ import unittest
from csst_dfs_api.cpic import Level1DataApi
class CPICResult1TestCase(unittest.TestCase):
class CPICLevel1TestCase(unittest.TestCase):
def setUp(self):
self.api = Level1DataApi()
......@@ -28,7 +28,7 @@ class CPICResult1TestCase(unittest.TestCase):
def test_write(self):
rec = self.api.write(
level0_id='0000223',
level0_id='400001223',
data_type = "sci",
prc_params = "/opt/dddasd.params",
prc_status = 3,
......
tests/msc/test_msc_cal_merge.py → tests/facility/test_facility_cal_merge.py
View file @ 1be038fd
import os
import unittest
from astropy.io import fits
from csst_dfs_api.msc.calmerge import CalMergeApi
from csst_dfs_api.facility.calmerge import CalMergeApi
class MSCCalMergeApiTestCase(unittest.TestCase):
......@@ -10,7 +8,7 @@ class MSCCalMergeApiTestCase(unittest.TestCase):
self.api = CalMergeApi()
def test_find(self):
recs = self.api.find(detector_no='CCD01',
recs = self.api.find(detector_no='01',
ref_type = "bias",
obs_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13"))
print('find:', recs)
......
tests/sls/test_sls_level0.py → tests/facility/test_facility_level0.py
View file @ 1be038fd
import os
import unittest
from astropy.io import fits
from csst_dfs_api.sls.level0 import Level0DataApi
from csst_dfs_api.facility.level0 import Level0DataApi
class SLSLevel0DataTestCase(unittest.TestCase):
class Level0DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0DataApi()
def test_find(self):
recs = self.api.find(obs_id = '000009', obs_type = 'sci', limit = 0)
recs = self.api.find(obs_id = '0000011', obs_type = 'sci', limit = 0)
print('find:', recs)
def test_get(self):
rec = self.api.get(id = 100)
rec = self.api.get(id = 3)
print('get:', rec)
rec = self.api.get(level0_id = '1000000102')
rec = self.api.get(level0_id = '000001101',obs_type='sci')
print('get:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(level0_id = '000001102', status = 6)
rec = self.api.update_proc_status(level0_id = '000001101', status = 6)
print('update_proc_status:', rec)
def test_update_qc0_status(self):
rec = self.api.update_qc0_status(level0_id = '000001102', status = 7)
rec = self.api.update_qc0_status(level0_id = '000001101', status = 7)
print('update_qc0_status:', rec)
def test_write(self):
......
tests/ifs/test_ifs_level0_prc.py → tests/facility/test_facility_level0_prc.py
View file @ 1be038fd
import os
import unittest
from astropy.io import fits
from csst_dfs_api.ifs.level0prc import Level0PrcApi
from csst_dfs_api.facility.level0prc import Level0PrcApi
class IFSLevel0PrcTestCase(unittest.TestCase):
class Level0PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0PrcApi()
......
tests/ifs/test_ifs_cal_merge.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.ifs.calmerge import CalMergeApi
class IFSCalMergeApiTestCase(unittest.TestCase):
def setUp(self):
self.api = CalMergeApi()
def test_find(self):
recs = self.api.find(detector_no='CCD01',
ref_type = "bias",
obs_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13"))
print('find:', recs)
def test_get_latest_by_l0(self):
rec = self.api.get_latest_by_l0(level0_id='000001102', ref_type = "bias")
print('get_latest_by_l0:', rec)
def test_get(self):
rec = self.api.get(id = 2)
print('get by id:', rec)
rec = self.api.get(cal_id = '2')
print('get by cal_id:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 3, status = 1)
print('update_proc_status:', rec)
def test_update_qc1_status(self):
rec = self.api.update_qc1_status(id = 3, status = 2)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
cal_id = '00002',
detector_no = '01',
ref_type = "bias",
obs_time = "2021-06-04 11:12:13",
exp_time = 150,
filename = "/opt/dddasd1.params",
file_path = "/opt/dddasd1.fits",
prc_status = 3,
prc_time = '2021-06-04 11:12:13',
level0_ids = ['1','2','3','4'])
print('write:', rec)
\ No newline at end of file
tests/ifs/test_ifs_level0.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.ifs.level0 import Level0DataApi
class IFSLevel0DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0DataApi()
# def test_find(self):
# result = self.api.find(ra = 248, dec = 87, radius = 2, limit = 0, obs_type = "sky")
# print('find:', result)
# def test_get(self):
# result = self.api.get(id = 1)
# print('get:', result.data)
# rec = self.api.get(level0_id = '300000145CCD231-c4')
# print('get:', rec)
# def test_update_proc_status(self):
# rec = self.api.update_proc_status(level0_id = '000001102', status = 6)
# print('update_proc_status:', rec)
# def test_update_qc0_status(self):
# rec = self.api.update_qc0_status(level0_id = '000001102', status = 7)
# print('update_qc0_status:', rec)
def test_write(self):
rec = self.api.write(file_path = "/Users/wsl/temp/csst/ifs_data/sky_Data/CSST_IFS_B_sky_20211225001425_20211225001925_300000013_X_L0_VER_I4203.fits")
print('write:', rec)
\ No newline at end of file
tests/ifs/test_ifs_level1_prc.py
View file @ 1be038fd
import os
import unittest
from astropy.io import fits
from csst_dfs_api.ifs.level1prc import Level1PrcApi
......
tests/mci/test_mci_cal_merge.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.mci.calmerge import CalMergeApi
class MCICalMergeApiTestCase(unittest.TestCase):
def setUp(self):
self.api = CalMergeApi()
def test_find(self):
recs = self.api.find(detector_no='CCD01',
ref_type = "bias",
obs_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13"))
print('find:', recs)
def test_get_latest_by_l0(self):
rec = self.api.get_latest_by_l0(level0_id='000001102', ref_type = "bias")
print('get_latest_by_l0:', rec)
def test_get(self):
rec = self.api.get(id = 2)
print('get by id:', rec)
rec = self.api.get(cal_id = '2')
print('get by cal_id:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 3, status = 1)
print('update_proc_status:', rec)
def test_update_qc1_status(self):
rec = self.api.update_qc1_status(id = 3, status = 2)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
cal_id = '00002',
detector_no = '01',
ref_type = "bias",
obs_time = "2021-06-04 11:12:13",
exp_time = 150,
filename = "/opt/dddasd1.params",
file_path = "/opt/dddasd1.fits",
prc_status = 3,
prc_time = '2021-06-04 11:12:13',
level0_ids = ['1','2','3','4'])
print('write:', rec)
\ No newline at end of file
tests/mci/test_mci_level0.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.mci.level0 import Level0DataApi
class MCILevel0DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0DataApi()
def test_find(self):
recs = self.api.find(limit = 0)
print('find:', recs)
def test_get(self):
rec = self.api.get(id = 1)
print('get:', rec)
rec = self.api.get(level0_id = '200000006-')
print('get:', rec)
# def test_update_proc_status(self):
# rec = self.api.update_proc_status(level0_id = '000001102', status = 6)
# print('update_proc_status:', rec)
# def test_update_qc0_status(self):
# rec = self.api.update_qc0_status(level0_id = '000001102', status = 7)
# print('update_qc0_status:', rec)
# def test_write(self):
# rec = self.api.write(
# file_path = "/opt/MSC_00001234.fits")
# print('write:', rec)
\ No newline at end of file
tests/mci/test_mci_level0_prc.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.mci.level0prc import Level0PrcApi
class MCILevel0PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0PrcApi()
def test_find(self):
recs = self.api.find(level0_id='134')
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 8, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='134',
pipeline_id = "P1",
prc_module = "QC0",
params_file_path = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-06-04 11:12:13',
result_file_path = "/opt/dddasd.header")
print('write:', rec)
\ No newline at end of file
tests/msc/test_msc_level0.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.msc.level0 import Level0DataApi
class MSCLevel0DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0DataApi()
def test_find(self):
recs = self.api.find(obs_id = '100000154', obs_type = 'sci', limit = 0)
print('find:', recs)
def test_get(self):
rec = self.api.get(id = 2)
print('get:', rec)
rec = self.api.get(level0_id = '10000000123')
print('get:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(level0_id = '000001102', status = 6)
print('update_proc_status:', rec)
def test_update_qc0_status(self):
rec = self.api.update_qc0_status(level0_id = '000001102', status = 7)
print('update_qc0_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id = '000001101',
obs_id = '0000011',
detector_no = "01",
obs_type = "sci",
obs_time = "2021-06-06 11:12:13",
exp_time = 150,
detector_status_id = 3,
filename = "MSC_00001234",
file_path = "/opt/MSC_00001234.fits")
print('write:', rec)
\ No newline at end of file
tests/msc/test_msc_level0_prc.py deleted 100644 → 0
View file @ 3b1aed9a
import os
import unittest
from astropy.io import fits
from csst_dfs_api.msc.level0prc import Level0PrcApi
class MSCLevel0PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level0PrcApi()
def test_find(self):
recs = self.api.find(level0_id='134')
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 8, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='134',
pipeline_id = "P1",
prc_module = "QC0",
params_file_path = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-06-04 11:12:13',
result_file_path = "/opt/dddasd.header")
print('write:', rec)
\ No newline at end of file
tests/msc/test_msc_level1.py
View file @ 1be038fd
......@@ -36,7 +36,7 @@ class MSCLevel1DataTestCase(unittest.TestCase):
prc_status = 3,
prc_time = '2021-10-22 11:12:13',
filename = "MSC_MS_210525121500_100000001_09_raw",
file_path = "/opt/temp/csst/MSC_MS_210525121500_100000001_09_raw.fits",
file_path = "/opt/temp/csst/msc/level1/MSC_210304093000_0000000_06_img.fits.fz",
pipeline_id = "P1",
refs = {'dark': 1, 'bias': 2, 'flat': 3 })
print('write:', rec)
\ No newline at end of file
tests/msc/test_msc_level2.py
View file @ 1be038fd
import os
import unittest
from astropy.io import fits
from csst_dfs_api.common.utils import to_table as to_fits_table
from csst_dfs_api.msc.level2 import Level2DataApi
class MSCLevel2DataTestCase(unittest.TestCase):
......@@ -9,42 +9,51 @@ class MSCLevel2DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level2DataApi()
def test_find(self):
recs = self.api.find(
level1_id=1,
create_time = ("2022-06-15 11:12:13","2022-06-16 11:12:13"))
print('find:', recs)
def test_catalog_query(self):
result = self.api.catalog_query(
obs_id='100000000',
obs_time = ("2021-05-24 11:12:13","2021-05-25 13:12:13"),
limit = 2)
if result.success:
dt = self.api.to_table(result)
dt.pprint()
else:
# def test_find(self):
# recs = self.api.find(
# level1_id=1,
# create_time = ("2022-06-15 11:12:13","2022-06-16 11:12:13"))
# print('find:', recs)
# def test_catalog_query(self):
# result = self.api.catalog_query(
# obs_id = '100000133',
# detector_no = '13',
# ra = 192,
# dec = 26,
# radius = 1,
# limit = 2)
# print(result)
# if result.success and result['totalCount'] > 0:
# dt = to_fits_table(result)
# dt.pprint()
def test_catalog_query_file(self):
result = self.api.catalog_query_file(
ra = 192,
dec = 26,
radius = 1)
print(result)
def test_get(self):
rec = self.api.get(id = 1)
print('get:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 1, status = 4)
print('update_proc_status:', rec)
def test_update_qc2_status(self):
rec = self.api.update_qc2_status(id = 1, status = 7)
print('update_qc2_status:', rec)
def test_write(self):
rec = self.api.write(
level1_id= 1,
data_type = "sci",
prc_status = 3,
prc_time = '2021-10-22 11:12:13',
filename = "MSC_MS_210525120000_100000000_20_cat.fits",
file_path = "/opt/temp/csst/msc_data/MSC_MS_210525120000_100000000_20_cat.fits"
)
print('write:', rec)
# def test_get(self):
# rec = self.api.get(id = 1)
# print('get:', rec)
# def test_update_proc_status(self):
# rec = self.api.update_proc_status(id = 1, status = 4)
# print('update_proc_status:', rec)
# def test_update_qc2_status(self):
# rec = self.api.update_qc2_status(id = 1, status = 7)
# print('update_qc2_status:', rec)
# def test_write(self):
# rec = self.api.write(
# level1_id= 1,
# data_type = "sci",
# prc_status = 3,
# prc_time = '2021-10-22 11:12:13',
# filename = "MSC_MS_210525120000_100000000_20_cat.fits",
# file_path = "/opt/temp/csst/msc_data/MSC_MS_210525120000_100000000_20_cat.fits"
# )
# print('write:', rec)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment