Coverage for tests / test_postprocess.py: 25%

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# (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 pytest 

23import unittest 

24 

25import lsst.utils.tests 

26from lsst.pipe.tasks.postprocess import ModelExtendednessColumnAction 

27 

28from astropy.table import Table 

29import numpy as np 

30 

31 

32class ModelExtendednessColumnActionTestCase(lsst.utils.tests.TestCase): 

33 """Demo test case.""" 

34 

35 def setUp(self): 

36 self.bands = ("g", "r", "i") 

37 action = ModelExtendednessColumnAction(bands=self.bands, min_n_good_to_shift_flux_ratio=1) 

38 self.action = action 

39 data = { 

40 action.size_column.format(axis="x"): [1., 3., 0.01, 0.4, 0.02], 

41 action.size_column.format(axis="y"): [0.2, 6, 0.01, 0.2, 0.01], 

42 } 

43 model = action.model_flux_name 

44 factors = np.array([1.01, 1.12, 0.995, 0.998, 1.6]) 

45 fluxes = np.array([1.5e3, 2.5e3, 6.8e3, 3.4e3, 5.5e3]) 

46 for column_flux, column_flux_err, factors in ( 

47 (action.model_column_flux, action.model_column_flux_err, factors), 

48 (action.psf_column_flux, action.psf_column_flux_err, None), 

49 ): 

50 for idx, band in enumerate(action.bands): 

51 flux = np.sqrt(idx + 1.)*fluxes 

52 if factors is not None: 

53 flux *= factors 

54 data[column_flux.format(band=band, model=model)] = flux 

55 data[column_flux_err.format(band=band, model=model)] = np.sqrt(flux) 

56 self.data = Table(data) 

57 

58 def testExtendednessColumnAction(self): 

59 action = self.action 

60 with pytest.raises(ValueError): 

61 action.validate() 

62 action.bands_combined = {"gri": "g,r,i"} 

63 action.validate() 

64 schema = action.getInputSchema() 

65 n_values = len(self.data[schema[0][0]]) 

66 assert all([len(self.data[col]) == n_values for col, _ in schema[1:]]) 

67 

68 result = self.action(self.data) 

69 columns_expected = [ 

70 action.output_column.format(band=band) 

71 for band in list(action.bands) + list(action.bands_combined.keys()) 

72 ] 

73 assert list(result.keys()) == columns_expected 

74 

75 for column, values in result.items(): 

76 assert len(values) == n_values 

77 assert all((values >= 0) & (values <= 1)) 

78 

79 

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

81 pass 

82 

83 

84def setup_module(module): 

85 lsst.utils.tests.init() 

86 

87 

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

89 lsst.utils.tests.init() 

90 unittest.main()