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

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 return modified 

194 

195 @classmethod 

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

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

198 supplied header. 

199 

200 Parameters 

201 ---------- 

202 header : `dict`-like 

203 Header to convert to standardized form. 

204 filename : `str`, optional 

205 Name of file being translated. 

206 

207 Returns 

208 ------- 

209 can : `bool` 

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

211 otherwise. 

212 """ 

213 # INSTRUME keyword might be of two types 

214 if "INSTRUME" in header: 

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

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

217 return True 

218 return False 

219 

220 @cache_translation 

221 def to_physical_filter(self): 

222 """Calculate the physical filter name. 

223 

224 Returns 

225 ------- 

226 filter : `str` 

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

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

229 are stripped. 

230 Returns "unknown" if no filter is declared. 

231 """ 

232 joined = super().to_physical_filter() 

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

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

235 

236 return joined 

237 

238 def _is_on_mountain(self): 

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

240 installed on the mountain. 

241 

242 Returns 

243 ------- 

244 is : `bool` 

245 `True` if instrument is on the mountain. 

246 

247 Notes 

248 ----- 

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

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

251 simulated file has come from the BTS. 

252 """ 

253 # phosim is always on mountain. 

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

255 return True 

256 

257 date = self.to_datetime_begin() 

258 if date > self._CAMERA_ON_TELESCOPE_DATE: 

259 return True 

260 

261 return False 

262 

263 @classmethod 

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

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

266 

267 Parameters 

268 ---------- 

269 observing_date : `astropy.time.Time` 

270 The date of the observation. Unused. 

271 

272 Returns 

273 ------- 

274 offset : `astropy.time.TimeDelta` 

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

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

277 In Chile the offset is always UTC-12. 

278 """ 

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

280 # is in the lab. 

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

282 return cls._ROLLOVER_TIME # 12 hours in base class 

283 

284 # Convert the date to a datetime UTC. 

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

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

287 

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

289 offset = pacific_time.utcoffset() * -1 

290 return astropy.time.TimeDelta(offset) 

291 

292 @cache_translation 

293 def to_exposure_time(self): 

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

295 # opens). 

296 if self.is_key_ok("SHUTTIME"): 

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

298 return shuttime * u.s 

299 return self.to_exposure_time_requested() 

300 

301 @cache_translation 

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

303 if not self._is_on_mountain(): 

304 # Lab data cannot see sky. 

305 return False 

306 return super().to_can_see_sky()