import datetime
from pathlib import Path
from typing import List
[docs]
class RecoPathStructure:
"""Implements the path structures to store RTA data for LST AUTO RTA
Attributes
----------
base_data_dir: Path
Directory into which all RTA data are written. A Night's data will be added in sub-directories.
night_date: datetime.datetime
Date of the observation night. If in the morning, the date modified to correspond to the
previous day (day of the start of the night).
"""
[docs]
def __init__(self, base_data_dir: str, night_date: datetime.datetime):
self.base_data_dir = Path(base_data_dir)
self.night_date = night_date
if self.night_date.hour < 12:
self.night_date = night_date.replace(day=self.night_date.day - 1)
@property
def night_data_dir(
self,
) -> Path:
"""Path to the night's data directory
Returns
-------
Path
Path of the night's data directory: base_data_dir/YYYY/MM/DD
"""
return self.base_data_dir.joinpath(
"{:04d}".format(self.night_date.year),
"{:02d}".format(self.night_date.month),
"{:02d}".format(self.night_date.day),
)
[docs]
def create_night_data_dir(self):
"""Create the night's data directory"""
self.night_data_dir.mkdir(parents=True, exist_ok=True)
def _new_obs_dir(self, obs_id: str, ensure_unique: bool = False) -> Path:
"""Get the Path to a new observation directory
Parameters
----------
obs_id : str
Observation ID
ensure_unique : bool, optional
If true, a suffix is appended to the default name to make sure that the new directory doesn't yet exists, by default False
Returns
-------
Path
Path to a new observation directory
"""
obs_data_dir = self.night_data_dir / obs_id
if ensure_unique:
suffix_number = 0
while obs_data_dir.exists():
obs_data_dir = self.night_data_dir / (obs_id + f"_{suffix_number}")
suffix_number += 1
return obs_data_dir
[docs]
def dl1_dir(self, obs_dir: Path | str) -> Path:
"""Path to the DL1 subdirectory in `obs_dir`
Parameters
----------
obs_dir : Path or str
Path to the observation directory
Returns
-------
Path
Path to the DL1 sub-directory of `obs_dir`.
"""
return Path(obs_dir) / "DL1"
[docs]
def dl2_dir(self, obs_dir: Path | str) -> Path:
"""Path to the DL2 subdirectory in `obs_dir`
Parameters
----------
obs_dir : Path or str
Path to the observation directory
Returns
-------
Path
Path to the DL2 sub-directory of `obs_dir`
"""
return Path(obs_dir) / "DL2"
[docs]
def dl3_dir(self, obs_dir: Path | str) -> Path:
"""Path to the DL3 subdirectory in `obs_dir`
Parameters
----------
obs_dir : Path or str
Path to the observation directory
Returns
-------
Path
Path to the DL3 sub-directory of `obs_dir`
"""
return Path(obs_dir) / "DL3"
[docs]
def log_dir(self, obs_dir: Path | str) -> Path:
"""Path to the logs subdirectory in `obs_dir`
Parameters
----------
obs_dir : Path or str
Path to the observation directory
Returns
-------
Path
Path to the log sub-directory of `obs_dir`
"""
return Path(obs_dir) / "logs"
[docs]
def plt_dir(self, obs_dir: Path | str) -> Path:
"""Path to the plot subdirectory in `obs_dir`
Parameters
----------
obs_dir : Path or str
Path to the observation directory
Returns
-------
Path
Path to the plot sub-directory of `obs_dir`
"""
return Path(obs_dir) / "plots"
[docs]
def create_observation_data_dirs(self, obs_id: str, ensure_unique: bool = False) -> Path:
"""Create the observation directory and sub-directories
Parameters
----------
obs_id : str
Observation ID
ensure_unique : bool, optional
If true, makes sure that the created observation directory is unique, otherwise
an existing repository can be re-used, by default False
Returns
-------
obs_data_dir : Path
Path of the observation directory.
"""
obs_data_dir = self._new_obs_dir(obs_id, ensure_unique)
obs_data_dir.mkdir(parents=True, exist_ok=True)
self.dl1_dir(obs_data_dir).mkdir(exist_ok=True)
self.dl2_dir(obs_data_dir).mkdir(exist_ok=True)
self.dl3_dir(obs_data_dir).mkdir(exist_ok=True)
self.log_dir(obs_data_dir).mkdir(exist_ok=True)
self.plt_dir(obs_data_dir).mkdir(exist_ok=True)
return obs_data_dir
[docs]
def obs_dirs(self, obs_id: str) -> List[Path]:
"""Get the observation directories with data from observation `obs_id`
Since several directories can exists when `ensure_unique = True`, this function
returns a list of all possible directories for an observation.
Parameters
----------
obs_id : str
Observation ID.
Returns
-------
List[Path]
List of Paths to directories containing data from observation `obs_id`.
"""
return [d for d in self.night_data_dir.iterdir() if d.is_dir() and d.startswith(obs_id)]