Coverage for tests/test_postprocess.py: 98%
134 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 09:36 +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/>.
22import pytest
23import tempfile
24import unittest
26import astropy.units as u
27from astropy.table import Table
28import numpy as np
30import lsst.afw.geom as afwGeom
31import lsst.afw.image as afwImage
32from lsst.afw.cameraGeom.testUtils import CameraWrapper
33import lsst.daf.butler
34import lsst.daf.butler.tests as butlerTests
35import lsst.geom
36import lsst.images
37import lsst.images.psfs
38import lsst.meas.base.tests
39import lsst.utils.tests
40from lsst.pipe.tasks.postprocess import ConsolidateVisitSummaryTask, ModelExtendednessColumnAction
42from utils import makeTestVisitInfo, make_exposure_record
45class ModelExtendednessColumnActionTestCase(lsst.utils.tests.TestCase):
46 """Demo test case."""
48 def setUp(self):
49 self.bands = ("g", "r", "i")
50 action = ModelExtendednessColumnAction(bands=self.bands, min_n_good_to_shift_flux_ratio=1)
51 self.action = action
52 data = {
53 action.size_column.format(axis="x"): [1., 3., 0.01, 0.4, 0.02],
54 action.size_column.format(axis="y"): [0.2, 6, 0.01, 0.2, 0.01],
55 }
56 model = action.model_flux_name
57 factors = np.array([1.01, 1.12, 0.995, 0.998, 1.6])
58 fluxes = np.array([1.5e3, 2.5e3, 6.8e3, 3.4e3, 5.5e3])
59 for column_flux, column_flux_err, factors in (
60 (action.model_column_flux, action.model_column_flux_err, factors),
61 (action.psf_column_flux, action.psf_column_flux_err, None),
62 ):
63 for idx, band in enumerate(action.bands):
64 flux = np.sqrt(idx + 1.)*fluxes
65 if factors is not None:
66 flux *= factors
67 data[column_flux.format(band=band, model=model)] = flux
68 data[column_flux_err.format(band=band, model=model)] = np.sqrt(flux)
69 self.data = Table(data)
71 def testExtendednessColumnAction(self):
72 action = self.action
73 with pytest.raises(ValueError):
74 action.validate()
75 action.bands_combined = {"gri": "g,r,i"}
76 action.validate()
77 schema = action.getInputSchema()
78 n_values = len(self.data[schema[0][0]])
79 assert all([len(self.data[col]) == n_values for col, _ in schema[1:]])
81 result = self.action(self.data)
82 columns_expected = [
83 action.output_column.format(band=band)
84 for band in list(action.bands) + list(action.bands_combined.keys())
85 ]
86 assert list(result.keys()) == columns_expected
88 for column, values in result.items():
89 assert len(values) == n_values
90 assert all((values >= 0) & (values <= 1))
93class ConsolidateVisitSummaryTestCase(lsst.utils.tests.TestCase):
94 """Test that ConsolidateVisitSummaryTask makes the same row from a legacy
95 Exposure and from the equivalent lsst.images.VisitImage.
96 """
98 def setUp(self):
99 instrument = "testCam"
100 self.visit = 1
101 self.band = "r"
102 physical_filter = "r_test"
104 # lsst.images needs raw amplifier geometry and a FIELD_ANGLE
105 # transform, which the TestDataset detector does not have.
106 detector = list(CameraWrapper().camera)[0]
107 dataset = lsst.meas.base.tests.TestDataset(detector.getBBox())
108 dataset.addSource(instFlux=1e4, centroid=lsst.geom.Point2D(50, 60))
109 exposure, _ = dataset.realize(noise=10.0, schema=dataset.makeMinimalSchema())
110 exposure.setDetector(detector)
111 exposure.info.setVisitInfo(makeTestVisitInfo(id=self.visit).copyWith(exposureTime=30.0))
112 exposure.setFilter(afwImage.FilterLabel(band=self.band, physical=physical_filter))
113 summaryStats = afwImage.ExposureSummaryStats()
114 summaryStats.psfSigma = 2.5
115 summaryStats.zeroPoint = 31.4
116 exposure.info.setSummaryStats(summaryStats)
117 bbox = lsst.geom.Box2D(exposure.getBBox())
118 exposure.info.setValidPolygon(afwGeom.Polygon(
119 [bbox.getMin(), lsst.geom.Point2D(bbox.maxX, bbox.minY), lsst.geom.Point2D(bbox.minX, bbox.maxY)]
120 ))
121 self.exposure = exposure
123 self.repo_path = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
124 self.addCleanup(self.repo_path.cleanup)
125 # A file datastore, so that reads go through the lsst.images
126 # formatter's component support.
127 config = lsst.daf.butler.Config()
128 config["datastore", "cls"] = "lsst.daf.butler.datastores.fileDatastore.FileDatastore"
129 self.repo = butlerTests.makeTestRepo(self.repo_path.name, config=config)
130 self.enterContext(self.repo)
131 instrumentRecord = self.repo.dimensions["instrument"].RecordClass(
132 name=instrument, visit_max=1e6, exposure_max=1e6, detector_max=128,
133 class_name="lsst.obs.base.instrument_tests.DummyCam",
134 )
135 self.repo.registry.syncDimensionData("instrument", instrumentRecord)
136 butlerTests.addDataIdValue(self.repo, "physical_filter", physical_filter, band=self.band)
137 butlerTests.addDataIdValue(self.repo, "detector", detector.getId())
138 butlerTests.addDataIdValue(self.repo, "visit", self.visit, physical_filter=physical_filter)
139 self.dataId = {"instrument": instrument, "visit": self.visit, "detector": detector.getId()}
140 dimensions = {"instrument", "visit", "detector"}
141 butlerTests.addDatasetType(self.repo, "legacy_image", dimensions, "ExposureF")
142 butlerTests.addDatasetType(self.repo, "future_image", dimensions, "VisitImage")
143 self.butler = butlerTests.makeTestCollection(self.repo, uniqueId=self.id())
144 self.enterContext(self.butler)
146 def _run(self, datasetType, image, input_image_type):
147 """Put one image in the butler and consolidate it into a visit
148 summary.
149 """
150 self.butler.put(image, datasetType, self.dataId)
151 ref = self.butler.find_dataset(datasetType, self.dataId, dimension_records=True)
152 config = ConsolidateVisitSummaryTask.ConfigClass()
153 config.do_refit_pointing = False
154 config.do_write_visit_geometry = False
155 config.input_image_type = input_image_type
156 task = ConsolidateVisitSummaryTask(config=config)
157 return task.run(visit=self.visit, handles=[self.butler.getDeferred(ref)]).visitSummary
159 def _checkFutureMatchesLegacy(self, unit):
160 """Check the future-mode row against the legacy-mode row.
162 Parameters
163 ----------
164 unit : `astropy.units.Unit`
165 Units of the image pixels.
167 Returns
168 -------
169 visitImage : `lsst.images.VisitImage`
170 The converted image the future-mode row was made from.
171 """
172 visitImage = lsst.images.VisitImage.from_legacy(
173 self.exposure, exposure_record=make_exposure_record(self.exposure), unit=unit
174 )
175 # The afw GaussianPsf from TestDataset cannot be serialized, and this
176 # task does not read the PSF.
177 visitImage.psf = lsst.images.psfs.GaussianPointSpreadFunction(
178 2.0, stamp_size=21, bounds=visitImage.bbox
179 )
180 legacy = self._run("legacy_image", self.exposure, "legacy")[0]
181 future = self._run("future_image", visitImage, "future")[0]
183 self.assertEqual(future.getId(), legacy.getId())
184 self.assertEqual(future["visit"], legacy["visit"])
185 self.assertEqual(future["physical_filter"], legacy["physical_filter"])
186 self.assertEqual(future["band"], legacy["band"])
187 self.assertEqual(future.getBBox(), legacy.getBBox())
188 self.assertEqual(future.getValidPolygon(), legacy.getValidPolygon())
189 self.assertWcsAlmostEqualOverBBox(future.getWcs(), legacy.getWcs(), legacy.getBBox())
190 self.assertFloatsAlmostEqual(
191 future.getPhotoCalib().getCalibrationMean(),
192 legacy.getPhotoCalib().getCalibrationMean(),
193 rtol=1e-5,
194 )
195 self.assertEqual(future.getVisitInfo().id, legacy.getVisitInfo().id)
196 self.assertEqual(future.getVisitInfo().exposureTime, legacy.getVisitInfo().exposureTime)
197 self.assertEqual(future["psfSigma"], legacy["psfSigma"])
198 self.assertEqual(future["zeroPoint"], legacy["zeroPoint"])
199 return visitImage
201 def testCalibratedImage(self):
202 """Test an image calibrated to nJy, as calibrateImage writes."""
203 self.exposure.setPhotoCalib(afwImage.PhotoCalib(1.0))
204 visitImage = self._checkFutureMatchesLegacy(u.nJy)
205 self.assertIsNone(visitImage.photometric_scaling)
207 def testUncalibratedImage(self):
208 """Test an image in instrumental units with a photometric scaling."""
209 visitImage = self._checkFutureMatchesLegacy(u.electron)
210 self.assertIsNotNone(visitImage.photometric_scaling)
212 def testFullRequiresLegacy(self):
213 """Test that full=True is rejected for VisitImage inputs."""
214 config = ConsolidateVisitSummaryTask.ConfigClass()
215 config.input_image_type = "future"
216 config.full = True
217 with self.assertRaises(ValueError):
218 config.validate()
221class MemoryTester(lsst.utils.tests.MemoryTestCase):
222 pass
225def setup_module(module):
226 lsst.utils.tests.init()
229if __name__ == "__main__": 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true
230 lsst.utils.tests.init()
231 unittest.main()