Skip to content

Commit

Permalink
Add unit test of quickLook.runQuantum
Browse files Browse the repository at this point in the history
  • Loading branch information
@isullivan
isullivan committed Jan 26, 2024
1 parent 549e6c2 commit c780214
Showing 1 changed file with 183 additions and 54 deletions.
237 changes: 183 additions & 54 deletions tests/test_quickLook.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,15 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

import contextlib
import tempfile
import unittest

import numpy as np

import lsst.afw.cameraGeom.testUtils as afwTestUtils
import lsst.afw.image as afwImage
from lsst.afw.image import TransmissionCurve
import lsst.daf.butler.tests as butlerTests
import lsst.ip.isr.isrMock as isrMock
import lsst.ip.isr as ipIsr
Expand All @@ -33,6 +38,10 @@


class QuickLookIsrTaskTestCase(unittest.TestCase):

"""Tests of the run method with fake data.
"""

def setUp(self):
self.mockConfig = isrMock.IsrMockConfig()
self.dataContainer = isrMock.MockDataContainer(config=self.mockConfig)
Expand All @@ -50,9 +59,9 @@ def setUp(self):
self.defects = self.dataContainer.get("defects")
self.ptc = ipIsr.PhotonTransferCurveDataset(ampNames=ampNames) # Mock PTC dataset
self.bfKernel = self.dataContainer.get("bfKernel")
self.bfGains = {}
self.newBFKernel = pipeBase.Struct(gain={})
for amp_i, amp in enumerate(ampNames):
self.bfGains[amp] = 0.9 + 0.1*amp_i
self.newBFKernel.gain[amp] = 0.9 + 0.1*amp_i
self.task = QuickLookIsrTask(config=QuickLookIsrTaskConfig())

def test_runQuickLook(self):
Expand All @@ -66,7 +75,7 @@ def test_runQuickLook(self):
linearizer=None,
crosstalk=None,
bfKernel=self.bfKernel,
bfGains=self.bfGains,
newBFKernel=self.newBFKernel,
ptc=self.ptc,
crosstalkSources=None
)
Expand All @@ -89,96 +98,216 @@ def test_runQuickLookBadDark(self):
dark=self.dark[bbox], flat=self.flat, defects=self.defects)


