Coverage for tests / test_warper.py: 33%

1# 

2# LSST Data Management System 

3# Copyright 2008, 2009, 2010 LSST Corporation. 

4# 

5# This product includes software developed by the 

6# LSST Project (http://www.lsst.org/). 

7# 

8# This program is free software: you can redistribute it and/or modify 

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

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

11# (at your option) any later version. 

12# 

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

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

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

16# GNU General Public License for more details. 

17# 

18# You should have received a copy of the LSST License Statement and 

19# the GNU General Public License along with this program. If not, 

20# see <http://www.lsstcorp.org/LegalNotices/>. 

21# 

22"""Basic test of Warp (the warping algorithm is thoroughly tested in lsst.afw.math) 

23""" 

24import os 

25import unittest 

26 

27import lsst.utils 

28import lsst.utils.logging 

29import lsst.utils.tests 

30import lsst.geom 

31import lsst.afw.geom as afwGeom 

32import lsst.afw.image as afwImage 

33import lsst.afw.math as afwMath 

34from lsst.log import Log 

35 

36# Change the level to Log.DEBUG to see debug messages 

37Log.getLogger("lsst.afw.image.Mask").setLevel(Log.INFO) 

38 

39# Change integer argument to adjust trace logging. 

40lsst.utils.logging.trace_set_at("lsst.afw.math.warp", -1) 

41 

42try: 

43 afwdataDir = lsst.utils.getPackageDir("afwdata") 

44except LookupError: 

45 afwdataDir = None 

46 dataDir = None 

47else: 

48 dataDir = os.path.join(afwdataDir, "data") 

49 originalExposureName = "medexp.fits" 

50 originalExposurePath = os.path.join(dataDir, originalExposureName) 

51 subExposureName = "medsub.fits" 

52 subExposurePath = os.path.join(dataDir, originalExposureName) 

53 originalFullExposureName = os.path.join( 

54 "CFHT", "D4", "cal-53535-i-797722_1.fits") 

55 originalFullExposurePath = os.path.join(dataDir, originalFullExposureName) 

56 

57 

58class WarpExposureTestCase(lsst.utils.tests.TestCase): 

59 """Test case for Warp 

60 """ 

61 

62 def testMatchSwarpLanczos2Exposure(self): 

63 """Test that warpExposure matches swarp using a lanczos2 warping kernel. 

64 """ 

65 self.compareToSwarp("lanczos2") 

66 

67 def testMatchSwarpLanczos2SubExposure(self): 

68 """Test that warpExposure matches swarp using a lanczos2 warping kernel with a subexposure 

69 """ 

70 for useDeepCopy in (False, True): 

71 self.compareToSwarp("lanczos2", useSubregion=True, 

72 useDeepCopy=useDeepCopy) 

73 

74 @unittest.skipIf(dataDir is None, "afwdata not setup") 

75 def testBBox(self): 

76 """Test that the default bounding box includes all warped pixels 

77 """ 

78 kernelName = "lanczos2" 

79 warper = afwMath.Warper(kernelName) 

80 originalExposure, swarpedImage, swarpedWcs = self.getSwarpedImage( 

81 kernelName=kernelName, useSubregion=True, useDeepCopy=False) 

82 

83 originalFilterLabel = afwImage.FilterLabel(band="i") 

84 originalPhotoCalib = afwImage.PhotoCalib(1.0e5, 1.0e3) 

85 originalExposure.setFilter(originalFilterLabel) 

86 originalExposure.setPhotoCalib(originalPhotoCalib) 

87 

88 warpedExposure1 = warper.warpExposure( 

89 destWcs=swarpedWcs, srcExposure=originalExposure) 

90 # the default size must include all good pixels, so growing the bbox 

91 # should not add any 

92 warpedExposure2 = warper.warpExposure( 

93 destWcs=swarpedWcs, srcExposure=originalExposure, border=1) 

94 # a bit of excess border is allowed, but surely not as much as 10 (in 

95 # fact it is approx. 5) 

96 warpedExposure3 = warper.warpExposure( 

97 destWcs=swarpedWcs, srcExposure=originalExposure, border=-10) 

98 # assert that warpedExposure and warpedExposure2 have the same number of non-no_data pixels 

99 # and that warpedExposure3 has fewer 

100 noDataBitMask = afwImage.Mask.getPlaneBitMask("NO_DATA") 

101 mask1Arr = warpedExposure1.getMaskedImage().getMask().getArray() 

102 mask2Arr = warpedExposure2.getMaskedImage().getMask().getArray() 

103 mask3Arr = warpedExposure3.getMaskedImage().getMask().getArray() 

104 nGood1 = (mask1Arr & noDataBitMask == 0).sum() 

105 nGood2 = (mask2Arr & noDataBitMask == 0).sum() 

106 nGood3 = (mask3Arr & noDataBitMask == 0).sum() 

107 self.assertEqual(nGood1, nGood2) 

108 self.assertLess(nGood3, nGood1) 

109 

110 self.assertEqual(warpedExposure1.getFilter().bandLabel, 

111 originalFilterLabel.bandLabel) 

112 self.assertEqual(warpedExposure1.getPhotoCalib(), originalPhotoCalib) 

113 

114 @unittest.skipIf(dataDir is None, "afwdata not setup") 

115 def testDestBBox(self): 

116 """Test that the destBBox argument works 

117 """ 

118 kernelName = "lanczos2" 

119 warper = afwMath.Warper(kernelName) 

120 originalExposure, swarpedImage, swarpedWcs = self.getSwarpedImage( 

121 kernelName=kernelName, useSubregion=True, useDeepCopy=False) 

