_rewrite_difference_image.py

Go to the documentation of this file.

# This file is part of pipe_tasks.
#
# Developed for the LSST Data Management System.
# This product includes software developed by the LSST Project
# (https://www.lsst.org).
# See the COPYRIGHT file at the top-level directory of this distribution
# for details of code ownership.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
from __future__ import annotations
 
__all__ = ("RewriteDifferenceImageConnections", "RewriteDifferenceImageTask", "RewriteDifferenceImageConfig")
 
import uuid
from collections.abc import Mapping
from typing import ClassVar
 
import astropy.units
 
import lsst.pipe.base.connectionTypes as cT
from lsst.afw.image import PhotoCalib
from lsst.afw.math import BackgroundList, Kernel
from lsst.daf.base import PropertyList
from lsst.daf.butler import DataCoordinate
from lsst.images import DifferenceImage, DifferenceImageTemplateInfo
from lsst.images.convolution_kernels import ImageBasisConvolutionKernel
from lsst.images.fields import field_from_legacy_background, field_from_legacy_photo_calib
from lsst.meas.algorithms import CoaddPsf
from lsst.pex.config import Field
from lsst.pipe.base import (
    InputQuantizedConnection,
    OutputQuantizedConnection,
    PipelineTask,
    PipelineTaskConfig,
    PipelineTaskConnections,
    QuantumContext,
    Struct,
)
 
 
class RewriteDifferenceImageConnections(
    PipelineTaskConnections,
    dimensions={"visit", "detector"},
    defaultTemplates={"legacy_prefix": "legacy_", "future_prefix": ""},
):
    legacy_exposure = cT.Input(
        "{legacy_prefix}difference_image",
        # We expect the repository storage class to be ExposureF, but we set
        # the storage class we want to DifferenceImage so the butler can do
        # most of the conversion on read, and in doing so preserve the
        # quantization so we don't have doubly-lossless compression.
        storageClass="DifferenceImage",
        dimensions={"visit", "detector"},
        deferLoad=True,  # So we can pass preserve_quantization=True as a parameter.
        doc="The input image to convert.",
    )
    template_psf = cT.Input(
        "template_detector.psf",
        storageClass="Psf",
        dimensions={"visit", "detector"},
        doc="The PSF of the detector-level template, which holds some provenance information.",
    )
    template_metadata = cT.Input(
        "template_detector.metadata",
        storageClass="PropertyList",
        dimensions={"visit", "detector"},
        doc="The metadata of the detector-level template, which holds some provenance information.",
    )
    template_coadds = cT.Input(
        "template_coadd",
        storageClass="ExposureF",
        dimensions={"tract", "patch", "band"},
        deferLoad=True,
        multiple=True,
        doc=(
            "The template coadds that may have gone into the detector-level template. "
            "This is a dummy input that is not actually loaded; it is needed only to connect "
            "data IDs to butler dataset IDs."
        ),
    )
    visit_summary = cT.Input(
        "visit_summary",
        storageClass="ExposureCatalog",
        dimensions={"visit"},
        doc="A visit summary catalog with the PhotoCalib that was already applied to the image's pixels.",
    )
    difference_kernel = cT.Input(
        "difference_kernel",
        storageClass="MatchingKernel",
        dimensions={"visit", "detector"},
        doc="The kernel used to match the template to the science image.",
    )
    background = cT.Input(
        "difference_image_background",
        storageClass="Background",
        dimensions={"visit", "detector"},
        doc="The background model that was subtracted from this image.",
    )
    future_difference_image = cT.Output(
        "{future_prefix}difference_image",
        storageClass="DifferenceImage",
        dimensions={"visit", "detector"},
        doc="The output difference image.",
    )
 
    config: RewriteDifferenceImageConfig
 
    def __init__(self, config: RewriteDifferenceImageConfig):
        super().__init__(config=config)
        if self.config.background_description is None:
            del self.background
 
 
class RewriteDifferenceImageConfig(
    PipelineTaskConfig,
    pipelineConnections=RewriteDifferenceImageConnections,
):
    instrumental_unit = Field(
        "Unit for instrumental flux pixels (i.e. uncalibrated side of a PhotoCalib).",
        dtype=str,
        default="electron",
    )
    background_description = Field(
        "Description of the subtracted background, to be stored with the image. "
        "If None, no background will be attached.",
        dtype=str,
        optional=True,
        default=None,
    )
 
 
class RewriteDifferenceImageTask(PipelineTask):
    ConfigClass: ClassVar[type[RewriteDifferenceImageConfig]] = RewriteDifferenceImageConfig
    config: RewriteDifferenceImageConfig
    _DefaultName = "rewriteDifferenceImage"
 
    def runQuantum(
        self,
        butlerQC: QuantumContext,
        inputRefs: InputQuantizedConnection,
        outputRefs: OutputQuantizedConnection,
    ) -> None:
        inputs = butlerQC.get(inputRefs)
        difference_image = inputs.pop("legacy_exposure").get(parameters={"preserve_quantization": True})
        visit_summary = inputs.pop("visit_summary")
        coadd_data_ids_by_uuid = {h.ref.id: h.ref.dataId for h in inputs.pop("template_coadds")}
        photo_calib: PhotoCalib = visit_summary.find(butlerQC.quantum.dataId["detector"]).getPhotoCalib()
        outputs = self.run(
            difference_image, photo_calib=photo_calib, **inputs, coadd_data_ids_by_uuid=coadd_data_ids_by_uuid
        )
        butlerQC.put(outputs, outputRefs)
 
    def run(
        self,
        difference_image: DifferenceImage,
        *,
        photo_calib: PhotoCalib,
        background: BackgroundList | None = None,
        difference_kernel: Kernel,
        template_psf: CoaddPsf,
        template_metadata: PropertyList,
        coadd_data_ids_by_uuid: Mapping[uuid.UUID, DataCoordinate],
    ) -> Struct:
        instrumental_unit = astropy.units.Unit(self.config.instrumental_unit)
        difference_image.photometric_scaling = field_from_legacy_photo_calib(
            photo_calib, bounds=difference_image.bbox, instrumental_unit=instrumental_unit
        )
        if background is not None and self.config.background_description:
            difference_image.backgrounds.add(
                "subtracted",
                field_from_legacy_background(
                    background, bounds=difference_image.bbox, unit=difference_image.unit
                ),
                self.config.background_description,
                is_subtracted=True,
            )
        difference_image.kernel = ImageBasisConvolutionKernel.from_legacy(difference_kernel)
        difference_image.templates = DifferenceImageTemplateInfo.from_legacy(
            difference_image.sky_projection.pixel_frame,
            template_psf,
            template_metadata,
            coadd_data_ids_by_uuid,
        )
        return Struct(future_difference_image=difference_image)