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_read_legacy_headers(self) -> None: 

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

55 `VisitImage.read_legacy`. 

56 """ 

57 # Check that we read the units from BUNIT. 

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

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

60 # keys we don't want. 

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

62 self.assertIn("EXPTIME", header) 

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

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

65 self.assertNotIn("A_ORDER", header) 

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

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

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

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

70 

71 def test_rewrite(self) -> None: 

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

73 lossy-compressed pixel values and components exactly. 

74 """ 

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

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

77 # wrote WCSs. 

78 fits = roundtrip.inspect() 

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

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

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

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

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

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

85 # Check a subimage read. 

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

87 subimage = roundtrip.get(bbox=subbox) 

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

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

90 with self.subTest(): 

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

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

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

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

95 self.assertEqual( 

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

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

98 ) 

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

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

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

102 with self.subTest(): 

103 if self.legacy_exposure is None: 

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

105 compare_visit_image_to_legacy( 

106 self, 

107 roundtrip.result, 

108 self.legacy_exposure, 

109 expect_view=False, 

110 plane_map=self.plane_map, 

111 **DP2_VISIT_DETECTOR_DATA_ID, 

112 alternates=alternates, 

113 ) 

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

115 compare_visit_image_to_legacy( 

116 self, 

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

118 self.legacy_exposure, 

119 expect_view=True, 

120 plane_map=self.plane_map, 

121 **DP2_VISIT_DETECTOR_DATA_ID, 

122 ) 

123 

124 def test_butler_converters(self) -> None: 

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

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

127 """ 

128 if self.legacy_exposure is None: 

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

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

131 from lsst.daf.butler import FileDataset 

132 

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

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

135 visit_image = helper.butler.get(visit_image_ref) 

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

137 self.assertEqual(bbox, visit_image.bbox) 

138 alternates = { 

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

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

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

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

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

144 # data ID standardization logic. 

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

146 } 

147 compare_visit_image_to_legacy( 

148 self, 

149 visit_image, 

150 self.legacy_exposure, 

151 expect_view=False, 

152 plane_map=self.plane_map, 

153 alternates=alternates, 

154 **DP2_VISIT_DETECTOR_DATA_ID, 

155 ) 

156 

157 

158if __name__ == "__main__": 

159 unittest.main()