Coverage for tests/test_footprint_conversions.py: 98%

113 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-06-21 02:07 -0700

1# This file is part of meas_extensions_scarlet. 

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/>. 

21 

22"""Round-trip tests for the afw ↔ scarlet footprint conversions. 

23 

24The footprints are built from circular masks produced by 

25``scl.utils.get_circle_mask`` rather than rectangles. A non-rectangular 

26span set turns the round-trip "spans match" assertion into a real 

27check — a rectangle would survive almost any broken conversion. 

28""" 

29 

30import unittest 

31 

32import lsst.geom as geom 

33import lsst.meas.extensions.scarlet as mes 

34import lsst.scarlet.lite as scl 

35import lsst.utils.tests 

36import numpy as np 

37from lsst.afw.detection import Footprint as afwFootprint 

38from lsst.afw.geom import SpanSet 

39from lsst.afw.image import Mask 

40from lsst.scarlet.lite.detect_pybind11 import Peak 

41 

42 

43def _circle_mask(diameter): 

44 """Return a ``(diameter, diameter)`` int32 mask of a circle.""" 

45 return scl.utils.get_circle_mask(diameter, dtype=np.int32) 

46 

47 

48def _afw_circle_footprint(min_point, diameter, peaks): 

49 """Build a circular afw Footprint with the given peaks. 

50 

51 Parameters 

52 ---------- 

53 min_point : `tuple` [`int`] 

54 ``(x, y)`` corner of the enclosing bbox. 

55 diameter : `int` 

56 Diameter of the circle (also the side length of the bbox). 

57 peaks : `list` [`tuple` [`int`, `int`, `float`]] 

58 ``(x, y, peakValue)`` triples, in the order afw's ``addPeak`` 

59 expects. 

60 """ 

61 afw_mask = Mask(_circle_mask(diameter), xy0=geom.Point2I(*min_point)) 

62 fp = afwFootprint(SpanSet.fromMask(afw_mask)) 

63 for x, y, v in peaks: 

64 fp.addPeak(x, y, v) 

65 return fp 

66 

67 

68def _scarlet_circle_footprint(origin, diameter, peaks): 

69 """Build a circular scarlet Footprint with the given peaks. 

70 

71 Parameters 

72 ---------- 

73 origin : `tuple` [`int`] 

74 ``(y_min, x_min)`` of the enclosing bbox (scarlet's `(y, x)` 

75 order). 

76 diameter : `int` 

77 Diameter of the circle (also the side length of the bbox). 

78 peaks : `list` [`tuple` [`int`, `int`, `float`]] 

79 ``(y, x, flux)`` triples, in the order scarlet's ``Peak`` ctor 

80 expects. 

81 """ 

82 bounds = scl.detect.bbox_to_bounds(scl.Box((diameter, diameter), origin)) 

83 scl_peaks = [Peak(y, x, v) for y, x, v in peaks] 

84 return scl.detect.Footprint(_circle_mask(diameter), scl_peaks, bounds) 

85 

86 

87class TestFootprintConversions(lsst.utils.tests.TestCase): 

88 """Round-trip tests for ``afwFootprintToScarlet``, 

89 ``scarletFootprintToAfw``, and ``scarletFootprintsToPeakCatalog`` 

90 in ``lsst.meas.extensions.scarlet.footprint``. 

91 

92 Peaks live in afw as ``(getIx(), getIy(), getPeakValue())`` and in 

93 scarlet as ``(peak.x, peak.y, peak.flux)``. The conversion functions 

94 cross between those conventions and also between afw's 

95 separate-axis ``Box2I`` and scarlet's ``(y, x)``-ordered ``Box``. 

96 """ 

97 

98 def test_afwFootprintToScarlet_peak_order(self): 

99 """afw → scarlet maps ``getIy() → peak.y`` and ``getIx() → peak.x``.""" 

100 peaks_in = [ 

101 # Peak at the centre of a diameter-7 circle. 

102 (3, 3, 100.0), 

103 # Off-centre asymmetric (x, y) — a silent x/y flip in the 

104 # conversion would land at (2, 1) instead. 

105 (1, 2, 50.0), 

106 # Opposite quadrant; flux < 1 confirms peakValue is kept 

107 # as float, not int. 

108 (5, 4, 0.5), 

109 ] 

110 fp = _afw_circle_footprint((0, 0), diameter=7, peaks=peaks_in) 

111 sf = mes.footprint.afwFootprintToScarlet(fp) 

112 self.assertEqual(len(sf.peaks), len(peaks_in)) 

113 for (x, y, v), p in zip(peaks_in, sf.peaks): 

114 self.assertEqual(p.y, y) 

115 self.assertEqual(p.x, x) 

116 self.assertEqual(p.flux, v) 

117 # The span data crosses over too — the scarlet footprint's mask 

118 # is the bbox-sized array of the afw spans. 

119 np.testing.assert_array_equal(sf.data, _circle_mask(7)) 

