Coverage for python / lsst / obs / base / _fitsRawFormatterBase.py: 28%

1# This file is part of obs_base. 

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# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22__all__ = ("FitsRawFormatterBase",) 

23 

24import logging 

25from abc import abstractmethod 

26from typing import Any, ClassVar 

27 

28from astro_metadata_translator import ObservationInfo, fix_header 

29 

30import lsst.afw.fits 

31import lsst.afw.geom 

32import lsst.afw.image 

33from lsst.daf.butler import FileDescriptor, FormatterNotImplementedError 

34from lsst.resources import ResourcePath 

35from lsst.utils.classes import cached_getter 

36 

37from .formatters.fitsExposure import FitsImageFormatterBase, standardizeAmplifierParameters 

38from .makeRawVisitInfoViaObsInfo import MakeRawVisitInfoViaObsInfo 

39from .utils import InitialSkyWcsError, createInitialSkyWcsFromBoresight 

40 

41log = logging.getLogger(__name__) 

42 

43 

44class FitsRawFormatterBase(FitsImageFormatterBase): 

45 """Abstract base class for reading and writing raw data to and from 

46 FITS files. 

47 """ 

48 

49 ReaderClass = lsst.afw.image.ImageFitsReader 

50 

51 wcsFlipX: ClassVar[bool] = False 

52 """Control whether the WCS is flipped in the X-direction (`bool`). 

53 """ 

54 

55 def __init__(self, *args, **kwargs): 

56 super().__init__(*args, **kwargs) 

57 self._metadata = None 

58 self._observationInfo = None 

59 

60 @classmethod 

61 def fromMetadata(cls, metadata, obsInfo=None, storageClass=None, location=None): 

62 """Construct a possibly-limited formatter from known metadata. 

63 

64 Parameters 

65 ---------- 

66 metadata : `lsst.daf.base.PropertyList` 

67 Raw header metadata, with any fixes (see 

68 `astro_metadata_translator.fix_header`) applied but nothing 

69 stripped. 

70 obsInfo : `astro_metadata_translator.ObservationInfo`, optional 

71 Structured information already extracted from ``metadata``. 

72 If not provided, will be read from ``metadata`` on first use. 

73 storageClass : `lsst.daf.butler.StorageClass`, optional 

74 StorageClass for this file. If not provided, the formatter will 

75 only support `makeWcs`, `makeVisitInfo`, `makeFilter`, and other 

76 operations that operate purely on metadata and not the actual file. 

77 location : `lsst.daf.butler.Location`, optional. 

78 Location of the file. If not provided, the formatter will only 

79 support `makeWcs`, `makeVisitInfo`, `makeFilter`, and other 

80 operations that operate purely on metadata and not the actual file. 

81 

82 Returns 

83 ------- 

84 formatter : `FitsRawFormatterBase` 

85 An instance of ``cls``. 

86 """ 

87 self = cls(FileDescriptor(location, storageClass)) 

88 self._metadata = metadata 

89 self._observationInfo = obsInfo 

90 return self 

91 

92 @property 

93 @abstractmethod 

94 def translatorClass(self): 

95 """`~astro_metadata_translator.MetadataTranslator` to translate 

96 metadata header to `~astro_metadata_translator.ObservationInfo`. 

97 """ 

98 return None 

99 

100 @property 

101 @abstractmethod 

102 def filterDefinitions(self): 

103 """`~lsst.obs.base.FilterDefinitions`, defining the filters for this 

104 instrument. 

105 """ 

106 return None 

107 

108 @property 

109 @cached_getter 

110 def checked_parameters(self): 

111 # Docstring inherited. 

112 parameters = super().checked_parameters 

113 if "bbox" in parameters: 

114 raise TypeError( 

115 "Raw formatters do not support reading arbitrary subimages, as some " 

116 "implementations may be assembled on-the-fly." 

117 ) 

118 return parameters 

119 

120 def readImage(self): 

121 """Read just the image component of the Exposure. 

122 

123 Returns 

124 ------- 

125 image : `~lsst.afw.image.Image` 

126 In-memory image component. 

127 """ 

128 return lsst.afw.image.ImageU(self._reader_path) 

129 

130 def isOnSky(self): 

131 """Boolean to determine if the exposure is thought to be on the sky. 

132 

133 Returns 

134 ------- 

135 onSky : `bool` 

136 Returns `True` if the observation looks like it was taken on the 

137 sky. Returns `False` if this observation looks like a calibration 

138 observation. 

139 

140 Notes 

141 ----- 

142 If there is tracking RA/Dec information associated with the 

143 observation it is assumed that the observation is on sky. 

144 Currently the observation type is not checked. 

145 """ 

