Commit a1e6ad7b authored by Wei Shoulin's avatar Wei Shoulin
Browse files

rm module level1

parent 4cd9a653
import unittest
from csst_dfs_api.cpic import Level1DataApi
class CPICLevel1TestCase(unittest.TestCase):
def setUp(self):
self.api = Level1DataApi()
def test_find(self):
recs = self.api.find(
level0_id='400001223',
create_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13")
)
print('find:', recs)
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_qc1_status(self):
rec = self.api.update_qc1_status(id = 1, status = 7)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='400001223',
data_type = "sci",
prc_params = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-10-22 11:12:13',
filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
pipeline_id = "P2",
refs = {'dark': 1, 'bias': 2, 'flat': 3 })
print('write:', rec)
\ No newline at end of file
import os
import unittest
from astropy.io import fits
from csst_dfs_api.cpic.level1prc import Level1PrcApi
class CPICLevel1PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1PrcApi()
def test_find(self):
recs = self.api.find(level1_id=1)
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 1, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(level1_id=1,
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
...@@ -2,9 +2,9 @@ import os ...@@ -2,9 +2,9 @@ import os
import unittest import unittest
from astropy.io import fits from astropy.io import fits
from csst_dfs_api.mbi.level1 import Level1DataApi from csst_dfs_api.facility.level1 import Level1DataApi
class MBILevel1DataTestCase(unittest.TestCase): class Level1DataTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
self.api = Level1DataApi() self.api = Level1DataApi()
...@@ -38,5 +38,5 @@ class MBILevel1DataTestCase(unittest.TestCase): ...@@ -38,5 +38,5 @@ class MBILevel1DataTestCase(unittest.TestCase):
filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits", filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits", file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
pipeline_id = "P1", pipeline_id = "P1",
refs = {'dark': 1, 'bias': 2, 'flat': 3 }) refs = {'dark': '1', 'bias': '2', 'flat': '3' })
print('write:', rec) print('write:', rec)
\ No newline at end of file
...@@ -2,9 +2,9 @@ import os ...@@ -2,9 +2,9 @@ import os
import unittest import unittest
from astropy.io import fits from astropy.io import fits
from csst_dfs_api.mci.level1prc import Level1PrcApi from csst_dfs_api.facility.level1prc import Level1PrcApi
class MCILevel1PrcTestCase(unittest.TestCase): class Level1PrcTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
self.api = Level1PrcApi() self.api = Level1PrcApi()
......
import unittest
from csst_dfs_api.hstdm import Level1DataApi
class HSTDMLevel1TestCase(unittest.TestCase):
def setUp(self):
self.api = Level1DataApi()
def test_find(self):
recs = self.api.find(
level0_id='50002981600',
create_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13")
)
print('find:', recs)
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_qc1_status(self):
rec = self.api.update_qc1_status(id = 1, status = 7)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='50002981600',
data_type = "sci",
prc_params = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-10-22 11:12:13',
filename = "MSC_MS_210525121500_100000001_09_raw",
file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
pipeline_id = "P2",
refs = {'dark': 1, 'bias': 2, 'flat': 3 })
print('write:', rec)
\ No newline at end of file
import unittest
from csst_dfs_api.hstdm.level1prc import Level1PrcApi
class HSTDMLevel1PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1PrcApi()
def test_find(self):
recs = self.api.find(level1_id=1)
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 1, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(level1_id=1,
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
import os
import unittest
from astropy.io import fits
from csst_dfs_api.ifs.level1 import Level1DataApi
class IFSLevel1DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1DataApi()
def test_find(self):
recs = self.api.find(
level0_id='000001201',
create_time = ("2021-10-01 11:12:13","2021-11-08 11:12:13"))
print('find:', recs)
def test_get(self):
rec = self.api.get(id = 2)
print('get:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 2, status = 4)
print('update_proc_status:', rec)
def test_update_qc1_status(self):
rec = self.api.update_qc1_status(id = 2, status = 7)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='000001201',
data_type = "sci",
cor_sci_id = 2,
prc_params = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-06-05 11:12:13',
filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
pipeline_id = "P2",
refs = {'dark': 1, 'bias': 2, 'flat': 3, 'sky': 33 })
print('write:', rec)
\ No newline at end of file
import unittest
from csst_dfs_api.ifs.level1prc import Level1PrcApi
class IFSLevel1PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1PrcApi()
def test_find(self):
recs = self.api.find(level1_id=1)
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 1, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(level1_id=1,
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
import os
import unittest
from astropy.io import fits
from csst_dfs_api.mbi.level1prc import Level1PrcApi
class MBILevel1PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1PrcApi()
def test_find(self):
recs = self.api.find(level1_id=1)
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 1, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(level1_id=1,
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
import os
import unittest
from astropy.io import fits
from csst_dfs_api.mci.level1 import Level1DataApi
class MCILevel1DataTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1DataApi()
def test_find(self):
recs = self.api.find(
level0_id='000001201',
create_time = ("2021-10-01 11:12:13","2021-11-08 11:12:13"))
print('find:', recs)
def test_get(self):
rec = self.api.get(id = 2)
print('get:', rec)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 2, status = 4)
print('update_proc_status:', rec)
def test_update_qc1_status(self):
rec = self.api.update_qc1_status(id = 2, status = 7)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='000001201',
data_type = "sci",
cor_sci_id = 2,
prc_params = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-06-05 11:12:13',
filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
pipeline_id = "P2",
refs = {'dark': 1, 'bias': 2, 'flat': 3, 'sky': 33 })
print('write:', rec)
\ No newline at end of file
import unittest
from csst_dfs_api.sls import Level1DataApi
class SLSResult1TestCase(unittest.TestCase):
def setUp(self):
self.api = Level1DataApi()
def test_find(self):
recs = self.api.find(
level0_id='0000223',
create_time = ("2021-06-01 11:12:13","2021-06-08 11:12:13")
)
print('find:', recs)
def test_find_by_prc_status(self):
recs = self.api.find_by_prc_status()
print('test_find_by_prc_status:', recs)
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_qc1_status(self):
rec = self.api.update_qc1_status(id = 1, status = 7)
print('update_qc1_status:', rec)
def test_write(self):
rec = self.api.write(
level0_id='0000223',
data_type = "sci",
prc_params = "/opt/dddasd.params",
prc_status = 3,
prc_time = '2021-10-22 11:12:13',
filename = "CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
file_path = "/opt/temp/csst/CSST_MSC_MS_SCI_20270810142128_20270810142358_100000101_20_img_L1.fits",
pipeline_id = "P2",
refs = {'dark': 1, 'bias': 2, 'flat': 3 })
print('write:', rec)
\ No newline at end of file
import os
import unittest
from astropy.io import fits
from csst_dfs_api.sls.level1prc import Level1PrcApi
class SLSLevel1PrcTestCase(unittest.TestCase):
def setUp(self):
self.api = Level1PrcApi()
def test_find(self):
recs = self.api.find(level1_id=1)
print('find:', recs)
def test_update_proc_status(self):
rec = self.api.update_proc_status(id = 1, status = 4)
print('update_proc_status:', rec)
def test_write(self):
rec = self.api.write(level1_id=1,
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
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