120 

121 def test_scarletFootprintToAfw_peak_order(self): 

122 """scarlet → afw maps ``peak.x → getIx()`` and ``peak.y → getIy()``.""" 

123 peaks_in = [ 

124 # Same three positions as the afw → scarlet test but with 

125 # the role of x and y swapped in the input tuple; a silent 

126 # flip would surface here. 

127 (3, 3, 100.0), 

128 (2, 1, 50.0), 

129 (4, 5, 0.5), 

130 ] 

131 sf = _scarlet_circle_footprint((0, 0), diameter=7, peaks=peaks_in) 

132 fp = mes.footprint.scarletFootprintToAfw(sf) 

133 self.assertEqual(len(fp.peaks), len(peaks_in)) 

134 for (y, x, v), p in zip(peaks_in, fp.peaks): 

135 self.assertEqual(p.getIy(), y) 

136 self.assertEqual(p.getIx(), x) 

137 self.assertEqual(p.getPeakValue(), v) 

138 np.testing.assert_array_equal(fp.spans.asArray(), _circle_mask(7)) 

139 

140 def test_roundtrip_footprint_with_negative_origin(self): 

141 """afw → scarlet → afw preserves a circle whose bbox corner is 

142 below ``(0, 0)``. 

143 """ 

144 fp = _afw_circle_footprint( 

145 min_point=(-5, -3), 

146 diameter=7, 

147 # (-2, 0) lands on the circle centre when the bbox origin 

148 # is (-5, -3): (x - (-5), y - (-3)) == (3, 3). 

149 peaks=[(-2, 0, 7.0)], 

150 ) 

151 sf = mes.footprint.afwFootprintToScarlet(fp) 

152 fp_back = mes.footprint.scarletFootprintToAfw(sf) 

153 self.assertEqual(fp_back.getBBox(), fp.getBBox()) 

154 np.testing.assert_array_equal( 

155 fp_back.spans.asArray(), fp.spans.asArray() 

156 ) 

157 self.assertEqual(len(fp_back.peaks), 1) 

158 peak = fp_back.peaks[0] 

159 self.assertEqual(peak.getIx(), -2) 

160 self.assertEqual(peak.getIy(), 0) 

161 self.assertEqual(peak.getPeakValue(), 7.0) 

162 

163 def test_roundtrip_footprint_empty_spans(self): 

164 """A footprint with no spans round-trips without error and the 

165 scarlet form is a 0×0 box with no peaks. 

166 """ 

167 # Empty isn't really a "circle" — the diameter that would 

168 # describe it is zero, so we build the empty Footprint directly. 

169 fp = afwFootprint(SpanSet()) 

170 sf = mes.footprint.afwFootprintToScarlet(fp) 

171 self.assertEqual(tuple(int(s) for s in sf.bbox.shape), (0, 0)) 

172 self.assertEqual(len(sf.peaks), 0) 

173 fp_back = mes.footprint.scarletFootprintToAfw(sf) 

174 self.assertEqual(len(fp_back.peaks), 0) 

175 # afw represents a degenerate footprint as min=(0,0), 

176 # max=(-1,-1); both round-trip endpoints agree on that. 

177 self.assertEqual(fp_back.getBBox(), fp.getBBox()) 

178 

179 def test_roundtrip_footprint_edge_pixels(self): 

180 """A circle with its bbox rooted at ``(0, 0)`` round-trips — 

181 pins the edge case where the conversion's signed offsets are zero. 

182 """ 

183 peaks_in = [ 

184 # (3, 0) is the apex of a diameter-7 circle's top row, 

185 # where mask[0, 3] == 1 — exercises the y == 0 edge. 

186 (3, 0, 100.0), 

187 # (0, 3) is the leftmost pixel of the circle's middle row, 

188 # where mask[3, 0] == 1 — exercises the x == 0 edge. 

189 (0, 3, 50.0), 

190 ] 

191 fp = _afw_circle_footprint((0, 0), diameter=7, peaks=peaks_in) 

192 sf = mes.footprint.afwFootprintToScarlet(fp) 

193 fp_back = mes.footprint.scarletFootprintToAfw(sf) 

194 self.assertEqual(fp_back.getBBox(), fp.getBBox()) 

195 np.testing.assert_array_equal( 

196 fp_back.spans.asArray(), _circle_mask(7) 

197 ) 

198 self.assertEqual(len(fp_back.peaks), len(peaks_in)) 

199 for (x, y, v), p in zip(peaks_in, fp_back.peaks): 

200 self.assertEqual(p.getIx(), x) 

201 self.assertEqual(p.getIy(), y) 

202 self.assertEqual(p.getPeakValue(), v) 

203 

204 def test_scarletFootprintsToPeakCatalog_schema(self): 

205 """The returned ``PeakCatalog`` has the default ``PeakTable`` 

206 schema and one row per input peak with matching values. 

207 """ 

208 # Two circular scarlet footprints, three peaks total, with one 

