Coverage for python/lsst/obs/lsst/translators/latiss.py: 16%

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.

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.

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

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

11"""Metadata translation code for LSST LATISS headers"""

13__all__ = ("LatissTranslator", )

15import logging

16import math

18import astropy.units as u

19from astropy.time import Time

20from astropy.coordinates import EarthLocation

22from astro_metadata_translator import cache_translation

23from astro_metadata_translator.translators.helpers import is_non_science

24from .lsst import LsstBaseTranslator, FILTER_DELIMITER

25from .lsstCam import is_non_science_or_lab

27log = logging.getLogger(__name__)

30# AuxTel is not the same place as LSST

31# These coordinates read from Apple Maps

32AUXTEL_LOCATION = EarthLocation.from_geodetic(-70.747698, -30.244728, 2663.0)

34# Date instrument is taking data at telescope

35# Prior to this date many parameters are automatically nulled out

36# since the headers have not historically been reliable

37TSTART = Time("2020-01-01T00:00", format="isot", scale="utc")

39# Define the sensor and group name for AuxTel globally so that it can be used

40# in multiple places. There is no raft but for consistency with other LSST

41# cameras we define one.

42_DETECTOR_GROUP_NAME = "RXX"

43_DETECTOR_NAME = "S00"

45# Date 068 detector was put in LATISS

46DETECTOR_068_DATE = Time("2019-06-24T00:00", format="isot", scale="utc")

48# IMGTYPE header is filled in after this date

49IMGTYPE_OKAY_DATE = Time("2019-11-07T00:00", format="isot", scale="utc")

51# OBJECT IMGTYPE really means ENGTEST until this date

52OBJECT_IS_ENGTEST = Time("2020-01-27T20:00", format="isot", scale="utc")

54# RA and DEC headers are in radians until this date

55RADEC_IS_RADIANS = Time("2020-01-28T22:00", format="isot", scale="utc")

57# RASTART/DECSTART/RAEND/DECEND used wrong telescope location before this

58# 2020-02-01T00:00 we fixed the telescope location, but RASTART is still

59# in mount coordinates, so off by pointing model.

60RASTART_IS_BAD = Time("2020-05-01T00:00", format="isot", scale="utc")

62# Between RASTART_IS_BAD and this time the RASTART header uses hours

63# instead of degrees.

64RASTART_IS_HOURS = Time("2021-02-11T18:45", format="isot", scale="utc")

66# From this date RASTART is correct as-is.

67RASTART_IS_OKAY = Time("2021-02-12T00:00", format="isot", scale="utc")

69# DATE-END is not to be trusted before this date

70DATE_END_IS_BAD = Time("2020-02-01T00:00", format="isot", scale="utc")

72# The convention for the reporting of ROTPA changed by 180 here

73ROTPA_CONVENTION_180_SWITCH1 = Time("2020-11-19T00:00", format="isot", scale="utc")

74ROTPA_CONVENTION_180_SWITCH2 = Time("2021-10-29T00:00", format="isot", scale="utc")

76# TARGET is set to start with 'spec:' for dispsered images before this date

77TARGET_STARTS_SPECCOLON = Time("2022-07-10T00:00", format="isot", scale="utc")

79# Scaling factor radians to degrees. Keep it simple.

80RAD2DEG = 180.0 / math.pi

83class LatissTranslator(LsstBaseTranslator):

84 """Metadata translator for LSST LATISS data from AuxTel.

86 For lab measurements many values are masked out.

87 """

89 name = "LSST_LATISS"

90 """Name of this translation class"""

92 supported_instrument = "LATISS"

93 """Supports the LATISS instrument."""

95 _const_map = {

96 "instrument": "LATISS",

97 "telescope": "Rubin Auxiliary Telescope",

98 "detector_group": _DETECTOR_GROUP_NAME,

99 "detector_num": 0,

100 "detector_name": _DETECTOR_NAME, # Single sensor

101 }

102

