Coverage for tests / test_guider.py: 22%
155 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-18 09:28 +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.metrics import GuiderMetricsBuilder
34from lsst.summit.utils.guiders.plotting import GuiderPlotter
35from lsst.summit.utils.guiders.reading import GuiderData, GuiderReader
36from lsst.summit.utils.guiders.seeing import CorrelationAnalysis, GuiderSeeing
37from lsst.summit.utils.guiders.tracking import GuiderStarTracker
38from lsst.summit.utils.utils import getSite
41class GuiderTestCase(unittest.TestCase):
42 """Tests of the run method with fake data."""
44 def setUp(self) -> None:
45 try:
46 if getSite() == "jenkins":
47 raise unittest.SkipTest("Skip running butler-driven tests in Jenkins.")
48 self.butler = makeDefaultButler("LSSTCam", embargo=False)
49 except FileNotFoundError:
50 raise unittest.SkipTest("Skipping tests that require the LSSTCam butler repo.")
51 self.assertIsInstance(self.butler, Butler)
53 self.dayObs = 20250629
54 self.seqNum = 340
55 self.expId = 2025062900340
57 self.reader = GuiderReader(self.butler, view="dvcs")
58 self.guiderData = self.reader.get(dayObs=self.dayObs, seqNum=self.seqNum)
59 self.tracker = GuiderStarTracker(self.guiderData)
60 self.stars = self.tracker.trackGuiderStars(refCatalog=None)
61 self.plotter = GuiderPlotter(self.guiderData, starsDf=self.stars)
62 self.metricsBuilder = GuiderMetricsBuilder(self.stars, self.guiderData.nMissingStamps)
64 def test_types(self) -> None:
65 self.assertIsInstance(self.guiderData.header, dict)
66 self.assertIsInstance(self.guiderData.stampsMap, dict)
68 expectedKeys = (
69 "R00_SG0",
70 "R00_SG1",
71 "R04_SG0",
72 "R04_SG1",
73 "R40_SG0",
74 "R40_SG1",
75 "R44_SG0",
76 "R44_SG1",
77 )
78 self.assertTrue(
79 all(key in self.guiderData.stampsMap for key in expectedKeys),
80 "Not all expected guider datasets are present in the data.",
81 )
83 detName = "R00_SG0"
84 single = self.guiderData[detName, 0]
85 self.assertIsInstance(single, np.ndarray)
86 stack = self.guiderData.getStampArrayCoadd(detName=detName)
87 self.assertIsInstance(stack, np.ndarray)
88 self.assertEqual(single.shape, (400, 400))
90 return
92 def test_reading(self) -> None:
93 """Test that the reader can read the expected data."""
94 self.assertTrue(self.guiderData.isMedianSubtracted, "isMedianSubtracted not set correctly")
95 for detName in self.guiderData.guiderNames:
96 single = self.guiderData[detName, 0]
97 self.assertIsInstance(single, np.ndarray)
98 self.assertEqual(single.shape, (400, 400))
99 self.assertLessEqual(abs(np.nanmedian(single)), 10, "median subtracted median is too high")
101 stack = self.guiderData.getStampArrayCoadd(detName=detName)
102 self.assertIsInstance(stack, np.ndarray)
103 self.assertEqual(stack.shape, (400, 400))
104 self.assertLessEqual(abs(np.nanmedian(stack)), 10, "stack median subtracted median is too high")
106 fullStamps = self.guiderData[detName]
107 self.assertIsInstance(fullStamps, Stamps)
109 noMedianSubtracted = self.reader.get(dayObs=self.dayObs, seqNum=self.seqNum, doSubtractMedian=False)
110 self.assertIsInstance(noMedianSubtracted, GuiderData)
111 self.assertFalse(noMedianSubtracted.isMedianSubtracted, "isMedianSubtracted not set correctly")
112 for detName in noMedianSubtracted.guiderNames:
113 single = noMedianSubtracted[detName, 0]
114 self.assertIsInstance(single, np.ndarray)
115 self.assertEqual(single.shape, (400, 400))
116 self.assertGreater(abs(np.nanmedian(single)), 500, "median un-subtracted median is too low")
118 stack = noMedianSubtracted.getStampArrayCoadd(detName=detName)
119 self.assertIsInstance(stack, np.ndarray)
120 self.assertEqual(stack.shape, (400, 400))
121 self.assertGreater(abs(np.nanmedian(single)), 500, "median un-subtracted median is too low")
123 fullStamps = noMedianSubtracted[detName]
124 self.assertIsInstance(fullStamps, Stamps)
126 def test_detection(self) -> None:
127 self.assertIsInstance(self.stars, pd.DataFrame)
128 requiredColumns = (
129 "xroi",
130 "yroi",
131 "dx",
132 "dy",
133 "dalt",
134 "daz",
135 "fwhm",
136 "trackid",
137 "expid",
138 )
139 self.assertTrue(
140 all(col in self.stars.columns for col in requiredColumns),
141 "Not all required columns are present in the stars DataFrame.",
142 )
144 maxStampIndex = max(self.stars["stamp"])
145 nStamps = len(self.guiderData) # we should make an attribute for this
147 # we skip the first stamp, so the max index should be nStamps - 1
148 self.assertEqual(maxStampIndex, nStamps - 1, "Did not get detections for all expected stamps")
150 def testPlotMosaicFullView(self) -> None:
151 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
152 self.plotter.plotMosaic(stampNum=-1, cutoutSize=-1, plo=50, phi=98, saveAs=tmp.name)
153 os.fsync(tmp.fileno()) # be strict
154 size = os.path.getsize(tmp.name)
155 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
157 def testPlotMosaicZoomView(self) -> None:
158 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
159 self.plotter.plotMosaic(stampNum=-1, cutoutSize=12, plo=50, phi=98, saveAs=tmp.name)
160 os.fsync(tmp.fileno()) # be strict
161 size = os.path.getsize(tmp.name)
162 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
164 def testPlotMosaicStampZoomView(self) -> None:
165 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
166 self.plotter.plotMosaic(stampNum=4, cutoutSize=12, plo=50, phi=98, saveAs=tmp.name)
167 os.fsync(tmp.fileno()) # be strict
168 size = os.path.getsize(tmp.name)
169 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
171 def testStripPlotPsf(self) -> None:
172 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
173 self.plotter.stripPlot(plotType="psf", saveAs=tmp.name)
174 os.fsync(tmp.fileno())
175 size = os.path.getsize(tmp.name)
176 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
178 def testStripPlotCentroidAltAz(self) -> None:
179 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
180 self.plotter.stripPlot(plotType="centroidAltAz", saveAs=tmp.name)
181 os.fsync(tmp.fileno())
182 size = os.path.getsize(tmp.name)
183 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
185 def testStripPlotFlux(self) -> None:
186 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
187 self.plotter.stripPlot(plotType="flux", saveAs=tmp.name)
188 os.fsync(tmp.fileno())
189 size = os.path.getsize(tmp.name)
190 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
192 def testStripPlotShape(self) -> None:
193 with tempfile.NamedTemporaryFile(suffix=".png", delete=True) as tmp:
194 self.plotter.stripPlot(plotType="ellip", saveAs=tmp.name)
195 os.fsync(tmp.fileno())
196 size = os.path.getsize(tmp.name)
197 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
199 def testMakeGif(self) -> None:
200 with tempfile.NamedTemporaryFile(suffix=".gif", delete=True) as tmp:
201 # test the crop and zoom as a gif
202 self.plotter.makeAnimation(cutoutSize=14, plo=50, phi=98, saveAs=tmp.name)
203 os.fsync(tmp.fileno())
204 size = os.path.getsize(tmp.name)
205 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
207 def testMakeMp4(self) -> None:
208 with tempfile.NamedTemporaryFile(suffix=".mp4", delete=True) as tmp:
209 # test the full frame as an mp4
210 self.plotter.makeAnimation(cutoutSize=-1, plo=50, phi=98, saveAs=tmp.name)
211 os.fsync(tmp.fileno())
212 size = os.path.getsize(tmp.name)
213 self.assertGreater(size, 1000, f"{tmp.name} too small: {size} bytes")
215 def test_metrics(self) -> None:
216 # Check that metrics can be built without error
217 metrics = self.metricsBuilder.buildMetrics(self.expId)
218 self.assertIsInstance(metrics, pd.DataFrame)
219 self.assertGreater(len(metrics), 0, "Metrics DataFrame is empty")
221 # These are what's currently there. Why is the 8th guider missing?
222 expectedColumns = (
223 "n_guiders",
224 "n_stars",
225 "n_missing_stamps",
226 "n_measurements",
227 "fraction_possible_measurements",
228 "exptime",
229 "R00_SG0",
230 "R00_SG1",
231 "R04_SG0",
232 "R04_SG1",
233 "R40_SG0",
234 "R40_SG1",
235 "R44_SG0",
236 "R44_SG1",
237 "az_drift_slope",
238 "az_drift_intercept",
239 "az_drift_trend_rmse",
240 "az_drift_global_std",
241 "az_drift_outlier_frac",
242 "az_drift_slope_significance",
243 "az_drift_nsize",
244 "alt_drift_slope",
245 "alt_drift_intercept",
246 "alt_drift_trend_rmse",
247 "alt_drift_global_std",
248 "alt_drift_outlier_frac",
249 "alt_drift_slope_significance",
250 "alt_drift_nsize",
251 "rotator_slope",
252 "rotator_intercept",
253 "rotator_trend_rmse",
254 "rotator_global_std",
255 "rotator_outlier_frac",
256 "rotator_slope_significance",
257 "rotator_nsize",
258 "mag_slope",
259 "mag_intercept",
260 "mag_trend_rmse",
261 "mag_global_std",
262 "mag_outlier_frac",
263 "mag_slope_significance",
264 "mag_nsize",
265 "psf_slope",
266 "psf_intercept",
267 "psf_trend_rmse",
268 "psf_global_std",
269 "psf_outlier_frac",
270 "psf_slope_significance",
271 "psf_nsize",
272 )
273 for col in expectedColumns:
274 self.assertIn(col, metrics.columns, f"Column {col} is missing from metrics DataFrame")
276 # check this runs without error
277 self.metricsBuilder.printSummary()
279 def testTomographicSeeing(self) -> None:
280 analysis = CorrelationAnalysis(self.stars, self.expId)
281 variance = np.nanmedian(analysis.measureVariance())
282 self.assertGreater(variance, 0, "Variance should be positive and non-zero")
283 self.assertIsInstance(variance, float)
285 seeing = analysis.measureTomographicSeeing()
286 self.assertIsInstance(seeing, GuiderSeeing)
289class TestMemory(lsst.utils.tests.MemoryTestCase):
290 pass
293def setup_module(module):
294 lsst.utils.tests.init()
297if __name__ == "__main__": 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 lsst.utils.tests.init()
299 unittest.main()