209 # in negative-coordinate territory to confirm the conversion 

210 # doesn't clamp. 

211 sf1 = _scarlet_circle_footprint( 

212 origin=(0, 0), 

213 diameter=7, 

214 peaks=[(3, 3, 100.0), (3, 0, 50.0)], 

215 ) 

216 sf2 = _scarlet_circle_footprint( 

217 origin=(-3, -2), 

218 diameter=7, 

219 # (0, 1) lands on the circle centre: (y - (-3), x - (-2)) 

220 # == (3, 3). 

221 peaks=[(0, 1, 25.0)], 

222 ) 

223 catalog = mes.footprint.scarletFootprintsToPeakCatalog([sf1, sf2]) 

224 

225 # Default PeakTable schema columns; the function uses a "dummy 

226 # Footprint" internally and so picks up the default schema. 

227 self.assertEqual( 

228 set(catalog.schema.getNames()), 

229 {"id", "i_x", "i_y", "f_x", "f_y", "peakValue"}, 

230 ) 

231 # Catalog rows are emitted in input order; each peak.x/.y/.flux 

232 # maps to getIx()/getIy()/getPeakValue(). 

233 expected = [(3, 3, 100.0), (0, 3, 50.0), (1, 0, 25.0)] 

234 self.assertEqual(len(catalog), len(expected)) 

235 for (x, y, v), row in zip(expected, catalog): 

236 self.assertEqual(row.getIx(), x) 

237 self.assertEqual(row.getIy(), y) 

238 self.assertEqual(row.getPeakValue(), v) 

239 

240 

241class TestScarletModelToHeavy(lsst.utils.tests.TestCase): 

242 """Tests for ``scarletModelToHeavy`` in 

243 ``lsst.meas.extensions.scarlet.footprint``. 

244 """ 

245 

246 def test_multiband_spanset_includes_negative_band_pixels(self): 

247 """A multi-band model's heavy footprint covers every pixel 

248 where any band is non-zero, including pixels that are 

249 negative-only in some bands. 

250 

251 The 2-band model has three non-zero pixels: 

252 

253 - ``(0, 0)``: ``[+1, 0]`` -- positive in g. 

254 - ``(1, 1)``: ``[-1, 0]`` -- negative-only in g 

255 (``np.max == 0`` would exclude this pixel under U-2's old 

256 idiom). 

257 - ``(2, 2)``: ``[+1, +1]`` -- positive in both bands. 

258 

259 A 1x1 PSF of value 1.0 per band makes 

260 ``observation.convolve(model, mode="real")`` the identity so 

261 the post-convolution model preserves these exact values. 

262 Under the U-18 bug the call raises ``AttributeError`` from 

263 the ``MultibandImage(blend.bands, ...)`` line; under the 

264 fix the SpanSet covers all three pixels and excludes the 

265 rest of the bbox. 

266 """ 

267 bands = ("g", "r") 

268 model_data = np.zeros((2, 3, 3), dtype=np.float32) 

269 model_data[0, 0, 0] = 1.0 

270 model_data[0, 1, 1] = -1.0 

271 model_data[:, 2, 2] = 1.0 

272 

273 psfs = np.ones((2, 1, 1), dtype=np.float32) 

274 obs_shape = (2, 5, 5) 

275 observation = scl.Observation( 

276 images=np.zeros(obs_shape, dtype=np.float32), 

277 variance=np.ones(obs_shape, dtype=np.float32), 

278 weights=np.ones(obs_shape, dtype=np.float32), 

279 psfs=psfs, 

280 bands=bands, 

281 ) 

282 image = scl.Image(model_data, yx0=(0, 0), bands=bands) 

283 component = scl.component.CubeComponent(model=image, peak=(2, 2)) 

284 source = scl.Source([component]) 

285 blend = scl.Blend(sources=[source], observation=observation) 

286 

287 heavy = mes.footprint.scarletModelToHeavy(source, blend, useFlux=False) 

288 

289 bbox = heavy.getBBox() 

290 # The 3x3 model bbox contains the three non-zero pixels. 

291 self.assertEqual(bbox.getMin(), geom.Point2I(0, 0)) 

292 self.assertEqual(bbox.getDimensions(), geom.Extent2I(3, 3)) 

293 mask = heavy.getSpans().asArray( 

294 shape=(bbox.getHeight(), bbox.getWidth()), 

295 xy0=bbox.getMin(), 

296 ) 

297 expected = np.array( 

298 [ 

299 [True, False, False], 

300 [False, True, False], 

301 [False, False, True], 

302 ] 

303 ) 

304 np.testing.assert_array_equal(mask, expected) 

305 

306 

307def setup_module(module): 

308 lsst.utils.tests.init() 

309 

310 

311class MemoryTester(lsst.utils.tests.MemoryTestCase): 

312 pass 

313 

314 

315if __name__ == "__main__": 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true

316 lsst.utils.tests.init() 

317 unittest.main()