refactor(level1): simplify write function signature and update tests

csst_dfs_client/level1.py

@@ -190,61 +190,18 @@ def update_prc_status(level1_id: str, data_model: str, prc_status: int, batch_id
     """
     return request.put(f"/api/level1/prc_status/{level1_id}", {'data_model': data_model, 'prc_status': prc_status, 'batch_id': batch_id})
 
-def write(local_file: Union[IO, str], 
-        instrument: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'],
-        obs_group: Optional[str],
-        obs_type: str,
-        level1_id: str,
-        data_model: str,
-        file_name: str,
-        dag: str,
-        pmapname: str,
-        build: int,
-        dataset: str,
-        batch_id: str,        
-        level0_id: Optional[str] = None,
-        qc_status: int = 0,
-        **extra_kwargs) -> Result:
+def write(local_file: Union[IO, str], **extra_kwargs) -> Result:
     '''
     将本地的1级数据文件写入到DFS中其他参数如rss_id, cube_id等，可通过extra_kwargs传入
     
     Args:
         local_file (Union[IO, str]): 文件路径或文件对象
-        instrument ['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC']其中一个，代表: 模块ID
-        obs_group (str): 项目ID
-        obs_type (str): 观测类型
-        level0_id (Optional[str]): 0级数据的ID默认为 None
-        level1_id (str): 1级数据的ID
-        data_model (str): 数据类型
-        file_name (str): 1级数据文件名
-        dag (str): DAG标识
-        pmapname (str): CCDS pmap名称
-        build (int): 构建号
-        dataset (str): 数据集名称
-        batch_id (str): 批次ID
-        qc_status (int): QC状态
         **kwargs: 额外的关键字参数，这些参数将传递给DFS
     
     Returns:
         Result: 操作的结果对象，包含操作是否成功以及相关的错误信息，成功返回data为1级数据对象
     '''
-    params = {
-        'instrument': instrument, 
-        'obs_group': obs_group,
-        'obs_type': obs_type,
-        'level0_id': level0_id, 
-        'level1_id': level1_id, 
-        'data_model': data_model, 
-        'file_name': file_name, 
-        'dag': dag, 
-        'pmapname': pmapname, 
-        'build': build,
-        'dataset': dataset,
-        'batch_id': batch_id,
-        'qc_status': qc_status
-    }
-    if not dataset or not batch_id:
-        raise ValueError("dataset and batch_id is required")
+    params = {}
     params.update(extra_kwargs)
     if local_file is None:
         raise ValueError("local_file is required")

tests/test_level1.py

@@ -33,28 +33,28 @@ class Level1TestCase(unittest.TestCase):
     #     print(result)
     #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
 
-    # def test_write(self):
-    #     file_path = "/Users/wsl/temp/csst/L1/MSC/CSST_MSC_MS_WIDE_20230413214557_20230413214827_10100070335_13_L1_V01.fits"
-    #     # from io import BytesIO
-    #     # with open(file_path, "rb") as file:
-    #     #     file_path = BytesIO(file.read())
+    def test_write(self):
+        file_path = "/Users/wsl/temp/csst/L1/MSC/CSST_MSC_MS_WIDE_20230413214557_20230413214827_10100070335_13_L1_V01.fits"
+        # from io import BytesIO
+        # with open(file_path, "rb") as file:
+        #     file_path = BytesIO(file.read())
             
-    #     result = level1.write(local_file = file_path, 
-    #         obs_group="10109100157991",
-    #         obs_type="WIDE",
-    #         instrument = "MSC",
-    #         level0_id="1010910015799127",
-    #         level1_id = "1010910015799127", 
-    #         dag = "csst-msc-l1-mbi",
-    #         file_name= "CSST_MSC_MS_WIDE_20230413214557_20230413214827_10100070335_13_L1_V01.fits",
-    #         pmapname="csst_000128.pmap",
-    #         data_model='csst-msc-l1-mbi', 
-    #         build=1,
-    #         qc_status=10,
-    #         dataset='msc-v093',
-    #         batch_id="inttest")
-    #     print(result)
-    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
+        result = level1.write(local_file = file_path, 
+            obs_group="10109100157991",
+            obs_type="WIDE",
+            instrument = "MSC",
+            level0_id="1010910015799127",
+            level1_id = "1010910015799127", 
+            dag = "csst-msc-l1-mbi",
+            file_name= "CSST_MSC_MS_WIDE_20230413214557_20230413214827_10100070335_13_L1_V01.fits",
+            pmapname="csst_000128.pmap",
+            data_model='csst-msc-l1-mbi', 
+            build=1,
+            qc_status=10,
+            dataset='msc-v093',
+            batch_id="inttest")
+        print(result)
+        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
 
     # def test_find_process(self):
     #     result = level1.find_process(level1_id="1060940003452925")
@@ -80,8 +80,8 @@ class Level1TestCase(unittest.TestCase):
     #     print(result)
     #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)
     
-    def test_count_by_dataset(self):
-        result = level1.count_by_dataset(dataset="msc-v093")
-        print(result)
-        self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)    
-        self.assertIsNotNone(result.data, "error message: " + result.message)
+    # def test_count_by_dataset(self):
+    #     result = level1.count_by_dataset(dataset="msc-v093")
+    #     print(result)
+    #     self.assertEqual(result.code, 200, "error code: " + str(result.code) + ", message: " + result.message)    
+    #     self.assertIsNotNone(result.data, "error message: " + result.message)
\ No newline at end of file