Coverage for python / lsst / obs / lsst / translators / ts8.py: 28%

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 LSST TestStand 8 headers""" 

12 

13__all__ = ("LsstTS8Translator", ) 

14 

15import logging 

16import re 

17 

18import astropy.units as u 

19from astropy.time import Time, TimeDelta 

20 

21from astro_metadata_translator import cache_translation 

22 

23from .lsst import LsstBaseTranslator, compute_detector_exposure_id_generic 

24 

25log = logging.getLogger(__name__) 

26 

27# First observation with new exposure ID is TS_C_20230524_000906. 

28_EXPOSURE_ID_DATE_CHANGE = Time("2023-05-24T23:00:00.0", format="isot", scale="tai") 

29_UNMODIFIED_DATE_OBS_HEADER = "HIERARCH LSST-TS8 DATE-OBS" 

30 

31 

32class LsstTS8Translator(LsstBaseTranslator): 

33 """Metadata translator for LSST Test Stand 8 data. 

34 """ 

35 

36 name = "LSST-TS8" 

37 """Name of this translation class""" 

38 

39 _const_map = { 

40 # TS8 is not attached to a telescope so many translations are null. 

41 "instrument": "LSST-TS8", 

42 "telescope": None, 

43 "location": None, 

44 "boresight_rotation_coord": None, 

45 "boresight_rotation_angle": None, 

46 "boresight_airmass": None, 

47 "tracking_radec": None, 

48 "altaz_begin": None, 

49 "object": "UNKNOWN", 

50 "relative_humidity": None, 

51 "temperature": None, 

52 "pressure": None, 

53 "can_see_sky": False, 

54 } 

55 

56 _trivial_map = { 

57 "science_program": "RUNNUM", 

58 "exposure_time": ("EXPTIME", dict(unit=u.s)), 

59 } 

60 

61 cameraPolicyFile = "policy/ts8.yaml" 

62 

63 _ROLLOVER_TIME = TimeDelta(8*60*60, scale="tai", format="sec") 

64 """Time delta for the definition of a Rubin Test Stand start of day.""" 

65 

66 @classmethod 

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

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

69 supplied header. 

70 

71 There is no ``INSTRUME`` header in TS8 data. Instead we use 

72 the ``TSTAND`` header. We also look at the file name to see if 

73 it starts with "ts8-". 

74 

75 Older data has no ``TSTAND`` header so we must use a combination 

76 of headers. 

77 

78 Parameters 

79 ---------- 

80 header : `dict`-like 

81 Header to convert to standardized form. 

82 filename : `str`, optional 

83 Name of file being translated. 

84 

85 Returns 

86 ------- 

87 can : `bool` 

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

89 otherwise. 

90 """ 

91 can = cls.can_translate_with_options(header, {"TSTAND": "TS8"}, filename=filename) 

92 if can: 

93 return True 

94 

95 if "LSST_NUM" in header and "REBNAME" in header and \ 

96 "CONTNUM" in header and \ 

97 header["CONTNUM"] in ("000018910e0c", "000018ee33b7", "000018ee0f35", "000018ee3b40", 

98 "00001891fcc7", "000018edfd65", "0000123b5ba8", "000018911b05", 

99 "00001891fa3e", "000018910d7f", "000018ed9f12", "000018edf4a7", 

100 "000018ee34e6", "000018ef1464", "000018eda120", "000018edf8a2", 

101 "000018ef3819", "000018ed9486", "000018ee02c8", "000018edfb24", 

102 "000018ee34c0", "000018edfb51", "0000123b51d1", "0000123b5862", 

103 "0000123b8ca9", "0000189208fa", "0000189111af", "0000189126e1", 

104 "000018ee0618", "000018ee3b78", "000018ef1534"): 

105 return True 

106 

107 return False 

108 

109 @classmethod 

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

111 """Fix TS8 headers. 

112 

113 Notes 

114 ----- 

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

116 process. 

117 """ 

118 modified = False 

119 

120 # Calculate the standard label to use for log messages 

121 log_label = cls._construct_log_prefix(obsid, filename) 

122 

123 if header.get("DATE-OBS", "OBS") == header.get("DATE-TRG", "TRG"): 

124 log.warning("%s: DATE-OBS detected referring to end of observation.", log_label) 

125 if "DATE-END" not in header: 

126 header["DATE-END"] = header["DATE-OBS"] 

127 header["MJD-END"] = header["MJD-OBS"] 

128 

129 # Time system used to be UTC and at some point became TAI. 

130 # Need to include the transition date and update the TIMESYS 

131 # header. 

132 timesys = header.get("TIMESYS", "utc").lower() 

133 

134 # Need to subtract exposure time from DATE-OBS. 

135 date_obs = None 

136 for (key, format) in (("MJD-OBS", "mjd"), ("DATE-OBS", "isot")): 

137 if date_val := header.get(key): 

138 date_obs = Time(date_val, format=format, scale=timesys) 

139 break 

140 

141 if date_obs: 

142 # The historical exposure ID calculation requires that we 

143 # have access to the unmodified DATE-OBS value. 

144 header[_UNMODIFIED_DATE_OBS_HEADER] = header["DATE-OBS"] 

145 

146 exptime = TimeDelta(header["EXPTIME"]*u.s, scale="tai") 

147 date_obs = date_obs - exptime 

148 header["MJD-OBS"] = float(date_obs.mjd) 

149 header["DATE-OBS"] = date_obs.isot 

150 header["DATE-BEG"] = header["DATE-OBS"] 

151 header["MJD-BEG"] = header["MJD-OBS"] 

152 

153 modified = True 

154 else: 

155 # This should never happen because DATE-OBS is already present. 

156 log.warning("%s: Unexpectedly failed to extract date from DATE-OBS/MJD-OBS", log_label) 

