An error occurred while loading the file. Please try again.
file.py 4.48 KiB
"""
Identifier:     csst_common/file.py
Name:           file.py
Description:    file operation
Author:         Bo Zhang
Created:        2023-12-13
Modified-History:
    2023-12-13, Bo Zhang, created
    2023-12-13, Bo Zhang, add File
    2023-12-15, Bo Zhang, use re to match file name
    2023-12-15, Bo Zhang, add module header
"""
import os
import re
from typing import Optional
class File:
    """
    CSST MSC standard file path parser.
    It can be used for parse CSST MSC file paths and generate corresponding output file paths.
    Parameters
    ----------
    file_path : str
        File path.
    new_dir : str
        New directory.
    """
    def __init__(
        self, file_path: str = "/path/to/file.fits", new_dir: Optional[str] = None
        self.file_path = file_path
        # get dir name
        self.dirname = os.path.dirname(self.file_path)
        # get file name
        self.file_name = os.path.basename(self.file_path)
        # default new dir
        self.new_dir = new_dir if new_dir is not None else self.dirname
        # match file name
        pattern = re.compile(
            r"(?P<telescope>[A-Z]{4})_"
            r"(?P<instrument>[A-Z]+)_"
            r"(?P<project>[A-Z]+)_"
            r"(?P<obs_type>[A-Z]+)_"
            r"(?P<exp_start>[0-9]{14})_"
            r"(?P<exp_stop>[0-9]{14})_"
            r"(?P<obs_id>[0-9]{11,14})_"
            r"(?P<detector>[0-9A-Z]+)_"
            r"L(?P<level>[0-9]{1})_"
            r"V(?P<version>[0-9]{2})"
            r"(?P<ext>[a-z.]+)"
        self.mo = re.fullmatch(pattern, self.file_name)
        assert self.mo is not None, f"Pattern does not match: {self.file_name}"
        # set attributes
        # for k, v in self.mo.groupdict().items():
        #     self.__setattr__(k, v)
        self.telescope = self.mo.groupdict()["telescope"]
        self.instrument = self.mo.groupdict()["instrument"]
        self.project = self.mo.groupdict()["project"]
        self.obs_type = self.mo.groupdict()["obs_type"]
        self.exp_start = self.mo.groupdict()["exp_start"]
        self.exp_stop = self.mo.groupdict()["exp_stop"]
        self.obs_id = self.mo.groupdict()["obs_id"]
        self.detector = self.mo.groupdict()["detector"]
        self.level = self.mo.groupdict()["level"]
self.version = self.mo.groupdict()["version"] self.ext = self.mo.groupdict()["ext"] def derive( self, ext: Optional[str] = None, new_dir: Optional[str] = None, telescope: Optional[str] = None, instrument: Optional[str] = None, project: Optional[str] = None, obs_type: Optional[str] = None, exp_start: Optional[str | int] = None, exp_stop: Optional[str | int] = None, obs_id: Optional[str | int] = None, detector: Optional[str | int] = None, level: Optional[str | int] = None, version: Optional[str | int] = None, ): if ext is None: ext = self.ext else: assert isinstance(ext, str) if not ext.startswith("."): ext = "_" + str(ext) return os.path.join( new_dir if new_dir is not None else self.new_dir, f"{self.telescope if telescope is None else str(telescope)}_" f"{self.instrument if instrument is None else str(instrument)}_" f"{self.project if project is None else str(project)}_" f"{self.obs_type if obs_type is None else str(obs_type)}_" f"{self.exp_start if exp_start is None else str(exp_start)}_" f"{self.exp_stop if exp_stop is None else str(exp_stop)}_" f"{self.obs_id if obs_id is None else str(obs_id)}_" f"{self.detector if detector is None else str(detector)}_" f"L{self.level if level is None else str(level)}_" f"V{self.version if version is None else str(version).zfill(2)}" f"{ext}", ) def derive0(self, *args, **kwargs): return self.derive(*args, **kwargs, level=0) def derive1(self, *args, **kwargs): return self.derive(*args, **kwargs, level=1) def derive2(self, *args, **kwargs): return self.derive(*args, **kwargs, level=2) def __repr__(self): return f"<File {self.file_path}>" @property def level0_id(self): if self.instrument == "MSC": return self.obs_id + self.detector if self.instrument == "CPIC": return self.obs_id + self.project raise ValueError(f"Unmatched instrument: {self.instrument}")