103 _trivial_map = {

104 "observation_id": (["OBSID", "IMGNAME"], dict(default=None, checker=is_non_science)),

105 "detector_serial": ["LSST_NUM", "DETSER"],

106 "object": ("OBJECT", dict(checker=is_non_science_or_lab, default="UNKNOWN")),

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

108 default=float("nan"), unit=u.deg)),

109 "science_program": ("PROGRAM", dict(default="unknown")),

110 }

111

112 DETECTOR_GROUP_NAME = _DETECTOR_GROUP_NAME

113 """Fixed name of detector group."""

114

115 DETECTOR_NAME = _DETECTOR_NAME

116 """Fixed name of single sensor."""

117

118 DETECTOR_MAX = 0

119 """Maximum number of detectors to use when calculating the

120 detector_exposure_id."""

121

122 _DEFAULT_LOCATION = AUXTEL_LOCATION

123 """Default telescope location in absence of relevant FITS headers."""

124

125 @classmethod

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

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

128 supplied header.

129

130 Parameters

131 ----------

132 header : `dict`-like

133 Header to convert to standardized form.

134 filename : `str`, optional

135 Name of file being translated.

136

137 Returns

138 -------

139 can : `bool`

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

141 otherwise.

142 """

143 # INSTRUME keyword might be of two types

144 if "INSTRUME" in header:

145 instrume = header["INSTRUME"]

146 for v in ("LSST_ATISS", "LATISS"):

147 if instrume == v:

148 return True

149 # Calibration files strip important headers at the moment so guess

150 if "DETNAME" in header and header["DETNAME"] == "RXX_S00":

151 return True

152 return False

153

154 @classmethod

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

156 """Fix an incorrect LATISS header.

157

158 Parameters

159 ----------

160 header : `dict`

161 The header to update. Updates are in place.

162 instrument : `str`

163 The name of the instrument.

164 obsid : `str`

165 Unique observation identifier associated with this header.

166 Will always be provided.

167 filename : `str`, optional

168 Filename associated with this header. May not be set since headers

169 can be fixed independently of any filename being known.

170

171 Returns

172 -------

173 modified = `bool`

174 Returns `True` if the header was updated.

175

176 Notes

177 -----

178 This method does not apply per-obsid corrections. The following

179 corrections are applied:

180

181 * On June 24th 2019 the detector was changed from ITL-3800C-098

182 to ITL-3800C-068. The header is intended to be correct in the

183 future.

184 * In late 2019 the DATE-OBS and MJD-OBS headers were reporting

185 1970 dates. To correct, the DATE/MJD headers are copied in to

186 replace them and the -END headers are cleared.

187 * Until November 2019 the IMGTYPE was set in the GROUPID header.

188 The value is moved to IMGTYPE.

189 * SHUTTIME is always forced to be `None`.

190

191 Corrections are reported as debug level log messages.

192

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

194 process.

