Coverage for lst_auto_rta/paths.py: 98%

1import datetime 

2from pathlib import Path 

3from typing import List 

4 

5 

6class RecoPathStructure: 

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

8 

9 Attributes 

10 ---------- 

11 base_data_dir: Path 

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

13 night_date: datetime.datetime 

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

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

16 """ 

17 

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

19 self.base_data_dir = Path(base_data_dir) 

20 self.night_date = night_date 

21 if self.night_date.hour < 12: 

22 self.night_date = night_date.replace(day=self.night_date.day - 1) 

23 

24 @property 

25 def night_data_dir( 

26 self, 

27 ) -> Path: 

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

29 

30 Returns 

31 ------- 

32 Path 

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

34 """ 

35 return self.base_data_dir.joinpath( 

36 "{:04d}".format(self.night_date.year), 

37 "{:02d}".format(self.night_date.month), 

38 "{:02d}".format(self.night_date.day), 

39 ) 

40 

41 def create_night_data_dir(self): 

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

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

44 

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

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

47 

48 Parameters 

49 ---------- 

50 obs_id : str 

51 Observation ID 

52 ensure_unique : bool, optional 

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

54 

55 Returns 

56 ------- 

57 Path 

58 Path to a new observation directory 

59 """ 

60 obs_data_dir = self.night_data_dir / obs_id 

61 if ensure_unique: 

62 suffix_number = 0 

63 while obs_data_dir.exists(): 

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

65 suffix_number += 1 

66 return obs_data_dir 

67 

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

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

70 

71 Parameters 

72 ---------- 

73 obs_dir : Path or str 

74 Path to the observation directory 

75 

76 Returns 

77 ------- 

78 Path 

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

80 """ 

81 return Path(obs_dir) / "DL1" 

82 

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

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

85 

86 Parameters 

87 ---------- 

88 obs_dir : Path or str 

89 Path to the observation directory 

90 

91 Returns 

92 ------- 

93 Path 

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

95 """ 

96 return Path(obs_dir) / "DL2" 

97 

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

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

100 

101 Parameters 

102 ---------- 

103 obs_dir : Path or str 

104 Path to the observation directory 

105 

106 Returns 

107 ------- 

108 Path 

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

110 """ 

111 return Path(obs_dir) / "DL3" 

112 

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

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

115 

116 Parameters 

117 ---------- 

118 obs_dir : Path or str 

119 Path to the observation directory 

120 

121 Returns 

122 ------- 

123 Path 

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

125 """ 

126 return Path(obs_dir) / "logs" 

127 

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

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

130 

131 Parameters 

132 ---------- 

133 obs_dir : Path or str 

134 Path to the observation directory 

135 

136 Returns 

137 ------- 

138 Path 

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

140 """ 

141 return Path(obs_dir) / "plots" 

142 

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

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

145 

146 Parameters 

147 ---------- 

148 obs_id : str 

149 Observation ID 

150 ensure_unique : bool, optional 

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

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

153 

154 Returns 

155 ------- 

156 obs_data_dir : Path 

157 Path of the observation directory. 

158 """ 

159 obs_data_dir = self._new_obs_dir(obs_id, ensure_unique) 

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

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

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

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

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

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

166 return obs_data_dir 

167 

168 def obs_dirs(self, obs_id: str) -> List[Path]: 

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

170 

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

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

173 

174 Parameters 

175 ---------- 

176 obs_id : str 

177 Observation ID. 

178 

179 Returns 

180 ------- 

181 List[Path] 

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

183 """ 

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