Coverage for python/lsst/drp/tasks/update_visit_summary.py: 19%
296 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 02:45 -0800
1# This file is part of drp_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/>.
22from __future__ import annotations
24__all__ = (
25 "UpdateVisitSummaryConnections",
26 "UpdateVisitSummaryConfig",
27 "UpdateVisitSummaryTask",
28 "PossiblyMultipleInput",
29 "PerTractInput",
30 "PerHealpixInput",
31 "GlobalInput",
32)
34import dataclasses
35import re
36from abc import ABC, abstractmethod
37from collections.abc import Iterable, Mapping
38from typing import Any
40import astropy.table
41import numpy as np
43import lsst.pipe.base.connectionTypes as cT
44from lsst.afw.geom import SkyWcs
45from lsst.afw.image import ExposureSummaryStats
46from lsst.afw.math import BackgroundList
47from lsst.afw.table import ExposureCatalog, ExposureRecord, SchemaMapper
48from lsst.daf.butler import Butler, DatasetRef, DeferredDatasetHandle
49from lsst.geom import Angle, Box2I, SpherePoint, degrees
50from lsst.pex.config import ChoiceField, ConfigurableField, Field
51from lsst.pipe.base import (
52 InputQuantizedConnection,
53 InvalidQuantumError,
54 OutputQuantizedConnection,
55 PipelineTask,
56 PipelineTaskConfig,
57 PipelineTaskConnections,
58 QuantumContext,
59 Struct,
60)
61from lsst.pipe.tasks.computeExposureSummaryStats import ComputeExposureSummaryStatsTask
62from lsst.skymap import BaseSkyMap, TractInfo
63from lsst.skymap.detail import makeSkyPolygonFromBBox
64from lsst.sphgeom import HealpixPixelization
67def compute_center_for_detector_record(
68 record: ExposureRecord, bbox: Box2I | None = None, wcs: SkyWcs | None = None
69) -> SpherePoint | None:
70 """Compute the sky coordinate center for a detector to be used when
71 testing distance to tract center.
73 Parameters
74 ----------
75 record : `lsst.afw.table.ExposureRecord`
76 Exposure record to obtain WCS and bbox from if not provided.
77 bbox : `lsst.geom.Box2I`, optional
78 Bounding box for the detector in its own pixel coordinates.
79 wcs : `lsst.afw.geom.SkyWcs`, optional
80 WCS that maps the detector's pixel coordinate system to celestial
81 coordinates.
83 Returns
84 -------
85 center : `lsst.geom.SpherePoint` or `None`
86 Center of the detector in sky coordinates, or `None` if no WCS was
87 given or present in the given record.
88 """
89 if bbox is None:
90 bbox = record.getBBox()
91 if wcs is None:
92 wcs = record.getWcs()
93 if wcs is None:
94 return None
95 region = makeSkyPolygonFromBBox(bbox, wcs)
96 return SpherePoint(region.getCentroid())
99def _validate_wcs_provider(name: str) -> bool:
100 """A helper function to validate the WCS provider at config time.
102 Parameters
103 ----------
104 name : str
105 The source or dimension of the source of the WCS.
107 Returns
108 -------
109 is_valid : bool
110 Whether the provided source is valid.
111 """
112 # Check if ``name`` is a healpix dimension.
113 match_healpix = re.match(re.compile(r"healpix\d+"), name)
114 # Check if ``name`` is one of the other source options.
115 match_other = name in ["input_summary", "global", "tract"]
116 return match_healpix or match_other
119class PossiblyMultipleInput(ABC):
120 """A helper ABC for handling input `~lsst.afw.table.ExposureCatalog`
121 datasets that may be multiple (one per tract/visit combination) or
122 unique/global (one per visit).
123 """
125 @abstractmethod
126 def best_for_detector(
127 self,
128 detector_id: int,
129 center: SpherePoint | None = None,
130 bbox: Box2I | None = None,
131 ) -> tuple[int, ExposureRecord | None]:
132 """Return the exposure record for this detector that is the best match
133 for this detector.
135 Parameters
136 ----------
137 detector_id : `int`
138 Detector ID; used to find the right row in the catalog or catalogs.
139 center : `lsst.geom.SpherePoint` or `None`
140 Center of the detector in sky coordinates. If not provided, one
141 will be computed via `compute_center_for_detector_record`.
142 bbox : `lsst.geom.Box2I`, optional
143 Bounding box for the detector in its own pixel coordinates.
145 Returns
146 -------
147 tract_id : `int`
148 ID of the tract that supplied this record, or `-1` if ``record`` is
149 `None` or if the input was not per-tract.
150 record : `lsst.afw.table.ExposureRecord` or `None`
151 Best record for this detector, or `None` if there either were no
152 records for this detector or no WCS available to compute a center.
153 """
154 raise NotImplementedError()
157@dataclasses.dataclass
158class PerTractInput(PossiblyMultipleInput):
159 """Wrapper class for input `~lsst.afw.table.ExposureCatalog` datasets
160 that are per-tract.
162 This selects the best tract via the minimum average distance (on the sky)
163 from the detector's corners to the tract center.
164 """
166 catalogs_by_tract: list[tuple[TractInfo, ExposureCatalog]]
167 """List of tuples of catalogs and the tracts they correspond to
168 (`list` [`tuple` [`lsst.skymap.TractInfo`,
169 `lsst.afw.table.ExposureCatalog`]]).
170 """
172 @classmethod
173 def load(
174 cls,
175 butler: QuantumContext | Butler,
176 sky_map: BaseSkyMap,
177 refs: Iterable[DatasetRef],
178 ) -> PerTractInput:
179 """Load and wrap input catalogs.
181 Parameters
182 ----------
183 butler : `lsst.pipe.base.QuantumContext`
184 Butler proxy used in `~lsst.pipe.base.PipelineTask.runQuantum`.
185 sky_map : `lsst.skymap.BaseSkyMap`
186 Definition of tracts and patches.
187 refs : `~collections.abc.Iterable` [`lsst.daf.butler.DatasetRef`]
188 References to the catalog datasets to load.
190 Returns
191 -------
192 wrapper : `PerTractInput`
193 Wrapper object for the loaded catalogs.
194 """
195 catalogs_by_tract = []
196 for ref in refs:
197 tract_id = ref.dataId["tract"]
198 tract_info = sky_map[tract_id]
199 catalogs_by_tract.append(
200 (
201 tract_info,
202 butler.get(ref),
203 )
204 )
205 return cls(catalogs_by_tract)
207 def best_for_detector(
208 self,
209 detector_id: int,
210 center: SpherePoint | None = None,
211 bbox: Box2I | None = None,
212 ) -> tuple[int, ExposureRecord | None]:
213 # Docstring inherited.
214 best_result: tuple[int, ExposureRecord | None] = (-1, None)
215 best_distance: Angle = float("inf") * degrees
216 for tract_info, catalog in self.catalogs_by_tract:
217 record = catalog.find(detector_id)
218 if record is None:
219 continue
220 if center is None:
221 center_for_record = compute_center_for_detector_record(record, bbox=bbox)
222 if center_for_record is None:
223 continue
224 else:
225 center_for_record = center
226 center_distance = tract_info.ctr_coord.separation(center_for_record)
227 if best_distance > center_distance:
228 best_result = (tract_info.tract_id, record)
229 best_distance = center_distance
230 return best_result
233@dataclasses.dataclass
234class PerHealpixInput(PossiblyMultipleInput):
235 """Wrapper class for input `~lsst.afw.table.ExposureCatalog` datasets
236 that are per-healpix pixel.
238 This selects the best pixel based on which one contains the detector
239 centroid. Since the pixel regions are ConvexPolygons that do not perfectly
240 describe the pixel borders, in case of ambiguity, the pixel is chosen via
241 the minimum distance (on the sky) from the detector's centroid to the pixel
242 centroid.
243 """
245 catalogs_by_pixel: list[int, tuple[int, SpherePoint, ExposureCatalog]]
246 """List of tuples of catalogs and the pixels they correspond to
247 (`list` [`tuple` [`int`, `lsst.geom.SpherePoint`,
248 `lsst.afw.table.ExposureCatalog`]]).
249 """
251 @classmethod
252 def load(
253 cls,
254 butler: QuantumContext | Butler,
255 refs: Iterable[DatasetRef],
256 ) -> PerHealpixInput:
257 """Load and wrap input catalogs.
259 Parameters
260 ----------
261 butler : `lsst.pipe.base.QuantumContext`
262 Butler proxy used in `~lsst.pipe.base.PipelineTask.runQuantum`.
263 refs : `~collections.abc.Iterable` [`lsst.daf.butler.DatasetRef`]
264 References to the catalog datasets to load.
266 Returns
267 -------
268 wrapper : `PerHealpixInput`
269 Wrapper object for the loaded catalogs.
270 """
271 catalogs_by_pixel = []
272 pixelization = HealpixPixelization(3)
273 for ref in refs:
274 pixel_id = ref.dataId["healpix3"]
275 pixel_region = pixelization.pixel(pixel_id)
276 catalogs_by_pixel.append(
277 (
278 pixel_id,
279 pixel_region,
280 butler.get(ref),
281 )
282 )
283 return cls(catalogs_by_pixel)
285 def best_for_detector(
286 self,
287 detector_id: int,
288 center: SpherePoint | None = None,
289 bbox: Box2I | None = None,
290 ) -> tuple[int, ExposureRecord | None]:
291 # Docstring inherited.
292 best_result: tuple[int, ExposureRecord | None] = (-1, None)
293 inPixel = []
294 distances = []
295 results = []
296 for pixel_id, pixel_region, catalog in self.catalogs_by_pixel:
297 record = catalog.find(detector_id)
298 if record is None:
299 continue
300 if center is None:
301 center_for_record = compute_center_for_detector_record(record, bbox=bbox)
302 if center_for_record is None:
303 continue
304 else:
305 center_for_record = center
307 inPixel.append(pixel_region.contains(center_for_record.getVector()))
308 distances.append(SpherePoint(pixel_region.getCentroid()).separation(center_for_record))
309 results.append((pixel_id, record))
311 if len(results) == 0:
312 # There were no entries at all; return the null result.
313 return best_result
315 inPixel = np.array(inPixel)
316 results = np.array(results)
317 distances = np.array(distances)
318 if inPixel.sum() == 1:
319 # There was one match in the pixel; return it.
320 best_result = results[inPixel][0]
321 elif inPixel.sum() == 0:
322 # There were no entries in the pixel, but choose the closest.
323 best_result = results[distances.argmin()]
324 else:
325 # There were multiple entries in the pixel. Choose the closest.
326 best_result = results[inPixel][distances[inPixel].argmin()]
327 return best_result
330@dataclasses.dataclass
331class GlobalInput(PossiblyMultipleInput):
332 """Wrapper class for input `~lsst.afw.table.ExposureCatalog` datasets
333 that are not per-tract.
334 """
336 catalog: ExposureCatalog
337 """Loaded per-visit catalog dataset (`lsst.afw.table.ExposureCatalog`).
338 """
340 def best_for_detector(
341 self,
342 detector_id: int,
343 center: SpherePoint | None = None,
344 bbox: Box2I | None = None,
345 ) -> tuple[int, ExposureRecord | None]:
346 # Docstring inherited.
347 return -1, self.catalog.find(detector_id)
350class UpdateVisitSummaryConnections(
351 PipelineTaskConnections,
352 dimensions=("instrument", "visit"),
353 defaultTemplates={
354 "skyWcsName": "gbdesAstrometricFit",
355 "photoCalibName": "fgcm",
356 },
357):
358 sky_map = cT.Input(
359 doc="Description of tract/patch geometry.",
360 name=BaseSkyMap.SKYMAP_DATASET_TYPE_NAME,
361 dimensions=("skymap",),
362 storageClass="SkyMap",
363 )
364 input_summary_schema = cT.InitInput(
365 doc="Schema for input_summary_catalog.",
366 name="visitSummary_schema",
367 storageClass="ExposureCatalog",
368 )
369 input_summary_catalog = cT.Input(
370 doc="Visit summary table to load and modify.",
371 name="visitSummary",
372 dimensions=("instrument", "visit"),
373 storageClass="ExposureCatalog",
374 )
375 input_exposures = cT.Input(
376 doc=(
377 "Per-detector images to obtain image, mask, and variance from "
378 "(embedded summary stats and other components are ignored)."
379 ),
380 name="calexp",
381 dimensions=("instrument", "detector", "visit"),
382 storageClass="ExposureF",
383 multiple=True,
384 deferLoad=True,
385 deferGraphConstraint=True,
386 )
387 psf_overrides = cT.Input(
388 doc="Visit-level catalog of updated PSFs to use.",
389 name="finalized_psf_ap_corr_catalog",
390 dimensions=("instrument", "visit"),
391 storageClass="ExposureCatalog",
392 deferGraphConstraint=True,
393 )
394 psf_star_catalog = cT.Input(
395 doc="Per-visit table of PSF reserved- and used-star measurements.",
396 name="finalized_src_table",
397 dimensions=("instrument", "visit"),
398 storageClass="ArrowAstropy",
399 deferGraphConstraint=True,
400 )
401 ap_corr_overrides = cT.Input(
402 doc="Visit-level catalog of updated aperture correction maps to use.",
403 name="finalized_psf_ap_corr_catalog",
404 dimensions=("instrument", "visit"),
405 storageClass="ExposureCatalog",
406 deferGraphConstraint=True,
407 )
408 photo_calib_overrides_tract = cT.Input(
409 doc="Per-Tract visit-level catalog of updated photometric calibration objects to use.",
410 name="{photoCalibName}PhotoCalibCatalog",
411 dimensions=("instrument", "visit", "tract"),
412 storageClass="ExposureCatalog",
413 multiple=True,
414 deferGraphConstraint=True,
415 )
416 photo_calib_overrides_global = cT.Input(
417 doc="Global visit-level catalog of updated photometric calibration objects to use.",
418 name="{photoCalibName}PhotoCalibCatalog",
419 dimensions=("instrument", "visit"),
420 storageClass="ExposureCatalog",
421 deferGraphConstraint=True,
422 )
423 wcs_overrides_tract = cT.Input(
424 doc="Per-tract visit-level catalog of updated astrometric calibration objects to use.",
425 name="{skyWcsName}SkyWcsCatalog",
426 dimensions=("instrument", "visit", "tract"),
427 storageClass="ExposureCatalog",
428 multiple=True,
429 deferGraphConstraint=True,
430 )
431 wcs_overrides_skypix = cT.Input(
432 doc="Per-skypix pixel visit-level catalog of updated astrometric calibration objects to use.",
433 name="gbdesHealpix3AstrometricFitSkyWcsCatalog",
434 dimensions=("instrument", "visit", "healpix3"),
435 storageClass="ExposureCatalog",
436 multiple=True,
437 deferGraphConstraint=True,
438 )
439 wcs_overrides_global = cT.Input(
440 doc="Global visit-level catalog of updated astrometric calibration objects to use.",
441 name="{skyWcsName}SkyWcsCatalog",
442 dimensions=("instrument", "visit"),
443 storageClass="ExposureCatalog",
444 deferGraphConstraint=True,
445 )
446 background_originals = cT.Input(
447 doc="Per-detector original background that has already been subtracted from 'input_exposures'.",
448 name="calexpBackground",
449 dimensions=("instrument", "visit", "detector"),
450 storageClass="Background",
451 multiple=True,
452 deferLoad=True,
453 deferGraphConstraint=True,
454 )
455 background_overrides = cT.Input(
456 doc="Per-detector background that can be subtracted directly from 'input_exposures'.",
457 name="skyCorr",
458 dimensions=("instrument", "visit", "detector"),
459 storageClass="Background",
460 multiple=True,
461 deferLoad=True,
462 deferGraphConstraint=True,
463 )
464 output_summary_schema = cT.InitOutput(
465 doc="Schema of the output visit summary catalog.",
466 name="finalVisitSummary_schema",
467 storageClass="ExposureCatalog",
468 )
469 output_summary_catalog = cT.Output(
470 doc="Visit-level catalog summarizing all image characterizations and calibrations.",
471 name="finalVisitSummary",
472 dimensions=("instrument", "visit"),
473 storageClass="ExposureCatalog",
474 )
476 def __init__(self, *, config: UpdateVisitSummaryConfig | None = None):
477 super().__init__(config=config)
478 match self.config.wcs_provider:
479 case "input_summary":
480 self.inputs.remove("wcs_overrides_tract")
481 self.inputs.remove("wcs_overrides_global")
482 self.inputs.remove("wcs_overrides_skypix")
483 case "tract":
484 self.inputs.remove("wcs_overrides_global")
485 self.inputs.remove("wcs_overrides_skypix")
486 case "global":
487 self.inputs.remove("wcs_overrides_tract")
488 self.inputs.remove("wcs_overrides_skypix")
489 case x if "healpix" in x:
490 self.inputs.remove("wcs_overrides_tract")
491 self.inputs.remove("wcs_overrides_global")
492 self.wcs_overrides_skypix = dataclasses.replace(
493 self.wcs_overrides_skypix, dimensions=("instrument", "visit", self.config.wcs_provider)
494 )
495 case bad:
496 raise ValueError(f"Invalid value wcs_provider={bad!r}; config was not validated.")
497 match self.config.photo_calib_provider:
498 case "input_summary":
499 self.inputs.remove("photo_calib_overrides_tract")
500 self.inputs.remove("photo_calib_overrides_global")
501 case "tract":
502 self.inputs.remove("photo_calib_overrides_global")
503 case "global":
504 self.inputs.remove("photo_calib_overrides_tract")
505 case bad:
506 raise ValueError(f"Invalid value photo_calib_provider={bad!r}; config was not validated.")
507 match self.config.background_provider:
508 case "input_summary":
509 self.inputs.remove("background_originals")
510 self.inputs.remove("background_overrides")
511 case "replacement":
512 pass
513 case bad:
514 raise ValueError(f"Invalid value background_provider={bad!r}; config was not validated.")
517class UpdateVisitSummaryConfig(PipelineTaskConfig, pipelineConnections=UpdateVisitSummaryConnections):
518 """Configuration for UpdateVisitSummaryTask.
520 Notes
521 -----
522 The configuration defaults for this task reflect a simple or "least common
523 denominator" pipeline, not the more complete, more sophisticated pipeline
524 we run on the instruments we support best. The expectation is that the
525 various full pipeline definitions will generally import the simpler
526 definition, so making the defaults correspond to any full pipeline would
527 just lead to the simple pipeline setting them back to the simple-pipeline
528 values and the full pipeline still having to then override them to the
529 full-pipeline values.
530 """
532 compute_summary_stats = ConfigurableField(
533 doc="Subtask that computes summary statistics from Exposure components.",
534 target=ComputeExposureSummaryStatsTask,
535 )
536 wcs_provider = Field(
537 doc=(
538 "Which connection and behavior to use when applying WCS overrides."
539 "The string should be one of the following:\n"
540 "- input_summary: Propagate the WCS from the input visit summary "
541 "catalog and do not recompute WCS-based summary statistics.\n"
542 "- tract: Use the 'wcs_overrides_tract' connection to load an "
543 "`ExposureCatalog` with {visit, tract} dimensions and per-"
544 "detector rows, and recommpute WCS-based summary statistics.\n"
545 "- global: Use the 'wcs_overrides_global' connection to load an "
546 "`ExposureCatalog` with {visit} dimensions and per-detector rows,"
547 " and recommpute WCS-based summary statistics.\n"
548 "- healpixN: Use the 'wcs_overrides_skypix' connection to load an "
549 "`ExposureCatalog` with {visit, healpixN} dimensions and per-"
550 "detector rows, and recommpute WCS-based summary statistics."
551 ),
552 # If needed, we could add options here to propagate the WCS from
553 # the input exposures and/or transfer WCS-based summary statistics
554 # from them as well. Right now there's no use case for that, since
555 # the input visit summary is always produced after the last time we
556 # write a new Exposure.
557 dtype=str,
558 default="input_summary",
559 optional=False,
560 check=_validate_wcs_provider,
561 )
562 photo_calib_provider = ChoiceField(
563 doc="Which connection and behavior to use when applying photometric calibration overrides.",
564 dtype=str,
565 allowed={
566 "input_summary": (
567 "Propagate the PhotoCalib from the input visit summary catalog "
568 "and do not recompute photometric calibration summary "
569 "statistics."
570 ),
571 "tract": {
572 "Use the 'photo_calib_overrides_tract' connection to load an "
573 "`ExposureCatalog` with {visit, tract} dimensions and per-"
574 "detector rows, and recommpute photometric calibration summary "
575 "statistics."
576 },
577 "global": {
578 "Use the 'photo_calib_overrides_global' connection to load an "
579 "`ExposureCatalog` with {visit} dimensions and per-"
580 "detector rows, and recommpute photometric calibration summary "
581 "statistics."
582 },
583 # If needed, we could add options here to propagate the PhotoCalib
584 # from the input exposures and/or transfer photometric calibration
585 # summary statistics them as well. Right now there's no use case
586 # for that, since the input visit summary is always produced after
587 # the last time we write a new Exposure.
588 },
589 default="input_summary",
590 optional=False,
591 )
592 background_provider = ChoiceField(
593 doc="Which connection(s) and behavior to use when applying background overrides.",
594 dtype=str,
595 allowed={
596 "input_summary": (
597 "The input visit summary catalog already includes summary "
598 "statistics for the final backgrounds that can be used as-is."
599 ),
600 "replacement": {
601 "The 'background_originals' connection refers to a background "
602 "model that has been superseded by the model referred to by "
603 "the 'background_overrides' connection."
604 },
605 # Could also imagine an option in which there is no original
606 # background and the new one stands alone; can add later if needed.
607 },
608 default="input_summary",
609 optional=False,
610 )
611 # Could imagine an option here to say that the original background has not
612 # been subtracted from the input exposures, allowing postISRCCD to be used
613 # as input exposures. Can add later if needed.
616class UpdateVisitSummaryTask(PipelineTask):
617 """A pipeline task that creates a new visit-summary table after all
618 `lsst.afw.image.Exposure` components have been finalized.
620 Notes
621 -----
622 This task is designed to be run just prior to making warps for coaddition,
623 as it aggregates all inputs other than the images and backgrounds into a
624 single ``ExposureCatalog`` dataset and recomputes summary statistics that
625 are useful in selecting which images should go into a coadd. Its output
626 can also be used to reconstruct a final processed visit image when combined
627 with a post-ISR image, the background model, and the final mask.
628 """
630 # The `run` method of this task can conditionally apply overrides for PSFs
631 # and aperture corrections, but its `PipelineTask` interface always applies
632 # them. We can always add the config options to make them optional later,
633 # if that turns out to be useful.
635 _DefaultName = "updateVisitSummary"
636 ConfigClass = UpdateVisitSummaryConfig
638 compute_summary_stats: ComputeExposureSummaryStatsTask
640 def __init__(self, *, initInputs: dict[str, Any] | None = None, **kwargs: Any):
641 super().__init__(initInputs=initInputs, **kwargs)
642 self.makeSubtask("compute_summary_stats")
643 if initInputs is None or "input_summary_schema" not in initInputs:
644 raise RuntimeError("Task requires 'input_summary_schema' in initInputs.")
645 input_summary_schema = initInputs["input_summary_schema"].schema
646 self.schema_mapper = SchemaMapper(input_summary_schema)
647 self.schema_mapper.addMinimalSchema(input_summary_schema)
648 self.schema = self.schema_mapper.getOutputSchema()
649 if self.config.wcs_provider == "tract":
650 self.schema.addField("wcsTractId", type="L", doc="ID of the tract that provided the WCS.")
651 elif "healpix" in self.config.wcs_provider:
652 self.schema.addField("wcsSkypixId", type="L", doc="ID of the skypix pixel that provided the WCS.")
653 if self.config.photo_calib_provider == "tract":
654 self.schema.addField(
655 "photoCalibTractId",
656 type="L",
657 doc="ID of the tract that provided the PhotoCalib.",
658 )
659 self.output_summary_schema = ExposureCatalog(self.schema)
661 def runQuantum(
662 self,
663 butlerQC: QuantumContext,
664 inputRefs: InputQuantizedConnection,
665 outputRefs: OutputQuantizedConnection,
666 ) -> None:
667 # Docstring inherited.
668 sky_map = butlerQC.get(inputRefs.sky_map)
669 del inputRefs.sky_map
670 inputs = {}
671 # Collapse the wcs_override_ and photo_calib_override_ connection pairs
672 # into individual inputs (either ExposureCatalog or PerTractInput
673 # objects).
674 match self.config.wcs_provider:
675 case "tract":
676 inputs["wcs_overrides"] = PerTractInput.load(butlerQC, sky_map, inputRefs.wcs_overrides_tract)
677 del inputRefs.wcs_overrides_tract
678 case "global":
679 inputs["wcs_overrides"] = GlobalInput(butlerQC.get(inputRefs.wcs_overrides_global))
680 del inputRefs.wcs_overrides_global
681 case "input_summary":
682 inputs["wcs_overrides"] = None
683 case x if "healpix" in x:
684 # TODO: match healpix of arbitrary int?
685 inputs["wcs_overrides"] = PerHealpixInput.load(butlerQC, inputRefs.wcs_overrides_skypix)
686 del inputRefs.wcs_overrides_skypix
687 match self.config.photo_calib_provider:
688 case "tract":
689 inputs["photo_calib_overrides"] = PerTractInput.load(
690 butlerQC, sky_map, inputRefs.photo_calib_overrides_tract
691 )
692 del inputRefs.photo_calib_overrides_tract
693 case "global":
694 inputs["photo_calib_overrides"] = GlobalInput(
695 butlerQC.get(inputRefs.photo_calib_overrides_global)
696 )
697 del inputRefs.photo_calib_overrides_global
698 case "input_summary":
699 inputs["photo_calib_overrides"] = None
700 # Load or make DeferredDatasetHandles for everything else.
701 inputs.update(butlerQC.get(inputRefs))
702 deferred_dataset_types = ["input_exposures"]
703 # Handle whether to look for background originals and overrides at all.
704 match self.config.background_provider:
705 case "replacement":
706 deferred_dataset_types.append("background_originals")
707 deferred_dataset_types.append("background_overrides")
708 # Transform the lists of DeferredDatasetHandles for the multiple=True,
709 # deferLoad=True connections into mappings keyed by detector ID.
710 for name in deferred_dataset_types:
711 handles_list = inputs[name]
712 inputs[name] = {handle.dataId["detector"]: handle for handle in handles_list}
713 for record in inputs["input_summary_catalog"]:
714 detector_id = record.getId()
715 if detector_id not in inputs[name]:
716 raise InvalidQuantumError(
717 f"No {name!r} with detector {detector_id} for visit "
718 f"{butlerQC.quantum.dataId['visit']} even though this detector is present "
719 "in the input visit summary catalog. "
720 "This is most likely to occur when the QuantumGraph that includes this task "
721 "was incorrectly generated with an explicit or implicit (from datasets) tract "
722 "constraint."
723 )
724 # Actually run the task and write the results.
725 outputs = self.run(**inputs)
726 butlerQC.put(outputs, outputRefs)
728 def run(
729 self,
730 input_summary_catalog: ExposureCatalog,
731 input_exposures: Mapping[int, DeferredDatasetHandle],
732 psf_overrides: ExposureCatalog,
733 psf_star_catalog: astropy.table.Table,
734 ap_corr_overrides: ExposureCatalog,
735 photo_calib_overrides: PossiblyMultipleInput | None = None,
736 wcs_overrides: PossiblyMultipleInput | None = None,
737 background_originals: Mapping[int, DeferredDatasetHandle] | None = None,
738 background_overrides: Mapping[int, DeferredDatasetHandle] | None = None,
739 ):
740 """Build an updated version of a visit summary catalog.
742 Parameters
743 ----------
744 input_summary_catalog : `lsst.afw.table.ExposureCatalog`
745 Input catalog. Each row in this catalog will be used to produce
746 a row in the output catalog. Any override parameter that is `None`
747 will leave the corresponding values unchanged from those in this
748 input catalog.
749 input_exposures : `collections.abc.Mapping` [`int`,
750 `lsst.daf.butler.DeferredDatasetHandle`]
751 Deferred-load objects that fetch `lsst.afw.image.Exposure`
752 instances. Only the image, mask, and variance are used; all other
753 components are assumed to be superceded by at least
754 ``input_summary_catalog`` and probably some ``_overrides``
755 arguments as well. This usually corresponds to the ``calexp``
756 dataset.
757 psf_overrides : `lsst.afw.table.ExposureCatalog`
758 Catalog with attached `lsst.afw.detection.Psf` objects that
759 supersede the input catalog's PSFs.
760 psf_star_catalog : `astropy.table.Table`
761 Table containing PSF stars for use in computing PSF summary
762 statistics.
763 ap_corr_overrides : `lsst.afw.table.ExposureCatalog`
764 Catalog with attached `lsst.afw.image.ApCorrMap` objects that
765 supersede the input catalog's aperture corrections.
766 photo_calib_overrides : `PossiblyMultipleInput`, optional
767 Catalog wrappers with attached `lsst.afw.image.PhotoCalib`
768 objects that supersede the input catalog's photometric
769 calibrations.
770 wcs_overrides : `PossiblyMultipleInput`, optional
771 Catalog wrappers with attached `lsst.afw.geom.SkyWcs` objects
772 that supersede the input catalog's astrometric calibrations.
773 background_originals : `collections.abc.Mapping` [`int`,
774 `lsst.daf.butler.DeferredDatasetHandle`], optional
775 Deferred-load objects that fetch `lsst.afw.math.BackgroundList`
776 instances. These should correspond to the background already
777 subtracted from ``input_exposures``. If not provided and
778 ``background_overrides`` is, it is assumed that the background in
779 ``input_exposures`` has not been subtracted. If provided, all keys
780 in ``background_overrides`` must also be present in
781 ``background_originals``.
782 background_overrides : `collections.abc.Mapping` [`int`,
783 `lsst.daf.butler.DeferredDatasetHandle`], optional
784 Deferred-load objects that fetch `lsst.afw.math.BackgroundList`
785 instances. These should correspond to the background that should
786 now be subtracted from``input_exposures`` to yield the final
787 background-subtracted image.
789 Returns
790 -------
791 output_summary_catalog : `lsst.afw.table.ExposureCatalog`
792 Output visit summary catalog.
794 Notes
795 -----
796 If any override parameter is provided but does not have a value for a
797 particular detector, that component will be set to `None` in the
798 returned catalog for that detector and all summary statistics derived
799 from that component will be reset (usually to ``NaN``) as well. Not
800 passing an override parameter at all will instead pass through the
801 original component and values from the input catalog unchanged.
802 """
803 output_summary_catalog = ExposureCatalog(self.schema)
804 output_summary_catalog.setMetadata(input_summary_catalog.getMetadata())
805 for input_record in input_summary_catalog:
806 detector_id = input_record.getId()
807 output_record = output_summary_catalog.addNew()
809 # Make a new ExposureSummaryStats from the input record. These
810 # might be full of NaNs if the summary stats were not computed
811 # originally because initial astrometry or photometry failed, but
812 # if that's possible there should be a flag for it.
813 summary_stats = ExposureSummaryStats.from_record(input_record)
815 # Also copy the input record values to output record; this copies
816 # many of the same values just copied into `summary_stats` (which
817 # will be overridden later by summary_stats.update_record), but it
818 # also copies fields that aren't part of summary_stats, including
819 # the actual components like Psf, Wcs, etc.
820 output_record.assign(input_record, self.schema_mapper)
822 exposure = input_exposures[detector_id].get()
823 bbox = exposure.getBBox()
825 if wcs_overrides:
826 wcs_region, wcs_record = wcs_overrides.best_for_detector(detector_id, bbox=bbox)
827 if wcs_record is not None:
828 wcs = wcs_record.getWcs()
829 else:
830 wcs = None
831 if self.config.wcs_provider == "tract":
832 output_record["wcsTractId"] = wcs_region
833 elif "healpix" in self.config.wcs_provider:
834 output_record["wcsSkypixId"] = wcs_region
835 output_record.setWcs(wcs)
836 self.compute_summary_stats.update_wcs_stats(
837 summary_stats, wcs, bbox, output_record.getVisitInfo()
838 )
839 else:
840 wcs = input_record.getWcs()
842 sources = None
844 if (psf_record := psf_overrides.find(detector_id)) is not None:
845 psf = psf_record.getPsf()
846 else:
847 psf = None
848 output_record.setPsf(psf)
849 sources = psf_star_catalog[psf_star_catalog["detector"] == detector_id]
850 if len(sources) == 0:
851 sources = None
853 if (ap_corr_record := ap_corr_overrides.find(detector_id)) is not None:
854 ap_corr = ap_corr_record.getApCorrMap()
855 else:
856 ap_corr = None
857 output_record.setApCorrMap(ap_corr)
858 self.compute_summary_stats.update_psf_stats(
859 summary_stats,
860 psf,
861 bbox,
862 sources,
863 image_mask=exposure.mask,
864 image_ap_corr_map=ap_corr,
865 sources_is_astropy=True,
866 )
868 if photo_calib_overrides:
869 center = compute_center_for_detector_record(output_record, bbox, wcs)
870 (
871 photo_calib_tract,
872 photo_calib_record,
873 ) = photo_calib_overrides.best_for_detector(detector_id, center=center)
874 if photo_calib_record is not None:
875 photo_calib = photo_calib_record.getPhotoCalib()
876 else:
877 photo_calib = None
878 if self.config.photo_calib_provider == "tract":
879 output_record["photoCalibTractId"] = photo_calib_tract
880 output_record.setPhotoCalib(photo_calib)
881 self.compute_summary_stats.update_photo_calib_stats(summary_stats, photo_calib)
883 if background_overrides is not None:
884 if (handle := background_overrides.get(detector_id)) is not None:
885 new_bkg = handle.get()
886 if background_originals is not None:
887 orig_bkg = background_originals[detector_id].get()
888 else:
889 orig_bkg = BackgroundList()
891 full_bkg = orig_bkg.clone()
892 for layer in new_bkg:
893 full_bkg.append(layer)
894 exposure.image -= new_bkg.getImage()
895 self.compute_summary_stats.update_background_stats(summary_stats, full_bkg)
896 self.compute_summary_stats.update_masked_image_stats(
897 summary_stats, exposure.getMaskedImage()
898 )
900 # Update the estimated depth and effective exposure time.
901 self.compute_summary_stats.update_magnitude_limit_stats(summary_stats, exposure)
902 self.compute_summary_stats.update_effective_time_stats(summary_stats, exposure)
904 summary_stats.update_record(output_record)
905 del exposure
907 return Struct(output_summary_catalog=output_summary_catalog)