added FileRecorder

    Examples

    --------

    >>> dm_mbi = CsstMbiDataManager(...)

    >>> dm_mbi = CsstMsDataManager(...)

    >>> # access L0 directory

    >>> dm_mbi.dir_l0

    >>> # access L1 directory

from collections import namedtuple

from astropy import table

import os

File = namedtuple(

    "FileRecord",

    ["filepath", "db", "comment", "existence"]

)

class FileRecorder(list):

    """

    File Recorder, inherited from the built-in ``list``.

    This recorder is used to record files generated by functional modules.

    In principle, a CSST pipeline module should return a status (CsstStatus)

    and a file recorder (``FileRecorder``) after it finishes data processing.

    Methods

    -------

    add_record(filepath: str = "", db: bool = False, comment: str = "")

        Add a file record, each record is a ``collections.namedtuple``.

        Users should provide a file path ``filepath``, whether this file should be

        written to database ``db``, a comment ``comment``.

        An existence boolean will be attached to check if the file exists.

    to_table()

        Convert to ``astropy.table.Table``.

    pprint(*args, **kwargs)

        Use ``astropy.table.Table.pprint`` to print file records in table format.

    pprint_all(*args, **kwargs)

        Use ``astropy.table.Table.pprint_all`` to print file records in table format.

    Examples

    --------

    >>> fr = FileRecorder()

    >>> for i in range(3):

    >>>     fr.add_record("test{:03d}.txt".format(i), db=True, comment="Test file {:d}".format(i))

    >>> fr.pprint_all()

    <FileRecorder length=3>

      filepath   db    comment   existence

    ----------- ---- ----------- ---------

    test000.txt True Test file 0     False

    test001.txt True Test file 1     False

    test002.txt True Test file 2     False

    >>> fr.pprint_all()

    <FileRecorder length=3>

      filepath   db    comment   existence

    ----------- ---- ----------- ---------

    test000.txt True Test file 0     False

    test001.txt True Test file 1     False

    test002.txt True Test file 2     False

    """

    def __init__(self):

        super(FileRecorder, self).__init__()

    def add_record(self, filepath: str = "", db: bool = False, comment: str = ""):

        existence = os.path.exists(filepath)

        assert isinstance(filepath, str)

        assert isinstance(db, bool)

        assert isinstance(comment, str)

        super().append(File(filepath=filepath, db=db, comment=comment, existence=existence))

    def to_table(self):

        return table.Table([_._asdict() for _ in self])

    def __repr__(self):

        return self.to_table().__repr__()

    def pprint(self, *args, **kwargs):

        return self.to_table().pprint(*args, **kwargs)

    def pprint_all(self, *args, **kwargs):

        print("<FileRecorder length={}>".format(len(self)))

        return self.to_table().pprint_all(*args, **kwargs)

from .data_manager import CsstMsDataManager

from .logger import get_logger

from .status import CsstStatus

from .file_recorder import FileRecorder

__all__ = ["CsstMsDataManager", "get_logger"]

__all__ = ["CsstMsDataManager", "get_logger", "FileRecorder"]

"""

Identifier:     KSC-SJ4-tests/test_data_manager.py

Name:           test_data_manager.py

Description:    data manager unit test

Author:         Bo Zhang

Created:        2022-09-13

Modified-History:

    2022-10-10, Bo Zhang, created

    2022-10-10, Bo Zhang, added test for FileRecorder

"""

import unittest

from csst_common.file_recorder import FileRecorder, File

# deprecated in some days

class TestCsstMbiDataManager(unittest.TestCase):

    def setUp(self) -> None:

        self.fr = FileRecorder()

    def test_add_record(self):

        self.fr.add_record(filepath="test1", db=False, comment="the first test")

        self.assertTrue(self.fr[0] == File("test1", db=False, comment="the first test", existence=False))