146 if self.observationInfo.tracking_radec is None: 

147 return False 

148 return True 

149 

150 @property 

151 def metadata(self): 

152 """The metadata read from this file. It will be stripped as 

153 components are extracted from it 

154 (`lsst.daf.base.PropertyList`). 

155 """ 

156 if self._metadata is None: 

157 self._metadata = self.readMetadata() 

158 return self._metadata 

159 

160 def readMetadata(self): 

161 """Read all header metadata directly into a PropertyList. 

162 

163 Returns 

164 ------- 

165 metadata : `~lsst.daf.base.PropertyList` 

166 Header metadata. 

167 """ 

168 md = self.reader.readMetadata() 

169 fix_header(md, translator_class=self.translatorClass) 

170 return md 

171 

172 def stripMetadata(self): 

173 """Remove metadata entries that are parsed into components.""" 

174 try: 

175 lsst.afw.geom.stripWcsMetadata(self.metadata) 

176 except TypeError as e: 

177 log.debug("Error caught and ignored while stripping metadata: %s", e.args[0]) 

178 

179 def makeVisitInfo(self): 

180 """Construct a VisitInfo from metadata. 

181 

182 Returns 

183 ------- 

184 visitInfo : `~lsst.afw.image.VisitInfo` 

185 Structured metadata about the observation. 

186 """ 

187 visit_info = MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.observationInfo) 

188 if self.data_id and "exposure" in self.data_id: 

189 # Special case exposure existing for this dataset type. 

190 # Want to ensure that the id stored in the visitInfo matches that 

191 # from the dataId from butler, regardless of what may have come 

192 # from the ObservationInfo. In some edge cases they might differ. 

193 exposure_id = self.data_id["exposure"] 

194 if exposure_id != visit_info.id: 

195 visit_info = visit_info.copyWith(id=exposure_id) 

196 

197 return visit_info 

198 

199 @abstractmethod 

200 def getDetector(self, id): 

201 """Return the detector that acquired this raw exposure. 

202 

203 Parameters 

204 ---------- 

205 id : `int` 

206 The identifying number of the detector to get. 

207 

208 Returns 

209 ------- 

210 detector : `~lsst.afw.cameraGeom.Detector` 

211 The detector associated with that ``id``. 

212 """ 

213 raise NotImplementedError("Must be implemented by subclasses.") 

214 

215 def makeWcs(self, visitInfo, detector): 

216 """Create a SkyWcs from information about the exposure. 

217 

218 If VisitInfo is not None, use it and the detector to create a SkyWcs, 

219 otherwise return the metadata-based SkyWcs (always created, so that 

220 the relevant metadata keywords are stripped). 

221 

222 Parameters 

223 ---------- 

224 visitInfo : `~lsst.afw.image.VisitInfo` 

225 The information about the telescope boresight and camera 

226 orientation angle for this exposure. 

227 detector : `~lsst.afw.cameraGeom.Detector` 

228 The detector used to acquire this exposure. 

229 

230 Returns 

231 ------- 

232 skyWcs : `~lsst.afw.geom.SkyWcs` 

233 Reversible mapping from pixel coordinates to sky coordinates. 

234 

235 Raises 

236 ------ 

237 InitialSkyWcsError 

238 Raised if there is an error generating the SkyWcs, chained from the 

239 lower-level exception if available. 

240 """ 

241 if not self.isOnSky(): 

242 # This is not an on-sky observation 

243 return None 

244 

245 if visitInfo is None: 

246 msg = "No VisitInfo; cannot access boresight information. Defaulting to metadata-based SkyWcs." 

247 log.warning(msg) 

248 skyWcs = self._createSkyWcsFromMetadata() 

249 if skyWcs is None: 

250 raise InitialSkyWcsError( 

251 "Failed to create both metadata and boresight-based SkyWcs." 

252 "See warnings in log messages for details." 

253 ) 

254 return skyWcs 

255 

256 return self.makeRawSkyWcsFromBoresight( 

257 visitInfo.getBoresightRaDec(), visitInfo.getBoresightRotAngle(), detector 

258 ) 

259 

260 @classmethod 

261 def makeRawSkyWcsFromBoresight(cls, boresight, orientation, detector): 

262 """Class method to make a raw sky WCS from boresight and detector. 

263 

264 Parameters 

265 ---------- 

266 boresight : `lsst.geom.SpherePoint` 

267 The ICRS boresight RA/Dec 

268 orientation : `lsst.geom.Angle` 

269 The rotation angle of the focal plane on the sky. 

270 detector : `lsst.afw.cameraGeom.Detector` 

271 Where to get the camera geometry from. 

272 

273 Returns 

274 ------- 

275 skyWcs : `~lsst.afw.geom.SkyWcs` 

276 Reversible mapping from pixel coordinates to sky coordinates. 

277 """ 

