Loading csst_common/pipeline.py +11 −0 Original line number Diff line number Diff line Loading @@ -18,11 +18,13 @@ import traceback import warnings from typing import Callable, NamedTuple, Optional, Any from astropy.time import Time, TimeDelta from astropy.io import fits from .utils import retry from .ccds import CCDS from .dfs import DFS from .file import File from .io import reformat_header from .logger import get_logger from .status import CsstStatus, CsstResult Loading Loading @@ -130,6 +132,15 @@ class Pipeline: print(f"PIPELINE_ID={self.pipeline_id}") print(f"BUILD={self.build}") print(f"CREATED={self.created}") print(f"VERBOSE={self.verbose}") def summary(self) -> fits.Header: """Summarize pipeline info into a `astropy.io.fits.Header`.""" h = fits.Header() h.set("PIPELINE_ID", self.pipeline_id, comment="pipeline ID") h.set("BUILD", self.build, comment="pipeline build number") h.set("CREATED", self.pipeline_id, comment="pipeline build time") return reformat_header(h, strip=False, comment="Pipeline summary") def clean_output(self): """Clean output directory.""" Loading Loading
csst_common/pipeline.py +11 −0 Original line number Diff line number Diff line Loading @@ -18,11 +18,13 @@ import traceback import warnings from typing import Callable, NamedTuple, Optional, Any from astropy.time import Time, TimeDelta from astropy.io import fits from .utils import retry from .ccds import CCDS from .dfs import DFS from .file import File from .io import reformat_header from .logger import get_logger from .status import CsstStatus, CsstResult Loading Loading @@ -130,6 +132,15 @@ class Pipeline: print(f"PIPELINE_ID={self.pipeline_id}") print(f"BUILD={self.build}") print(f"CREATED={self.created}") print(f"VERBOSE={self.verbose}") def summary(self) -> fits.Header: """Summarize pipeline info into a `astropy.io.fits.Header`.""" h = fits.Header() h.set("PIPELINE_ID", self.pipeline_id, comment="pipeline ID") h.set("BUILD", self.build, comment="pipeline build number") h.set("CREATED", self.pipeline_id, comment="pipeline build time") return reformat_header(h, strip=False, comment="Pipeline summary") def clean_output(self): """Clean output directory.""" Loading