Coverage for python / lsst / obs / lsst / translators / lsstCam.py: 19%

1# This file is currently part of obs_lsst but is written to allow it 

2# to be migrated to the astro_metadata_translator package at a later date. 

3# 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the LICENSE file in this directory for details of code ownership. 

7# 

8# Use of this source code is governed by a 3-clause BSD-style 

9# license that can be found in the LICENSE file. 

10 

11"""Metadata translation code for the main LSST Camera""" 

12 

13__all__ = ("LsstCamTranslator", ) 

14 

15import logging 

16 

17import pytz 

18import astropy.time 

19import astropy.units as u 

20from astropy.coordinates import Angle 

21 

22from astro_metadata_translator import cache_translation 

23from astro_metadata_translator.translators.helpers import is_non_science 

24 

25from .lsst import LsstBaseTranslator, SIMONYI_TELESCOPE 

26 

27log = logging.getLogger(__name__) 

28 

29# Normalized name of the LSST Camera 

30LSST_CAM = "LSSTCam" 

31 

32 

33def is_non_science_or_lab(self): 

34 """Pseudo method to determine whether this is a lab or non-science 

35 header. 

36 

37 Raises 

38 ------ 

39 KeyError 

40 If this is a science observation and on the mountain. 

41 """ 

42 # Return without raising if this is not a science observation 

43 # since the defaults are fine. 

44 try: 

45 # This will raise if it is a science observation. 

46 is_non_science(self) 

47 return 

48 except KeyError: 

49 pass 

50 

51 # We are still in the lab, return and use the default. 

52 if not self._is_on_mountain(): 

53 return 

54 

55 # This is a science observation on the mountain so we should not 

56 # use defaults. 

57 raise KeyError(f"{self._log_prefix}: Required key is missing and this is a mountain science observation") 

58 

59 

60class LsstCamTranslator(LsstBaseTranslator): 

61 """Metadata translation for the main LSST Camera.""" 

62 

63 name = LSST_CAM 

64 """Name of this translation class""" 

65 

66 supported_instrument = LSST_CAM 

67 """Supports the lsstCam instrument.""" 

68 

69 _const_map = { 

70 "instrument": LSST_CAM, 

71 "telescope": SIMONYI_TELESCOPE, 

72 } 

73 

74 _trivial_map = { 

75 "detector_group": "RAFTBAY", 

76 "detector_name": "CCDSLOT", 

77 "observation_id": "OBSID", 

78 "exposure_time_requested": ("EXPTIME", dict(unit=u.s)), 

79 "detector_serial": "LSST_NUM", 

80 "object": ("OBJECT", dict(default="UNKNOWN")), 

81 "science_program": (["PROGRAM", "RUNNUM"], dict(default="unknown")), 

82 "boresight_rotation_angle": (["ROTPA", "ROTANGLE"], dict(checker=is_non_science_or_lab, 

83 default=0.0, unit=u.deg)), 

84 } 

85 

86 # Use Imsim raft definitions until a true lsstCam definition exists 

87 cameraPolicyFile = "policy/lsstCam.yaml" 

88 

89 # Date (YYYYMM) the camera changes from using lab day_offset (Pacific time) 

90 # to summit day_offset (12 hours). 

91 _CAMERA_SHIP_DATE = 202405 

92 

93 # Date we know camera is in Chile and potentially taking on-sky data. 

94 _CAMERA_ON_TELESCOPE_DATE = astropy.time.Time("2025-03-01T00:00", format="isot", scale="utc") 

95 

96 # Not allowed to use obstype for can_see_sky fallback. 

97 _can_check_obstype_for_can_see_sky = False 

98 

99 @classmethod 

100 def fix_header(cls, header, instrument, obsid, filename=None): 

101 """Fix LSSTCam headers. 

102 

103 Notes 

104 ----- 

105 See `~astro_metadata_translator.fix_header` for details of the general 

106 process. 

107 """ 

108 

109 modified = False 

110 

111 # Calculate the standard label to use for log messages 

112 log_label = cls._construct_log_prefix(obsid, filename) 

113 

114 if "FILTER" not in header and header.get("FILTER2") is not None: 

115 ccdslot = header.get("CCDSLOT", "unknown") 

116 raftbay = header.get("RAFTBAY", "unknown") 

