Coverage for tests / test_ssoAssociationTask.py: 22%

1# This file is part of ap_association. 

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 <https://www.gnu.org/licenses/>. 

21 

22import numpy as np 

23from astropy.table import Table 

24import astropy.table as tb 

25import unittest 

26 

27from lsst.pipe.tasks.ssoAssociation import SolarSystemAssociationTask 

28import astshim as ast 

29import lsst.geom as geom 

30import lsst.afw.geom as afwGeom 

31import lsst.meas.base.tests as measTests 

32import lsst.utils.tests 

33 

34 

35class TestSolarSystemAssociation(unittest.TestCase): 

36 

37 def setUp(self): 

38 # Make fake sources 

39 self.nSources = 10 

40 self.bbox = geom.Box2I(geom.Point2I(0, 0), 

41 geom.Extent2I(1024, 1153)) 

42 self.xyLoc = 100 

43 dataset = measTests.TestDataset(self.bbox) 

44 for srcIdx in range(self.nSources): 

45 dataset.addSource(100000.0, 

46 geom.Point2D(srcIdx*self.xyLoc, 

47 srcIdx*self.xyLoc)) 

48 schema = dataset.makeMinimalSchema() 

49 schema.addField("base_PixelFlags_flag", type="Flag") 

50 schema.addField("base_PixelFlags_flag_offimage", type="Flag") 

51 self.exposure, catalog = dataset.realize( 

52 10.0, schema, randomSeed=1234) 

53 for src in catalog: 

54 src.setCoord(self.exposure.getWcs().pixelToSky(src.getCentroid())) 

55 # Non-invertible WCS to test robustness to distortions. 

56 # Coefficients transform (x - 1e-7*x^3 -> x, y -> y); see docs for PolyMap. 

57 pixelCoeffs = np.array([[-1.0e-7, 1, 3, 0], 

58 [1.0, 1, 1, 0], 

59 [1.0, 2, 0, 1], 

60 ]) 

61 self.exposure.setWcs(afwGeom.makeModifiedWcs( 

62 afwGeom.TransformPoint2ToPoint2(ast.PolyMap(pixelCoeffs, 2, options="IterInverse=1")), 

63 self.exposure.wcs, 

64 modifyActualPixels=False 

65 )) 

66 

67 # Convert to task required format 

68 self.testDiaSources = catalog.asAstropy() 

69 self.testDiaSources.rename_columns(["coord_ra", "coord_dec"], ["ra", "dec"]) 

70 self.testDiaSources["ra"] = np.rad2deg(self.testDiaSources["ra"]) 

71 self.testDiaSources["dec"] = np.rad2deg(self.testDiaSources["dec"]) 

72 self.testDiaSources["midpointMjdTai"] = self.testDiaSources["ra"]*0 + 65000.0 

73 self.testDiaSources["extendedness"] = np.zeros(len(self.testDiaSources)) 

74 self.testDiaSources["band"] = np.full(len(self.testDiaSources), 'r', dtype='U1') 

75 self.testDiaSources["psfFlux"] = np.ones(len(self.testDiaSources)) * 3. 

76 self.testDiaSources["psfFluxErr"] = np.ones(len(self.testDiaSources)) * 0.1 

77 self.testDiaSources = tb.vstack([self.testDiaSources, Table([[45], [45]], names=["ra", "dec"])]) 

78 self.testDiaSources["ssObjectId"] = 0 

79 self.testDiaSources["diaSourceId"] = [i for i in range(len(self.testDiaSources))] 

80 

81 # Grab a subset to treat as solar system objects 

82 self.testSsObjects = Table() 

83 self.testSsObjects["ObjID"] = ["K15MK4Q"] 

84 self.testSsObjects["MPCORB_unpacked_primary_provisional_designation"] = self.testSsObjects["ObjID"] 

85 

86 self.testSsObjects["ssObjectId"] = [21164728253101137] 

87 self.testSsObjects["ephRa"] = [45] 

88 self.testSsObjects["ephDec"] = [45] 

89 self.testSsObjects["tmin"] = [-1] 

90 self.testSsObjects["tmax"] = [1] 

91 self.testSsObjects["obs_x_poly"] = [np.array([0])] 

92 self.testSsObjects["obs_y_poly"] = [np.array([0])] 

93 self.testSsObjects["obs_z_poly"] = [np.array([0])] 

94 self.testSsObjects["obj_x_poly"] = [np.array([1])] 

95 self.testSsObjects["obj_y_poly"] = [np.array([1])] 

96 self.testSsObjects["obj_z_poly"] = [np.array([2 ** 0.5])] 

97 self.testSsObjects["Err(arcsec)"] = np.ones(len(self.testSsObjects)) 

98 self.testSsObjects["trailedSourceMagTrue"] = 22 

99 

100 def test_run(self): 

101 """Test that association and id assignment work as expected. 

102 """ 

103 ssAssocTask = SolarSystemAssociationTask() 

104 results = ssAssocTask.run(self.testDiaSources, 

105 self.testSsObjects, 

106 self.exposure.visitInfo, 

107 self.exposure.getBBox(), 

108 self.exposure.wcs) 

109 self.assertEqual(len(results.ssoAssocDiaSources), 1) 

110 self.assertEqual(results.ssoAssocDiaSources['ra'][0], 45.0) 

111 self.assertEqual(results.ssoAssocDiaSources['dec'][0], 45.0) 

112 self.assertEqual(results.ssoAssocDiaSources['ssObjectId'][0], 21164728253101137) 

113 

114 def test_mask(self): 

115 """Test that masking against the CCD bounding box works as expected. 

116 """ 

117 ssAssocTask = SolarSystemAssociationTask() 

118 # Test will all inside ccd 

119 maskedObjects = ssAssocTask._maskToCcdRegion(self.testSsObjects, 

120 self.exposure.getBBox(), 

121 self.exposure.wcs, 

122 1.0) 

123 self.assertEqual(len(maskedObjects), len(self.testSsObjects)) 

124 

125 # Add a new SolarSystemObjects outside of the bbox and test that it 

126 # is excluded. 

127 testObjects = tb.vstack([self.testSsObjects[:1] for i in range(3)]) 

128 testObjects[0]["ephRa"] = 150 

129 testObjects[0]["ephDec"] = 80 

130 testObjects[1]["ephRa"] = 150 

131 testObjects[1]["ephDec"] = -80 

132 # Coordinates are chosen so that the inverse WCS erroneously maps them 

133 # to inside the box (to (74.5, 600.6) instead of around (1745, 600.6)). 

134 testObjects[2]["ephRa"] = 44.91215199831453 

135 testObjects[2]["ephDec"] = 45.001331943391406 

136 maskedObjects = ssAssocTask._maskToCcdRegion( 

137 tb.vstack([self.testSsObjects, testObjects]), 

138 self.exposure.getBBox(), 

139 self.exposure.wcs, 

140 1.0) 

141 self.assertEqual(len(maskedObjects), len(self.testSsObjects)) 

142 

143 

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

145 pass 

146 

147 

148def setup_module(module): 

149 lsst.utils.tests.init() 

150 

151 

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

153 lsst.utils.tests.init() 

154 unittest.main()