Coverage for tests/test_nightReport.py: 30%
148 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-09-28 10:40 +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 datetime
23import itertools
24import os
25import tempfile
26import unittest
27from unittest import mock
29import matplotlib as mpl
30from astro_metadata_translator import ObservationInfo
31from numpy.random import shuffle
33import lsst.utils.tests
35mpl.use("Agg")
37import lsst.summit.utils.butlerUtils as butlerUtils # noqa: E402
38from lsst.daf.butler import Butler # noqa: E402
39from lsst.summit.utils.nightReport import ColorAndMarker, NightReport # noqa: E402
40from lsst.summit.utils.utils import getSite # noqa: E402
43class NightReportTestCase(lsst.utils.tests.TestCase):
44 # class attributes populated in setUpClass
45 butler: Butler
46 dayObs: int
47 report: NightReport
48 nImages: int
49 seqNums: list[int]
51 @classmethod
52 def setUpClass(cls) -> None:
53 try:
54 if getSite() == "jenkins": 54 ↛ 56line 54 didn't jump to line 56 because the condition on line 54 was always true
55 raise unittest.SkipTest("Skip running butler-driven tests in Jenkins.")
56 cls.butler = butlerUtils.makeDefaultLatissButler()
57 except FileNotFoundError:
58 raise unittest.SkipTest("Skipping tests that require the LATISS butler repo.")
60 cls.dayObs = 20200314 # has 377 images and data also exists on the TTS & summit
62 # Do the init in setUpClass because this takes about 35s for 20200314
63 cls.report = NightReport(cls.butler, cls.dayObs)
64 # number of images isn't necessarily the same as the number for the
65 # the dayObs in the registry becacuse of the test stands/summit
66 # having partial data, so get the number of images from the length
67 # of the scraped data. Not ideal, but best that can be done due to
68 # only having partial days in the test datasets.
69 cls.nImages = len(cls.report.data.keys())
70 cls.seqNums = list(cls.report.data.keys())
72 def test_saveAndLoad(self) -> None:
73 """Test that a NightReport can save itself, and be loaded back."""
74 writeDir = tempfile.mkdtemp()
75 saveFile = os.path.join(writeDir, f"testNightReport_{self.dayObs}.pickle")
76 self.report.save(saveFile)
77 self.assertTrue(os.path.exists(saveFile))
79 loaded = NightReport(self.butler, self.dayObs, saveFile)
80 self.assertIsInstance(loaded, lsst.summit.utils.nightReport.NightReport)
81 self.assertGreaterEqual(len(loaded.data), 1)
82 self.assertEqual(loaded.dayObs, self.dayObs)
84 # TODO: add a self.assertRaises on a mismatched dayObs
86 def test_getSortedData(self) -> None:
87 """Test the _getSortedData returns the seqNums in order."""
88 shuffledKeys = list(self.report.data.keys())
89 shuffle(shuffledKeys)
90 shuffledData = {k: self.report.data[k] for k in shuffledKeys}
92 sortedData = self.report._getSortedData(shuffledData)
93 sortedKeys = sorted(list(sortedData.keys()))
94 self.assertEqual(sortedKeys, list(self.report.data.keys()))
95 return
97 def test_getExpRecordDictForDayObs(self) -> None:
98 """Test getExpRecordDictForDayObs.
100 Test it returns a dict of dicts, keyed by integer seqNums.
101 """
102 expRecDict = self.report.getExpRecordDictForDayObs(self.dayObs)
103 self.assertIsInstance(expRecDict, dict)
104 self.assertGreaterEqual(len(expRecDict), 1)
106 # check all the keys are ints
107 seqNums = list(expRecDict.keys())
108 self.assertTrue(all(isinstance(s, int) for s in seqNums))
110 # check all the values are dicts
111 self.assertTrue(all(isinstance(expRecDict[s], dict) for s in seqNums))
112 return
114 def test_getObsInfoAndMetadataForSeqNum(self) -> None:
115 """Test that getObsInfoAndMetadataForSeqNum returns the correct
116 types.
117 """
118 seqNum = self.seqNums[0]
119 obsInfo, md = self.report.getObsInfoAndMetadataForSeqNum(seqNum)
120 self.assertIsInstance(obsInfo, ObservationInfo)
121 self.assertIsInstance(md, dict)
122 return
124 def test_rebuild(self) -> None:
125 """Test that rebuild does nothing, as no data will be being added.
127 NB Do not call full=True on this, as it will double the length of the
128 tests and they're already extremely slow.
129 """
130 lenBefore = len(self.report.data)
131 self.report.rebuild()
132 self.assertEqual(len(self.report.data), lenBefore)
133 return
135 def test_getExposureMidpoint(self) -> None:
136 """Test the exposure midpoint calculation"""
137 # we would like a non-zero exptime exposure really
138 seqNumToUse = 0
139 expTime = 0
140 for seqNum in self.report.data.keys():
141 expTime = self.report.data[seqNum]["exposure_time"]
142 if expTime > 0:
143 seqNumToUse = seqNum
144 break
146 midPoint = self.report.getExposureMidpoint(seqNumToUse)
147 record = self.report.data[seqNumToUse]
149 if expTime == 0:
150 self.assertGreaterEqual(midPoint, record["datetime_begin"].to_datetime())
151 self.assertLessEqual(midPoint, record["datetime_end"].to_datetime())
152 else:
153 self.assertGreater(midPoint, record["datetime_begin"].to_datetime())
154 self.assertLess(midPoint, record["datetime_end"].to_datetime())
155 return
157 def test_getTimeDeltas(self) -> None:
158 """Test the time delta calculation returns a dict."""
159 dts = self.report.getTimeDeltas()
160 self.assertIsInstance(dts, dict)
161 return
163 def test_makeStarColorAndMarkerMap(self) -> None:
164 """Test the color map maker returns a dict of ColorAndMarker
165 objects.
166 """
167 cMap = self.report.makeStarColorAndMarkerMap(self.report.stars)
168 self.assertEqual(len(cMap), len(self.report.stars))
169 self.assertIsInstance(cMap, dict)
170 values = list(cMap.values())
171 self.assertTrue(all(isinstance(value, ColorAndMarker) for value in values))
172 return
174 def test_printObsTable(self) -> None:
175 """Test that a the printObsTable() method prints out the correct
176 number of lines.
177 """
178 with mock.patch("sys.stdout") as fake_stdout:
179 self.report.printObsTable()
181 # newline for each row plus header line, plus the line with dashes
182 self.assertEqual(len(fake_stdout.mock_calls), 2 * (self.nImages + 2))
184 def test_plotPerObjectAirMass(self) -> None:
185 """Test that a the per-object airmass plots runs."""
186 # We assume matplotlib is making plots, so just check that these
187 # don't crash.
189 # Default plotting:
190 self.report.plotPerObjectAirMass()
191 # plot with only one object as a str not a list of str
192 self.report.plotPerObjectAirMass(objects=self.report.stars[0])
193 # plot with first two objects as a list
194 self.report.plotPerObjectAirMass(objects=self.report.stars[0:2])
195 # flip y axis option
196 self.report.plotPerObjectAirMass(airmassOneAtTop=True)
197 # flip and select stars
198 self.report.plotPerObjectAirMass(objects=self.report.stars[0], airmassOneAtTop=True) # both
200 def test_makeAltAzCoveragePlot(self) -> None:
201 """Test that a the polar coverage plotting code runs."""
202 # We assume matplotlib is making plots, so just check that these
203 # don't crash.
205 # test the default case
206 self.report.makeAltAzCoveragePlot()
207 # plot with only one object as a str not a list of str
208 self.report.makeAltAzCoveragePlot(objects=self.report.stars[0])
209 # plot with first two objects as a list
210 self.report.makeAltAzCoveragePlot(objects=self.report.stars[0:2])
211 # test turning lines off
212 self.report.makeAltAzCoveragePlot(objects=self.report.stars[0:2], withLines=False)
214 def test_calcShutterTimes(self) -> None:
215 timings = self.report.calcShutterTimes()
216 if not timings:
217 return # if the day has no on-sky observations, this returns None
218 efficiency = 100 * (timings["scienceTimeTotal"] / timings["nightLength"])
219 self.assertGreater(efficiency, 0)
220 self.assertLessEqual(efficiency, 100)
222 def test_getDatesForSeqNums(self) -> None:
223 dateTimeDict = self.report.getDatesForSeqNums()
224 self.assertIsInstance(dateTimeDict, dict)
225 self.assertTrue(all(isinstance(seqNum, int) for seqNum in dateTimeDict.keys()))
226 self.assertTrue(all(isinstance(seqNum, datetime.datetime) for seqNum in dateTimeDict.values()))
228 def test_doesNotRaise(self) -> None:
229 """Tests for things which are hard to test, so just make sure they
230 run.
231 """
232 self.report.printShutterTimes()
233 for sample, includeRaw in itertools.product((True, False), (True, False)):
234 self.report.printAvailableKeys(sample=sample, includeRaw=includeRaw)
235 self.report.printObsTable()
236 for threshold, includeCalibs in itertools.product((0, 1, 10), (True, False)):
237 self.report.printObsGaps(threshold=threshold, includeCalibs=includeCalibs)
239 def test_internals(self) -> None:
240 startNum = self.report.getObservingStartSeqNum()
241 self.assertIsInstance(startNum, int)
242 assert startNum is not None
243 self.assertGreater(startNum, 0) # the day starts at 1, so zero would be an error of some sort
245 starsFromGetter = self.report.getObservedObjects()
246 self.assertIsInstance(starsFromGetter, list)
247 self.assertSetEqual(set(starsFromGetter), set(self.report.stars))
249 starsFromGetter = self.report.getObservedObjects(ignoreTileNum=True)
250 self.assertLessEqual(len(starsFromGetter), len(self.report.stars))
252 # check the internal color map has the right number of items
253 self.assertEqual(len(self.report.cMap), len(starsFromGetter))
256class TestMemory(lsst.utils.tests.MemoryTestCase):
257 pass
260def setup_module(module: object) -> None:
261 lsst.utils.tests.init()
264if __name__ == "__main__": 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 lsst.utils.tests.init()
266 unittest.main()