Coverage for tests/test_edgeBleed.py: 99%
269 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 09:20 +0000
1# This file is part of ip_isr.
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/>.
21import unittest
23import numpy as np
25import lsst.geom as geom
26import lsst.afw.cameraGeom as cameraGeom
27import lsst.afw.image as afwImage
28import lsst.utils.tests
29import lsst.ip.isr.isrFunctions as ipIsrFunctions
30from lsst.ip.isr.isrTask import IsrTaskConfig
31from lsst.ip.isr.masking import DECamEdgeBleedMaskTask
33AMP_WIDTH = 200
34AMP_HEIGHT = 800
35SKY = 1000.0
36NOISE = 10.0
37DIP = 200.0
40def makeDetector():
41 """Two side-by-side amplifiers: A reads out at the top, B at the bottom."""
42 camBuilder = cameraGeom.Camera.Builder("testCam")
43 detBuilder = camBuilder.add("testDet", 0)
44 detBuilder.setBBox(geom.Box2I(geom.Point2I(0, 0), geom.Extent2I(2*AMP_WIDTH, AMP_HEIGHT)))
45 detBuilder.setPixelSize(geom.Extent2D(0.015, 0.015))
46 detBuilder.setOrientation(cameraGeom.Orientation())
47 for name, x0, corner in (("A", 0, cameraGeom.ReadoutCorner.UL),
48 ("B", AMP_WIDTH, cameraGeom.ReadoutCorner.LL)):
49 bbox = geom.Box2I(geom.Point2I(x0, 0), geom.Extent2I(AMP_WIDTH, AMP_HEIGHT))
50 amp = cameraGeom.Amplifier.Builder()
51 amp.setName(name)
52 amp.setBBox(bbox)
53 amp.setRawBBox(bbox)
54 amp.setRawDataBBox(bbox)
55 amp.setReadoutCorner(corner)
56 amp.setGain(1.0)
57 amp.setReadNoise(5.0)
58 amp.setSaturation(60000.0)
59 detBuilder.append(amp)
60 return camBuilder.finish()["testDet"]
63class MaskDECamEdgeBleedTestCase(lsst.utils.tests.TestCase):
64 def setUp(self):
65 self.detector = makeDetector()
66 self.exposure = afwImage.ExposureF(self.detector.getBBox())
67 self.exposure.setDetector(self.detector)
68 rng = np.random.default_rng(12345)
69 self.exposure.image.array[:] = SKY + rng.normal(0.0, NOISE, self.exposure.image.array.shape)
70 self.exposure.variance.array[:] = NOISE**2
71 self.satBit = self.exposure.mask.getPlaneBitMask("SAT")
73 def addSatBlock(self, x0, y0, width, height):
74 """Mark a width x height block as SAT (image value irrelevant)."""
75 self.exposure.mask.array[y0:y0 + height, x0:x0 + width] |= self.satBit
76 self.exposure.image.array[y0:y0 + height, x0:x0 + width] = 60000.0
78 def addDip(self, x0, y0, width, height):
79 """Depress a width x height block below sky."""
80 self.exposure.image.array[y0:y0 + height, x0:x0 + width] -= DIP
82 def runMasking(self):
83 before = self.exposure.mask.array.copy()
84 ipIsrFunctions.maskDECamEdgeBleed(self.exposure)
85 return before, self.exposure.mask.array
87 def test_topReadEdgeBleedIsMasked(self):
88 # 100 x 200 = 20000 SAT pixels touching the top edge of amp A.
89 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
90 # Dip in the 100 rows nearest the top edge, full width of amp A.
91 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 100)
93 before, after = self.runMasking()
95 # height 100 -> margin int(100*0.125) + 1 = 13 -> 113 rows masked.
96 expected = before.copy()
97 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
98 np.testing.assert_array_equal(after, expected)
100 def test_noDipIsNotMasked(self):
101 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
103 before, after = self.runMasking()
105 np.testing.assert_array_equal(after, before)
107 def test_smallFootprintIsNotMasked(self):
108 # 50 x 100 = 5000 SAT pixels: below satMinArea.
109 self.addSatBlock(50, AMP_HEIGHT - 100, 50, 100)
110 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 100)
112 before, after = self.runMasking()
114 np.testing.assert_array_equal(after, before)
116 def test_largeFootprintIsMasked(self):
117 # 300 x 400 = 120000 SAT pixels: tall DECam bleeds have footprints of
118 # 1e5-2e5 pixels, so there is no upper area cut. Only x 0-49 of the
119 # dip rows are unsaturated, still enough low pixels per row to be
120 # detectable.
121 self.addSatBlock(50, 400, 300, 400)
122 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 100)
124 before, after = self.runMasking()
126 expected = before.copy()
127 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
128 np.testing.assert_array_equal(after, expected)
130 def test_dipInFewUsablePixelsIsMasked(self):
131 # Saturation covers 160 of the 200 columns in the 100 rows nearest
132 # the read edge, and the dip is in 25 of the remaining 40 columns:
133 # 25 low pixels per row is far below the old count threshold of 30,
134 # but 62 per cent of the usable pixels are low.
135 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
136 self.addSatBlock(40, AMP_HEIGHT - 100, 160, 100)
137 self.addDip(0, AMP_HEIGHT - 100, 25, 100)
139 before, after = self.runMasking()
141 expected = before.copy()
142 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
143 np.testing.assert_array_equal(after, expected)
145 def test_sparseLowPixelsAreNotADip(self):
146 # 2 low pixels per row, 2 per cent of the 100 usable columns in the
147 # check rows, is below minLowFraction: not an edge bleed.
148 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
149 self.addDip(0, AMP_HEIGHT - 100, 2, 100)
151 before, after = self.runMasking()
153 np.testing.assert_array_equal(after, before)
155 def test_shallowDipInFirstRowsIsMasked(self):
156 # A dip in 10 of the 100 usable columns over the 4 rows nearest the
157 # read edge: 2 per cent averaged over the 20 check rows, but 8 per
158 # cent over the 5-row short window. 4 rows -> int(4*0.125) + 1 = 1
159 # -> 5 rows masked.
160 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
161 self.addDip(0, AMP_HEIGHT - 4, 10, 4)
163 before, after = self.runMasking()
165 expected = before.copy()
166 expected[AMP_HEIGHT - 5:, 0:AMP_WIDTH] |= self.satBit
167 np.testing.assert_array_equal(after, expected)
169 def test_shortWindowCanBeDisabled(self):
170 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
171 self.addDip(0, AMP_HEIGHT - 4, 10, 4)
173 before = self.exposure.mask.array.copy()
174 ipIsrFunctions.maskDECamEdgeBleed(self.exposure, nRowsCheckShort=0)
176 np.testing.assert_array_equal(self.exposure.mask.array, before)
178 def test_blockedRowsBeyondBorderAreMasked(self):
179 # Both edges of amp A carry a 10-row BAD border (as DECam defects
180 # do). A drained read register leaves rows 10-59 from the read edge
181 # fully BAD as well, with no measurable dip beyond them: those 50
182 # blocked rows are themselves the edge bleed, so the mask covers
183 # them plus the margin (60 -> int(60*0.125) + 1 = 8 -> 68 rows).
184 badBit = self.exposure.mask.getPlaneBitMask("BAD")
185 self.exposure.mask.array[:10, 0:AMP_WIDTH] |= badBit
186 self.exposure.mask.array[AMP_HEIGHT - 60:, 0:AMP_WIDTH] |= badBit
187 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
189 before, after = self.runMasking()
191 expected = before.copy()
192 expected[AMP_HEIGHT - 68:, 0:AMP_WIDTH] |= self.satBit
193 np.testing.assert_array_equal(after, expected)
195 def test_normalBorderIsNotABleed(self):
196 # The same 10-row BAD border at both edges, but nothing beyond it:
197 # the read-edge border is not counted as blocked rows.
198 badBit = self.exposure.mask.getPlaneBitMask("BAD")
199 self.exposure.mask.array[:10, 0:AMP_WIDTH] |= badBit
200 self.exposure.mask.array[AMP_HEIGHT - 10:, 0:AMP_WIDTH] |= badBit
201 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
203 before, after = self.runMasking()
205 np.testing.assert_array_equal(after, before)
207 def test_fewBlockedRowsNeedADip(self):
208 # 3 blocked rows beyond the border are below minBlockedRows; with no
209 # dip beyond them nothing is masked.
210 badBit = self.exposure.mask.getPlaneBitMask("BAD")
211 self.exposure.mask.array[:10, 0:AMP_WIDTH] |= badBit
212 self.exposure.mask.array[AMP_HEIGHT - 13:, 0:AMP_WIDTH] |= badBit
213 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
215 before, after = self.runMasking()
217 np.testing.assert_array_equal(after, before)
219 def test_blockedRowsExtendAMeasuredDip(self):
220 # 30 blocked rows beyond a 10-row border, then a 40-row dip: the
221 # height is measured from the physical edge through the dip
222 # (80 -> int(80*0.125) + 1 = 11 -> 91 rows).
223 badBit = self.exposure.mask.getPlaneBitMask("BAD")
224 self.exposure.mask.array[:10, 0:AMP_WIDTH] |= badBit
225 self.exposure.mask.array[AMP_HEIGHT - 40:, 0:AMP_WIDTH] |= badBit
226 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
227 self.addDip(0, AMP_HEIGHT - 80, AMP_WIDTH, 40)
229 before, after = self.runMasking()
231 expected = before.copy()
232 expected[AMP_HEIGHT - 91:, 0:AMP_WIDTH] |= self.satBit
233 np.testing.assert_array_equal(after, expected)
235 def test_footprintFarFromEdgeIsNotMasked(self):
236 # Footprint top at row 599, 200 rows short of the read edge.
237 self.addSatBlock(50, 400, 100, 200)
238 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 100)
240 before, after = self.runMasking()
242 np.testing.assert_array_equal(after, before)
244 def test_bottomReadEdgeBleedIsMasked(self):
245 # Amp B reads out at the bottom.
246 self.addSatBlock(AMP_WIDTH + 50, 0, 100, 200)
247 self.addDip(AMP_WIDTH, 0, AMP_WIDTH, 100)
249 before, after = self.runMasking()
251 expected = before.copy()
252 # height 100 -> margin int(100*0.125) + 1 = 13 -> 113 rows masked.
253 expected[:113, AMP_WIDTH:2*AMP_WIDTH] |= self.satBit
254 np.testing.assert_array_equal(after, expected)
256 def test_dipInOtherAmpIsNotMasked(self):
257 # Footprint in amp A (top read edge); dip only in amp B's top rows,
258 # which is not B's read edge and has no footprint.
259 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
260 self.addDip(AMP_WIDTH, AMP_HEIGHT - 100, AMP_WIDTH, 100)
262 before, after = self.runMasking()
264 np.testing.assert_array_equal(after, before)
266 def test_unusableEdgeRowsAreSkippedButMasked(self):
267 # The 20 rows nearest the read edge are NO_DATA (as after trimming
268 # on DECam), so the dip can only be seen from row 20 inward. The
269 # measured height still counts from the physical edge.
270 noDataBit = self.exposure.mask.getPlaneBitMask("NO_DATA")
271 self.exposure.mask.array[AMP_HEIGHT - 20:, 0:AMP_WIDTH] |= noDataBit
272 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
273 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 80)
275 before, after = self.runMasking()
277 expected = before.copy()
278 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
279 np.testing.assert_array_equal(after, expected)
281 def test_suspectPixelsCountAsLow(self):
282 # DECam flags the rows nearest the read edge SUSPECT; those pixels
283 # must still be counted when confirming the dip.
284 suspectBit = self.exposure.mask.getPlaneBitMask("SUSPECT")
285 self.exposure.mask.array[AMP_HEIGHT - 35:, 0:AMP_WIDTH] |= suspectBit
286 self.addSatBlock(50, AMP_HEIGHT - 200, 100, 200)
287 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 100)
289 before, after = self.runMasking()
291 expected = before.copy()
292 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
293 np.testing.assert_array_equal(after, expected)
295 def test_nonDefaultParameters(self):
296 # A 100-row footprint ending 150 rows short of the edge is only a
297 # candidate with a larger approachRows; a 50 ADU dip is only seen
298 # with a smaller nSigma; a zero marginFraction masks height + 1 rows.
299 self.addSatBlock(50, AMP_HEIGHT - 250, 200, 100)
300 self.exposure.image.array[AMP_HEIGHT - 100:, 0:AMP_WIDTH] -= 50.0
302 before = self.exposure.mask.array.copy()
303 ipIsrFunctions.maskDECamEdgeBleed(self.exposure, approachRows=150, nSigma=3.0,
304 marginFraction=0.0)
306 expected = before.copy()
307 expected[AMP_HEIGHT - 101:, 0:AMP_WIDTH] |= self.satBit
308 np.testing.assert_array_equal(self.exposure.mask.array, expected)
310 def test_alternateSaturatedMaskName(self):
311 self.exposure.mask.addMaskPlane("MYSAT")
312 mySatBit = self.exposure.mask.getPlaneBitMask("MYSAT")
313 self.exposure.mask.array[AMP_HEIGHT - 200:, 50:150] |= mySatBit
314 self.addDip(0, AMP_HEIGHT - 100, AMP_WIDTH, 100)
316 before = self.exposure.mask.array.copy()
317 ipIsrFunctions.maskDECamEdgeBleed(self.exposure, saturatedMaskName="MYSAT")
319 expected = before.copy()
320 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= mySatBit
321 np.testing.assert_array_equal(self.exposure.mask.array, expected)
323 def test_partialAmpIsSkipped(self):
324 # An exposure covering only the bottom 600 rows of amp B has a
325 # qualifying footprint and dip, but the amp is not fully contained
326 # so nothing is masked (rather than failing on the sub-image).
327 self.addSatBlock(AMP_WIDTH + 50, 0, 100, 200)
328 self.addDip(AMP_WIDTH, 0, AMP_WIDTH, 100)
329 partial = geom.Box2I(geom.Point2I(AMP_WIDTH, 0), geom.Extent2I(AMP_WIDTH, 600))
330 subExposure = self.exposure[partial]
332 before = subExposure.mask.array.copy()
333 ipIsrFunctions.maskDECamEdgeBleed(subExposure)
335 np.testing.assert_array_equal(subExposure.mask.array, before)
337 def test_noDetectorRaises(self):
338 exposure = afwImage.ExposureF(geom.Box2I(geom.Point2I(0, 0), geom.Extent2I(50, 50)))
339 with self.assertRaises(RuntimeError):
340 ipIsrFunctions.maskDECamEdgeBleed(exposure)
343class DECamEdgeBleedMaskTaskTestCase(lsst.utils.tests.TestCase):
344 def setUp(self):
345 self.detector = makeDetector()
346 self.exposure = afwImage.ExposureF(self.detector.getBBox())
347 self.exposure.setDetector(self.detector)
348 rng = np.random.default_rng(12345)
349 self.exposure.image.array[:] = SKY + rng.normal(0.0, NOISE, self.exposure.image.array.shape)
350 self.exposure.variance.array[:] = NOISE**2
351 self.satBit = self.exposure.mask.getPlaneBitMask("SAT")
353 def test_defaults(self):
354 config = DECamEdgeBleedMaskTask.ConfigClass()
355 self.assertEqual(config.satMinArea, 10000)
356 self.assertEqual(config.approachRows, 20)
357 self.assertEqual(config.nSigma, 5.0)
358 self.assertEqual(config.nRowsCheck, 20)
359 self.assertEqual(config.nRowsCheckShort, 5)
360 self.assertEqual(config.minUsablePixelsPerRow, 30)
361 self.assertEqual(config.minLowFraction, 0.03)
362 self.assertEqual(config.minLowFractionExtent, 0.01)
363 self.assertEqual(config.minBlockedRows, 5)
364 self.assertEqual(config.marginFraction, 0.125)
365 self.assertEqual(config.saturatedMaskName, "SAT")
366 config.validate()
368 def test_validateExtentFractionNotAboveDipFraction(self):
369 config = DECamEdgeBleedMaskTask.ConfigClass()
370 config.minLowFractionExtent = 2*config.minLowFraction
371 with self.assertRaises(ValueError):
372 config.validate()
374 def test_validateFractionsInRange(self):
375 config = DECamEdgeBleedMaskTask.ConfigClass()
376 config.minLowFraction = 1.5
377 with self.assertRaises(ValueError):
378 config.validate()
380 def test_runPassesNewParameters(self):
381 # minLowFraction above the 62 per cent dip fraction of a 25-column dip
382 # in 40 usable columns must switch masking off through the task.
383 self.exposure.mask.array[AMP_HEIGHT - 200:, 50:150] |= self.satBit
384 self.exposure.mask.array[AMP_HEIGHT - 100:, 40:AMP_WIDTH] |= self.satBit
385 self.exposure.image.array[AMP_HEIGHT - 100:, 0:25] -= DIP
386 before = self.exposure.mask.array.copy()
388 config = DECamEdgeBleedMaskTask.ConfigClass()
389 config.minLowFraction = 0.9
390 config.minLowFractionExtent = 0.5
391 DECamEdgeBleedMaskTask(config=config).run(self.exposure)
392 np.testing.assert_array_equal(self.exposure.mask.array, before)
394 DECamEdgeBleedMaskTask().run(self.exposure)
395 expected = before.copy()
396 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
397 np.testing.assert_array_equal(self.exposure.mask.array, expected)
399 def test_runMasksEdgeBleed(self):
400 self.exposure.mask.array[AMP_HEIGHT - 200:, 50:150] |= self.satBit
401 self.exposure.image.array[AMP_HEIGHT - 100:, 0:AMP_WIDTH] -= DIP
402 before = self.exposure.mask.array.copy()
404 task = DECamEdgeBleedMaskTask()
405 task.run(self.exposure)
407 expected = before.copy()
408 expected[AMP_HEIGHT - 113:, 0:AMP_WIDTH] |= self.satBit
409 np.testing.assert_array_equal(self.exposure.mask.array, expected)
411 def test_retargetIntoIsrTaskConfig(self):
412 config = IsrTaskConfig()
413 config.doCameraSpecificMasking = True
414 config.masking.retarget(DECamEdgeBleedMaskTask)
415 config.masking.nSigma = 3.0
416 config.validate()
417 self.assertEqual(config.masking.nSigma, 3.0)
418 self.assertIs(config.masking.target, DECamEdgeBleedMaskTask)
421class MemoryTester(lsst.utils.tests.MemoryTestCase):
422 pass
425def setup_module(module):
426 lsst.utils.tests.init()
429if __name__ == "__main__": 429 ↛ 430line 429 didn't jump to line 430 because the condition on line 429 was never true
430 lsst.utils.tests.init()
431 unittest.main()