Coverage for tests / test_make_coadd_input_summary.py: 17%

1# This file is part of drp_tasks. 

2# 

3# LSST Data Management System 

4# This product includes software developed by the 

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

6# See COPYRIGHT file at the top of the source tree. 

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 <https://www.lsstcorp.org/LegalNotices/>. 

21 

22import unittest 

23 

24import numpy as np 

25 

26from lsst.drp.tasks.make_coadd_input_summary import ( 

27 MakeCoaddInputSummary, 

28 MakeCoaddInputSummaryTract, 

29) 

30from lsst.pipe.base import InMemoryDatasetHandle 

31from lsst.pipe.tasks.coaddInputRecorder import CoaddInputRecorderTask 

32 

33 

34class MakeCoaddInputSummaryTestCase(unittest.TestCase): 

35 def setUp(self): 

36 self.recorder_task = CoaddInputRecorderTask(name="recorder") 

37 

38 def _make_coadd_input_handle(self, tract, patch, band, n_input): 

39 coadd_inputs = self.recorder_task.makeCoaddInputs() 

40 

41 coadd_inputs.ccds.resize(n_input) 

42 coadd_inputs.ccds["id"] = np.arange(n_input) 

43 coadd_inputs.ccds["ccd"] = np.arange(n_input) 

44 coadd_inputs.ccds["visit"] = np.arange(100, 100 + n_input) 

45 coadd_inputs.ccds["goodpix"][:] = 1000 

46 coadd_inputs.ccds["weight"][:] = 1.0 

47 

48 handle = InMemoryDatasetHandle( 

49 coadd_inputs, 

50 dataId={"tract": tract, "patch": patch, "band": band}, 

51 storageClass="CoaddInputs", 

52 ) 

53 

54 return handle 

55 

56 def test_make_coadd_input_summary_tract(self): 

57 handles = [self._make_coadd_input_handle(100, patch, "g", 10) for patch in [0, 10, 20]] 

58 

59 task = MakeCoaddInputSummaryTract() 

60 summary_tract = task.run(coadd_inputs_handles=handles).coadd_input_summary_tract 

61 

62 self.assertEqual(len(summary_tract), 10 * 3) 

63 np.testing.assert_array_equal(summary_tract["tract"], 100) 

64 np.testing.assert_array_equal(summary_tract["band"], "g") 

65 

66 index = 0 

67 for handle in handles: 

68 inputs = handle.get().ccds 

69 

70 np.testing.assert_array_equal( 

71 summary_tract["patch"][index : index + len(inputs)], 

72 handle.dataId["patch"], 

73 ) 

74 np.testing.assert_array_equal( 

75 summary_tract["visit"][index : index + len(inputs)], 

76 inputs["visit"], 

77 ) 

78 np.testing.assert_array_equal( 

79 summary_tract["detector"][index : index + len(inputs)], 

80 inputs["ccd"], 

81 ) 

82 np.testing.assert_array_equal( 

83 summary_tract["goodpix"][index : index + len(inputs)], 

84 inputs["goodpix"], 

85 ) 

86 np.testing.assert_array_almost_equal( 

87 summary_tract["weight"][index : index + len(inputs)], 

88 inputs["weight"], 

89 ) 

90 index += len(inputs) 

91 

92 def test_make_coadd_input_summary(self): 

93 task1 = MakeCoaddInputSummaryTract() 

94 

95 summary_tract_handles = [] 

96 

97 for tract in [100, 200]: 

98 handles = [self._make_coadd_input_handle(tract, patch, "g", 10) for patch in [0, 10, 20]] 

99 

100 summary_tract = task1.run(coadd_inputs_handles=handles).coadd_input_summary_tract 

101 

102 summary_tract_handles.append( 

103 InMemoryDatasetHandle( 

104 summary_tract, 

105 dataId={"tract": tract, "band": "g"}, 

106 storageClass="ArrowAstropy", 

107 ), 

108 ) 

109 

110 task2 = MakeCoaddInputSummary() 

111 summary = task2.run(coadd_input_summary_tract_handles=summary_tract_handles).coadd_input_summary 

112 

113 self.assertEqual(len(summary), 2 * 10 * 3) 

114 

115 index = 0 

116 for handle in summary_tract_handles: 

117 summary_tract = handle.get() 

118 

119 for name in ["tract", "patch", "visit", "detector", "goodpix"]: 

120 np.testing.assert_array_equal( 

121 summary[name][index : index + len(summary_tract)], 

122 summary_tract[name], 

123 ) 

124 

125 np.testing.assert_array_almost_equal( 

126 summary["weight"][index : index + len(summary_tract)], 

127 summary_tract["weight"], 

128 ) 

129 

130 index += len(summary_tract) 

131 

132 

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

134 unittest.main()