Coverage for tests / test_transformObject.py: 19%

1# This file is part of pipe_tasks. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://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 <http://www.gnu.org/licenses/>. 

21 

22import os 

23import unittest 

24import astropy.table 

25import pandas as pd 

26import numpy as np 

27 

28import lsst.utils.tests 

29 

30from lsst.pipe.base import InMemoryDatasetHandle 

31from lsst.pipe.tasks.functors import CoordColumn, Column 

32from lsst.pipe.tasks.postprocess import TransformObjectCatalogTask, TransformObjectCatalogConfig 

33 

34ROOT = os.path.abspath(os.path.dirname(__file__)) 

35 

36 

37def setup_module(module): 

38 lsst.utils.tests.init() 

39 

40 

41class TransformObjectCatalogTestCase(unittest.TestCase): 

42 def setUp(self): 

43 # Note that this test input includes HSC-G, HSC-R, and HSC-I data 

44 df = pd.read_csv(os.path.join(ROOT, 'data', 'test_multilevel_parq.csv.gz'), 

45 header=[0, 1, 2], index_col=0) 

46 

47 self.dataId = {"tract": 9615, "patch": "4,4"} 

48 self.handle = InMemoryDatasetHandle(df, storageClass="DataFrame", dataId=self.dataId) 

49 n_rows = len(df) 

50 tab_epoch = astropy.table.Table({df.index.name: df.index, "r_epoch": [0.]*n_rows}) 

51 tab_ref = astropy.table.Table({df.index.name: df.index, "refBand": ["r"]*n_rows}) 

52 tab_exp = astropy.table.Table({df.index.name: df.index, "exp_n_iter": [0] * n_rows}) 

53 tab_sersic = astropy.table.Table({df.index.name: df.index, "sersic_n_iter": [0] * n_rows}) 

54 self.kwargs_task = { 

55 "handle_epoch": InMemoryDatasetHandle( 

56 tab_epoch, storageClass="ArrowAstropy", dataId=self.dataId, 

57 ), 

58 "handle_ref": InMemoryDatasetHandle( 

59 tab_ref, storageClass="ArrowAstropy", dataId=self.dataId, 

60 ), 

61 "handle_Exp_multiprofit": InMemoryDatasetHandle( 

62 tab_exp, storageClass="ArrowAstropy", dataId=self.dataId, 

63 ), 

64 "handle_Sersic_multiprofit": InMemoryDatasetHandle( 

65 tab_sersic, storageClass="ArrowAstropy", dataId=self.dataId, 

66 ), 

67 } 

68 self.funcs_multi = { 

69 "epoch": Column("r_epoch", dataset="epoch"), 

70 "refBand": Column("refBand", dataset="ref"), 

71 "exp_n_iter": Column("exp_n_iter", dataset="Exp_multiprofit"), 

72 "sersic_n_iter": Column("sersic_n_iter", dataset="Sersic_multiprofit"), 

73 } 

74 

75 def testNullFilter(self): 

76 """Test that columns for all filters are created despite they may not 

77 exist in the input data. 

78 """ 

79 config = TransformObjectCatalogConfig() 

80 config.camelCase = True 

81 # Want y band columns despite the input data do not have them 

82 # Exclude g band columns despite the input data have them 

83 config.outputBands = ["r", "i", "y"] 

84 # Arbitrarily choose a boolean flag column to be "good" 

85 config.goodFlags = ['GoodFlagColumn'] 

86 task = TransformObjectCatalogTask(config=config) 

87 # Add in a float column, an integer column, a good flag, and 

88 # a bad flag. It does not matter which columns we choose, just 

89 # that they have the appropriate type. 

90 funcs = {'FloatColumn': CoordColumn('coord_ra', dataset='meas'), 

91 'IntColumn': Column('base_InputCount_value', dataset='meas'), 

92 'GoodFlagColumn': Column('slot_GaussianFlux_flag', dataset='meas'), 

93 'BadFlagColumn': Column('slot_Centroid_flag', dataset='meas')} 

94 funcs.update(self.funcs_multi) 

95 tbl = task.run(self.handle, funcs=funcs, dataId=self.dataId, **self.kwargs_task).outputCatalog 

96 self.assertIsInstance(tbl, astropy.table.Table) 

97 

98 for filt in config.outputBands: 