class CalibrateImageTaskRunQuantumTests(lsst.utils.tests.TestCase):
"""Tests of ``CalibrateImageTask.runQuantum``, which need a test butler,
class QuickLookIsrTaskRunQuantumTests(lsst.utils.tests.TestCase):
"""Tests of ``QuickLookIsrTask.runQuantum``, which need a test butler,
but do not need real images.
Adapted from the unit tests of ``CalibrateImageTask.runQuantum``
"""
def setUp(self):
QuickLookIsrTaskTestCase.setUp(self)
instrument = "testCam"
exposure = 100
exposureId = 100
visit = 100101
detector = self.detector.getId()
detector = 0
physical_filter = "testCam_filter"
band = "X"

# Map the isrTask connection names to the names of the Butler dataset
# inputs
ccdExposure = "raw"
camera = "camera"
bias = "bias"
dark = "dark"
flat = "flat"
defects = "defects"
bfKernel = "bfKernel"
newBFKernel = "brighterFatterKernel"
ptc = "ptc"
filterTransmission = "transmission_filter"
deferredChargeCalib = "cpCtiCalib"
opticsTransmission = "transmission_optics"
strayLightData = "yBackground"
atmosphereTransmission = "transmission_atmosphere"
crosstalk = "crosstalk"
illumMaskedImage = "illum"
linearizer = "linearizer"
fringes = "fringe"
sensorTransmission = "transmission_sensor"
crosstalkSources = "isrOverscanCorrected"

# outputs
outputExposure = "postISRCCD"

# quickLook-only outputs
exposure = "quickLookExp"

# Create a and populate a test butler for runQuantum tests.
self.repo_path = tempfile.TemporaryDirectory()
self.repo_path = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
self.repo = butlerTests.makeTestRepo(self.repo_path.name)

# dataIds for fake data
butlerTests.addDataIdValue(self.repo, "instrument", instrument)
butlerTests.addDataIdValue(self.repo, "exposure", exposure)
butlerTests.addDataIdValue(self.repo, "physical_filter", physical_filter)
butlerTests.addDataIdValue(self.repo, "physical_filter", physical_filter, band=band)
butlerTests.addDataIdValue(self.repo, "detector", detector)
# butlerTests.addDataIdValue(self.repo, "detector", detector + 1)
butlerTests.addDataIdValue(self.repo, "exposure", exposureId, physical_filter=physical_filter)
butlerTests.addDataIdValue(self.repo, "visit", visit)

# inputs
butlerTests.addDatasetType(self.repo, "ccdExposure", {"instrument", "exposure", "detector"},
"Exposure")
butlerTests.addDatasetType(self.repo, "camera", {"instrument"}, "Camera")
butlerTests.addDatasetType(self.repo, "bias", {"instrument", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, "dark", {"instrument", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, "flat", {"instrument", "physical_filter", "detector"},
"Exposure")
butlerTests.addDatasetType(self.repo, "defects", {"instrument", "detector"}, "Defects")
butlerTests.addDatasetType(self.repo, "linearizer", {"instrument", "detector"}, "Linearizer")
butlerTests.addDatasetType(self.repo, "crosstalk", {"instrument", "detector"}, "CrosstalkCalib")
butlerTests.addDatasetType(self.repo, "bfKernel", {"instrument"}, "NumpyArray")
butlerTests.addDatasetType(self.repo, "newBFKernel", {"instrument", "detector"},
"BrighterFatterKernel")
butlerTests.addDatasetType(self.repo, "ptc", {"instrument", "detector"}, "PhotonTransferCurveDataset")
butlerTests.addDatasetType(self.repo, "crosstalkSources", {"instrument", "exposure", "detector"},
"Exposure")
butlerTests.addDatasetType(self.repo, ccdExposure,
{"instrument", "exposure", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, camera,
{"instrument"}, "Camera")
butlerTests.addDatasetType(self.repo, bias,
{"instrument", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, dark,
{"instrument", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, flat,
{"instrument", "physical_filter", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, defects,
{"instrument", "detector"}, "Defects")
butlerTests.addDatasetType(self.repo, linearizer,
{"instrument", "detector"}, "Linearizer")
butlerTests.addDatasetType(self.repo, crosstalk,
{"instrument", "detector"}, "CrosstalkCalib")
butlerTests.addDatasetType(self.repo, bfKernel,
{"instrument"}, "NumpyArray")
butlerTests.addDatasetType(self.repo, newBFKernel,
{"instrument", "detector"}, "BrighterFatterKernel")
butlerTests.addDatasetType(self.repo, ptc,
{"instrument", "detector"}, "PhotonTransferCurveDataset")
butlerTests.addDatasetType(self.repo, filterTransmission,
{"instrument", "physical_filter"}, "TransmissionCurve")
butlerTests.addDatasetType(self.repo, opticsTransmission,
{"instrument"}, "TransmissionCurve")
butlerTests.addDatasetType(self.repo, deferredChargeCalib,
{"instrument", "detector"}, "IsrCalib")
butlerTests.addDatasetType(self.repo, strayLightData,
{"instrument", "physical_filter", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, atmosphereTransmission,
{"instrument"}, "TransmissionCurve")
butlerTests.addDatasetType(self.repo, illumMaskedImage,
{"instrument", "physical_filter", "detector"}, "MaskedImage")
butlerTests.addDatasetType(self.repo, fringes,
{"instrument", "physical_filter", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, sensorTransmission,
{"instrument", "detector"}, "TransmissionCurve")
butlerTests.addDatasetType(self.repo, crosstalkSources,
{"instrument", "exposure", "detector"}, "Exposure")

# outputs
butlerTests.addDatasetType(self.repo, outputExposure,
{"instrument", "exposure", "detector"}, "Exposure")
butlerTests.addDatasetType(self.repo, exposure,
{"instrument", "exposure", "detector"}, "Exposure")

# dataIds
self.exposure_id = self.repo.registry.expandDataId(
{"instrument": instrument, "exposure": exposure, "detector": detector})
{"instrument": instrument, "exposure": exposureId, "detector": detector,
"physical_filter": physical_filter})
self.instrument_id = self.repo.registry.expandDataId(
{"instrument": instrument})
self.flat_id = self.repo.registry.expandDataId(
{"instrument": instrument, "physical_filter": physical_filter, "detector": detector})
self.detector_id = self.repo.registry.expandDataId(
{"instrument": instrument, "detector": detector})
self.visit_id = self.repo.registry.expandDataId(
{"instrument": instrument, "visit": visit, "detector": detector})
self.filter_id = self.repo.registry.expandDataId(
{"instrument": instrument, "physical_filter": physical_filter})

# put empty data
transmissionCurve = TransmissionCurve.makeSpatiallyConstant(np.ones(2), np.linspace(0, 1, 2), 0., 0.)
self.butler = butlerTests.makeTestCollection(self.repo)
self.butler.put(self.ccdExposure, "ccdExposure", self.exposure_id)
self.butler.put(self.camera, "camera", self.instrument_id)
self.butler.put(self.bias, "bias", self.detector_id)
self.butler.put(self.dark, "dark", self.detector_id)
self.butler.put(self.flat, "flat", self.flat_id)
self.butler.put(self.defects, "defects", self.detector_id)
# self.butler.put(self.ccdExposure, "linearizer", self.detector_id)
# self.butler.put(self.ccdExposure, "crosstalk", self.detector_id)
self.butler.put(self.bfKernel, "bfKernel", self.instrument_id)
# self.butler.put(self.bfGains, "newBFKernel", self.detector_id)
self.butler.put(self.ptc, "ptc", self.detector_id)
# self.butler.put(self.ccdExposure, "crosstalkSources", self.exposure_id)
self.butler.put(afwImage.ExposureF(), ccdExposure, self.exposure_id)
self.butler.put(afwTestUtils.CameraWrapper().camera, camera, self.instrument_id)
self.butler.put(afwImage.ExposureF(), bias, self.detector_id)
self.butler.put(afwImage.ExposureF(), dark, self.detector_id)
self.butler.put(afwImage.ExposureF(), flat, self.flat_id)
self.butler.put(lsst.ip.isr.Defects(), defects, self.detector_id)
self.butler.put(np.zeros(2), bfKernel, self.instrument_id)
self.butler.put(lsst.ip.isr.brighterFatterKernel.BrighterFatterKernel(),
newBFKernel, self.detector_id)
self.butler.put(ipIsr.PhotonTransferCurveDataset(), ptc, self.detector_id)
self.butler.put(transmissionCurve, filterTransmission, self.filter_id)
self.butler.put(lsst.ip.isr.calibType.IsrCalib(), deferredChargeCalib, self.detector_id)
self.butler.put(transmissionCurve, opticsTransmission, self.instrument_id)
self.butler.put(afwImage.ExposureF(), strayLightData, self.flat_id)
self.butler.put(transmissionCurve, atmosphereTransmission, self.instrument_id)
self.butler.put(lsst.ip.isr.crosstalk.CrosstalkCalib(), crosstalk, self.detector_id)
self.butler.put(afwImage.ExposureF().maskedImage, illumMaskedImage, self.flat_id)
self.butler.put(lsst.ip.isr.linearize.Linearizer(), linearizer, self.detector_id)
self.butler.put(afwImage.ExposureF(), fringes, self.flat_id)
self.butler.put(transmissionCurve, sensorTransmission, self.detector_id)
self.butler.put(afwImage.ExposureF(), crosstalkSources, self.exposure_id)

def tearDown(self):
del self.repo_path # this removes the temporary directory

def test_runQuantum(self):
task = self.task
config = ipIsr.IsrTaskConfig()
# Remove some outputs
config.doBinnedExposures = False
config.doSaveInterpPixels = False
config.qa.doThumbnailOss = False
config.qa.doThumbnailFlattened = False
config.doCalculateStatistics = False

# Turn on all optional inputs
config.doAttachTransmissionCurve = True
config.doIlluminationCorrection = True
config.doStrayLight = True
config.doDeferredCharge = True
config.usePtcReadNoise = True
config.doCrosstalk = True
config.doBrighterFatter = False

# Override a method in IsrTask that is executed early, to instead raise
# a custom exception called ExitMock that we can catch and ignore.
isrTask = ipIsr.IsrTask
isrTask.ensureExposure = raiseExitMockError
task = QuickLookIsrTask(isrTask=isrTask)
lsst.pipe.base.testUtils.assertValidInitOutput(task)

# Use the names of the connections here, not the Butler dataset name
quantum = lsst.pipe.base.testUtils.makeQuantum(
task, self.butler, self.exposure_id,
{"blah": [self.exposure_id],
"camera": [self.instrument_id],
"bias": [self.detector_id],
"dark": [self.detector_id],
"flat": [self.flat_id],
"defects": [self.detector_id],
"bfKernel": [self.instrument_id],
# "newBFKernel": [self.detector_id],
"ptc": [self.detector_id],
{"ccdExposure": self.exposure_id,
"camera": self.instrument_id,
"bias": self.detector_id,
"dark": self.detector_id,
"flat": self.flat_id,
"defects": self.detector_id,
"bfKernel": self.instrument_id,
"newBFKernel": self.detector_id,
"ptc": self.detector_id,
"filterTransmission": self.filter_id,
"deferredChargeCalib": self.detector_id,
"opticsTransmission": self.instrument_id,
"strayLightData": self.flat_id,
"atmosphereTransmission": self.instrument_id,
"crosstalk": self.detector_id,
"illumMaskedImage": self.flat_id,
"linearizer": self.detector_id,
"fringes": self.flat_id,
"sensorTransmission": self.detector_id,
"crosstalkSources": [self.exposure_id, self.exposure_id],
# outputs
"outputExposure": self.exposure_id,
"exposure": self.exposure_id,
})
mock_run = lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum)
# Check that the proper kwargs are passed to run().
self.assertEqual(mock_run.call_args.kwargs.keys(), {"exposures"})
with contextlib.suppress(ExitMockError):
lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum, mockRun=False)


def raiseExitMockError(*args):
"""Raise a custom exception.
"""
raise ExitMockError


class ExitMockError(Exception):

"""A custom exception to catch during a unit test.
"""

pass

0 comments on commit c780214

Please sign in to comment.