Coverage for tests/test_ics.py: 92%

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

21 

22import asyncio 

23import logging 

24import os 

25import tempfile 

26import unittest 

27from typing import Any 

28 

29import matplotlib.pyplot as plt 

30from utils import getVcr 

31 

32import lsst.utils.tests 

33from lsst.summit.utils.efdUtils import makeEfdClient 

34from lsst.summit.utils.m1m3.inertia_compensation_system import evaluate_m1m3_ics_single_slew 

35from lsst.summit.utils.m1m3.plots.plot_ics import FIGURE_HEIGHT, FIGURE_WIDTH, plot_hp_measured_data 

36from lsst.summit.utils.tmaUtils import TMAEvent, TMAEventMaker 

37 

38vcr = getVcr() 

39 

40 

41@vcr.use_cassette() 

42class M1M3ICSTestCase(lsst.utils.tests.TestCase): 

43 # class attributes populated in setUp 

44 client: Any 

45 dayObs: int 

46 seqNumToPlot: int 

47 tmaEventMaker: TMAEventMaker 

48 events: list[TMAEvent] 

49 sampleData: Any 

50 outputDir: str 

51 log: logging.Logger 

52 

53 @classmethod 

54 @vcr.use_cassette() 

55 def setUp(cls) -> None: 

56 try: 

57 cls.client = makeEfdClient(testing=True) 

58 except RuntimeError: 

59 raise unittest.SkipTest("Could not instantiate an EFD client") 

60 

61 cls.dayObs = 20230728 # need a day with M1M3 data 

62 cls.seqNumToPlot = 38 

63 

64 cls.tmaEventMaker = TMAEventMaker(cls.client) 

65 cls.events = cls.tmaEventMaker.getEvents(cls.dayObs) # does the fetch 

66 cls.sampleData = cls.tmaEventMaker._data[cls.dayObs] # pull the data from the object and test length 

67 cls.outputDir = tempfile.mkdtemp() 

68 cls.log = logging.getLogger(__name__) 

69 

70 @vcr.use_cassette() 

71 def tearDown(self) -> None: 

72 loop = asyncio.get_event_loop() 

73 if self.client.influx_client is not None: 73 ↛ exitline 73 didn't return from function 'tearDown' because the condition on line 73 was always true

74 loop.run_until_complete(self.client.influx_client.close()) 

75 

76 @vcr.use_cassette() 

77 def test_analysis(self) -> None: 

78 self.log.info(f"Writing temp output files to {self.outputDir}") 

79 plotFilename = os.path.join(self.outputDir, "testPlotting_exp.jpg") 

80 statFilename = os.path.join(self.outputDir, "m1m3_ics_stats.csv") 

81 dataFilename = os.path.join(self.outputDir, "m1m3_ics_df.csv") 

82 

83 event = self.events[self.seqNumToPlot] 

84 

85 results = evaluate_m1m3_ics_single_slew(event, self.client) 

86 results.stats.to_csv(statFilename) 

87 results.df.to_csv(dataFilename) 

88 

89 self.assertTrue(os.path.isfile(dataFilename)) 

90 # data is big, about 2.5MB at time of writing 

91 self.assertTrue(os.path.getsize(dataFilename) > 400_000) 

92 # stats are small, about 1.8kB at time of writing 

93 self.assertTrue(os.path.isfile(statFilename)) 

94 self.assertTrue(os.path.getsize(statFilename) > 1000) 

95 

96 dpi = 300 

97 fig = plt.figure(figsize=(FIGURE_WIDTH, FIGURE_HEIGHT), dpi=dpi) 

98 fig = plot_hp_measured_data(results, fig) 

99 fig.savefig(plotFilename) 

100 self.assertTrue(os.path.isfile(plotFilename)) 

101 self.assertTrue(os.path.getsize(plotFilename) > 200_000) # plot is about 400kB at 300 dpi 

102 

103 

104class TestMemory(lsst.utils.tests.MemoryTestCase): 

105 pass 

106 

107 

108def setup_module(module: object) -> None: 

109 lsst.utils.tests.init() 

110 

111 

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

113 lsst.utils.tests.init() 

114 unittest.main()