195 """

196 modified = False

197

198 # Calculate the standard label to use for log messages

199 log_label = cls._construct_log_prefix(obsid, filename)

200

201 if "OBSID" not in header:

202 # Very old data used IMGNAME

203 header["OBSID"] = obsid

204 modified = True

205 # We are reporting the OBSID so no need to repeat it at start

206 # of log message. Use filename if we have it.

207 log_prefix = f"{filename}: " if filename else ""

208 log.debug("%sAssigning OBSID to a value of '%s'", log_prefix, header["OBSID"])

209

210 if "DAYOBS" not in header:

211 # OBS-NITE could have the value for DAYOBS but it is safer

212 # for older data to set it from the OBSID. Fall back to OBS-NITE

213 # if we have no alternative

214 dayObs = None

215 try:

216 dayObs = obsid.split("_", 3)[2]

217 except (AttributeError, ValueError):

218 # did not split as expected

219 pass

220 if dayObs is None or len(dayObs) != 8:

221 if "OBS-NITE" in header:

222 dayObs = header["OBS-NITE"]

223 log.debug("%s: Setting DAYOBS to '%s' from OBS-NITE header", log_label, dayObs)

224 else:

225 log.debug("%s: Unable to determine DAYOBS from header", log_label)

226 else:

227 log.debug("%s: Setting DAYOBS to '%s' from OBSID", log_label, dayObs)

228 if dayObs:

229 header["DAYOBS"] = dayObs

230 modified = True

231

232 if "SEQNUM" not in header:

233 try:

234 seqnum = obsid.split("_", 3)[3]

235 except (AttributeError, ValueError):

236 # did not split as expected

237 pass

238 else:

239 header["SEQNUM"] = int(seqnum)

240 modified = True

241 log.debug("%s: Extracting SEQNUM of '%s' from OBSID", log_label, header["SEQNUM"])

242

243 # The DATE-OBS / MJD-OBS keys can be 1970

244 if "DATE-OBS" in header and header["DATE-OBS"].startswith("1970"):

245 # Copy the headers from the DATE and MJD since we have no other

246 # choice.

247 header["DATE-OBS"] = header["DATE"]

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

249 header["MJD-OBS"] = header["MJD"]

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

251

252 # And clear the DATE-END and MJD-END -- the translator will use

253 # EXPTIME instead.

254 header["DATE-END"] = None

255 header["MJD-END"] = None

256

257 log.debug("%s: Forcing 1970 dates to '%s'", log_label, header["DATE"])

258 modified = True

259

260 # Create a translator since we need the date

261 translator = cls(header)

262 date = translator.to_datetime_begin()

263 if date > DETECTOR_068_DATE:

264 header["LSST_NUM"] = "ITL-3800C-068"

265 log.debug("%s: Forcing detector serial to %s", log_label, header["LSST_NUM"])

266 modified = True

267

268 if date < DATE_END_IS_BAD:

269 # DATE-END may or may not be in TAI and may or may not be

270 # before DATE-BEG. Simpler to clear it

271 if header.get("DATE-END"):

272 header["DATE-END"] = None

273 header["MJD-END"] = None

274

275 log.debug("%s: Clearing DATE-END as being untrustworthy", log_label)

276 modified = True

277

278 # Up until a certain date GROUPID was the IMGTYPE

279 if date < IMGTYPE_OKAY_DATE:

280 groupId = header.get("GROUPID")

281 if groupId and not groupId.startswith("test"):

282 imgType = header.get("IMGTYPE")

283 if not imgType:

284 if "_" in groupId:

285 # Sometimes have the form dark_0001_0002

286 # in this case we pull the IMGTYPE off the front and

287 # do not clear groupId (although groupId may now

288 # repeat on different days).

289 groupId, _ = groupId.split("_", 1)

290 elif groupId.upper() != "FOCUS" and groupId.upper().startswith("FOCUS"):

291 # If it is exactly FOCUS we want groupId cleared

292 groupId = "FOCUS"

293 else:

294 header["GROUPID"] = None

295 header["IMGTYPE"] = groupId

296 log.debug("%s: Setting IMGTYPE to '%s' from GROUPID", log_label, header["IMGTYPE"])

297 modified = True

298 else:

299 # Someone could be fixing headers in old data

300 # and we do not want GROUPID == IMGTYPE

301 if imgType == groupId:

302 # Clear the group so we default to original

303 header["GROUPID"] = None

304

305 # We were using OBJECT for engineering observations early on

306 if date < OBJECT_IS_ENGTEST:

307 imgType = header.get("IMGTYPE")

308 if imgType == "OBJECT":

309 header["IMGTYPE"] = "ENGTEST"

310 log.debug("%s: Changing OBJECT observation type to %s",

311 log_label, header["IMGTYPE"])

312 modified = True

313

314 # Early on the RA/DEC headers were stored in radians

315 if date < RADEC_IS_RADIANS:

316 if header.get("RA") is not None:

317 header["RA"] *= RAD2DEG

318 log.debug("%s: Changing RA header to degrees", log_label)

319 modified = True

320 if header.get("DEC") is not None:

321 header["DEC"] *= RAD2DEG

322 log.debug("%s: Changing DEC header to degrees", log_label)

323 modified = True

324

325 if header.get("SHUTTIME"):

326 log.debug("%s: Forcing SHUTTIME header to be None", log_label)

327 header["SHUTTIME"] = None

328 modified = True

329

330 if "OBJECT" not in header:

331 # Only patch OBJECT IMGTYPE

332 if "IMGTYPE" in header and header["IMGTYPE"] == "OBJECT":

333 log.debug("%s: Forcing OBJECT header to exist", log_label)

334 header["OBJECT"] = "NOTSET"

335 modified = True

336

337 if date < TARGET_STARTS_SPECCOLON:

338 if (obj := header.get("OBJECT", None)) is not None:

339 header["OBJECT"] = obj.replace('spec:', '')

340 modified = True

341

342 if "RADESYS" in header:

343 if header["RADESYS"] == "":

344 # Default to ICRS

345 header["RADESYS"] = "ICRS"

346 log.debug("%s: Forcing blank RADESYS to '%s'", log_label, header["RADESYS"])

347 modified = True

348

349 if "TRACKSYS" in header:

350 if header["TRACKSYS"] == "":

351 # Force it to a consistent undefined value. Could try to

352 # see if RADESYS is defined.

353 header["TRACKSYS"] = None

354 log.debug("%s: Forcing blank TRACKSYS to undef", log_label)

355 modified = True

356 elif header["TRACKSYS"] == "SIDEREAL":

357 header["TRACKSYS"] = "RADEC"

358 log.debug("%s: Conforming SIDEREAL TRACKSYS to '%s'", log_label, header["TRACKSYS"])

359 modified = True

360

361 if date < RASTART_IS_HOURS:

362 # Avoid two checks for case where RASTART is fine

363 if date < RASTART_IS_BAD:

364 # The wrong telescope position was used. Unsetting these will

365 # force the RA/DEC demand headers to be used instead.

366 for h in ("RASTART", "DECSTART", "RAEND", "DECEND"):

367 header[h] = None

368 log.debug("%s: Forcing derived RA/Dec headers to undefined", log_label)

369 modified = True

370 else:

371 # Correct hours to degrees

372 for h in ("RASTART", "RAEND"):

373 if header[h]:

374 header[h] *= 15.0

375 log.debug("%s: Correcting RASTART/END from hours to degrees", log_label)

376 modified = True

377

378 # RASTART/END headers have a TAI/UTC confusion causing an offset

379 # of 37 seconds for a period of time.

380 if RASTART_IS_BAD < date < RASTART_IS_OKAY:

381 modified = True

382 offset = (37.0 / 3600.0) * 15.0

383 for epoch in ("START", "END"):

384 h = "RA" + epoch

385 if header[h]:

386 header[h] += offset

387

388 if date < ROTPA_CONVENTION_180_SWITCH2 and date > ROTPA_CONVENTION_180_SWITCH1:

389 header['ROTPA'] = header['ROTPA'] - 180

390 modified = True

391

392 if obsgeo := header.get("OBSGEO-Z"):

393 try:

394 if obsgeo > 0.0:

395 obsgeo *= -1.0

396 header["OBSGEO-Z"] = obsgeo

397 modified = True

398 except TypeError:

399 pass

400

401 return modified

402

403 def _is_on_mountain(self):

404 date = self.to_datetime_begin()

405 if date > TSTART:

406 return True

407 return False

408

409 @staticmethod

410 def compute_detector_exposure_id(exposure_id, detector_num):

411 # Docstring inherited.

412 if detector_num != 0:

413 log.warning("Unexpected non-zero detector number for LATISS")

414 return LsstBaseTranslator.compute_detector_exposure_id(exposure_id, detector_num)

415

416 @cache_translation

417 def to_dark_time(self):

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

419

420 # Always compare with exposure time

421 # We may revisit this later if there is a cutoff date where we

422 # can always trust the header.

423 exptime = self.to_exposure_time()

424

425 if self.is_key_ok("DARKTIME"):

426 darktime = self.quantity_from_card("DARKTIME", u.s)

427 if darktime >= exptime:

428 return darktime

429 reason = "Dark time less than exposure time."

430 else:

431 reason = "Dark time not defined."

432

433 log.warning("%s: %s Setting dark time to the exposure time.",

434 self._log_prefix, reason)

435 return exptime

436

437 @cache_translation

438 def to_exposure_time(self):

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

440 # Some data is missing a value for EXPTIME.

441 # Have to be careful we do not have circular logic when trying to

442 # guess. SHUTTIME for LATISS is always the same as EXPTIME so requested

443 # and actual are identical.

444 if self.is_key_ok("EXPTIME"):

445 return self.quantity_from_card("EXPTIME", u.s)

446

447 # A missing or undefined EXPTIME is problematic. Set to -1

448 # to indicate that none was found.

449 log.warning("%s: Insufficient information to derive exposure time. Setting to -1.0s",

450 self._log_prefix)

451 return -1.0 * u.s

452

453 @cache_translation

454 def to_observation_type(self):

455 """Determine the observation type.

