Coverage for python/lsst/images/_obs_info_from_legacy.py: 62%

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

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

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

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

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

11 

12from __future__ import annotations 

13 

14__all__ = ("obs_info_from_legacy",) 

15 

16import logging 

17from collections.abc import MutableMapping 

18from typing import TYPE_CHECKING, Any 

19 

20import astropy.units as u 

21from astro_metadata_translator import ( 

22 FITS_FLAT_DICT_PREFIX, 

23 FITS_FLAT_DICT_SEP, 

24 ObservationInfo, 

25 VisitInfoTranslator, 

26) 

27from astropy.coordinates import AltAz 

28 

29_LOG = logging.getLogger("lsst.images") 

30 

31if TYPE_CHECKING: 

32 try: 

33 from lsst.afw.cameraGeom import Detector as LegacyDetector 

34 from lsst.afw.image import VisitInfo as LegacyVisitInfo 

35 from lsst.daf.butler import DimensionRecord 

36 except ImportError: 

37 type LegacyDetector = Any # type: ignore[no-redef] 

38 type LegacyVisitInfo = Any # type: ignore[no-redef] 

39 type DimensionRecord = Any # type: ignore[no-redef] 

40 

41 

42def obs_info_from_legacy( 

43 visit_info: LegacyVisitInfo, 

44 exposure_record: DimensionRecord | None, 

45 detector: LegacyDetector, 

46 *, 

47 detector_exposure_id: int, 

48 telescope: str | None = None, 

49 physical_filter: str | None = None, 

50 header: MutableMapping[str, Any] | None = None, 

51) -> ObservationInfo: 

52 """Recover the observation info struct of a legacy 

53 `lsst.afw.image.Exposure`. 

54 

55 If the legacy metadata carries an embedded 

56 `~astro_metadata_translator.ObservationInfo` (the flat ``OBS_INFO`` cards 

57 written by the raw formatter), that is returned as is. Otherwise as much 

58 as possible is reconstructed from the `~lsst.afw.image.VisitInfo`, an 

59 ``exposure`` dimension record, and the camera geometry. 

60 

61 Parameters 

62 ---------- 

63 visit_info 

64 The legacy visit info struct. 

65 exposure_record 

66 The ``exposure`` `~lsst.daf.butler.DimensionRecord`, or `None` to 

67 convert without one. Without a record, fields it would authoritatively 

68 supply are set to `None` and a warning is logged. 

69 detector 

70 The legacy detector object; the source for detector identifiers. 

71 detector_exposure_id 

72 The combination detector + exposure ID. 

73 telescope 

74 Value of the FITS ``TELESCOP`` card, if available from the legacy 

75 metadata. `~lsst.afw.image.VisitInfo` carries no telescope 

76 information, so without this the telescope is reported as 

77 ``"Unknown"``. 

78 physical_filter 

79 Full name of the physical bandpass filter (the ``physicalLabel`` of 

80 the legacy exposure's `~lsst.afw.image.FilterLabel`), if available. 

81 Only used when there is no ``exposure_record``. 

82 header 

83 The legacy exposure's metadata (its primary FITS header as an 

84 `astropy.io.fits.Header` or `lsst.daf.base.PropertyList`), if 

85 available. When it holds an embedded ``ObservationInfo`` the 

86 ``OBS_INFO`` cards are removed from it in place, so they do not leak 

87 into other metadata, and the other parameters are only used to check 

88 that the ``exposure_record`` describes the same exposure. 

89 

90 Returns 

91 ------- 

92 obs_info : `~astro_metadata_translator.ObservationInfo` 

93 The recovered observation metadata. 

94 

95 Raises 

96 ------ 

97 ValueError 

98 Raised if ``exposure_record`` is given and does not describe the same 

99 instrument and exposure as the embedded ``ObservationInfo`` or the 

100 ``visit_info``. 

101 """ 

102 flat_dict_kwargs: dict[str, Any] = {"prefix": FITS_FLAT_DICT_PREFIX, "sep": FITS_FLAT_DICT_SEP} 

103 if header is not None and ObservationInfo.has_flat_dict(header, **flat_dict_kwargs): 

104 obs_info = ObservationInfo.from_flat_dict(header, **flat_dict_kwargs) 

105 ObservationInfo.strip_from_flat_dict(header, **flat_dict_kwargs) 

106 if exposure_record is not None: 

107 if obs_info.instrument != exposure_record.instrument: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 raise ValueError( 

109 f"Instrument {obs_info.instrument!r} from embedded ObservationInfo does not match " 

110 f"instrument {exposure_record.instrument!r} of the given exposure record." 

111 ) 

112 if obs_info.exposure_id != exposure_record.id: 

