Coverage for tests/test_fit_coadd.py: 61%
105 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 10:40 +0000
1# This file is part of meas_extensions_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/>.
22import os
24import numpy as np
25import pytest
26from astropy.table import Table
28import lsst.gauss2d.fit as g2f
29import lsst.meas.extensions.multiprofit.fit_coadd_multiband as fitCMB
30import lsst.meas.extensions.multiprofit.fit_coadd_psf as fitCP
31from lsst.afw.image import ExposureF
32from lsst.afw.table import SourceCatalog
33from lsst.daf.butler.formatters.parquet import arrow_to_astropy
34from lsst.multiprofit.componentconfig import (
35 CentroidConfig,
36 GaussianComponentConfig,
37 ParameterConfig,
38 SersicComponentConfig,
39 SersicIndexParameterConfig,
40)
41from lsst.multiprofit.fitting.fit_psf import CatalogPsfFitterConfig
42from lsst.multiprofit.modelconfig import ModelConfig
43from lsst.multiprofit.sourceconfig import ComponentGroupConfig, SourceConfig
44from lsst.pipe.tasks.fit_coadd_psf import CatalogExposurePsf
46# If this env var is set, some additional convenience tests will be run
47# on an exposure-measurement catalog pair assumed to be from imsim.
48# This should be faster than debugging any CI package.
49# If this env var is not set, the optional tests do nothing.
50ROOT = os.environ.get("TESTDATA_CI_IMSIM_MINI_DIR", None)
51has_files = (ROOT is not None) and os.path.isdir(ROOT)
53# These default names assume the files came from ci_imsim run with the
54# --config-use-skymap-small flag, but any other patch could be used.
55filename_cat = os.path.join(ROOT, "data", "deepCoadd_meas_0_24_r_2k_ci_imsim.fits") if has_files else None
56filename_exp = os.path.join(ROOT, "data", "deepCoadd_calexp_0_24_r_2k_ci_imsim.fits") if has_files else None
58band = "r"
59channel = g2f.Channel.get(band)
60dataId = {"band": band}
61do_exp_fixedcen = False
62include_ps = False
63n_test = 5
66@pytest.fixture(scope="module")
67def catalog():
68 """Return a source catalog if the mini imsim path exists."""
69 if not has_files: 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was always true
70 return None
71 catalog = SourceCatalog.readFits(filename_cat)
72 good = (catalog["detect_isPrimary"] == 1) & (catalog["merge_peak_sky"] == 0)
73 good[np.where(good)[0][n_test:]] = False
74 return catalog[good]
77@pytest.fixture(scope="module")
78def exposure():
79 """Return exposures if available."""
80 if not has_files: 80 ↛ 82line 80 didn't jump to line 82 because the condition on line 80 was always true
81 return None
82 return ExposureF.readFits(filename_exp)
85@pytest.fixture(scope="module")
86def psf_fit_config():
87 """Return a default PSF fitting config."""
88 return fitCP.MultiProFitPsfConfig()
91@pytest.fixture(scope="module")
92def psf_fit_results(catalog, exposure, psf_fit_config):
93 """Fit the imsim PSFs if available."""
94 if not has_files: 94 ↛ 96line 94 didn't jump to line 96 because the condition on line 94 was always true
95 return None
96 catexp = CatalogExposurePsf(dataId=dataId, catalog=catalog, exposure=exposure)
97 task = fitCP.MultiProFitPsfTask(config=psf_fit_config)
98 results = task.run(catexp).output
99 return arrow_to_astropy(results)
102@pytest.fixture(scope="module")
103def source_fit_exp_fixedcen_config():
104 """Return a fixed-centroid exponential source fit config."""
105 config = fitCMB.MultiProFitSourceConfig(
106 bands_fit=(band,),
107 config_model=ModelConfig(
108 sources={
109 "": SourceConfig(
110 component_groups={
111 "": ComponentGroupConfig(
112 centroids={
113 "default": CentroidConfig(
114 x=ParameterConfig(fixed=True),
115 y=ParameterConfig(fixed=True),
116 )
117 },
118 components_sersic={
119 "exp": SersicComponentConfig(
120 sersic_index=SersicIndexParameterConfig(value_initial=1.0, fixed=True),
121 )
122 },
123 ),
124 }
125 ),
126 },
127 ),
128 )
129 config.validate()
130 return config
133@pytest.fixture(scope="module")
134def source_fit_ser_config():
135 """Return a Sersic source fit config."""
136 config = fitCMB.MultiProFitSourceConfig(
137 bands_fit=(band,),
138 config_model=ModelConfig(
139 sources={
140 "": SourceConfig(
141 component_groups={
142 "": ComponentGroupConfig(
143 components_gauss=(
144 {
145 "ps": GaussianComponentConfig(
146 size_x=ParameterConfig(value_initial=0.0, fixed=True),
147 size_y=ParameterConfig(value_initial=0.0, fixed=True),
148 rho=ParameterConfig(value_initial=0.0, fixed=True),
149 )
150 }
151 if include_ps
152 else {}
153 ),
154 components_sersic={
155 "ser": SersicComponentConfig(
156 sersic_index=SersicIndexParameterConfig(value_initial=1.0),
157 )
158 },
159 ),
160 }
161 ),
162 },
163 ),
164 )
165 config.validate()
166 return config
169@pytest.fixture(scope="module")
170def source_fit_exp_fixedcen_results(
171 catalog,
172 exposure,
173 psf_fit_results,
174 psf_fit_config,
175 source_fit_exp_fixedcen_config,
176) -> Table:
177 """Return the exponential fit results if data exists."""
178 if not has_files: 178 ↛ 180line 178 didn't jump to line 180 because the condition on line 178 was always true
179 return None
180 if not do_exp_fixedcen:
181 return None
182 catexp = fitCMB.CatalogExposurePsfs(
183 dataId=dataId,
184 catalog=catalog,
185 exposure=exposure,
186 table_psf_fits=psf_fit_results,
187 channel=channel,
188 config_fit=source_fit_exp_fixedcen_config,
189 )
190 task = fitCMB.MultiProFitSourceTask(config=source_fit_exp_fixedcen_config)
191 results = task.run(catalog_multi=catalog, catexps=[catexp])
192 return arrow_to_astropy(results.output)
195@pytest.fixture(scope="module")
196def source_fit_ser_results(
197 catalog,
198 exposure,
199 psf_fit_results,
200 psf_fit_config,
201 source_fit_ser_config,
202) -> Table:
203 """Return the Sersic fit results if data exists."""
204 if not has_files: 204 ↛ 206line 204 didn't jump to line 206 because the condition on line 204 was always true
205 return None
206 catexp = fitCMB.CatalogExposurePsfs(
207 dataId=dataId,
208 catalog=catalog,
209 exposure=exposure,
210 table_psf_fits=psf_fit_results,
211 channel=channel,
212 config_fit=source_fit_ser_config,
213 )
214 task = fitCMB.MultiProFitSourceTask(config=source_fit_ser_config)
215 results = task.run(catalog_multi=catalog, catexps=[catexp])
216 return arrow_to_astropy(results.output)
219@pytest.fixture(scope="module")
220def source_fit_ser_shapelet_psf_results(
221 catalog,
222 exposure,
223 psf_fit_results,
224 psf_fit_config,
225 source_fit_ser_config,
226) -> Table:
227 """Return the Sersic fits using shapelet PSF parameters if data exists."""
228 if not has_files: 228 ↛ 230line 228 didn't jump to line 230 because the condition on line 228 was always true
229 return None
230 table_psf = Table(
231 meta=dict(config=CatalogPsfFitterConfig().toDict()),
232 )
233 catexp = fitCMB.CatalogExposurePsfs(
234 dataId=dataId,
235 catalog=catalog,
236 exposure=exposure,
237 table_psf_fits=table_psf,
238 channel=channel,
239 config_fit=source_fit_ser_config,
240 )
241 source_fit_ser_config.action_psf = fitCMB.SourceTablePsfComponentsAction()
242 task = fitCMB.MultiProFitSourceTask(config=source_fit_ser_config)
243 results = task.run(catalog_multi=catalog, catexps=[catexp])
244 return arrow_to_astropy(results.output)
247@pytest.fixture(scope="module")
248def source_fits_all(
249 source_fit_exp_fixedcen_results,
250 source_fit_ser_results,
251 source_fit_ser_shapelet_psf_results,
252):
253 """Return all of the fits."""
254 return (
255 source_fit_exp_fixedcen_results,
256 source_fit_ser_results,
257 source_fit_ser_shapelet_psf_results,
258 )
261def test_psf_fits(psf_fit_results):
262 """Test all available PSF fits."""
263 if psf_fit_results is not None: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 assert len(psf_fit_results) == n_test
265 for column in psf_fit_results.columns:
266 assert column and np.all(np.isfinite(psf_fit_results[column]))
267 # TODO: Determine what checks can be done against previous values
270def test_source_fits(source_fits_all):
271 """Test all available source fits."""
272 for results in source_fits_all:
273 if results is not None: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 assert len(results) == n_test
275 prefix = results.meta["config"]["prefix_column"]
276 good = ~results[f"{prefix}unknown_flag"]
277 assert np.sum(~good) == 0
278 for column in results.columns:
279 if column.startswith(prefix):
280 assert column and (np.sum(~np.isfinite(results[column][good])) == 0)
281 # TODO: Determine what checks can be done against previous values