456

457 In the absence of an ``IMGTYPE`` header, assumes lab data is always a

458 dark if exposure time is non-zero, else bias.

459

460 Returns

461 -------

462 obstype : `str`

463 Observation type.

464 """

465

466 # LATISS observation type is documented to appear in OBSTYPE

467 # but for historical reasons prefers IMGTYPE.

468 # Test the keys in order until we find one that contains a

469 # defined value.

470 obstype_keys = ["OBSTYPE", "IMGTYPE"]

471

472 obstype = None

473 for k in obstype_keys:

474 if self.is_key_ok(k):

475 obstype = self._header[k]

476 self._used_these_cards(k)

477 obstype = obstype.lower()

478 break

479

480 if obstype is not None:

481 if obstype == "object" and not self._is_on_mountain():

482 # Do not map object to science in lab since most

483 # code assume science is on sky with RA/Dec.

484 obstype = "labobject"

485 elif obstype in ("skyexp", "object"):

486 obstype = "science"

487

488 return obstype

489

490 # In the absence of any observation type information, return

491 # unknown unless we think it might be a bias.

492 exptime = self.to_exposure_time()

493 if exptime == 0.0:

494 obstype = "bias"

495 else:

496 obstype = "unknown"

497 log.warning("%s: Unable to determine observation type. Guessing '%s'",

498 self._log_prefix, obstype)

499 return obstype

500

501 @cache_translation

502 def to_physical_filter(self):

503 """Calculate the physical filter name.

