Coverage for python / lsst / pipe / tasks / visualizeVisit.py: 54%
129 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-21 10:40 +0000
1# This file is part of pipe_tasks.
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/>.
22__all__ = [
23 "VisualizeBinExpConfig",
24 "VisualizeBinExpTask",
25 "VisualizeMosaicExpConfig",
26 "VisualizeMosaicExpTask",
27 "VisualizeBinCalibConfig",
28 "VisualizeBinCalibTask",
29 "VisualizeMosaicCalibConfig",
30 "VisualizeMosaicCalibTask",
31 "VisualizeBinCalibFilterConfig",
32 "VisualizeBinCalibFilterTask",
33 "VisualizeMosaicCalibFilterConfig",
34 "VisualizeMosaicCalibFilterTask",
35]
37import dataclasses
39import lsst.afw.cameraGeom.utils as afwUtils
40import lsst.afw.image as afwImage
41import lsst.afw.math as afwMath
42import lsst.pex.config as pexConfig
43import lsst.pipe.base as pipeBase
44import lsst.pipe.base.connectionTypes as cT
45import numpy as np
48# VisualizeBinExp (here) & VisualizeMosaicExp (below):
49class VisualizeBinExpConnections(
50 pipeBase.PipelineTaskConnections, dimensions=("instrument", "visit", "detector")
51):
52 camera = cT.PrerequisiteInput(
53 name="camera",
54 doc="Input camera to use for mosaic geometry.",
55 storageClass="Camera",
56 dimensions=("instrument",),
57 isCalibration=True,
58 )
59 inputExp = cT.Input(
60 name="calexp",
61 doc="Input exposure data to mosaic.",
62 storageClass="ExposureF",
63 dimensions=("instrument", "visit", "detector"),
64 )
65 outputExp = cT.Output(
66 name="calexpBin",
67 doc="Output binned image.",
68 storageClass="ExposureF",
69 dimensions=("instrument", "visit", "detector"),
70 )
72 def __init__(self, *, config=None):
73 """Customize connections for a specific instance.
75 This customization enables for dynamic setup at runtime,
76 allowing VisualizeBinExpTask to work with different types of
77 Exposures such as postISRCCDs and calexps.
79 Parameters
80 ----------
81 config : `VisualizeBinExpConfig`
82 A config for `VisualizeBinExpTask`.
83 """
84 super().__init__(config=config)
85 if config:
86 # Update the dimensions of the task
87 self.dimensions.clear()
88 self.dimensions.update(config.dimensions)
89 # Update the storage classes and dimensions for inputs/outputs
90 self.inputExp = dataclasses.replace(
91 self.inputExp,
92 storageClass=config.storageClass,
93 dimensions=config.dimensions,
94 )
95 self.outputExp = dataclasses.replace(
96 self.outputExp,
97 storageClass=config.storageClass,
98 dimensions=config.dimensions,
99 )
102class VisualizeBinExpConfig(pipeBase.PipelineTaskConfig, pipelineConnections=VisualizeBinExpConnections):
103 """Configuration for focal plane visualization."""
105 storageClass = pexConfig.Field(
106 default=VisualizeBinExpConnections.inputExp.storageClass,
107 dtype=str,
108 doc=(
109 "The storageClasses of the input and output exposures. "
110 "Must be of type lsst.afw.Image.Exposure, or one of its subtypes."
111 ),
112 )
113 dimensions = pexConfig.ListField(
114 # Sort to ensure default order is consistent between runs
115 default=sorted(VisualizeBinExpConnections.dimensions),
116 dtype=str,
117 doc="The task dimensions, also applied to the input/output exposures. ",
118 )
119 binning = pexConfig.Field(
120 dtype=int,
121 default=8,
122 doc="Binning factor to apply to each input exposure's image data.",
123 )
124 detectorKeyword = pexConfig.Field(
125 dtype=str,
126 default="DET-ID",
127 doc="Metadata keyword to use to find detector if not available from input.",
128 )
131class VisualizeBinExpTask(pipeBase.PipelineTask):
132 """Bin the detectors of an exposure.
134 The outputs of this task should be passed to
135 VisualizeMosaicExpTask to be mosaicked into a full focal plane
136 visualization image.
137 """
139 ConfigClass = VisualizeBinExpConfig
140 _DefaultName = "VisualizeBinExp"
142 def run(self, inputExp, camera):
143 """Bin input image, attach associated detector.
145 Parameters
146 ----------
147 inputExp : `lsst.afw.image.Exposure`
148 Input exposure data to bin.
149 camera : `lsst.afw.cameraGeom.Camera`
150 Input camera to use for mosaic geometry.
152 Returns
153 -------
154 output : `lsst.pipe.base.Struct`
155 Results struct with attribute:
157 ``outputExp``
158 Binned version of input image (`lsst.afw.image.Exposure`).
159 """
160 if inputExp.getDetector() is None:
161 detectorId = inputExp.getMetadata().get(self.config.detectorKeyword)
162 if detectorId is not None:
163 inputExp.setDetector(camera[detectorId])
165 binned = inputExp.getMaskedImage()
166 binned = afwMath.binImage(binned, self.config.binning)
167 outputExp = afwImage.makeExposure(binned)
169 outputExp.setInfo(inputExp.getInfo())
171 return pipeBase.Struct(outputExp=outputExp)
174# VisualizeBinExp (above) & VisualizeMosaicExp (here):
175class VisualizeMosaicExpConnections(pipeBase.PipelineTaskConnections, dimensions=("instrument", "visit")):
176 camera = cT.PrerequisiteInput(
177 name="camera",
178 doc="Input camera to use for mosaic geometry.",
179 storageClass="Camera",
180 dimensions=("instrument",),
181 isCalibration=True,
182 )
183 inputExps = cT.Input(
184 name="calexpBin",
185 doc="Input binned images to mosaic.",
186 storageClass="ExposureF",
187 dimensions=("instrument", "visit", "detector"),
188 multiple=True,
189 )
190 outputData = cT.Output(
191 name="calexpFocalPlane",
192 doc="Output binned mosaicked frame.",
193 storageClass="ImageF",
194 dimensions=("instrument", "visit"),
195 )
197 def __init__(self, *, config=None):
198 """Customize connections for a specific instance.
200 This customization enables for dynamic setup at runtime,
201 allowing VisualizeMosaicExpTask to work with different types of
202 Exposures such as postISRCCDs and calexps.
204 Parameters
205 ----------
206 config : `VisualizeMosaicExpTask`
207 A config for `VisualizeMosaicExpTask`.
208 """
209 super().__init__(config=config)
210 if config:
211 # Update the dimensions of the task
212 self.dimensions.clear()
213 self.dimensions.update(config.dimensions)
214 # Update the storage classes and dimensions for inputs/outputs
215 inputExpsDimensions = list(config.dimensions) + ["detector"]
216 self.inputExps = dataclasses.replace(
217 self.inputExps,
218 storageClass=config.storageClass,
219 dimensions=inputExpsDimensions,
220 )
221 self.outputData = dataclasses.replace(
222 self.outputData,
223 dimensions=config.dimensions,
224 )
227class VisualizeMosaicExpConfig(
228 pipeBase.PipelineTaskConfig, pipelineConnections=VisualizeMosaicExpConnections
229):
230 """Configuration for focal plane visualization."""
232 storageClass = pexConfig.Field(
233 default=VisualizeMosaicExpConnections.inputExps.storageClass,
234 dtype=str,
235 doc=(
236 "The storageClass of the input exposures. "
237 "Must be of type lsst.afw.Image.Exposure, or one of its subtypes."
238 ),
239 )
240 dimensions = pexConfig.ListField(
241 # Sort to ensure default order is consistent between runs
242 default=sorted(VisualizeMosaicExpConnections.dimensions),
243 dtype=str,
244 doc="The task dimensions, also applied to the input/output exposures. "
245 "Note: the input exposure will have 'detector' appended by default.",
246 )
247 binning = pexConfig.Field(
248 dtype=int,
249 default=8,
250 doc="Binning factor previously applied to input exposures.",
251 )
254class VisualizeMosaicExpTask(pipeBase.PipelineTask):
255 """Task to mosaic binned products.
257 The config.binning parameter must match that used in the
258 VisualizeBinExpTask. Otherwise there will be a mismatch between
259 the input image size and the expected size of that image in the
260 full focal plane frame.
261 """
263 ConfigClass = VisualizeMosaicExpConfig
264 _DefaultName = "VisualizeMosaicExp"
266 def makeCameraImage(self, inputExps, camera, binning):
267 """Make an image of an entire focal plane.
269 Parameters
270 ----------
271 exposures: `dict` [`int`, `lsst.afw.image.Exposure`]
272 CCD exposures, binned by `binning`. The keys are the
273 detectorIDs, with the values the binned image exposure.
275 Returns
276 -------
277 image : `lsst.afw.image.Image`
278 Image mosaicked from the individual binned images for each
279 detector.
280 """
281 image = afwUtils.makeImageFromCamera(
282 camera, imageSource=ImageSource(inputExps), imageFactory=afwImage.ImageF, binSize=binning
283 )
284 return image
286 def run(self, inputExps, camera, inputIds=None):
287 """Mosaic inputs together to create focal plane image.
289 Parameters
290 ----------
291 inputExps : `list` [`lsst.afw.image.Exposure`]
292 Input exposure data to bin.
293 camera : `lsst.afw.cameraGeom.Camera`
294 Input camera to use for mosaic geometry.
295 inputIds : `list` [`int`], optional
296 Optional list providing exposure IDs corresponding to input
297 exposures. Will be generated via the exposure data `getDetector`
298 method if not provided.
300 Returns
301 -------
302 output : `lsst.pipe.base.Struct`
303 Results struct with attribute:
305 ``outputExp``
306 Binned version of input image (`lsst.afw.image.Exposure`).
307 """
308 if not inputIds:
309 inputIds = [exp.getDetector().getId() for exp in inputExps]
310 expDict = {id: exp for id, exp in zip(inputIds, inputExps)}
311 image = self.makeCameraImage(expDict, camera, self.config.binning)
313 return pipeBase.Struct(outputData=image)
316class ImageSource:
317 """Source of images for makeImageFromCamera"""
319 def __init__(self, exposures):
320 self.exposures = exposures
321 self.isTrimmed = True
322 self.background = np.nan
324 def getCcdImage(self, detector, imageFactory, binSize):
325 """Provide image of CCD to makeImageFromCamera
327 Parameters
328 ----------
329 detector : `int`
330 Detector ID to get image data for.
331 imageFactory : `lsst.afw.image.Image`
332 Type of image to construct.
333 binSize : `int`
334 Binsize to use to recompute dimensions.
336 Returns
337 -------
338 image : `lsst.afw.image.Image`
339 Appropriately rotated, binned, and transformed
340 image to be mosaicked.
341 detector : `lsst.afw.cameraGeom.Detector`
342 Camera detector that the returned image data
343 belongs to.
344 """
345 detId = detector.getId()
347 if detId not in self.exposures:
348 dims = detector.getBBox().getDimensions() / binSize
349 image = imageFactory(*[int(xx) for xx in dims])
350 image.set(self.background)
351 else:
352 image = self.exposures[detector.getId()]
353 if hasattr(image, "getMaskedImage"):
354 image = image.getMaskedImage()
355 if hasattr(image, "getMask"):
356 mask = image.getMask()
357 isBad = mask.getArray() & mask.getPlaneBitMask("NO_DATA") > 0
358 image = image.clone()
359 image.getImage().getArray()[isBad] = self.background
360 if hasattr(image, "getImage"):
361 image = image.getImage()
363 # afwMath.rotateImageBy90 checks NQuarter values,
364 # so we don't need to here.
365 image = afwMath.rotateImageBy90(image, detector.getOrientation().getNQuarter())
366 return image, detector
369# VisualizeBinCalib (here) & VisualizeMosaicCalib (below):
370# Inputs to bin task have dimensions: {instrument, detector}
371# Output of the mosaic task have: {instrument, }
372class VisualizeBinCalibConnections(pipeBase.PipelineTaskConnections, dimensions=("instrument", "detector")):
373 inputExp = cT.Input(
374 name="bias",
375 doc="Input exposure data to mosaic.",
376 storageClass="ExposureF",
377 dimensions=("instrument", "detector"),
378 isCalibration=True,
379 )
380 camera = cT.PrerequisiteInput(
381 name="camera",
382 doc="Input camera to use for mosaic geometry.",
383 storageClass="Camera",
384 dimensions=("instrument",),
385 isCalibration=True,
386 )
388 outputExp = cT.Output(
389 name="biasBin",
390 doc="Output binned image.",
391 storageClass="ExposureF",
392 dimensions=("instrument", "detector"),
393 )
396class VisualizeBinCalibConfig(VisualizeBinExpConfig, pipelineConnections=VisualizeBinCalibConnections):
397 pass
400class VisualizeBinCalibTask(VisualizeBinExpTask):
401 """Bin the detectors of an calibration.
403 The outputs of this task should be passed to
404 VisualizeMosaicCalibTask to be mosaicked into a full focal plane
405 visualization image.
406 """
408 ConfigClass = VisualizeBinCalibConfig
409 _DefaultName = "VisualizeBinCalib"
411 pass
414# VisualizeBinCalib (above) & VisualizeMosaicCalib (here):
415# Inputs to bin task have dimensions: {instrument, detector}
416# Output of the mosaic task have: {instrument, }
417class VisualizeMosaicCalibConnections(pipeBase.PipelineTaskConnections, dimensions=("instrument",)):
418 inputExps = cT.Input(
419 name="biasBin",
420 doc="Input binned images mosaic.",
421 storageClass="ExposureF",
422 dimensions=("instrument", "detector"),
423 multiple=True,
424 )
425 camera = cT.PrerequisiteInput(
426 name="camera",
427 doc="Input camera to use for mosaic geometry.",
428 storageClass="Camera",
429 dimensions=("instrument",),
430 isCalibration=True,
431 )
433 outputData = cT.Output(
434 name="biasFocalPlane",
435 doc="Output binned mosaicked frame.",
436 storageClass="ImageF",
437 dimensions=("instrument",),
438 )
441class VisualizeMosaicCalibConfig(
442 VisualizeMosaicExpConfig, pipelineConnections=VisualizeMosaicCalibConnections
443):
444 pass
447class VisualizeMosaicCalibTask(VisualizeMosaicExpTask):
448 """Task to mosaic binned products.
450 The config.binning parameter must match that used in the
451 VisualizeBinCalibTask. Otherwise there will be a mismatch between
452 the input image size and the expected size of that image in the
453 full focal plane frame.
454 """
456 ConfigClass = VisualizeMosaicCalibConfig
457 _DefaultName = "VisualizeMosaicCalib"
459 pass
462# VisualizeBinCalibFilter (here) & VisualizeMosaicCalibFilter (below):
463# Inputs to bin task have dimensions: {instrument, detector, physical_filter}
464# Output of the mosaic task have: {instrument, physical_filter}
465class VisualizeBinCalibFilterConnections(
466 pipeBase.PipelineTaskConnections, dimensions=("instrument", "detector", "physical_filter")
467):
468 inputExp = cT.Input(
469 name="flat",
470 doc="Input exposure data to mosaic.",
471 storageClass="ExposureF",
472 dimensions=("instrument", "detector", "physical_filter"),
473 isCalibration=True,
474 )
475 camera = cT.PrerequisiteInput(
476 name="camera",
477 doc="Input camera to use for mosaic geometry.",
478 storageClass="Camera",
479 dimensions=("instrument",),
480 isCalibration=True,
481 )
483 outputExp = cT.Output(
484 name="flatBin",
485 doc="Output binned image.",
486 storageClass="ExposureF",
487 dimensions=("instrument", "detector", "physical_filter"),
488 )
491class VisualizeBinCalibFilterConfig(
492 VisualizeBinExpConfig, pipelineConnections=VisualizeBinCalibFilterConnections
493):
494 pass
497class VisualizeBinCalibFilterTask(VisualizeBinExpTask):
498 """Bin the detectors of an calibration.
500 The outputs of this task should be passed to
501 VisualizeMosaicCalibTask to be mosaicked into a full focal plane
502 visualization image.
503 """
505 ConfigClass = VisualizeBinCalibFilterConfig
506 _DefaultName = "VisualizeBinCalibFilter"
508 pass
511# VisualizeBinCalibFilter (above) & VisualizeMosaicCalibFilter (here):
512# Inputs to bin task have dimensions: {instrument, detector, physical_filter}
513# Output of the mosaic task have: {instrument, physical_filter}
514class VisualizeMosaicCalibFilterConnections(
515 pipeBase.PipelineTaskConnections,
516 dimensions=("instrument", "physical_filter"),
517):
518 inputExps = cT.Input(
519 name="flatBin",
520 doc="Input binned images mosaic.",
521 storageClass="ExposureF",
522 dimensions=("instrument", "detector", "physical_filter"),
523 multiple=True,
524 )
525 camera = cT.PrerequisiteInput(
526 name="camera",
527 doc="Input camera to use for mosaic geometry.",
528 storageClass="Camera",
529 dimensions=("instrument",),
530 isCalibration=True,
531 )
533 outputData = cT.Output(
534 name="flatFocalPlane",
535 doc="Output binned mosaicked frame.",
536 storageClass="ImageF",
537 dimensions=("instrument", "physical_filter"),
538 )
541class VisualizeMosaicCalibFilterConfig(
542 VisualizeMosaicExpConfig, pipelineConnections=VisualizeMosaicCalibFilterConnections
543):
544 pass
547class VisualizeMosaicCalibFilterTask(VisualizeMosaicExpTask):
548 """Task to mosaic binned products.
550 The config.binning parameter must match that used in the
551 VisualizeBinCalibFilterTask. Otherwise there will be a mismatch between
552 the input image size and the expected size of that image in the
553 full focal plane frame.
554 """
556 ConfigClass = VisualizeMosaicCalibFilterConfig
557 _DefaultName = "VisualizeMosaicCalibFilter"
559 pass