Coverage for python / lsst / ap / pipe / createApFakes.py: 42%
67 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-26 09:39 +0000
1# This file is part of ap_pipe.
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 numpy as np
23import pandas as pd
24import uuid
26import lsst.pex.config as pexConfig
27from lsst.pipe.base import PipelineTask, PipelineTaskConnections, Struct
28import lsst.pipe.base.connectionTypes as connTypes
29from lsst.pipe.tasks.insertFakes import InsertFakesConfig
30from lsst.skymap import BaseSkyMap
32from lsst.source.injection import generate_injection_catalog
34from deprecated.sphinx import deprecated
36__all__ = ["CreateRandomApFakesTask",
37 "CreateRandomApFakesConfig",
38 "CreateRandomApFakesConnections"]
41class CreateRandomApFakesConnections(PipelineTaskConnections,
42 defaultTemplates={"fakesType": "fakes_"},
43 dimensions=("tract", "skymap")):
44 skyMap = connTypes.Input(
45 doc="Input definition of geometry/bbox and projection/wcs for "
46 "template exposures",
47 name=BaseSkyMap.SKYMAP_DATASET_TYPE_NAME,
48 dimensions=("skymap",),
49 storageClass="SkyMap",
50 )
51 fakeCat = connTypes.Output(
52 doc="Catalog of fake sources to draw inputs from.",
53 name="{fakesType}fakeSourceCat",
54 storageClass="DataFrame",
55 dimensions=("tract", "skymap")
56 )
59@deprecated(
60 reason="This task will be removed in v28.0 as it is replaced by `source_injection` tasks.",
61 version="v28.0",
62 category=FutureWarning,
63)
64class CreateRandomApFakesConfig(
65 InsertFakesConfig,
66 pipelineConnections=CreateRandomApFakesConnections):
67 """Config for CreateRandomApFakesTask. Copy from the InsertFakesConfig to
68 assert that columns created with in this task match that those expected in
69 the InsertFakes and related tasks.
70 """
71 fakeDensity = pexConfig.RangeField(
72 doc="Goal density of random fake sources per square degree. Default "
73 "value is roughly the density per square degree for ~10k sources "
74 "visit.",
75 dtype=float,
76 default=1000,
77 min=0,
78 )
79 filterSet = pexConfig.ListField(
80 doc="Set of Abstract filter names to produce magnitude columns for.",
81 dtype=str,
82 default=["u", "g", "r", "i", "z", "y"],
83 )
84 fraction = pexConfig.RangeField(
85 doc="Fraction of the created source that should be inserted into both "
86 "the visit and template images. Values less than 1 will result in "
87 "(1 - fraction) / 2 inserted into only visit or the template.",
88 dtype=float,
89 default=1/3,
90 min=0,
91 max=1,
92 )
93 magMin = pexConfig.RangeField(
94 doc="Minimum magnitude the mag distribution. All magnitudes requested "
95 "are set to the same value.",
96 dtype=float,
97 default=20,
98 min=1,
99 max=40,
100 )
101 magMax = pexConfig.RangeField(
102 doc="Maximum magnitude the mag distribution. All magnitudes requested "
103 "are set to the same value.",
104 dtype=float,
105 default=30,
106 min=1,
107 max=40,
108 )
109 visitSourceFlagCol = pexConfig.Field(
110 doc="Name of the column flagging objects for insertion into the visit "
111 "image.",
112 dtype=str,
113 default="isVisitSource"
114 )
115 templateSourceFlagCol = pexConfig.Field(
116 doc="Name of the column flagging objects for insertion into the "
117 "template image.",
118 dtype=str,
119 default="isTemplateSource"
120 )
123@deprecated(
124 reason="This task will be removed in v28.0 as it is replaced by `source_injection` tasks.",
125 version="v28.0",
126 category=FutureWarning,
127)
128class CreateRandomApFakesTask(PipelineTask):
129 """Create and store a set of spatially uniform star fakes over the sphere
130 for use in AP processing. Additionally assign random magnitudes to said
131 fakes and assign them to be inserted into either a visit exposure or
132 template exposure.
133 """
135 _DefaultName = "createApFakes"
136 ConfigClass = CreateRandomApFakesConfig
138 def runQuantum(self, butlerQC, inputRefs, outputRefs):
139 inputs = butlerQC.get(inputRefs)
140 inputs["tractId"] = butlerQC.quantum.dataId["tract"]
142 outputs = self.run(**inputs)
143 butlerQC.put(outputs, outputRefs)
145 def run(self, tractId, skyMap):
146 """Create a set of uniform random points that covers a tract.
148 Parameters
149 ----------
150 tractId : `int`
151 Tract id to produce randoms over.
152 skyMap : `lsst.skymap.SkyMap`
153 Skymap to produce randoms over.
155 Returns
156 -------
157 randoms : `pandas.DataFrame`
158 Catalog of random points covering the given tract. Follows the
159 columns and format expected in `lsst.pipe.tasks.InsertFakes`.
160 """
161 # Use the tractId as the random seed.
162 rng = np.random.default_rng(tractId)
164 tract = skyMap.generateTract(tractId)
165 tractArea = tract.getOuterSkyPolygon().getBoundingBox().getArea()
166 tractArea *= (180 / np.pi) ** 2
167 tractWcs = tract.getWcs()
168 vertexList = tract.getVertexList()
169 vertexRas = [vertex.getRa().asDegrees() for vertex in vertexList]
170 vertexDecs = [vertex.getDec().asDegrees() for vertex in vertexList]
172 catalog = generate_injection_catalog(
173 ra_lim=sorted([np.min(vertexRas), np.max(vertexRas)]),
174 dec_lim=sorted([np.min(vertexDecs), np.max(vertexDecs)]),
175 density=self.config.fakeDensity,
176 source_type="Star",
177 seed=str(tractId),
178 wcs=tractWcs
179 )
181 nFakes = len(catalog)
183 self.log.info(
184 f"Creating {nFakes} star fakes over tractId={tractId} with "
185 f" RA in ({sorted([np.min(vertexRas), np.max(vertexRas)])} "
186 f" Dec in ({sorted([np.min(vertexDecs), np.max(vertexDecs)])}), "
187 f"area={tractArea:.4f} deg^2 and "
188 f"magnitude range: [{self.config.magMin, self.config.magMax}]")
190 onesColumn = np.ones(nFakes, dtype="float")
191 zerosColumn = np.zeros(nFakes, dtype="float")
192 # Concatenate the data and add dummy values for the unused variables.
193 # Set all data to PSF like objects.
194 randData = {
195 "fakeId": [uuid.uuid4().int & (1 << 64) - 1 for n in range(nFakes)],
196 self.config.ra_col: catalog['ra'].value,
197 self.config.dec_col: catalog['dec'].value,
198 **self.createVisitCoaddSubdivision(nFakes),
199 **self.createRandomMagnitudes(nFakes, rng),
200 self.config.disk_semimajor_col: onesColumn,
201 self.config.bulge_semimajor_col: onesColumn,
202 self.config.disk_n_col: onesColumn,
203 self.config.bulge_n_col: onesColumn,
204 self.config.disk_axis_ratio_col: onesColumn,
205 self.config.bulge_axis_ratio_col: onesColumn,
206 self.config.disk_pa_col: zerosColumn,
207 self.config.bulge_pa_col: onesColumn,
208 self.config.sourceType: catalog['source_type'].value,
209 "source_type": catalog['source_type'].value,
210 "injection_id": catalog['injection_id'].value
211 }
213 return Struct(fakeCat=pd.DataFrame(data=randData))
215 def createVisitCoaddSubdivision(self, nFakes):
216 """Assign a given fake either a visit image or coadd or both based on
217 the ``faction`` config value.
219 Parameters
220 ----------
221 nFakes : `int`
222 Number of fakes to create.
224 Returns
225 -------
226 output : `dict`[`str`, `numpy.ndarray`]
227 Dictionary of boolean arrays specifying which image to put a
228 given fake into.
229 """
230 nBoth = int(self.config.fraction * nFakes)
231 nOnly = int((1 - self.config.fraction) / 2 * nFakes)
232 isVisitSource = np.zeros(nFakes, dtype=bool)
233 isTemplateSource = np.zeros(nFakes, dtype=bool)
234 if nBoth > 0:
235 isVisitSource[:nBoth] = True
236 isTemplateSource[:nBoth] = True
237 if nOnly > 0:
238 isVisitSource[nBoth:(nBoth + nOnly)] = True
239 isTemplateSource[(nBoth + nOnly):] = True
241 return {self.config.visitSourceFlagCol: isVisitSource,
242 self.config.templateSourceFlagCol: isTemplateSource}
244 def createRandomMagnitudes(self, nFakes, rng):
245 """Create a random distribution of magnitudes for out fakes.
247 Parameters
248 ----------
249 nFakes : `int`
250 Number of fakes to create.
251 rng : `numpy.random.Generator`
252 Initialized random number generator.
254 Returns
255 -------
256 randMags : `dict`[`str`, `numpy.ndarray`]
257 Dictionary of magnitudes in the bands set by the ``filterSet``
258 config option.
259 """
260 mags = rng.uniform(self.config.magMin,
261 self.config.magMax,
262 size=nFakes)
263 randMags = {}
264 for fil in self.config.filterSet:
265 randMags[self.config.mag_col % fil] = mags
266 # adding a non-filter column for magnitudes
267 randMags["mag"] = mags
268 return randMags