157 

158 return modified 

159 

160 @classmethod 

161 def compute_detector_exposure_id(cls, exposure_id, detector_num): 

162 # Docstring inherited from LsstBaseTranslator. 

163 return compute_detector_exposure_id_generic(exposure_id, detector_num, max_num=cls.DETECTOR_MAX) 

164 

165 @classmethod 

166 def max_exposure_id(cls): 

167 """The maximum exposure ID expected from this instrument. 

168 

169 The TS8 implementation is non-standard because TS8 data can create 

170 two different forms of exposure_id based on the date but we need 

171 the largest form to be the one returned. 

172 

173 Returns 

174 ------- 

175 max_exposure_id : `int` 

176 The maximum value. 

177 """ 

178 max_date = "2050-12-31T23:59.999" 

179 return int(re.sub(r"\D", "", max_date[:21])) 

180 

181 @cache_translation 

182 def to_detector_name(self): 

183 # Docstring will be inherited. Property defined in properties.py 

184 serial = self.to_detector_serial() 

185 detector_info = self.compute_detector_info_from_serial(serial) 

186 return detector_info[1] 

187 

188 def to_detector_group(self): 

189 """Returns the name of the raft. 

190 

191 Extracted from RAFTNAME header. 

192 

193 Raftname should be of the form: 'LCA-11021_RTM-011-Dev' and 

194 the resulting name will have the form "RTM-NNN". 

195 

196 Returns 

197 ------- 

198 name : `str` 

199 Name of raft. 

200 """ 

201 raft_name = self._header["RAFTNAME"] 

202 self._used_these_cards("RAFTNAME") 

203 match = re.search(r"(RTM-\d\d\d)", raft_name) 

204 if match: 

205 return match.group(0) 

206 raise ValueError(f"{self._log_prefix}: RAFTNAME has unexpected form of '{raft_name}'") 

207 

208 @cache_translation 

209 def to_detector_serial(self): 

210 """Returns the serial number of the detector. 

211 

212 Returns 

213 ------- 

214 serial : `str` 

215 LSST assigned serial number. 

216 

217 Notes 

218 ----- 

219 This is the LSST assigned serial number (``LSST_NUM``), and not 

220 the manufacturer's serial number (``CCD_SERN``). 

221 """ 

222 serial = self._header["LSST_NUM"] 

223 self._used_these_cards("LSST_NUM") 

224 

225 # this seems to be appended more or less at random and should be 

226 # removed. 

227 serial = re.sub("-Dev$", "", serial) 

228 return serial 

229 

230 @cache_translation 

231 def to_physical_filter(self): 

232 """Return the filter name. 

233 

234 Uses the FILTPOS header for older TS8 data. Newer data can use 

235 the base class implementation. 

236 

237 Returns 

238 ------- 

239 filter : `str` 

240 The filter name. Returns "NONE" if no filter can be determined. 

241 

242 Notes 

243 ----- 

244 The FILTPOS handling is retained for backwards compatibility. 

245 """ 

246 

247 default = "unknown" 

248 try: 

249 filter_pos = self._header["FILTPOS"] 

250 self._used_these_cards("FILTPOS") 

251 except KeyError: 

252 # TS8 data from 2023-05-09 and later should be following 

253 # DM-38882 conventions. 

254 physical_filter = super().to_physical_filter() 

255 # Some TS8 taken prior to 2023-05-09 have the string 

256 # 'unspecified' as the FILTER keyword and don't really 

257 # follow any established convention. 

258 if 'unspecified' in physical_filter: 

259 return default 

260 return physical_filter 

261 

262 try: 

263 return { 

264 2: 'g', 

265 3: 'r', 

266 4: 'i', 

267 5: 'z', 

268 6: 'y', 

269 }[filter_pos] 

270 except KeyError: 

271 log.warning("%s: Unknown filter position (assuming %s): %d", 

272 self._log_prefix, default, filter_pos) 

273 return default 

274 

275 @cache_translation 

276 def to_exposure_id(self): 

277 """Generate a unique exposure ID number 

278 

279 Modern TS8 data conforms to standard LSSTCam OBSID, using the "C" 

280 controller variant (all TS8 uses "C" controller). Due to existing 

281 ingests, data taken before 2023-05-25 must use the old style 

282 timestamp ID. 

283 

284 For older data SEQNUM is not unique for a given day in TS8 data 

285 so instead we convert the ISO date of observation directly to an 

286 integer. 

287 

288 Returns 

289 ------- 

290 exposure_id : `int` 

291 Unique exposure number. 

292 """ 

293 begin = self.to_datetime_begin() 

294 

295 if begin > _EXPOSURE_ID_DATE_CHANGE: 

296 obsid = self.to_observation_id() 

297 if obsid.startswith("TS_C_"): 

298 return super().to_exposure_id() 

299 

300 iso = self._header.get(_UNMODIFIED_DATE_OBS_HEADER, self._header["DATE-OBS"]) 

301 self._used_these_cards("DATE-OBS") 

302 

303 # There is worry that seconds are too coarse so use 10th of second 

304 # and read the first 21 characters. 

305 exposure_id = re.sub(r"\D", "", iso[:21]) 

306 return int(exposure_id) 

307 

308 # For now assume that visit IDs and exposure IDs are identical 

309 to_visit_id = to_exposure_id 

310 

311 @cache_translation 

312 def to_observation_id(self): 

313 # Docstring will be inherited. Property defined in properties.py 

314 if self.is_key_ok("OBSID"): 

315 observation_id = self._header["OBSID"] 

316 self._used_these_cards("OBSID") 

317 return observation_id 

318 filename = self._header["FILENAME"] 

319 self._used_these_cards("FILENAME") 

320 return filename[:filename.rfind(".")]