504

505 Returns

506 -------

507 filter : `str`

508 Name of filter. A combination of FILTER and GRATING

509 headers joined by a "~". The filter and grating are always

510 combined. The filter or grating part will be "none" if no value

511 is specified. Uses "empty" if any of the filters or gratings

512 indicate an "empty_N" name. "unknown" indicates that the filter is

513 not defined anywhere but we think it should be. "none" indicates

514 that the filter was not defined but the observation is a dark

515 or bias.

516 """

517

518 physical_filter = self._determine_primary_filter()

519

520 if self.is_key_ok("GRATING"):

521 grating = self._header["GRATING"]

522 self._used_these_cards("GRATING")

523

524 if not grating or grating.lower().startswith("empty"):

525 grating = "empty"

526 else:

527 # Be explicit about having no knowledge of the grating

528 grating = "unknown"

529

530 physical_filter = f"{physical_filter}{FILTER_DELIMITER}{grating}"

531

532 return physical_filter

533

534 @cache_translation

535 def to_altaz_end(self):

536 # For over a year we wrote AAZEND/AELEND by mistake with blank

537 # values because of a typo in the header configuration. In this

538 # scenario we return None since it is not possible to calculate

539 # anything regardless of observing mode.

540 if "AELEND" in self._header and "ELEND" not in self._header:

541 return None

542 return super().to_altaz_end()

Coverage for python / lsst / obs / lsst / translators / latiss.py: 16%

257 statements