Coverage for tests/test_createApFakes.py: 98%
187 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-23 09:01 +0000
1#
2# This file is part of ap_pipe.
3#
4# Developed for the LSST Data Management System.
5# This product includes software developed by the LSST Project
6# (http://www.lsst.org).
7# See the COPYRIGHT file at the top-level directory of this distribution
8# for details of code ownership.
9#
10# This program is free software: you can redistribute it and/or modify
11# it under the terms of the GNU General Public License as published by
12# the Free Software Foundation, either version 3 of the License, or
13# (at your option) any later version.
14#
15# This program is distributed in the hope that it will be useful,
16# but WITHOUT ANY WARRANTY; without even the implied warranty of
17# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18# GNU General Public License for more details.
19#
20# You should have received a copy of the GNU General Public License
21# along with this program. If not, see <http://www.gnu.org/licenses/>.
22#
24import numpy as np
25import shutil
26import tempfile
27import unittest
28from unittest.mock import MagicMock
30import lsst.daf.butler.tests as butlerTests
31import lsst.geom as geom
32from astropy.table import Table
33from lsst.pipe.base import testUtils
34import lsst.skymap as skyMap
35import lsst.utils.tests
37from lsst.ap.pipe.createApFakes import (
38 CreateRandomApFakesTask,
39 CreateRandomApFakesConfig,
40 CreateVisitDetectorFakesTask,
41 CreateVisitDetectorFakesConfig,
42)
45class TestCreateApFakes(lsst.utils.tests.TestCase):
47 def setUp(self):
48 """
49 """
50 self.tractId = 0
51 self.rng = np.random.default_rng(self.tractId)
52 simpleMapConfig = skyMap.discreteSkyMap.DiscreteSkyMapConfig()
53 simpleMapConfig.raList = [10]
54 simpleMapConfig.decList = [-1]
55 simpleMapConfig.radiusList = [0.1]
56 self.simpleMap = skyMap.DiscreteSkyMap(simpleMapConfig)
57 self.tract = self.simpleMap.generateTract(self.tractId)
59 bBox = self.tract.getOuterSkyPolygon().getBoundingBox()
60 self.nSources = 50
61 self.sourceDensity = (self.nSources
62 / (bBox.getArea() * (180 / np.pi) ** 2))
63 self.fraction = 0.5
64 self.nInVisit = (int(self.nSources * self.fraction)
65 + int((1 - self.fraction) / 2 * self.nSources))
66 self.nInTemplate = (self.nSources - self.nInVisit
67 + int(self.nSources * self.fraction))
69 def testRunQuantum(self):
70 """Test the run quantum method with a gen3 butler.
71 """
72 root = tempfile.mkdtemp()
73 dimensions = {"instrument": ["notACam"],
74 "skymap": ["skyMap"],
75 "tract": [0, 42],
76 }
77 testRepo = butlerTests.makeTestRepo(root, dimensions)
78 with self.assertWarns(FutureWarning):
79 fakesTask = CreateRandomApFakesTask()
80 connections = fakesTask.config.ConnectionsClass(
81 config=fakesTask.config)
82 butlerTests.addDatasetType(
83 testRepo,
84 connections.skyMap.name,
85 connections.skyMap.dimensions,
86 connections.skyMap.storageClass)
87 butlerTests.addDatasetType(
88 testRepo,
89 connections.fakeCat.name,
90 connections.fakeCat.dimensions,
91 connections.fakeCat.storageClass)
93 dataId = {"skymap": "skyMap", "tract": 0}
94 butler = butlerTests.makeTestCollection(testRepo)
95 butler.put(self.simpleMap, "skyMap", {"skymap": "skyMap"})
97 quantum = testUtils.makeQuantum(
98 fakesTask, butler, dataId,
99 {"skyMap": {"skymap": dataId["skymap"]}, "fakeCat": dataId})
100 run = testUtils.runTestQuantum(fakesTask, butler, quantum, True)
101 # Actual input dataset omitted for simplicity
102 run.assert_called_once_with(tractId=dataId["tract"], skyMap=self.simpleMap)
103 shutil.rmtree(root, ignore_errors=True)
105 def testRun(self):
106 """Test the run method.
107 """
108 with self.assertWarns(FutureWarning):
109 fakesConfig = CreateRandomApFakesConfig()
110 fakesConfig.fraction = 0.5
111 fakesConfig.fakeDensity = self.sourceDensity
113 with self.assertWarns(FutureWarning):
114 fakesTask = CreateRandomApFakesTask(config=fakesConfig)
115 bBox = self.tract.getOuterSkyPolygon().getBoundingBox()
116 result = fakesTask.run(self.tractId, self.simpleMap)
117 fakeCat = result.fakeCat
118 self.assertEqual(len(fakeCat), self.nSources)
119 self.assertIn("injection_id", fakeCat.columns)
120 self.assertIn("source_type", fakeCat.columns)
121 self.assertIn("mag", fakeCat.columns)
122 self.assertTrue(np.issubdtype(fakeCat["injection_id"].dtype, np.integer))
123 self.assertEqual(fakeCat["injection_id"].nunique(), len(fakeCat))
124 self.assertTrue(np.all(fakeCat["source_type"] == "Star"))
126 for idx, row in fakeCat.iterrows():
127 self.assertTrue(
128 bBox.contains(
129 geom.SpherePoint(row[fakesTask.config.ra_col],
130 row[fakesTask.config.dec_col],
131 geom.degrees).getVector()))
132 self.assertEqual(fakeCat[fakesConfig.visitSourceFlagCol].sum(),
133 self.nInVisit)
134 self.assertEqual(fakeCat[fakesConfig.templateSourceFlagCol].sum(),
135 self.nInTemplate)
136 for f in fakesConfig.filterSet:
137 filterMags = fakeCat[fakesConfig.mag_col % f]
138 self.assertEqual(self.nSources, len(filterMags))
139 self.assertTrue(
140 np.all(fakesConfig.magMin <= filterMags))
141 self.assertTrue(
142 np.all(fakesConfig.magMax > filterMags))
143 self.assertTrue(np.allclose(filterMags, fakeCat["mag"]))
145 def testVisitCoaddSubdivision(self):
146 """Test that the number of assigned visit to template objects is
147 correct.
148 """
149 with self.assertWarns(FutureWarning):
150 fakesConfig = CreateRandomApFakesConfig()
151 fakesConfig.fraction = 0.5
152 with self.assertWarns(FutureWarning):
153 fakesTask = CreateRandomApFakesTask(config=fakesConfig)
154 subdivision = fakesTask.createVisitCoaddSubdivision(self.nSources)
155 self.assertEqual(
156 subdivision[fakesConfig.visitSourceFlagCol].sum(),
157 self.nInVisit)
158 self.assertEqual(
159 subdivision[fakesConfig.templateSourceFlagCol].sum(),
160 self.nInTemplate)
163def _make_mock_visit_image(visitId=2024111100094, detId=3,
164 xmin=0, xmax=4096, ymin=0, ymax=4096,
165 magLim=25.0, ra_center=10.0, dec_center=-1.0):
166 """Build a minimal MagicMock that satisfies CreateVisitDetectorFakesTask.run."""
167 img = MagicMock()
169 # visit / detector IDs
170 img.getInfo().getVisitInfo().id = visitId
171 img.detector.getId.return_value = detId
173 # bounding box
174 bbox = MagicMock()
175 bbox.getMinX.return_value = xmin
176 bbox.getMaxX.return_value = xmax
177 bbox.getMinY.return_value = ymin
178 bbox.getMaxY.return_value = ymax
179 img.getBBox.return_value = bbox
181 # summary stats — magLim drives max_mag
182 stats = MagicMock()
183 stats.magLim = magLim
184 img.getInfo().getSummaryStats.return_value = stats
186 # convex polygon for density calculation (returns a bbox in steradians)
187 poly_bbox = MagicMock()
188 # ~1e-5 sr ≈ small patch, dense enough to yield O(10) fakes at density=1000
189 poly_bbox.getArea.return_value = 1e-5
190 img.getConvexPolygon().getBoundingBox.return_value = poly_bbox
192 # WCS: pixelToSkyArray returns ra/dec arrays of the right length
193 wcs = MagicMock()
195 def _pix_to_sky(xs, ys, degrees=True):
196 ra = ra_center + xs * 1e-4
197 dec = dec_center + ys * 1e-4
198 return np.asarray(ra), np.asarray(dec)
199 wcs.pixelToSkyArray.side_effect = _pix_to_sky
200 img.getWcs.return_value = wcs
202 # photoCalib — not used in non-hosted paths, but must exist
203 img.getPhotoCalib.return_value = MagicMock()
205 return img
208class TestCreateVisitDetectorFakesTask(lsst.utils.tests.TestCase):
210 def setUp(self):
211 self.visit_image = _make_mock_visit_image()
212 self.source_cat = MagicMock() # not used unless doAddHostedFakes
214 def _make_task(self, **config_overrides):
215 cfg = CreateVisitDetectorFakesConfig()
216 cfg.doAddRandomVisitFakes = True
217 cfg.doAddRandomTemplateFakes = False
218 cfg.doAddHostedFakes = False
219 cfg.doAddVariableFakes = False
220 cfg.doAddModelFakes = False
221 cfg.nRandomFakes = 50
222 for k, v in config_overrides.items():
223 setattr(cfg, k, v)
224 return CreateVisitDetectorFakesTask(config=cfg)
226 # ------------------------------------------------------------------
227 # Basic smoke test: random-only path
228 # ------------------------------------------------------------------
229 def testRunRandomOnly(self):
230 task = self._make_task()
231 result = task.run(self.source_cat, self.visit_image)
232 cat = result.outputCat
234 self.assertIsInstance(cat, Table)
235 self.assertEqual(len(cat), 50)
236 # Required columns
237 for col in ("ra", "dec", "mag", "x", "y", "source_type",
238 "injection_id", "visit", "detector"):
239 self.assertIn(col, cat.colnames)
240 # All sources are stars
241 self.assertTrue(np.all(cat["source_type"] == "Star"))
242 # IDs are unique
243 self.assertEqual(len(np.unique(cat["injection_id"])), len(cat))
244 # visit/detector are set correctly
245 self.assertTrue(np.all(cat["visit"] == 2024111100094))
246 self.assertTrue(np.all(cat["detector"] == 3))
248 # ------------------------------------------------------------------
249 # Template-fraction split
250 # ------------------------------------------------------------------
251 def testTemplateFakeFraction(self):
252 task = self._make_task(
253 doAddRandomTemplateFakes=True,
254 templateFakeFraction=0.25,
255 nRandomFakes=200,
256 )
257 result = task.run(self.source_cat, self.visit_image)
258 cat = result.outputCat
260 n_template = int(np.sum(cat["isTemplateSource"]))
261 # No source should be both visit AND template (mutually exclusive after split)
262 self.assertFalse(np.any(cat["isVisitSource"] & cat["isTemplateSource"]))
263 # template fraction should be roughly 25 %
264 frac = n_template / len(cat)
265 self.assertAlmostEqual(frac, 0.25, delta=0.10)
267 # ------------------------------------------------------------------
268 # Variable fakes: twin_id bookkeeping
269 # ------------------------------------------------------------------
270 def testVariableFakesTwinTracking(self):
271 task = self._make_task(
272 doAddVariableFakes=True,
273 variableFakeFraction=0.2,
274 nRandomFakes=100,
275 )
276 result = task.run(self.source_cat, self.visit_image)
277 cat = result.outputCat
279 # Catalog is larger than the base 100 fakes
280 self.assertGreater(len(cat), 100)
281 # twin_id column exists
282 self.assertIn("twin_id", cat.colnames)
283 # Every injection_id is unique
284 self.assertEqual(len(np.unique(cat["injection_id"])), len(cat))
285 # mag_offset column is present on variable rows
286 self.assertIn("mag_offset", cat.colnames)
288 # ------------------------------------------------------------------
289 # Empty-mode guard raises RuntimeError
290 # ------------------------------------------------------------------
291 def testEmptyCatalogRaises(self):
292 task = self._make_task(
293 doAddRandomVisitFakes=False,
294 doAddHostedFakes=False,
295 doAddModelFakes=False,
296 )
297 with self.assertRaises(RuntimeError):
298 task.run(self.source_cat, self.visit_image)
300 # ------------------------------------------------------------------
301 # Hosted fakes: sparse-host cap (fewer hosts than minHostedFakes)
302 # ------------------------------------------------------------------
303 def testHostedFakesSparseHostCap(self):
304 """When n_hosts < minHostedFakes the task should clamp, not crash."""
305 cfg = CreateVisitDetectorFakesConfig()
306 cfg.doAddRandomVisitFakes = False
307 cfg.doAddHostedFakes = True
308 cfg.doAddRandomTemplateFakes = False
309 cfg.doAddVariableFakes = False
310 cfg.doAddModelFakes = False
311 cfg.fracHostedFakes = 0.99 # near-100 % but within [0, 1) range
312 cfg.minHostedFakes = 50 # request 50 but we'll only supply 5 hosts
313 task = CreateVisitDetectorFakesTask(config=cfg)
315 n_hosts = 5
316 host_table = Table({
317 "slot_Centroid_x": np.full(n_hosts, 2048.0),
318 "slot_Centroid_y": np.full(n_hosts, 2048.0),
319 "slot_ModelFlux_mag": np.full(n_hosts, 20.0),
320 "slot_ModelFlux_flux": np.full(n_hosts, 1e4),
321 "slot_ModelFlux_fluxErr": np.full(n_hosts, 100.0),
322 "slot_Shape_xx": np.full(n_hosts, 4.0),
323 "slot_Shape_xy": np.zeros(n_hosts),
324 "slot_Shape_yy": np.full(n_hosts, 4.0),
325 "id": np.arange(n_hosts, dtype=np.int64),
326 "coord_ra": np.deg2rad(np.full(n_hosts, 10.0)),
327 "coord_dec": np.deg2rad(np.full(n_hosts, -1.0)),
328 "sky_source": np.zeros(n_hosts, dtype=bool),
329 "base_ClassificationSizeExtendedness_flag": np.zeros(n_hosts, dtype=bool),
330 "base_ClassificationExtendedness_flag": np.zeros(n_hosts, dtype=bool),
331 "slot_Shape_flag": np.zeros(n_hosts, dtype=bool),
332 "slot_Centroid_flag": np.zeros(n_hosts, dtype=bool),
333 "base_PixelFlags_flag": np.zeros(n_hosts, dtype=bool),
334 "base_ClassificationSizeExtendedness_value": np.ones(n_hosts),
335 "base_ClassificationExtendedness_value": np.ones(n_hosts, dtype=int),
336 })
338 # Patch photoCalib so calibrateCatalog().asAstropy() returns our table
339 img = _make_mock_visit_image()
340 img.getPhotoCalib().calibrateCatalog.return_value.asAstropy.return_value = host_table
342 result = task.run(self.source_cat, img)
343 cat = result.outputCat
345 # Number of hosted fakes should be clamped to n_hosts, not minHostedFakes
346 self.assertEqual(len(cat), n_hosts)
347 self.assertIn("hosted_fake", cat.colnames)
348 self.assertTrue(np.all(cat["hosted_fake"]))
349 self.assertEqual(len(np.unique(cat["injection_id"])), n_hosts)
351 # ------------------------------------------------------------------
352 # Hosted fakes: zero valid hosts — warning issued, no crash when
353 # random fakes are also enabled as fallback
354 # ------------------------------------------------------------------
355 def testHostedFakesNoHostsWarning(self):
356 task = self._make_task(
357 doAddRandomVisitFakes=True,
358 nRandomFakes=10,
359 doAddHostedFakes=True,
360 )
361 empty_table = Table({
362 "slot_Centroid_x": np.array([]),
363 "slot_Centroid_y": np.array([]),
364 "slot_ModelFlux_mag": np.array([]),
365 "slot_ModelFlux_flux": np.array([]),
366 "slot_ModelFlux_fluxErr": np.array([]),
367 "slot_Shape_xx": np.array([]),
368 "slot_Shape_xy": np.array([]),
369 "slot_Shape_yy": np.array([]),
370 "id": np.array([], dtype=np.int64),
371 "coord_ra": np.array([]),
372 "coord_dec": np.array([]),
373 "sky_source": np.array([], dtype=bool),
374 "base_ClassificationSizeExtendedness_flag": np.array([], dtype=bool),
375 "base_ClassificationExtendedness_flag": np.array([], dtype=bool),
376 "slot_Shape_flag": np.array([], dtype=bool),
377 "slot_Centroid_flag": np.array([], dtype=bool),
378 "base_PixelFlags_flag": np.array([], dtype=bool),
379 "base_ClassificationSizeExtendedness_value": np.array([]),
380 "base_ClassificationExtendedness_value": np.array([], dtype=int),
381 })
382 self.visit_image.getPhotoCalib().calibrateCatalog.return_value.asAstropy.return_value = (
383 empty_table
384 )
386 with self.assertLogs("lsst.ap.pipe.createApFakes", level="WARNING") as cm:
387 result = task.run(self.source_cat, self.visit_image)
389 self.assertTrue(any("no valid hosts" in msg.lower() for msg in cm.output))
390 # Random fakes still produced
391 self.assertEqual(len(result.outputCat), 10)
394class MemoryTester(lsst.utils.tests.MemoryTestCase):
395 pass
398def setup_module(module):
399 lsst.utils.tests.init()
402if __name__ == "__main__": 402 ↛ 403line 402 didn't jump to line 403 because the condition on line 402 was never true
403 lsst.utils.tests.init()
404 unittest.main()