Coverage for tests / test_modelconfig.py: 29%

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 lsst.gauss2d as g2 

23import lsst.gauss2d.fit as g2f 

24from lsst.multiprofit.componentconfig import ( 

25 CentroidConfig, 

26 GaussianComponentConfig, 

27 ParameterConfig, 

28 SersicComponentConfig, 

29 SersicIndexParameterConfig, 

30) 

31from lsst.multiprofit.modelconfig import ModelConfig 

32from lsst.multiprofit.observationconfig import ObservationConfig 

33from lsst.multiprofit.sourceconfig import ComponentGroupConfig, SourceConfig 

34import numpy as np 

35import pytest 

36 

37 

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

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

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

41 

42 

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

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

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

46 observations = [] 

47 for band in channels: 

48 config.band = band 

49 observations.append(config.make_observation()) 

50 return g2f.DataD(observations) 

51 

52 

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

54def psf_model(): 

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

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

57 

58 n_components = 3 

59 flux_total = 2.0 * (n_components + 1) 

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

61 

62 config = SourceConfig( 

63 component_groups={ 

64 "src": ComponentGroupConfig( 

65 components_gauss={ 

66 str(idx): GaussianComponentConfig( 

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

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

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

70 ) 

71 for idx in range(n_components) 

72 }, 

73 ) 

74 }, 

75 ) 

76 config.validate() 

77 channel = g2f.Channel.NONE 

78 psf_model, priors = config.make_psf_model( 

79 [ 

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

81 ], 

82 ) 

83 return psf_model 

84 

85 

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

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

88 return [psf_model] * len(channels) 

89 

90 

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

92def modelconfig_fluxes(channels): 

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

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

95 

96 components_sersic = {} 

97 fluxes_mix = [] 

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

99 components_sersic[name] = SersicComponentConfig( 

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

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

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

103 sersic_index=SersicIndexParameterConfig( 

104 value_initial=sersicn + idx * dsersicn, 

105 fixed=idx == 0, 

106 prior_mean=None, 

107 ), 

108 ) 

109 fluxes_comp = { 

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

111 } 

112 fluxes_mix.append(fluxes_comp) 

113 

114 modelconfig = ModelConfig( 

115 sources={ 

116 "src": SourceConfig( 

117 component_groups={ 

118 "mix": ComponentGroupConfig( 

119 centroids={ 

120 "default": CentroidConfig( 

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

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

123 ), 

124 }, 

125 components_sersic=components_sersic, 

126 ), 

127 } 

128 ), 

129 }, 

130 ) 

131 return modelconfig, fluxes_mix 

132 

133 

134def test_ModelConfig(modelconfig_fluxes, data, psf_models): 

135 modelconfig, fluxes = modelconfig_fluxes 

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

137 assert model is not None 

138 assert model.data is data 

139 for observation in model.data: 

140 observation.sigma_inv.fill(1.0) 

141 observation.mask_inv.fill(1) 

142 

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

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

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

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

147 model.evaluate() 

148 model.setup_evaluators(g2f.EvaluatorMode.loglike) 

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