117 

118 log.warning("%s %s_%s: No FILTER key found but FILTER2=\"%s\" (removed)", 

119 log_label, raftbay, ccdslot, header["FILTER2"]) 

120 header["FILTER2"] = None 

121 modified = True 

122 

123 day_obs = header.get("DAYOBS") 

124 i_day_obs = int(day_obs) if day_obs else None 

125 if ( 

126 day_obs in ("20231107", "20231108", "20241015", "20241016") 

127 and header["FILTER"] == "ph_05" 

128 ): 

129 header["FILTER"] = "ph_5" 

130 modified = True 

131 

132 # For first ~ week of observing the ROTPA in the header was the ComCam 

133 # value and needed to be adjusted by 90 degrees to match LSSTCam. 

134 # Fixed for day_obs 20250422. 

135 rotpa_fixed_on_day = 20250422 

136 if i_day_obs and i_day_obs > 20250301 and i_day_obs < rotpa_fixed_on_day: 

137 if rotpa := header.get("ROTPA"): 

138 header["ROTPA"] = rotpa - 90.0 

139 modified = True 

140 log.debug("%s: Correcting ROTPA by -90.0", log_label) 

141 

142 # For half a night the ROTPA was out by 180 degrees. 

143 if i_day_obs == 20250422: 

144 seq_num = header["SEQNUM"] 

145 if seq_num < 251 and (rotpa := header.get("ROTPA")): 

146 rotpa_corrected = rotpa - 180.0 

147 angle = Angle(rotpa_corrected * u.deg) 

148 header["ROTPA"] = float(angle.wrap_at("180d").value) 

149 modified = True 

150 log.debug( 

151 "%s: Correcting ROTPA of %f by 180 degrees to %f", log_label, rotpa, header["ROTPA"] 

152 ) 

153 

154 # For the night of 20250518 the dome was closed but many 

155 # calibs had the wrong header because of dome/TMA faults. 

156 if i_day_obs == 20250518: 

157 if header["VIGN_MIN"] != "FULLY": 

158 header["VIGN_MIN"] = "FULLY" 

159 modified = True 

160 log.debug("%s: Correcting VIGN_MIN to FULLY", log_label) 

161 

162 # DM-51847: For several nights in July, the dome was closed but many 

163 # flats had the wrong header because of a dome CSC regression 

164 fix_ranges = { 

165 20250703: [(743, 745)], 

166 20250704: [(832, 834)], 

167 20250705: [(1, 736)], 

168 20250707: [(784, 823), (744, 783), (864, 903), (904, 943)], 

169 20250714: [(258, 781)], 

170 20250715: [(205, 1218)], 

171 20250819: [(4, 1052)], # DM-52249 

172 } 

173 if i_day_obs in fix_ranges: 

174 i_seq_num = header["SEQNUM"] 

175 for seq_range in fix_ranges[i_day_obs]: 

176 if (seq_range[0] <= i_seq_num <= seq_range[1] 

177 and header["VIGN_MIN"] != "FULLY"): 

178 header["VIGN_MIN"] = "FULLY" 

179 modified = True 

180 log.debug("%s: Correcting VIGN_MIN to FULLY", log_label) 

181 

182 # DM-52711: The filter was incorrect for the start of the night. 

183 if i_day_obs == 20250609: 

184 i_seq_num = header["SEQNUM"] 

185 if i_seq_num >= 76 and i_seq_num <= 578: 

186 header["FILTER"] = "z_20" 

187 header["FILTBAND"] = "z" 

188 header["FILTPOS"] = 201.0 

189 header["FILTSLOT"] = 4 

190 modified = True 

191 log.debug("%s: Correcting filter to z", log_label) 

192 

193 # DM-53949: Incorrect block number. 

194 if i_day_obs in [ 

195 20250415, 20250416, 20250423, 20250417, 20250418, 20250421, 

196 20250424, 20250425, 20250504, 20250503, 20250512, 20250522, 

197 20250524, 20250826, 20250827, 20250828 

198 ]: 

199 if header["PROGRAM"] == "BLOCK-417": 

200 header["PROGRAM"] = "BLOCK-T417" 

201 modified = True 

202 log.debug("%s: Correcting BLOCK-417 to BLOCK-T417", log_label) 

203 

204 if i_day_obs == 20260315: 

