Coverage for src / lstautorta / paths.py: 98%

1import datetime 

2from pathlib import Path 

3 

4 

5class RecoPathStructure: 

6 """Implements the path structures to store RTA data for LST AUTO RTA 

7 

8 Attributes 

9 ---------- 

10 base_data_dir: Path 

11 Directory into which all RTA data are written. A Night's data will be added in sub-directories. 

12 night_date: datetime.datetime 

13 Date of the observation night. If in the morning, the date modified to correspond to the 

14 previous day (day of the start of the night). 

15 """ 

16 

17 def __init__(self, base_data_dir: str, night_date: datetime.datetime): 

18 self.base_data_dir = Path(base_data_dir) 

19 self.night_date = night_date 

20 if self.night_date.hour < 12: 

21 self.night_date = self.night_date - datetime.timedelta(days=1) 

22 

23 @property 

24 def night_data_dir( 

25 self, 

26 ) -> Path: 

27 """Path to the night's data directory 

28 

29 Returns 

30 ------- 

31 Path 

32 Path of the night's data directory: base_data_dir/YYYY/MM/DD 

33 """ 

34 return self.base_data_dir.joinpath( 

35 f"{self.night_date.year:04d}", 

36 f"{self.night_date.month:02d}", 

37 f"{self.night_date.day:02d}", 

38 ) 

39 

40 def create_night_data_dir(self): 

41 """Create the night's data directory""" 

42 self.night_data_dir.mkdir(parents=True, exist_ok=True) 

43 

44 def _new_obs_dir(self, obs_id: str, ensure_unique: bool = False) -> Path: 

45 """Get the Path to a new observation directory 

46 

47 Parameters 

48 ---------- 

49 obs_id : str 

50 Observation ID 

51 ensure_unique : bool, optional 

52 If true, a suffix is appended to the default name to make sure that the new directory doesn't yet exists, by default False 

53 

54 Returns 

55 ------- 

56 Path 

57 Path to a new observation directory 

58 """ 

59 obs_data_dir = self.night_data_dir / obs_id 

60 if ensure_unique: 

61 suffix_number = 0 

62 while obs_data_dir.exists(): 

63 obs_data_dir = self.night_data_dir / (obs_id + f"_{suffix_number}") 

64 suffix_number += 1 

65 return obs_data_dir 

66 

67 def dl1_dir(self, obs_dir: Path | str) -> Path: 

68 """Path to the DL1 subdirectory in `obs_dir` 

69 

70 Parameters 

71 ---------- 

72 obs_dir : Path or str 

73 Path to the observation directory 

74 

75 Returns 

76 ------- 

77 Path 

78 Path to the DL1 sub-directory of `obs_dir`. 

79 """ 

80 return Path(obs_dir) / "DL1" 

81 

82 def dl2_dir(self, obs_dir: Path | str) -> Path: 

83 """Path to the DL2 subdirectory in `obs_dir` 

84 

85 Parameters 

86 ---------- 

87 obs_dir : Path or str 

88 Path to the observation directory 

89 

90 Returns 

91 ------- 

92 Path 

93 Path to the DL2 sub-directory of `obs_dir` 

94 """ 

95 return Path(obs_dir) / "DL2" 

96 

97 def dl3_dir(self, obs_dir: Path | str) -> Path: 

98 """Path to the DL3 subdirectory in `obs_dir` 

99 

100 Parameters 

101 ---------- 

102 obs_dir : Path or str 

103 Path to the observation directory 

104 

105 Returns 

106 ------- 

107 Path 

108 Path to the DL3 sub-directory of `obs_dir` 

109 """ 

110 return Path(obs_dir) / "DL3" 

111 

112 def log_dir(self, obs_dir: Path | str) -> Path: 

113 """Path to the logs subdirectory in `obs_dir` 

114 

115 Parameters 

116 ---------- 

117 obs_dir : Path or str 

118 Path to the observation directory 

119 

120 Returns 

121 ------- 

122 Path 

123 Path to the log sub-directory of `obs_dir` 

124 """ 

125 return Path(obs_dir) / "logs" 

126 

127 def plt_dir(self, obs_dir: Path | str) -> Path: 

128 """Path to the plot subdirectory in `obs_dir` 

129 

130 Parameters 

131 ---------- 

132 obs_dir : Path or str 

133 Path to the observation directory 

134 

135 Returns 

136 ------- 

137 Path 

138 Path to the plot sub-directory of `obs_dir` 

139 """ 

140 return Path(obs_dir) / "plots" 

141 

142 def create_observation_data_dirs(self, obs_id: str, ensure_unique: bool = False) -> Path: 

143 """Create the observation directory and sub-directories 

144 

145 Parameters 

146 ---------- 

147 obs_id : str 

148 Observation ID 

149 ensure_unique : bool, optional 

150 If true, makes sure that the created observation directory is unique, otherwise 

151 an existing repository can be re-used, by default False 

152 

153 Returns 

154 ------- 

155 obs_data_dir : Path 

156 Path of the observation directory. 

157 """ 

158 obs_data_dir = self._new_obs_dir(obs_id, ensure_unique) 

159 obs_data_dir.mkdir(parents=True, exist_ok=True) 

160 self.dl1_dir(obs_data_dir).mkdir(exist_ok=True) 

161 self.dl2_dir(obs_data_dir).mkdir(exist_ok=True) 

162 self.dl3_dir(obs_data_dir).mkdir(exist_ok=True) 

163 self.log_dir(obs_data_dir).mkdir(exist_ok=True) 

164 self.plt_dir(obs_data_dir).mkdir(exist_ok=True) 

165 return obs_data_dir 

166 

167 def obs_dirs(self, obs_id: str) -> list[Path]: 

168 """Get the observation directories with data from observation `obs_id` 

169 

170 Since several directories can exists when `ensure_unique = True`, this function 

171 returns a list of all possible directories for an observation. 

172 

173 Parameters 

174 ---------- 

175 obs_id : str 

176 Observation ID. 

177 

178 Returns 

179 ------- 

180 List[Path] 

181 List of Paths to directories containing data from observation `obs_id`. 

182 """ 

183 return [d for d in self.night_data_dir.iterdir() if d.is_dir() and d.startswith(obs_id)]