Coverage for tests/test_modelconfig.py: 100%

1# This file is part of multiprofit. 

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 

23import pytest 

24 

25import lsst.gauss2d as g2 

26import lsst.gauss2d.fit as g2f 

27from lsst.multiprofit.componentconfig import ( 

28 CentroidConfig, 

29 GaussianComponentConfig, 

30 ParameterConfig, 

31 SersicComponentConfig, 

32 SersicIndexParameterConfig, 

33) 

34from lsst.multiprofit.modelconfig import ModelConfig 

35from lsst.multiprofit.observationconfig import ObservationConfig 

36from lsst.multiprofit.sourceconfig import ComponentGroupConfig, SourceConfig 

37 

38 

39@pytest.fixture(scope="module") 

40def channels() -> dict[str, g2f.Channel]: 

41 """Return dict of generic RGB channels.""" 

42 return {band: g2f.Channel.get(band) for band in ("R", "G", "B")} 

43 

44 

45@pytest.fixture(scope="module") 

46def data(channels) -> g2f.DataD: 

47 """Return initialized data in all bands.""" 

48 config = ObservationConfig(n_rows=13, n_cols=19) 

49 observations = [] 

50 for band in channels: 

51 config.band = band 

52 observations.append(config.make_observation()) 

53 return g2f.DataD(observations) 

54 

55 

56@pytest.fixture(scope="module") 

57def psf_model(): 

58 """Return a triple Gaussian PSF model.""" 

59 rho, size_x, size_y = 0.25, 1.6, 1.2 

60 drho, dsize_x, dsize_y = -0.4, 1.1, 1.9 

61 

62 n_components = 3 

63 flux_total = 2.0 * (n_components + 1) 

64 fluxes = [x / flux_total for x in range(1, 1 + n_components)] 

65 

66 config = SourceConfig( 

67 component_groups={ 

68 "src": ComponentGroupConfig( 

69 components_gauss={ 

70 str(idx): GaussianComponentConfig( 

71 rho=ParameterConfig(value_initial=rho + idx * drho), 

72 size_x=ParameterConfig(value_initial=size_x + idx * dsize_x), 

73 size_y=ParameterConfig(value_initial=size_y + idx * dsize_y), 

74 ) 

75 for idx in range(n_components) 

76 }, 

77 ) 

78 }, 

79 ) 

80 config.validate() 

81 channel = g2f.Channel.NONE 

82 psf_model, priors = config.make_psf_model( 

83 [ 

84 [{channel: flux} for flux in fluxes], 

85 ], 

86 ) 

87 return psf_model 

88 

89 

90@pytest.fixture(scope="module") 

91def psf_models(psf_model, channels) -> list[g2f.PsfModel]: 

92 """Return a list of PSF models per channel.""" 

93 return [psf_model] * len(channels) 

94 

95 

96@pytest.fixture(scope="module") 

97def modelconfig_fluxes(channels): 

98 """Return model configs and flux values.""" 

99 rho, size_x, size_y, sersicn, flux = 0.4, 1.5, 1.9, 0.5, 4.7 

100 drho, dsize_x, dsize_y, dsersicn, dflux = -0.9, 2.5, 5.4, 2.8, 13.9 

101 

102 components_sersic = {} 

103 fluxes_mix = [] 

104 for idx, name in enumerate(("PS", "Sersic")): 

105 components_sersic[name] = SersicComponentConfig( 

106 rho=ParameterConfig(value_initial=rho + idx * drho), 

107 size_x=ParameterConfig(value_initial=size_x + idx * dsize_x), 

108 size_y=ParameterConfig(value_initial=size_y + idx * dsize_y), 

109 sersic_index=SersicIndexParameterConfig( 

110 value_initial=sersicn + idx * dsersicn, 

111 fixed=idx == 0, 

112 prior_mean=None, 

113 ), 

114 ) 

115 fluxes_comp = { 

116 channel: flux + idx_channel * dflux * idx for idx_channel, channel in enumerate(channels.values()) 

117 } 

118 fluxes_mix.append(fluxes_comp) 

119 

120 modelconfig = ModelConfig( 

121 sources={ 

122 "src": SourceConfig( 

123 component_groups={ 

124 "mix": ComponentGroupConfig( 

125 centroids={ 

126 "default": CentroidConfig( 

127 x=ParameterConfig(value_initial=15.8, fixed=True), 

128 y=ParameterConfig(value_initial=14.3, fixed=False), 

129 ), 

130 }, 

131 components_sersic=components_sersic, 

132 ), 

133 } 

134 ), 

135 }, 

136 ) 

137 return modelconfig, fluxes_mix 

138 

139 

140def test_ModelConfig(modelconfig_fluxes, data, psf_models): 

141 """Test ModelConfig init and funcs.""" 

142 modelconfig, fluxes = modelconfig_fluxes 

143 model = modelconfig.make_model([[fluxes]], data=data, psf_models=psf_models) 

144 assert model is not None 

145 assert model.data is data 

146 for observation in model.data: 

147 observation.sigma_inv.fill(1.0) 

148 observation.mask_inv.fill(1) 

149 

150 # Set the outputs to new images that refer to the existing data 

151 # because obs.image will not return a holding pointer 

152 outputs = [[g2.ImageD(obs.image.data)] for obs in model.data] 

153 model.setup_evaluators(g2f.EvaluatorMode.image, outputs=outputs) 

154 model.evaluate() 

155 model.setup_evaluators(g2f.EvaluatorMode.loglike) 

156 assert np.sum(model.evaluate()) == 0