Commit 3570fb46 authored by BO ZHANG's avatar BO ZHANG 🏀
Browse files

configurable log paths

parent a3521701

csst_common/pipeline.py

+20 −7
Original line number Diff line number Diff line
@@ -12,6 +12,7 @@ Modified-History: 
"""

import json

import os

import shutil

import subprocess

import warnings

from typing import Any
@@ -24,6 +25,7 @@ from .file import File 
from .logger import get_logger





# TODO: unit test in need

class Pipeline:

    """

    CSST pipeline configuration class.
@@ -62,6 +64,8 @@ class Pipeline: 
        dfs_root: str = "/dfs_root",

        ccds_root: str = "/ccds_root",

        ccds_cache: str = "/pipeline/ccds_cache",

        pipeline_log: str = "pipeline.log",

        module_log: str = "module.log",

        filter_warnings: bool = False,

        dfs: bool = True,

        ccds: bool = False,
@@ -86,11 +90,11 @@ class Pipeline: 
        # set logger

        self.pipeline_logger = get_logger(

            name="pipeline logger",

            filename=os.path.join(self.dir_output, "pipeline.log"),

            filename=os.path.join(self.dir_output, pipeline_log),

        )

        self.module_logger = get_logger(

            name="module logger",

            filename=os.path.join(self.dir_output, "module.log"),

            filename=os.path.join(self.dir_output, module_log),

        )



        # change working directory
@@ -108,10 +112,7 @@ class Pipeline: 
        else:

            self.dfs = None

        if ccds:

            self.ccds = CCDS(

                ccds_root=ccds_root,

                ccds_cache=ccds_cache,

            )

            self.ccds = CCDS(ccds_root=ccds_root, ccds_cache=ccds_cache)

        else:

            self.ccds = None
@@ -144,15 +145,27 @@ class Pipeline: 


    @staticmethod

    def reset_warnings(self):

        # Reset warning filters

        """Reset warning filters."""

        warnings.resetwarnings()



    def file(self, file_path):

        """Initialize File object."""

        return File(file_path, new_dir=self.dir_output)



    def new(self, file_name="test.fits"):

        """Create new file in output directory."""

        return os.path.join(self.dir_output, file_name)



    @staticmethod

    def move(file_src: str, file_dst: str):

        """Move file `file_src` to `file_dist`."""

        shutil.move(file_src, file_dst)



    @staticmethod

    def copy(file_src: str, file_dst: str):

        """Move file `file_src` to `file_dist`."""

        shutil.copy(file_src, file_dst)





# class ErrorTrace:

#     """Write error trace to file."""