Coverage for tests / test_visit_image.py: 21%

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 

14import os 

15import unittest 

16from typing import Any 

17 

18import astropy.io.fits 

19 

20from lsst.images import Box, VisitImage, get_legacy_visit_image_mask_planes 

21from lsst.images.fits import ExtensionKey 

22from lsst.images.tests import ( 

23 DP2_VISIT_DETECTOR_DATA_ID, 

24 RoundtripFits, 

25 TemporaryButler, 

26 assert_masked_images_equal, 

27 compare_visit_image_to_legacy, 

28) 

29 

30DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None) 

31 

32 

33@unittest.skipUnless(DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not in the environment.") 

34class VisitImageTestCase(unittest.TestCase): 

35 """Tests for the VisitImage class and the basics of the archive system.""" 

36 

37 @classmethod 

38 def setUpClass(cls) -> None: 

39 assert DATA_DIR is not None, "Guaranteed by decorator." 

40 cls.filename = os.path.join(DATA_DIR, "dp2", "legacy", "visit_image.fits") 

41 cls.plane_map = plane_map = get_legacy_visit_image_mask_planes() 

42 cls.visit_image = VisitImage.read_legacy( 

43 cls.filename, preserve_quantization=True, plane_map=plane_map 

44 ) 

45 cls.legacy_exposure: Any = None 

46 try: 

47 from lsst.afw.image import ExposureFitsReader 

48 

49 cls.legacy_exposure = ExposureFitsReader(cls.filename).read() 

50 except ImportError: 

51 pass 

52 

53 def test_obs_info(self) -> None: 

54 """Check that ObservationInfo has been constructured.""" 

55 legacy = VisitImage.from_legacy(self.legacy_exposure, plane_map=self.plane_map) 

56 self.assertIsNotNone(legacy.obs_info) 

57 self.maxDiff = None 

58 self.assertEqual(legacy.obs_info, self.visit_image.obs_info) 

59 assert legacy.obs_info is not None # for mypy. 

60 self.assertEqual(legacy.obs_info.instrument, "LSSTCam") 

61 

62 def test_read_legacy_headers(self) -> None: 

63 """Test that headers were correctly stripped and interpreted in 

64 `VisitImage.read_legacy`. 

65 """ 

66 # Check that we read the units from BUNIT. 

67 self.assertEqual(self.visit_image.unit, astropy.units.nJy) 

68 # Check that the primary header has the keys we want, and none of the 

69 # keys we don't want. 

70 header = self.visit_image._opaque_metadata.headers[ExtensionKey()] 

71 self.assertIn("EXPTIME", header) 

72 self.assertNotIn("LSST BUTLER ID", header) 

73 self.assertNotIn("AR HDU", header) 

74 self.assertNotIn("A_ORDER", header) 

75 # Check that the extension HDUs do not have any custom headers. 

76 self.assertFalse(self.visit_image._opaque_metadata.headers[ExtensionKey("IMAGE")]) 

77 self.assertFalse(self.visit_image._opaque_metadata.headers[ExtensionKey("MASK")]) 

78 self.assertFalse(self.visit_image._opaque_metadata.headers[ExtensionKey("VARIANCE")]) 

79 

80 def test_rewrite(self) -> None: 

81 """Test that we can rewrite the visit image and preserve both 

82 lossy-compressed pixel values and components exactly. 

83 """ 

84 with RoundtripFits(self, self.visit_image, "VisitImage") as roundtrip: 

85 # Check that we're still using the right compression, and that we 

86 # wrote WCSs. 

87 fits = roundtrip.inspect() 

88 self.assertEqual(fits[1].header["ZCMPTYPE"], "RICE_1") 

89 self.assertEqual(fits[1].header["CTYPE1"], "RA---TAN-SIP") 

90 self.assertEqual(fits[2].header["ZCMPTYPE"], "GZIP_2") 

91 self.assertEqual(fits[2].header["CTYPE1"], "RA---TAN-SIP") 

92 self.assertEqual(fits[3].header["ZCMPTYPE"], "RICE_1") 

93 self.assertEqual(fits[3].header["CTYPE1"], "RA---TAN-SIP") 

94 # Check a subimage read. 

95 subbox = Box.factory[8:13, 9:30] 

96 subimage = roundtrip.get(bbox=subbox) 

97 assert_masked_images_equal(self, subimage, self.visit_image[subbox], expect_view=False) 

98 alternates: dict[str, Any] = {} 

99 with self.subTest(): 

100 self.assertEqual(roundtrip.get("bbox"), self.visit_image.bbox) 

101 alternates = {k: roundtrip.get(k) for k in ["projection", "image", "mask", "variance", "psf"]} 

102 assert_masked_images_equal(self, roundtrip.result, self.visit_image, expect_view=False) 

103 # Check that the round-tripped headers are the same (up to card order). 

104 self.assertEqual( 

105 dict(self.visit_image._opaque_metadata.headers[ExtensionKey()]), 

106 dict(roundtrip.result._opaque_metadata.headers[ExtensionKey()]), 

107 ) 

108 self.assertFalse(roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]) 

109 self.assertFalse(roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]) 

110 self.assertFalse(roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]) 

111 with self.subTest(): 

112 if self.legacy_exposure is None: 

113 raise unittest.SkipTest("'lsst.afw.image' could not be imported.") from None 

114 compare_visit_image_to_legacy( 

115 self, 

116 roundtrip.result, 

117 self.legacy_exposure, 

118 expect_view=False, 

119 plane_map=self.plane_map, 

120 **DP2_VISIT_DETECTOR_DATA_ID, 

121 alternates=alternates, 

122 ) 

123 # Check converting from the legacy object in-memory. 

124 compare_visit_image_to_legacy( 

125 self, 

126 VisitImage.from_legacy(self.legacy_exposure, plane_map=self.plane_map), 

127 self.legacy_exposure, 

128 expect_view=True, 

129 plane_map=self.plane_map, 

130 **DP2_VISIT_DETECTOR_DATA_ID, 

131 ) 

132 

133 def test_butler_converters(self) -> None: 

134 """Test that we can read a VisitImage and its components from a butler 

135 dataset written as an `lsst.afw.image.Exposure`. 

136 """ 

137 if self.legacy_exposure is None: 

138 raise unittest.SkipTest("lsst.afw.image.afw could not be imported.") 

139 with TemporaryButler(legacy="ExposureF") as helper: 

140 from lsst.daf.butler import FileDataset 

141 

142 helper.butler.ingest(FileDataset(path=self.filename, refs=[helper.legacy]), transfer="symlink") 

143 visit_image_ref = helper.legacy.overrideStorageClass("VisitImage") 

144 visit_image = helper.butler.get(visit_image_ref) 

145 bbox = helper.butler.get(visit_image_ref.makeComponentRef("bbox")) 

146 self.assertEqual(bbox, visit_image.bbox) 

147 alternates = { 

148 k: helper.butler.get(visit_image_ref.makeComponentRef(k)) 

149 # TODO: including "projection" here fails because there's 

150 # code in daf_butler that expects any component to be valid 

151 # for the *internal* storage class, not the requested one, 

152 # and that's difficult to fix because it's tied up with the 

153 # data ID standardization logic. 

154 for k in ["image", "mask", "variance", "psf"] 

155 } 

156 compare_visit_image_to_legacy( 

157 self, 

158 visit_image, 

159 self.legacy_exposure, 

160 expect_view=False, 

161 plane_map=self.plane_map, 

162 alternates=alternates, 

163 **DP2_VISIT_DETECTOR_DATA_ID, 

164 ) 

165 

166 

167if __name__ == "__main__": 

168 unittest.main()