205 i_seq_num = header["SEQNUM"] 

206 if i_seq_num >=49 and i_seq_num <= 109: 

207 header["FILTER"] = "i_39" 

208 header["FILTBAND"] = "i" 

209 header["FILTPOS"] = 304.0 

210 header["FILTSLOT"] = 1 

211 modified = True 

212 

213 return modified 

214 

215 @classmethod 

216 def can_translate(cls, header, filename=None): 

217 """Indicate whether this translation class can translate the 

218 supplied header. 

219 

220 Parameters 

221 ---------- 

222 header : `dict`-like 

223 Header to convert to standardized form. 

224 filename : `str`, optional 

225 Name of file being translated. 

226 

227 Returns 

228 ------- 

229 can : `bool` 

230 `True` if the header is recognized by this class. `False` 

231 otherwise. 

232 """ 

233 # INSTRUME keyword might be of two types 

234 if "INSTRUME" in header: 

235 instrume = header["INSTRUME"].lower() 

236 if instrume == cls.supported_instrument.lower(): 

237 return True 

238 return False 

239 

240 @cache_translation 

241 def to_physical_filter(self): 

242 """Calculate the physical filter name. 

243 

244 Returns 

245 ------- 

246 filter : `str` 

247 Name of filter. Can be a combination of FILTER, FILTER1, and 

248 FILTER2 headers joined by a "~". Trailing "~empty" components 

249 are stripped. 

250 Returns "unknown" if no filter is declared. 

251 """ 

252 joined = super().to_physical_filter() 

253 while joined.endswith("~empty"): 

254 joined = joined.removesuffix("~empty") 

255 

256 return joined 

257 

258 def _is_on_mountain(self): 

259 """Indicate whether these data are coming from the instrument 

260 installed on the mountain. 

261 

262 Returns 

263 ------- 

264 is : `bool` 

265 `True` if instrument is on the mountain. 

266 

267 Notes 

268 ----- 

269 LSSTCam was installed on the Simonyi Telescope in March 2025. 

270 This flag is true even if the telescope is on the test stand or a 

271 simulated file has come from the BTS. 

272 """ 

273 # phosim is always on mountain. 

274 if self._get_controller_code() == "H": 

275 return True 

276 

277 date = self.to_datetime_begin() 

278 if date > self._CAMERA_ON_TELESCOPE_DATE: 

279 return True 

280 

281 return False 

282 

283 @classmethod 

284 def observing_date_to_offset(cls, observing_date: astropy.time.Time) -> astropy.time.TimeDelta | None: 

285 """Return the offset to use when calculating the observing day. 

286 

287 Parameters 

288 ---------- 

289 observing_date : `astropy.time.Time` 

290 The date of the observation. Unused. 

291 

292 Returns 

293 ------- 

294 offset : `astropy.time.TimeDelta` 

295 The offset to apply. During lab testing the offset is Pacific 

296 Time which can mean UTC-7 or UTC-8 depending on daylight savings. 

297 In Chile the offset is always UTC-12. 

298 """ 

299 # Timezone calculations are slow. Only do this if the instrument 

300 # is in the lab. 

301 if int(observing_date.strftime("%Y%m")) >= cls._CAMERA_SHIP_DATE: 

302 return cls._ROLLOVER_TIME # 12 hours in base class 

303 

304 # Convert the date to a datetime UTC. 

305 pacific_tz = pytz.timezone("US/Pacific") 

306 pacific_time = observing_date.utc.to_datetime(timezone=pacific_tz) 

307 

308 # We need the offset to go the other way. 

309 offset = pacific_time.utcoffset() * -1 

310 return astropy.time.TimeDelta(offset) 

311 

312 @cache_translation 

313 def to_exposure_time(self): 

314 # Use shutter time if greater than 0 (for a dark the shutter never 

315 # opens). 

316 if self.is_key_ok("SHUTTIME"): 

317 if (shuttime := self._header["SHUTTIME"]) > 0.0: 

318 return shuttime * u.s 

319 return self.to_exposure_time_requested() 

320 

321 @cache_translation 

322 def to_can_see_sky(self) -> bool | None: 

323 if not self._is_on_mountain(): 

324 # Lab data cannot see sky. 

325 return False 

326 return super().to_can_see_sky()