122 

123 bbox = lsst.geom.Box2I(lsst.geom.Point2I(100, 25), lsst.geom.Extent2I(3, 7)) 

124 warpedExposure = warper.warpExposure( 

125 destWcs=swarpedWcs, 

126 srcExposure=originalExposure, 

127 destBBox=bbox, 

128 # should be ignored 

129 border=-2, 

130 # should be ignored 

131 maxBBox=lsst.geom.Box2I(lsst.geom.Point2I(1, 2), 

132 lsst.geom.Extent2I(8, 9)), 

133 ) 

134 self.assertEqual(bbox, warpedExposure.getBBox(afwImage.PARENT)) 

135 

136 @unittest.skipIf(dataDir is None, "afwdata not setup") 

137 def getSwarpedImage(self, kernelName, useSubregion=False, useDeepCopy=False): 

138 """ 

139 Inputs: 

140 - kernelName: name of kernel in the form used by afwImage.makeKernel 

141 - useSubregion: if True then the original source exposure (from which the usual 

142 test exposure was extracted) is read and the correct subregion extracted 

143 - useDeepCopy: if True then the copy of the subimage is a deep copy, 

144 else it is a shallow copy; ignored if useSubregion is False 

145 

146 Returns: 

147 - originalExposure 

148 - swarpedImage 

149 - swarpedWcs 

150 """ 

151 if useSubregion: 

152 originalFullExposure = afwImage.ExposureF(originalExposurePath) 

153 # "medsub" is a subregion of med starting at 0-indexed pixel (40, 150) of size 145 x 200 

154 bbox = lsst.geom.Box2I(lsst.geom.Point2I(40, 150), 

155 lsst.geom.Extent2I(145, 200)) 

156 originalExposure = afwImage.ExposureF( 

157 originalFullExposure, bbox, afwImage.LOCAL, useDeepCopy) 

158 swarpedImageName = f"medsubswarp1{kernelName}.fits" 

159 else: 

160 originalExposure = afwImage.ExposureF(originalExposurePath) 

161 swarpedImageName = f"medswarp1{kernelName}.fits" 

162 

163 swarpedImagePath = os.path.join(dataDir, swarpedImageName) 

164 swarpedDecoratedImage = afwImage.DecoratedImageF(swarpedImagePath) 

165 swarpedImage = swarpedDecoratedImage.getImage() 

166 swarpedMetadata = swarpedDecoratedImage.getMetadata() 

167 swarpedWcs = afwGeom.makeSkyWcs(swarpedMetadata) 

168 return (originalExposure, swarpedImage, swarpedWcs) 

169 

170 @unittest.skipIf(dataDir is None, "afwdata not setup") 

171 def compareToSwarp(self, kernelName, 

172 useSubregion=False, useDeepCopy=False, 

173 interpLength=10, cacheSize=100000, 

174 rtol=4e-05, atol=1e-2): 

175 """Compare warpExposure to swarp for given warping kernel. 

176 

177 Note that swarp only warps the image plane, so only test that plane. 

178 

179 Inputs: 

180 - kernelName: name of kernel in the form used by afwImage.makeKernel 

181 - useSubregion: if True then the original source exposure (from which the usual 

182 test exposure was extracted) is read and the correct subregion extracted 

183 - useDeepCopy: if True then the copy of the subimage is a deep copy, 

184 else it is a shallow copy; ignored if useSubregion is False 

185 - interpLength: interpLength argument for lsst.afw.math.warpExposure 

186 - cacheSize: cacheSize argument for lsst.afw.math.SeparableKernel.computeCache; 

187 0 disables the cache 

188 10000 gives some speed improvement but less accurate results (atol must be increased) 

189 100000 gives better accuracy but no speed improvement in this test 

190 - rtol: relative tolerance as used by numpy.allclose 

191 - atol: absolute tolerance as used by numpy.allclose 

192 """ 

193 warper = afwMath.Warper(kernelName) 

194 

195 originalExposure, swarpedImage, swarpedWcs = self.getSwarpedImage( 

196 kernelName=kernelName, useSubregion=useSubregion, useDeepCopy=useDeepCopy) 

197 maxBBox = lsst.geom.Box2I( 

198 lsst.geom.Point2I(swarpedImage.getX0(), swarpedImage.getY0()), 

199 lsst.geom.Extent2I(swarpedImage.getWidth(), swarpedImage.getHeight())) 

200 

201 # warning: this test assumes that the swarped image is smaller than it needs to be 

202 # to hold all of the warped pixels 

203 afwWarpedExposure = warper.warpExposure( 

204 destWcs=swarpedWcs, 

205 srcExposure=originalExposure, 

206 maxBBox=maxBBox, 

207 ) 

208 afwWarpedMaskedImage = afwWarpedExposure.getMaskedImage() 

209 

210 afwWarpedMask = afwWarpedMaskedImage.getMask() 

211 noDataBitMask = afwImage.Mask.getPlaneBitMask("NO_DATA") 

212 noDataMask = afwWarpedMask.getArray() & noDataBitMask 

213 

214 msg = "afw and swarp %s-warped %s (ignoring bad pixels)" 

215 self.assertImagesAlmostEqual(afwWarpedMaskedImage.getImage(), swarpedImage, 

216 skipMask=noDataMask, rtol=rtol, atol=atol, msg=msg) 

217 

218 

219class MemoryTester(lsst.utils.tests.MemoryTestCase): 

220 pass 

221 

222 

223def setup_module(module): 

224 lsst.utils.tests.init() 

225 

226 

227if __name__ == "__main__": 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true

228 lsst.utils.tests.init() 

229 unittest.main()