99 self.assertIn(filt + 'FloatColumn', tbl.columns) 

100 self.assertIn(filt + 'IntColumn', tbl.columns) 

101 self.assertIn(filt + 'BadFlagColumn', tbl.columns) 

102 self.assertIn(filt + 'GoodFlagColumn', tbl.columns) 

103 

104 # Check that the default filling has worked. 

105 self.assertNotIn('gFloatColumn', tbl.columns) 

106 self.assertTrue(np.all(np.ma.is_masked(tbl['yFloatColumn']))) 

107 self.assertFalse(np.any(np.ma.is_masked(tbl['iFloatColumn']))) 

108 self.assertTrue(np.all(tbl['iIntColumn'] >= 0)) 

109 self.assertTrue(np.all(tbl['yIntColumn'] < 0)) 

110 self.assertTrue(np.all(~tbl['yGoodFlagColumn'])) 

111 self.assertTrue(np.all(tbl['yBadFlagColumn'])) 

112 

113 # Check that the datatypes are preserved. 

114 self.assertEqual(tbl['iFloatColumn'].dtype, np.dtype(np.float64)) 

115 self.assertEqual(tbl['yFloatColumn'].dtype, np.dtype(np.float64)) 

116 self.assertEqual(tbl['iIntColumn'].dtype, np.dtype(np.int64)) 

117 self.assertEqual(tbl['yIntColumn'].dtype, np.dtype(np.int64)) 

118 self.assertEqual(tbl['iGoodFlagColumn'].dtype, np.dtype(np.bool_)) 

119 self.assertEqual(tbl['yGoodFlagColumn'].dtype, np.dtype(np.bool_)) 

120 self.assertEqual(tbl['iBadFlagColumn'].dtype, np.dtype(np.bool_)) 

121 self.assertEqual(tbl['yBadFlagColumn'].dtype, np.dtype(np.bool_)) 

122 

123 def testUnderscoreColumnFormat(self): 

124 """Test the per-filter column format with an underscore""" 

125 config = TransformObjectCatalogConfig() 

126 config.outputBands = ["g", "r", "i"] 

127 config.camelCase = False 

128 task = TransformObjectCatalogTask(config=config) 

129 funcs = {'ra': CoordColumn('coord_ra', dataset='meas')} 

130 funcs.update(self.funcs_multi) 

131 tbl = task.run(self.handle, funcs=funcs, dataId=self.dataId, **self.kwargs_task).outputCatalog 

132 self.assertIsInstance(tbl, astropy.table.Table) 

133 for filt in config.outputBands: 

134 self.assertIn(filt + '_ra', tbl.columns) 

135 

136 def testMultilevelOutput(self): 

137 """Test the non-flattened result dataframe with a multilevel column index""" 

138 config = TransformObjectCatalogConfig() 

139 config.outputBands = ["r", "i"] 

140 config.multilevelOutput = True 

141 task = TransformObjectCatalogTask(config=config) 

142 funcs = {'ra': CoordColumn('coord_ra', dataset='meas')} 

143 funcs.update(self.funcs_multi) 

144 df = task.run(self.handle, funcs=funcs, dataId=self.dataId, **self.kwargs_task).outputCatalog 

145 self.assertIsInstance(df, pd.DataFrame) 

146 self.assertNotIn('g', df) 

147 for filt in config.outputBands: 

148 self.assertIsInstance(df[filt], pd.DataFrame) 

149 self.assertIn('ra', df[filt].columns) 

150 

151 def testNoOutputBands(self): 

152 """All the input bands should go into the output, and nothing else. 

153 """ 

154 config = TransformObjectCatalogConfig() 

155 task = TransformObjectCatalogTask(config=config) 

156 funcs = {'ra': CoordColumn('coord_ra', dataset='meas')} 

157 funcs.update(self.funcs_multi) 

158 tbl = task.run(self.handle, funcs=funcs, dataId=self.dataId, **self.kwargs_task).outputCatalog 

159 self.assertIsInstance(tbl, astropy.table.Table) 

160 self.assertNotIn('HSC-G_Fwhm', tbl.columns) 

161 for filt in ['g', 'r', 'i']: 

162 self.assertIn(f'{filt}_ra', tbl.columns) 

163 

164 

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

166 lsst.utils.tests.init() 

167 unittest.main()