Coverage for tests/test_guider.py: 40%
195 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-19 10:48 +0000
1# This file is part of summit_utils.
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 os
23import tempfile
24import unittest
26import numpy as np
27import pandas as pd
29import lsst.utils.tests
30from lsst.daf.butler import Butler
31from lsst.meas.algorithms.stamps import Stamps
32from lsst.summit.utils.butlerUtils import makeDefaultButler
33from lsst.summit.utils.guiders.detection import GuiderStarTrackerConfig, isBlankImage
34from lsst.summit.utils.guiders.metrics import GuiderMetricsBuilder
35from lsst.summit.utils.guiders.plotting import GuiderPlotter
36from lsst.summit.utils.guiders.reading import GuiderData, GuiderReader
37from lsst.summit.utils.guiders.seeing import CorrelationAnalysis, GuiderSeeing
38from lsst.summit.utils.guiders.tracking import GuiderStarTracker, _diagnoseQualityCutRejections
39from lsst.summit.utils.utils import getSite
42class GuiderTestCase(unittest.TestCase):
43 """Tests of the run method with fake data."""
45 def setUp(self) -> None:
46 try:
47 if getSite() == "jenkins": 47 ↛ 49line 47 didn't jump to line 49 because the condition on line 47 was always true
48 raise unittest.SkipTest("Skip running butler-driven tests in Jenkins.")
49 self.butler = makeDefaultButler("LSSTCam", embargo=False)
50 except FileNotFoundError:
51 raise unittest.SkipTest("Skipping tests that require the LSSTCam butler repo.")
52 self.assertIsInstance(self.butler, Butler)
54 self.dayObs = 20250629
55 self.seqNum = 340
56 self.expId = 2025062900340
58 self.reader = GuiderReader(self.butler, view="dvcs")
59 self.guiderData = self.reader.get(dayObs=self.dayObs, seqNum=self.seqNum)
60 self.tracker = GuiderStarTracker(self.guiderData)
61 self.stars = self.tracker.trackGuiderStars(refCatalog=None)
62 self.plotter = GuiderPlotter(self.guiderData, starsDf=self.stars)
63 self.metricsBuilder = GuiderMetricsBuilder(self.stars, self.guiderData.nMissingStamps)
65 def test_types(self) -> None:
66 self.assertIsInstance(self.guiderData.header, dict)
67 self.assertIsInstance(self.guiderData.stampsMap, dict)
69 expectedKeys = (
70 "R00_SG0",
71 "R00_SG1",
72 "R04_SG0",
73 "R04_SG1",
74 "R40_SG0",
75 "R40_SG1",
76 "R44_SG0",
77 "R44_SG1",
78 )
79 self.assertTrue(
80 all(key in self.guiderData.stampsMap for key in expectedKeys),
81 "Not all expected guider datasets are present in the data.",
82 )
84 detName = "R00_SG0"
85 single = self.guiderData[detName, 0]
86 self.assertIsInstance(single, np.ndarray)
87 stack = self.guiderData.getStampArrayCoadd(detName=detName)
88 self.assertIsInstance(stack, np.ndarray)
89 self.assertEqual(single.shape, (400, 400))
91 return
93 def test_reading(self) -> None:
94 """Test that the reader can read the expected data."""
95 self.assertTrue(self.guiderData.isMedianSubtracted, "isMedianSubtracted not set correctly")
96 for detName in self.guiderData.guiderNames:
97 single = self.guiderData[detName, 0]
98 self.assertIsInstance(single, np.ndarray)
99 self.assertEqual(single.shape, (400, 400))
100 self.assertLessEqual(abs(np.nanmedian(single)), 10, "median subtracted median is too high")
102 stack = self.guiderData.getStampArrayCoadd(detName=detName)
103 self.assertIsInstance(stack, np.ndarray)
104 self.assertEqual(stack.shape, (400, 400))
105 self.assertLessEqual(abs(np.nanmedian(stack)), 10, "stack median subtracted median is too high")
107 fullStamps = self.guiderData[detName]
108 self.assertIsInstance(fullStamps, Stamps)
110 noMedianSubtracted = self.reader.get(dayObs=self.dayObs, seqNum=self.seqNum, doSubtractMedian=False)
111 self.assertIsInstance(noMedianSubtracted, GuiderData)
112 self.assertFalse(noMedianSubtracted.isMedianSubtracted, "isMedianSubtracted not set correctly")
113 for detName in noMedianSubtracted.guiderNames:
114 single = noMedianSubtracted[detName, 0]
115 self.assertIsInstance(single, np.ndarray)
116 self.assertEqual(single.shape, (400, 400))
117 self.assertGreater(abs(np.nanmedian(single)), 500, "median un-subtracted median is too low")
119 stack = noMedianSubtracted.getStampArrayCoadd(detName=detName)
120 self.assertIsInstance(stack, np.ndarray)
121 self.assertEqual(stack.shape, (400, 400))
122 self.assertGreater(abs(np.nanmedian(single)), 500, "median un-subtracted median is too low")
124 fullStamps = noMedianSubtracted[detName]
125 self.assertIsInstance(fullStamps, Stamps)
127 def test_detection(self) -> None:
128 self.assertIsInstance(self.stars, pd.DataFrame)
129 requiredColumns = (
130 "xroi",
131 "yroi",
132 "dx",
133 "dy",
134 "dalt",
135 "daz",
136 "fwhm",
137 "trackid",
138 "expid",
139 )
140 self.assertTrue(
141 all(col in self.stars.columns for col in requiredColumns),
142 "Not all required columns are present in the stars DataFrame.",
143 )
145 maxStampIndex = max(self.stars["stamp"])
146 nStamps = len(self.guiderData) # we should make an attribute for this
148 # we skip the first stamp, so the max index should be nStamps - 1
149 self.assertEqual(maxStampIndex, nStamps - 1, "Did not get detections for all expected stamps")
151 def testPlotMosaicFullView(self) -> None:
152 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
153 self.plotter.plotMosaic(stampNum=-1, cutoutSize=-1, plo=50, phi=98, saveAs=tmp.name)
154 os.fsync(tmp.fileno()) # be strict
155 size = os.path.getsize(tmp.name)
156 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
158 def testPlotMosaicZoomView(self) -> None:
159 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
160 self.plotter.plotMosaic(stampNum=-1, cutoutSize=12, plo=50, phi=98, saveAs=tmp.name)
161 os.fsync(tmp.fileno()) # be strict
162 size = os.path.getsize(tmp.name)
163 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
165 def testPlotMosaicStampZoomView(self) -> None:
166 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
167 self.plotter.plotMosaic(stampNum=4, cutoutSize=12, plo=50, phi=98, saveAs=tmp.name)
168 os.fsync(tmp.fileno()) # be strict
169 size = os.path.getsize(tmp.name)
170 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
172 def testStripPlotPsf(self) -> None:
173 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
174 self.plotter.stripPlot(plotType="psf", saveAs=tmp.name)
175 os.fsync(tmp.fileno())
176 size = os.path.getsize(tmp.name)
177 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
179 def testStripPlotCentroidAltAz(self) -> None:
180 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
181 self.plotter.stripPlot(plotType="centroidAltAz", saveAs=tmp.name)
182 os.fsync(tmp.fileno())
183 size = os.path.getsize(tmp.name)
184 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
186 def testStripPlotFlux(self) -> None:
187 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
188 self.plotter.stripPlot(plotType="flux", saveAs=tmp.name)
189 os.fsync(tmp.fileno())
190 size = os.path.getsize(tmp.name)
191 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
193 def testStripPlotShape(self) -> None:
194 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
195 self.plotter.stripPlot(plotType="ellip", saveAs=tmp.name)
196 os.fsync(tmp.fileno())
197 size = os.path.getsize(tmp.name)
198 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
200 def testMakeGif(self) -> None:
201 with tempfile.NamedTemporaryFile(suffix=".gif", delete=True) as tmp:
202 # test the crop and zoom as a gif
203 self.plotter.makeAnimation(cutoutSize=14, plo=50, phi=98, saveAs=tmp.name)
204 os.fsync(tmp.fileno())
205 size = os.path.getsize(tmp.name)
206 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
208 def testMakeMp4(self) -> None:
209 with tempfile.NamedTemporaryFile(suffix=".mp4", delete=True) as tmp:
210 # test the full frame as an mp4
211 self.plotter.makeAnimation(cutoutSize=-1, plo=50, phi=98, saveAs=tmp.name)
212 os.fsync(tmp.fileno())
213 size = os.path.getsize(tmp.name)
214 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
216 def test_metrics(self) -> None:
217 # Check that metrics can be built without error
218 metrics = self.metricsBuilder.buildMetrics(self.expId)
219 self.assertIsInstance(metrics, pd.DataFrame)
220 self.assertGreater(len(metrics), 0, "Metrics DataFrame is empty")
222 # These are what's currently there. Why is the 8th guider missing?
223 expectedColumns = (
224 "n_guiders",
225 "n_stars",
226 "n_missing_stamps",
227 "n_measurements",
228 "fraction_possible_measurements",
229 "exptime",
230 "R00_SG0",
231 "R00_SG1",
232 "R04_SG0",
233 "R04_SG1",
234 "R40_SG0",
235 "R40_SG1",
236 "R44_SG0",
237 "R44_SG1",
238 "az_drift_slope",
239 "az_drift_intercept",
240 "az_drift_trend_rmse",
241 "az_drift_global_std",
242 "az_drift_outlier_frac",
243 "az_drift_slope_significance",
244 "az_drift_nsize",
245 "alt_drift_slope",
246 "alt_drift_intercept",
247 "alt_drift_trend_rmse",
248 "alt_drift_global_std",
249 "alt_drift_outlier_frac",
250 "alt_drift_slope_significance",
251 "alt_drift_nsize",
252 "rotator_slope",
253 "rotator_intercept",
254 "rotator_trend_rmse",
255 "rotator_global_std",
256 "rotator_outlier_frac",
257 "rotator_slope_significance",
258 "rotator_nsize",
259 "mag_slope",
260 "mag_intercept",
261 "mag_trend_rmse",
262 "mag_global_std",
263 "mag_outlier_frac",
264 "mag_slope_significance",
265 "mag_nsize",
266 "psf_slope",
267 "psf_intercept",
268 "psf_trend_rmse",
269 "psf_global_std",
270 "psf_outlier_frac",
271 "psf_slope_significance",
272 "psf_nsize",
273 )
274 for col in expectedColumns:
275 self.assertIn(col, metrics.columns, f"Column {col} is missing from metrics DataFrame")
277 # check this runs without error
278 self.metricsBuilder.printSummary()
280 def testTomographicSeeing(self) -> None:
281 analysis = CorrelationAnalysis(self.stars, self.expId)
282 variance = np.nanmedian(analysis.measureVariance())
283 self.assertGreater(variance, 0, "Variance should be positive and non-zero")
284 self.assertIsInstance(variance, float)
286 seeing = analysis.measureTomographicSeeing()
287 self.assertIsInstance(seeing, GuiderSeeing)
290class IsBlankImageTestCase(unittest.TestCase):
291 """Pure-function tests for isBlankImage (no butler)."""
293 def test_blank_image_is_blank(self) -> None:
294 rng = np.random.default_rng(0)
295 image = 1000 + rng.normal(0, 2, (50, 50))
296 self.assertTrue(isBlankImage(image))
298 def test_bright_source_fails_flux_check(self) -> None:
299 rng = np.random.default_rng(0)
300 image = 1000 + rng.normal(0, 2, (50, 50))
301 image[25, 25] += 1000
302 self.assertFalse(isBlankImage(image, fluxMin=300))
304 def test_faint_source_caught_by_snr_check(self) -> None:
305 # Peak well under fluxMin, but many sigma above the noise -> not blank.
306 rng = np.random.default_rng(0)
307 image = 1000 + rng.normal(0, 1, (50, 50))
308 image[25, 25] += 50
309 self.assertFalse(isBlankImage(image, fluxMin=300, peakSnrMin=5.0))
311 def test_zero_std_is_blank(self) -> None:
312 image = np.full((10, 10), 5.0)
313 self.assertTrue(isBlankImage(image))
316class DiagnoseQualityCutRejectionsTestCase(unittest.TestCase):
317 """Pure-function tests for _diagnoseQualityCutRejections (no butler)."""
319 def setUp(self) -> None:
320 self.config = GuiderStarTrackerConfig()
321 self.shape = (400.0, 400.0)
323 def test_low_snr_reported(self) -> None:
324 stars = pd.DataFrame(
325 {
326 "snr": [1.0],
327 "flux": [100.0],
328 "flux_err": [10.0],
329 "e1": [0.0],
330 "e2": [0.0],
331 "xroi": [200.0],
332 "yroi": [200.0],
333 }
334 )
335 reasons = _diagnoseQualityCutRejections(stars, self.shape, self.config)
336 self.assertEqual(len(reasons), 1)
337 self.assertIn("snr=", reasons[0])
339 def test_high_ellipticity_reported(self) -> None:
340 stars = pd.DataFrame(
341 {
342 "snr": [50.0],
343 "flux": [100.0],
344 "flux_err": [10.0],
345 "e1": [0.9],
346 "e2": [0.0],
347 "xroi": [200.0],
348 "yroi": [200.0],
349 }
350 )
351 reasons = _diagnoseQualityCutRejections(stars, self.shape, self.config)
352 self.assertIn("e=", reasons[0])
354 def test_edge_position_reported(self) -> None:
355 stars = pd.DataFrame(
356 {
357 "snr": [50.0],
358 "flux": [100.0],
359 "flux_err": [10.0],
360 "e1": [0.0],
361 "e2": [0.0],
362 "xroi": [1.0],
363 "yroi": [1.0],
364 }
365 )
366 reasons = _diagnoseQualityCutRejections(stars, self.shape, self.config)
367 self.assertIn("edge", reasons[0])
369 def test_passing_star_has_no_reason_string(self) -> None:
370 stars = pd.DataFrame(
371 {
372 "snr": [50.0],
373 "flux": [100.0],
374 "flux_err": [10.0],
375 "e1": [0.0],
376 "e2": [0.0],
377 "xroi": [200.0],
378 "yroi": [200.0],
379 }
380 )
381 reasons = _diagnoseQualityCutRejections(stars, self.shape, self.config)
382 self.assertEqual(reasons[0], "?")
385class TestMemory(lsst.utils.tests.MemoryTestCase):
386 pass
389def setup_module(module: object) -> None:
390 lsst.utils.tests.init()
393if __name__ == "__main__": 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true
394 lsst.utils.tests.init()
395 unittest.main()