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 = self.night_date - datetime.timedelta(days=1) 

23 

24 

25 @property 

26 def night_data_dir( 

27 self, 

28 ) -> Path: 

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

30 

31 Returns 

32 ------- 

33 Path 

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

35 """ 

36 return self.base_data_dir.joinpath( 

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

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

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

40 ) 

41 

42 def create_night_data_dir(self): 

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

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

45 

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

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

48 

49 Parameters 

50 ---------- 

51 obs_id : str 

52 Observation ID 

53 ensure_unique : bool, optional 

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

55 

56 Returns 

57 ------- 

58 Path 

59 Path to a new observation directory 

60 """ 

61 obs_data_dir = self.night_data_dir / obs_id 

62 if ensure_unique: 

63 suffix_number = 0 

64 while obs_data_dir.exists(): 

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

66 suffix_number += 1 

67 return obs_data_dir 

68 

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

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

71 

72 Parameters 

73 ---------- 

74 obs_dir : Path or str 

75 Path to the observation directory 

76 

77 Returns 

78 ------- 

79 Path 

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

81 """ 

82 return Path(obs_dir) / "DL1" 

83 

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

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

86 

87 Parameters 

88 ---------- 

89 obs_dir : Path or str 

90 Path to the observation directory 

91 

92 Returns 

93 ------- 

94 Path 

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

96 """ 

97 return Path(obs_dir) / "DL2" 

98 

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

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

101 

102 Parameters 

103 ---------- 

104 obs_dir : Path or str 

105 Path to the observation directory 

106 

107 Returns 

108 ------- 

109 Path 

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

111 """ 

112 return Path(obs_dir) / "DL3" 

113 

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

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

116 

117 Parameters 

118 ---------- 

119 obs_dir : Path or str 

120 Path to the observation directory 

121 

122 Returns 

123 ------- 

124 Path 

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

126 """ 

127 return Path(obs_dir) / "logs" 

128 

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

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

131 

132 Parameters 

133 ---------- 

134 obs_dir : Path or str 

135 Path to the observation directory 

136 

137 Returns 

138 ------- 

139 Path 

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

141 """ 

142 return Path(obs_dir) / "plots" 

143 

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

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

146 

147 Parameters 

148 ---------- 

149 obs_id : str 

150 Observation ID 

151 ensure_unique : bool, optional 

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

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

154 

155 Returns 

156 ------- 

157 obs_data_dir : Path 

158 Path of the observation directory. 

159 """ 

160 obs_data_dir = self._new_obs_dir(obs_id, ensure_unique) 

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

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

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

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

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

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

167 return obs_data_dir 

168 

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

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

171 

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

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

174 

175 Parameters 

176 ---------- 

177 obs_id : str 

178 Observation ID. 

179 

180 Returns 

181 ------- 

182 List[Path] 

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

184 """ 

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