add Pipeline.download_oss_file

csst_common/pipeline.py

@@ -9,6 +9,7 @@ Modified-History:
     2023-07-11, Bo Zhang, add Pipeline class
     2023-12-10, Bo Zhang, update Pipeline
     2023-12-15, Bo Zhang, add module header
+    2025-12-27, Bo Zhang, rewrite Pipeline class
 """
 
 import json
@@ -21,14 +22,15 @@ from typing import Callable, NamedTuple, Optional, Any, Union
 from astropy.time import Time, TimeDelta
 from astropy.io import fits
 
-import csst_dfs_client as dfs1
-import csst_fs as dfs2
+import csst_dfs_client
+import csst_fs
 
 from .ccds import CCDS
 from .utils import retry
 from .file import File
 from .logger import get_logger
 from .status import CsstStatus, CsstResult
+from .fits import s3_options
 
 
 def print_directory_tree(directory="."):
@@ -51,10 +53,16 @@ EXIT_CODES = {
     "data_product_invalid": 30,
     "data_product_ingestion_error": 31,
 }
-1
 
 
 class Pipeline:
+    """
+    Examples
+    --------
+    >>> p = Pipeline(msg="{'a':1}")
+    >>> p.msg_dict
+    {'a': 1}
+    """
 
     def __init__(self, msg: str = "", **env_vars: Any):
         # record start time
@@ -106,8 +114,8 @@ class Pipeline:
             self.filter_warnings("ignore")
 
         # DFS1, DFS2 & CCDS
-        self.dfs1 = dfs1
-        self.dfs2 = dfs2
+        self.dfs1 = csst_dfs_client
+        self.dfs2 = csst_fs
         self.ccds = CCDS(ccds_root=self.ccds_root, ccds_cache=self.ccds_cache)
 
         # exit code
@@ -180,15 +188,14 @@ class Pipeline:
         """Move file `file_src` to `file_dist`."""
         shutil.copy(file_src, file_dst)
 
-    def copy_to_output(self, file_paths: list):
-        for file_path in file_paths:
-            pass
-
-    def download_oss_file(self, oss_file_path: str) -> None:
-        """Download an OSS file from OSS to local path."""
-        # TODO
-        # self.dfs.download_file(oss_file_path, local_path)
-        pass
+    def download_oss_file(self, oss_file_path: str) -> str:
+        """Download an OSS file from OSS to output directory."""
+        local_file_path = os.path.join(self.dir_output, os.path.basename(oss_file_path))
+        csst_fs.s3_fs.get(oss_file_path, local_file_path, s3_options=s3_options)
+        assert os.path.exists(
+            local_file_path
+        ), f"Failed to download {oss_file_path} to {local_file_path}"
+        return local_file_path
 
     def call(self, func: Callable, *args: Any, **kwargs: Any):
         self.logger.info(f"=====================================================")