278 return createInitialSkyWcsFromBoresight(boresight, orientation, detector, flipX=cls.wcsFlipX) 

279 

280 def _createSkyWcsFromMetadata(self): 

281 """Create a SkyWcs from the FITS header metadata in an Exposure. 

282 

283 Returns 

284 ------- 

285 skyWcs: `lsst.afw.geom.SkyWcs`, or None 

286 The WCS that was created from ``self.metadata``, or None if that 

287 creation fails due to invalid metadata. 

288 """ 

289 if not self.isOnSky(): 

290 # This is not an on-sky observation 

291 return None 

292 

293 try: 

294 return lsst.afw.geom.makeSkyWcs(self.metadata, strip=True) 

295 except TypeError as e: 

296 log.warning("Cannot create a valid WCS from metadata: %s", e.args[0]) 

297 return None 

298 

299 def makeFilterLabel(self): 

300 """Construct a FilterLabel from metadata. 

301 

302 Returns 

303 ------- 

304 filter : `~lsst.afw.image.FilterLabel` 

305 Object that identifies the filter for this image. 

306 """ 

307 physical = self.observationInfo.physical_filter 

308 band = self.filterDefinitions.physical_to_band[physical] 

309 return lsst.afw.image.FilterLabel(physical=physical, band=band) 

310 

311 def readComponent(self, component): 

312 # Docstring inherited. 

313 _ = self.checked_parameters # just for checking; no supported parameters. 

314 if component == "image": 

315 return self.readImage() 

316 elif component == "filter": 

317 return self.makeFilterLabel() 

318 elif component == "visitInfo": 

319 return self.makeVisitInfo() 

320 elif component == "detector": 

321 return self.getDetector(self.observationInfo.detector_num) 

322 elif component == "wcs": 

323 detector = self.getDetector(self.observationInfo.detector_num) 

324 visitInfo = self.makeVisitInfo() 

325 return self.makeWcs(visitInfo, detector) 

326 elif component == "metadata": 

327 self.stripMetadata() 

328 return self.metadata 

329 return None 

330 

331 def readFull(self): 

332 # Docstring inherited. 

333 amplifier, detector, _ = standardizeAmplifierParameters( 

334 self.checked_parameters, 

335 self.getDetector(self.observationInfo.detector_num), 

336 ) 

337 if amplifier is not None: 

338 amplifier_isolator = lsst.afw.cameraGeom.AmplifierIsolator( 

339 amplifier, 

340 self.reader.readBBox(), 

341 detector, 

342 ) 

343 subimage = amplifier_isolator.transform_subimage( 

344 self.reader.read(bbox=amplifier_isolator.subimage_bbox) 

345 ) 

346 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(subimage)) 

347 exposure.setDetector(amplifier_isolator.make_detector()) 

348 else: 

349 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(self.readImage())) 

350 exposure.setDetector(detector) 

351 self.attachComponentsFromMetadata(exposure) 

352 return exposure 

353 

354 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None: 

355 raise FormatterNotImplementedError("Raw data cannot be `put`.") 

356 

357 @property 

358 def observationInfo(self): 

359 """The `~astro_metadata_translator.ObservationInfo` extracted from 

360 this file's metadata (`~astro_metadata_translator.ObservationInfo`, 

361 read-only). 

362 """ 

363 if self._observationInfo is None: 

364 # Use the primary path rather than any local variant that the 

365 # formatter might be using. 

366 location = self.file_descriptor.location 

367 path = location.path if location is not None else None 

368 self._observationInfo = ObservationInfo( 

369 self.metadata, translator_class=self.translatorClass, filename=path 

370 ) 

371 return self._observationInfo 

372 

373 def attachComponentsFromMetadata(self, exposure): 

374 """Attach all `lsst.afw.image.Exposure` components derived from 

375 metadata (including the stripped metadata itself). 

376 

377 Parameters 

378 ---------- 

379 exposure : `lsst.afw.image.Exposure` 

380 Exposure to attach components to (modified in place). Must already 

381 have a detector attached. 

382 """ 

383 info = exposure.getInfo() 

384 info.id = self.observationInfo.detector_exposure_id 

385 info.setFilter(self.makeFilterLabel()) 

386 info.setVisitInfo(self.makeVisitInfo()) 

387 info.setWcs(self.makeWcs(info.getVisitInfo(), info.getDetector())) 

388 

389 self.stripMetadata() 

390 exposure.setMetadata(self.metadata)