tweaks

csst_common/pipeline.py

@@ -44,10 +44,12 @@ def print_directory_tree(directory="."):
             print(f"{subindent}{file}")
 
 
+# reason: code
 EXIT_CODES = {
     "success": 0,
     "ccds_reference_file_error": 10,  # CCDS 参考文件错误
-    "reference_catalog_error": 11,  # 参考星表错误
+    "ccds_pmap_error": 11,  # CCDS pmap 错误
+    "reference_catalog_error": 12,  # 参考星表错误
     "input_data_invalid": 20,  # 输入数据错误
     "input_data_not_unique": 21,  # 输入数据不唯一
     "algorithm_error": 30,  # 算法错误
@@ -89,6 +91,7 @@ class Pipeline:
             "CREATED": os.getenv("CREATED", "-"),
             # additional settings
             "VERBOSE": os.getenv("VERBOSE", "false").lower() == "true",
+            "INTTEST": os.getenv("INTTEST", "false").lower() == "true",
             "IGNORE_WARNINGS": os.getenv("IGNORE_WARNINGS", "true").lower() == "true",
             "USE_OSS": (os.getenv("USE_OSS", "false")).lower() == "true",
         }
@@ -110,7 +113,7 @@ class Pipeline:
         self.dfs2 = csst_fs
         self.ccds = CCDS(ccds_root=self.ccds_root, ccds_cache=self.ccds_cache)
 
-        # exit code
+        # exit code -> p.exit(reason="ccds_error")
         self.EXIT_CODES = EXIT_CODES
 
         # record start time
@@ -223,7 +226,7 @@ class Pipeline:
                 # CCDS
                 return os.path.join(self.ccds_root, file_path)
 
-    def download_dfs_file(self, file_path: str, dir_dst: str = None):
+    def download_dfs_file(self, file_path: str, dir_dst: str = None) -> str:
         """Copy DFS file to output directory."""
         if dir_dst is None:
             dir_dst = self.dir_output
@@ -311,11 +314,31 @@ class Pipeline:
 
     @property
     def pmapname(self):
-        """CCDS `.pmap` name (operational context)."""
-        if self.ccds is not None:
+        """Final CCDS `.pmap` name."""
+        task_pmapname = self.task.get("pmapname", None)
+        # TODO: validate this pmapname
+        if task_pmapname and self.ccds.validate(task_pmapname):
+            # task specified pmap
+            return task_pmapname
+        else:
+            # CCDS recommended pmap
             return self.ccds.operational_context
+
+    @property
+    def ref_cat(self):
+        """Final DFS catalog name."""
+        task_ref_cat = self.task.get("ref_cat", None)
+        if task_ref_cat and task_ref_cat in self.dfs1.catalog.all_catalog_names:
+            # task specified catalog
+            return task_ref_cat
         else:
-            raise ValueError("CCDS client not initialized!")
+            # DFS recommended catalog
+            if self.docker_image == "csst-msc-l1-mbi":
+                return "trilegal_093"
+            else:
+                raise ValueError(
+                    f"Invalid ref_cat: {task_ref_cat} not in {self.dfs1.catalog.all_catalog_names}"
+                )
 
     # automatically log pipeline information
     def info(self):
@@ -324,6 +347,24 @@ class Pipeline:
         self.logger.info(f"BUILD={self.build}")
         self.logger.info(f"CREATED={self.created}")
         self.logger.info(f"VERBOSE={self.verbose}")
+        self.logger.info(f"INTTEST={self.inttest}")
+
+    def exit(self, reason: str = None):
+        """Exit pipeline with reason.
+
+        Examples
+        --------
+        >>> p.exit(reason="ccds_error")
+        >>> p.exit(reason="success")
+        """
+        assert (
+            reason in self.EXIT_CODES.keys()
+        ), f"Reason {reason} not in {self.EXIT_CODES.keys()}"
+        sys.exit(self.EXIT_CODES[reason])
+
+    def add_metadata(self, meta: dict):
+        """Add metadata to pipeline."""
+        raise NotImplementedError()
 
 
 class ModuleResult(NamedTuple):