Coverage for tests / test_cell_coadd.py: 39%

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 pickle 

16import tempfile 

17import unittest 

18from typing import Any 

19 

20import numpy as np 

21 

22from lsst.images import YX, Box, Interval, fits, get_legacy_deep_coadd_mask_planes 

23from lsst.images.cells import CellCoadd, CellIJ 

24from lsst.images.formatters import CellCoaddFormatter 

25from lsst.images.tests import ( 

26 DP2_COADD_DATA_ID, 

27 DP2_COADD_MISSING_CELL, 

28 RoundtripFits, 

29 assert_masked_images_equal, 

30 assert_psfs_equal, 

31 compare_cell_coadd_to_legacy, 

32 make_test_formatter, 

33) 

34from lsst.resources import ResourcePath 

35 

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

37 

38 

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

40class CellCoaddTestCase(unittest.TestCase): 

41 """Tests for the CellCoadd class and its many component classes.""" 

42 

43 @classmethod 

44 def setUpClass(cls) -> None: 

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

46 cls.filename = os.path.join(DATA_DIR, "dp2", "legacy", "deep_coadd_cell_predetection.fits") 

47 cls.plane_map = get_legacy_deep_coadd_mask_planes() 

48 cls.missing_cell = CellIJ(**DP2_COADD_MISSING_CELL) 

49 try: 

50 from lsst.cell_coadds import MultipleCellCoadd 

51 

52 cls.legacy_cell_coadd = MultipleCellCoadd.read_fits(cls.filename) 

53 except ImportError: 

54 raise unittest.SkipTest("lsst.cell_coadds could not be imported.") from None 

55 with open(os.path.join(DATA_DIR, "dp2", "legacy", "skyMap.pickle"), "rb") as stream: 

56 cls.skymap = pickle.load(stream) 

57 cls.cell_coadd = CellCoadd.from_legacy( 

58 cls.legacy_cell_coadd, 

59 plane_map=cls.plane_map, 

60 tract_info=cls.skymap[DP2_COADD_DATA_ID["tract"]], 

61 ) 

62 

63 def make_psf_points(self, bbox: Box) -> YX[np.ndarray]: 

64 """Make arrays of points to test PSFs at, given a bbox that is assumed 

65 to be snapped to the cell_coadd grid. 

66 """ 

67 xc, yc = np.meshgrid( 

68 np.arange( 

69 bbox.x.start + self.cell_coadd.grid.cell_shape.x * 0.5, 

70 bbox.x.stop, 

71 self.cell_coadd.grid.cell_shape.x, 

72 ), 

73 np.arange( 

74 bbox.y.start + self.cell_coadd.grid.cell_shape.y * 0.5, 

75 bbox.y.stop, 

76 self.cell_coadd.grid.cell_shape.y, 

77 ), 

78 ) 

79 return YX( 

80 y=yc.ravel() + self.rng.uniform(-0.4, 0.4, size=yc.size), 

81 x=xc.ravel() + self.rng.uniform(-0.4, 0.4, size=xc.size), 

82 ) 

83 

84 def setUp(self) -> None: 

85 self.rng = np.random.default_rng(44) 

86 self.psf_points = self.make_psf_points(self.cell_coadd.bbox) 

87 

88 def test_from_legacy(self) -> None: 

89 """Test constructing a CellCoadd by converting a legacy 

90 lsst.cell_coadds.MultipleCellCoadd. 

91 """ 

92 self.assertEqual(self.cell_coadd.bounds.missing, {self.missing_cell}) 

93 self.assertEqual(self.cell_coadd.bbox, Box.factory[12900:13500, 9600:10050]) 

94 compare_cell_coadd_to_legacy( 

95 self, 

96 self.cell_coadd, 

97 self.legacy_cell_coadd, 

98 tract_bbox=Box.from_legacy(self.skymap[DP2_COADD_DATA_ID["tract"]].getBBox()), 

99 plane_map=self.plane_map, 

100 psf_points=self.psf_points, 

101 ) 

102 

103 def test_roundtrip(self) -> None: 

104 """Test serializing a CellCoadd and reading it back in, including 

105 subimage and component reads. 

106 """ 

107 with RoundtripFits(self, self.cell_coadd, "CellCoadd") as roundtrip: 

108 # Check a subimage read. The subbox only overlaps (but does not 

109 # fully cover) the middle 2 (of 4) cells in y, while covering 

110 # exactly the last column of cells in x. It does not cover the 

111 # missing cell. 

112 subbox = Box.factory[ 

113 self.cell_coadd.bbox.y.start + 252 : self.cell_coadd.bbox.y.stop - 175, 

114 self.cell_coadd.bbox.x.stop - 150 : self.cell_coadd.bbox.x.stop, 

115 ] 

116 subimage = roundtrip.get(bbox=subbox) 

117 assert_masked_images_equal(self, subimage, self.cell_coadd[subbox], expect_view=False) 

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

119 with self.subTest(): 

120 subpsf = roundtrip.get("psf", bbox=subbox) 

121 self.assertEqual( 

122 subpsf.bounds.bbox, 

123 Box( 

124 y=Interval.factory[ 

125 self.cell_coadd.bbox.y.start + 150 : self.cell_coadd.bbox.y.stop - 150 

126 ], 

127 x=subbox.x, 

128 ), 

129 ) 

130 assert_psfs_equal(self, subpsf, self.cell_coadd.psf, points=self.make_psf_points(subbox)) 

131 self.assertEqual(roundtrip.get("bbox"), self.cell_coadd.bbox) 

132 alternates = { 

133 k: roundtrip.get(k) 

134 for k in ["projection", "image", "mask", "variance", "psf", "provenance"] 

135 } 

136 self.assertEqual(self.cell_coadd.bounds.missing, {self.missing_cell}) 

137 self.assertEqual(self.cell_coadd.bbox, Box.factory[12900:13500, 9600:10050]) 

138 compare_cell_coadd_to_legacy( 

139 self, 

140 roundtrip.result, 

141 self.legacy_cell_coadd, 

142 tract_bbox=Box.from_legacy(self.skymap[DP2_COADD_DATA_ID["tract"]].getBBox()), 

143 plane_map=self.plane_map, 

144 alternates=alternates, 

145 psf_points=self.psf_points, 

146 ) 

147 

148 

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

150class CellCoaddFormatterComponentReadTestCase(unittest.TestCase): 

151 """CellCoaddFormatter reads psf/provenance components from FITS. 

152 

153 Reuses `CellCoaddTestCase`'s class-level fixture rather than 

154 inheriting from it, so the parent's tests don't run twice. 

155 """ 

156 

157 @classmethod 

158 def setUpClass(cls) -> None: 

159 CellCoaddTestCase.setUpClass() 

160 cls.cell_coadd = CellCoaddTestCase.cell_coadd 

161 

162 def test_fits_psf_component(self): 

163 with tempfile.NamedTemporaryFile(suffix=".fits", delete_on_close=False) as tmp: 

164 tmp.close() 

165 fits.write(self.cell_coadd, tmp.name) 

166 formatter = make_test_formatter(CellCoaddFormatter, CellCoadd) 

167 psf = formatter._read_component_from_uri("psf", ResourcePath(tmp.name)) 

168 self.assertIsNotNone(psf) 

169 

170 

171if __name__ == "__main__": 

172 unittest.main()