113 raise ValueError( 

114 f"Exposure ID {obs_info.exposure_id} from embedded ObservationInfo does not match " 

115 f"exposure ID {exposure_record.id} of the given exposure record." 

116 ) 

117 return obs_info 

118 

119 from lsst.afw.image import setVisitInfoMetadata 

120 from lsst.daf.base import PropertyList 

121 

122 if exposure_record is not None: 122 ↛ 134line 122 didn't jump to line 134 because the condition on line 122 was always true

123 if (label := visit_info.getInstrumentLabel()) and label != exposure_record.instrument: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 raise ValueError( 

125 f"Instrument label {label!r} from VisitInfo does not match instrument " 

126 f"{exposure_record.instrument!r} of the given exposure record." 

127 ) 

128 if (visit_id := visit_info.getId()) != exposure_record.id: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

129 raise ValueError( 

130 f"Exposure ID {visit_id} from VisitInfo does not match exposure ID " 

131 f"{exposure_record.id} of the given exposure record." 

132 ) 

133 else: 

134 _LOG.warning( 

135 "Converting without an exposure dimension record: ObservationInfo " 

136 "fields the record would supply are set to None." 

137 ) 

138 

139 # Start by exporting the VisitInfo to FITS header cards and translating 

140 # those. This populates a lot of fields correctly, a lot of fields not at 

141 # all, and some *incorrectly*, at least as compared to the original 

142 # ObservationInfo, because VisitInfoTranslator tries too hard to 

143 # reconstruct information that has been lost. 

144 pl = PropertyList() 

145 setVisitInfoMetadata(pl, visit_info) 

146 if telescope is not None: 

147 # TELESCOP is a standard FITS card that VisitInfo knows nothing about, 

148 # but VisitInfoTranslator uses it to populate the telescope field. 

149 pl.add("TELESCOP", telescope) 

150 obs_info = ObservationInfo.from_header(pl, translator_class=VisitInfoTranslator, quiet=True) 

151 with obs_info.edit_copy() as obs_info: 

152 obs_info.detector_exposure_id = detector_exposure_id 

153 if exposure_record is not None: 153 ↛ 185line 153 didn't jump to line 185 because the condition on line 153 was always true

154 # Since the dimension record is created directly from the original 

155 # ObservationInfo, anything it actually holds is authoritative. 

156 obs_info.datetime_begin = exposure_record.timespan.begin 

157 obs_info.datetime_end = exposure_record.timespan.end 

158 obs_info.observing_day = exposure_record.day_obs 

159 obs_info.observation_id = exposure_record.obs_id 

160 obs_info.exposure_group = exposure_record.group 

161 obs_info.physical_filter = exposure_record.physical_filter 

162 # The record's exposure_time is the requested exposure time; 

163 # the VisitInfo translator's exposure_time is the actual 

164 # shutter-open time. 

165 obs_info.exposure_time_requested = ( 

166 None if exposure_record.exposure_time is None else exposure_record.exposure_time * u.s 

167 ) 

168 obs_info.observation_counter = exposure_record.seq_num 

169 obs_info.group_counter_start = exposure_record.seq_start 

170 obs_info.group_counter_end = exposure_record.seq_end 

171 obs_info.can_see_sky = exposure_record.can_see_sky 

172 if (azimuth := exposure_record.azimuth) is not None and ( 172 ↛ 197line 172 didn't jump to line 197 because the condition on line 172 was always true

173 zenith_angle := exposure_record.zenith_angle 

174 ) is not None: 

175 # The record's azimuth/zenith_angle were ingested from 

176 # obsInfo.altaz_begin (begin-of-exposure). The VisitInfo stores 

177 # mid-exposure az/alt, and neither stores altaz_end. 

178 obs_info.altaz_begin = AltAz( 

179 az=azimuth * u.deg, 

180 alt=(90.0 - zenith_angle) * u.deg, 

181 obstime=exposure_record.timespan.begin, 

182 location=obs_info.location, 

183 ) 

184 else: 

185 obs_info.datetime_begin = None 

186 obs_info.datetime_end = None 

187 obs_info.observing_day = None 

188 obs_info.observation_id = None 

189 obs_info.exposure_group = None 

190 obs_info.physical_filter = physical_filter 

191 obs_info.exposure_time_requested = None 

192 obs_info.observation_counter = None 

193 obs_info.group_counter_start = None 

194 obs_info.group_counter_end = None 

195 obs_info.can_see_sky = None 

196 obs_info.altaz_begin = None 

197 obs_info.detector_num = detector.getId() 

198 obs_info.detector_unique_name = detector.getName() 

199 obs_info.detector_serial = detector.getSerial() 

200 return obs_info