lsst.pipe.tasks.make_psf_matched_warp Namespace Reference

Classes
class	MakePsfMatchedWarpConnections

Variables
	direct_warp : `lsst.afw.image.Exposure`
	struct : `lsst.pipe.base.Struct`

Variable Documentation

direct_warp

lsst.pipe.tasks.make_psf_matched_warp.direct_warp : `lsst.afw.image.Exposure`

    sky_map = Input(
        doc="Input definition of geometry/bbox and projection/wcs for warps.",
        name=BaseSkyMap.SKYMAP_DATASET_TYPE_NAME,
        storageClass="SkyMap",
        dimensions=("skymap",),
    )

    direct_warp = Input(
        doc="Direct warped exposure produced by resampling calexps onto the skyMap patch geometry",
        name="{coaddName}Coadd_directWarp",
        storageClass="ExposureF",
        dimensions=("tract", "patch", "skymap", "instrument", "visit"),
    )

    psf_matched_warp = Output(
        doc=(
            "Output PSF-Matched warped exposure, produced by resampling ",
            "calexps onto the skyMap patch geometry and PSF-matching to a model PSF.",
        ),
        name="{coaddName}Coadd_psfMatchedWarp",
        storageClass="ExposureF",
        dimensions=("tract", "patch", "skymap", "visit", "instrument"),
    )


class MakePsfMatchedWarpConfig(
    PipelineTaskConfig,
    pipelineConnections=MakePsfMatchedWarpConnections,
):

    modelPsf = GaussianPsfFactory.makeField(doc="Model Psf factory")
    psfMatch = ConfigurableField(
        target=ModelPsfMatchTask,
        doc="Task to warp and PSF-match calexp",
    )

    def setDefaults(self):
        super().setDefaults()
        self.psfMatch.kernel["AL"].alardSigGauss = [1.0, 2.0, 4.5]
        self.modelPsf.defaultFwhm = 7.7


class MakePsfMatchedWarpTask(PipelineTask):
    ConfigClass = MakePsfMatchedWarpConfig
    _DefaultName = "makePsfMatchedWarp"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.makeSubtask("psfMatch")

    def runQuantum(self, butlerQC, inputRefs, outputRefs):
        # Docstring inherited.

        # Read in all inputs.
        inputs = butlerQC.get(inputRefs)

        sky_map = inputs.pop("sky_map")

        quantumDataId = butlerQC.quantum.dataId
        sky_info = makeSkyInfo(
            sky_map,
            tractId=quantumDataId["tract"],
            patchId=quantumDataId["patch"],
        )

        results = self.run(inputs["direct_warp"], sky_info.bbox)
        butlerQC.put(results, outputRefs)

    @timeMethod
    def run(self, direct_warp: Exposure, bbox: geom.Box2I):

Definition at line 150 of file make_psf_matched_warp.py.

struct

lsst.pipe.tasks.make_psf_matched_warp.struct : `lsst.pipe.base.Struct`

Definition at line 155 of file make_psf_matched_warp.py.