Coverage for tests/test_tmaUtils.py: 17%

356 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-11-07 03:09 -0800

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 

22"""Test cases for utils.""" 

23 

24import asyncio 

25import os 

26import unittest 

27 

28import matplotlib.pyplot as plt 

29import numpy as np 

30import pandas as pd 

31from astropy.time import TimeDelta 

32from utils import getVcr 

33 

34import lsst.utils.tests 

35from lsst.summit.utils.efdUtils import calcNextDay, getDayObsStartTime, makeEfdClient 

36from lsst.summit.utils.enums import PowerState 

37from lsst.summit.utils.tmaUtils import ( 

38 AxisMotionState, 

39 TMAEvent, 

40 TMAEventMaker, 

41 TMAState, 

42 TMAStateMachine, 

43 _initializeTma, 

44 filterBadValues, 

45 getAxisAndType, 

46 getAzimuthElevationDataForEvent, 

47 getCommandsDuringEvent, 

48 getSlewsFromEventList, 

49 getTracksFromEventList, 

50 plotEvent, 

51) 

52 

53__all__ = [ 

54 "writeNewTmaEventTestTruthValues", 

55] 

56 

57TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

58vcr = getVcr() 

59 

60 

61def getTmaEventTestTruthValues(): 

62 """Get the current truth values for the TMA event test cases. 

63 

64 Returns 

65 ------- 

66 seqNums : `np.array` of `int` 

67 The sequence numbers of the events. 

68 startRows : `np.array` of `int` 

69 The _startRow numbers of the events. 

70 endRows : `np.array` of `int` 

71 The _endRow numbers of the events. 

72 types : `np.array` of `str` 

73 The event types, as a string, i.e. the ``TMAEvent.name`` of the event's 

74 ``event.type``. 

75 endReasons : `np.array` of `str` 

76 The event end reasons, as a string, i.e. the ``TMAEvent.name`` of the 

77 event's ``event.endReason``. 

78 """ 

79 dataFilename = os.path.join(TESTDIR, "data", "tmaEventData.txt") 

80 

81 seqNums, startRows, endRows, types, endReasons = np.genfromtxt( 

82 dataFilename, delimiter=",", dtype=None, names=True, encoding="utf-8", unpack=True 

83 ) 

84 return seqNums, startRows, endRows, types, endReasons 

85 

86 

87def writeNewTmaEventTestTruthValues(): 

88 """This function is used to write out the truth values for the test cases. 

89 

90 If the internal event creation logic changes, these values can change, and 

91 will need to be updated. Run this function, and check the new values into 

92 git. 

93 

94 Note: if you have cause to update values with this function, make sure to 

95 update the version number on the TMAEvent class. 

96 """ 

97 dayObs = 20241210 # obviously must match the day in the test class 

98 

99 eventMaker = TMAEventMaker() 

100 events = eventMaker.getEvents(dayObs) 

101 

102 dataFilename = os.path.join(TESTDIR, "data", "tmaEventData.txt") 

103 

104 columnHeader = "seqNum,startRow,endRow,type,endReason" 

105 with open(dataFilename, "w") as f: 

106 f.write(columnHeader + "\n") 

107 for event in events: 

108 line = ( 

109 f"{event.seqNum},{event._startRow},{event._endRow},{event.type.name}," 

110 f"{event.endReason.name}" 

111 ) 

112 f.write(line + "\n") 

113 

114 

115def makeValid(tma): 

116 """Helper function to turn a TMA into a valid state.""" 

117 for name, value in tma._parts.items(): 

118 if value == tma._UNINITIALIZED_VALUE: 

119 tma._parts[name] = 1 

120 

121 

122def _turnOn(tma): 

123 """Helper function to turn TMA axes on for testing. 

124 

125 Do not call directly in normal usage or code, as this just arbitrarily 

126 sets values to turn the axes on. 

127 

128 Parameters 

129 ---------- 

130 tma : `lsst.summit.utils.tmaUtils.TMAStateMachine` 

131 The TMA state machine model to initialize. 

132 """ 

133 tma._parts["azimuthSystemState"] = PowerState.ON 

134 tma._parts["elevationSystemState"] = PowerState.ON 

135 

136 

137class TmaUtilsTestCase(lsst.utils.tests.TestCase): 

138 def test_tmaInit(self): 

139 tma = TMAStateMachine() 

140 self.assertFalse(tma._isValid) 

141 

142 # setting one axis should not make things valid 

143 tma._parts["azimuthMotionState"] = 1 

144 self.assertFalse(tma._isValid) 

145 

146 # setting all the other components should make things valid 

147 tma._parts["azimuthInPosition"] = 1 

148 tma._parts["azimuthSystemState"] = 1 

149 tma._parts["elevationInPosition"] = 1 

150 tma._parts["elevationMotionState"] = 1 

151 tma._parts["elevationSystemState"] = 1 

152 self.assertTrue(tma._isValid) 

153 

154 def test_tmaReferences(self): 

155 """Check the linkage between the component lists and the _parts 

156 dict. 

157 """ 

158 tma = TMAStateMachine() 

159 

160 # setting one axis should not make things valid 

161 self.assertEqual(tma._parts["azimuthMotionState"], tma._UNINITIALIZED_VALUE) 

162 self.assertEqual(tma._parts["elevationMotionState"], tma._UNINITIALIZED_VALUE) 

163 tma.motion[0] = AxisMotionState.TRACKING # set azimuth to 0 

164 tma.motion[1] = AxisMotionState.TRACKING # set azimuth to 0 

165 self.assertEqual(tma._parts["azimuthMotionState"], AxisMotionState.TRACKING) 

166 self.assertEqual(tma._parts["elevationMotionState"], AxisMotionState.TRACKING) 

167 

168 def test_getAxisAndType(self): 

169 # check both the long and short form names work 

170 for s in ["azimuthMotionState", "lsst.sal.MTMount.logevent_azimuthMotionState"]: 

171 self.assertEqual(getAxisAndType(s), ("azimuth", "MotionState")) 

172 

173 # check in position, and use elevation instead of azimuth to test that 

174 for s in ["elevationInPosition", "lsst.sal.MTMount.logevent_elevationInPosition"]: 

175 self.assertEqual(getAxisAndType(s), ("elevation", "InPosition")) 

176 

177 for s in ["azimuthSystemState", "lsst.sal.MTMount.logevent_azimuthSystemState"]: 

178 self.assertEqual(getAxisAndType(s), ("azimuth", "SystemState")) 

179 

180 def test_initStateLogic(self): 

181 tma = TMAStateMachine() 

182 self.assertFalse(tma._isValid) 

183 self.assertFalse(tma.isMoving) 

184 self.assertFalse(tma.canMove) 

185 self.assertFalse(tma.isTracking) 

186 self.assertFalse(tma.isSlewing) 

187 self.assertEqual(tma.state, TMAState.UNINITIALIZED) 

188 

189 _initializeTma(tma) # we're valid, but still aren't moving and can't 

190 self.assertTrue(tma._isValid) 

191 self.assertNotEqual(tma.state, TMAState.UNINITIALIZED) 

192 self.assertTrue(tma.canMove) 

193 self.assertTrue(tma.isNotMoving) 

194 self.assertFalse(tma.isMoving) 

195 self.assertFalse(tma.isTracking) 

196 self.assertFalse(tma.isSlewing) 

197 

198 _turnOn(tma) # can now move, still valid, but not in motion 

199 self.assertTrue(tma._isValid) 

200 self.assertTrue(tma.canMove) 

201 self.assertTrue(tma.isNotMoving) 

202 self.assertFalse(tma.isMoving) 

203 self.assertFalse(tma.isTracking) 

204 self.assertFalse(tma.isSlewing) 

205 

206 # consider manipulating the axes by hand here and testing these? 

207 # it's likely not worth it, given how much this exercised elsewhere, 

208 # but these are the only functions not yet being directly tested 

209 # tma._axesInFault() 

210 # tma._axesOff() 

211 # tma._axesOn() 

212 # tma._axesInMotion() 

213 # tma._axesTRACKING() 

214 # tma._axesInPosition() 

215 

216 

217@vcr.use_cassette() 

218class TMAEventMakerTestCase(lsst.utils.tests.TestCase): 

219 @classmethod 

220 @vcr.use_cassette() 

221 def setUpClass(cls): 

222 try: 

223 cls.client = makeEfdClient(testing=True) 

224 except RuntimeError: 

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

226 

227 cls.dayObs = 20241210 

228 cls.dayObsWithBlockInfo = 20230615 

229 # get a sample expRecord here to test expRecordToTimespan 

230 cls.tmaEventMaker = TMAEventMaker(cls.client) 

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

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

233 

234 @vcr.use_cassette() 

235 def tearDown(self): 

236 loop = asyncio.get_event_loop() 

237 if self.client.influx_client is not None: 

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

239 

240 @vcr.use_cassette() 

241 def test_events(self): 

242 data = self.sampleData 

243 self.assertIsInstance(data, pd.DataFrame) 

244 self.assertEqual(len(data), 800) 

245 

246 @vcr.use_cassette() 

247 def test_rowDataForValues(self): 

248 rowsFor = set(self.sampleData["rowFor"]) 

249 self.assertEqual(len(rowsFor), 6) 

250 

251 # hard coding these ensures that you can't extend the axes/model 

252 # without being explicit about it here. 

253 correct = { 

254 "azimuthInPosition", 

255 "azimuthMotionState", 

256 "azimuthSystemState", 

257 "elevationInPosition", 

258 "elevationMotionState", 

259 "elevationSystemState", 

260 } 

261 self.assertSetEqual(rowsFor, correct) 

262 

263 @vcr.use_cassette() 

264 def test_monotonicTimeInDataframe(self): 

265 # ensure that each row is later than the previous 

266 times = self.sampleData["private_efdStamp"] 

267 self.assertTrue(np.all(np.diff(times) > 0)) 

268 

269 @vcr.use_cassette() 

270 def test_monotonicTimeApplicationOfRows(self): 

271 # ensure you can apply rows in the correct order 

272 tma = TMAStateMachine() 

273 row1 = self.sampleData.iloc[0] 

274 row2 = self.sampleData.iloc[1] 

275 

276 # just running this check it is OK 

277 tma.apply(row1) 

278 tma.apply(row2) 

279 

280 # and that if you apply them in reverse order then things will raise 

281 tma = TMAStateMachine() 

282 with self.assertRaises(ValueError): 

283 tma.apply(row2) 

284 tma.apply(row1) 

285 

286 @vcr.use_cassette() 

287 def test_fullDaySequence(self): 

288 # make sure we can apply all the data from the day without falling 

289 # through the logic sieve 

290 for engineering in (True, False): 

291 tma = TMAStateMachine(engineeringMode=engineering) 

292 

293 _initializeTma(tma) 

294 

295 for rowNum, row in self.sampleData.iterrows(): 

296 tma.apply(row) 

297 

298 @vcr.use_cassette() 

299 def test_endToEnd(self): 

300 eventMaker = self.tmaEventMaker 

301 events = eventMaker.getEvents(self.dayObs) 

302 self.assertIsInstance(events, list) 

303 self.assertEqual(len(events), 320) 

304 self.assertIsInstance(events[0], TMAEvent) 

305 

306 slews = [e for e in events if e.type == TMAState.SLEWING] 

307 tracks = [e for e in events if e.type == TMAState.TRACKING] 

308 self.assertEqual(len(slews), 172) 

309 self.assertEqual(len(tracks), 148) 

310 

311 seqNums, startRows, endRows, types, endReasons = getTmaEventTestTruthValues() 

312 for eventNum, event in enumerate(events): 

313 self.assertEqual(event.seqNum, seqNums[eventNum]) 

314 self.assertEqual(event._startRow, startRows[eventNum]) 

315 self.assertEqual(event._endRow, endRows[eventNum]) 

316 self.assertEqual(event.type.name, types[eventNum]) 

317 self.assertEqual(event.endReason.name, endReasons[eventNum]) 

318 

319 eventSet = set(slews) # check we can hash 

320 eventSet.update(slews) # check it ignores duplicates 

321 self.assertEqual(len(eventSet), len(slews)) 

322 

323 @vcr.use_cassette() 

324 def test_noDataBehaviour(self): 

325 eventMaker = self.tmaEventMaker 

326 noDataDayObs = 19600101 # do not use 19700101 - there is data for that day! 

327 with self.assertLogs(level="WARNING") as cm: 

328 correctMsg = f"No EFD data found for dayObs={noDataDayObs}" 

329 events = eventMaker.getEvents(noDataDayObs) 

330 self.assertIsInstance(events, list) 

331 self.assertEqual(len(events), 0) 

332 msg = cm.output[0] 

333 self.assertIn(correctMsg, msg) 

334 

335 @vcr.use_cassette() 

336 def test_helperFunctions(self): 

337 eventMaker = self.tmaEventMaker 

338 events = eventMaker.getEvents(self.dayObs) 

339 

340 slews = [e for e in events if e.type == TMAState.SLEWING] 

341 tracks = [e for e in events if e.type == TMAState.TRACKING] 

342 foundSlews = getSlewsFromEventList(events) 

343 foundTracks = getTracksFromEventList(events) 

344 self.assertEqual(slews, foundSlews) 

345 self.assertEqual(tracks, foundTracks) 

346 

347 def test_filterBadValues(self): 

348 # NB: if you add enough spurious values that the median is no longer 

349 # the value around which your "good" values are oscillating the first 

350 # two points will get replaced and this can be very confusing! 

351 

352 # test no bad values 

353 # mean = median = 1.0 

354 values = np.array([1.0, 0.96, 1.0, 1.04, 0.95, 1.0, 1.05, 1.0, 1.05, 1.0, 0.95]) 

355 mean = np.mean(values) 

356 nReplaced = filterBadValues(values) 

357 self.assertEqual(nReplaced, 0) 

358 self.assertEqual(np.mean(values), mean) 

359 

360 # test with one bad values 

361 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 1.0, 1.05, 1.0, 1.05, 1.0, 0.95]) 

362 nReplaced = filterBadValues(values) 

363 self.assertEqual(nReplaced, 1) 

364 

365 # test with two consecutive bad values 

366 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 1.05, 1.0, 1.05, 1.0, 0.95]) 

367 nReplaced = filterBadValues(values) 

368 self.assertEqual(nReplaced, 2) 

369 

370 # test with three consecutive bad values 

371 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 1.0, 1.05, 1.0, 0.95]) 

372 nReplaced = filterBadValues(values) 

373 self.assertEqual(nReplaced, 3) 

374 

375 # test with three consecutive bad values and another at the end 

376 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 1.0, 1.05, 1.0, 3.95]) 

377 nReplaced = filterBadValues(values) 

378 self.assertEqual(nReplaced, 4) 

379 

380 # test with more than three consecutive bad values 

381 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 0.95]) 

382 nReplaced = filterBadValues(values) 

383 self.assertEqual(nReplaced, 3) 

384 self.assertIn(5.0, values) # check the last bad value is still there specifically 

385 

386 # test with more than three consecutive bad values and another bad 

387 # value at the end 

388 values = np.array([1.0, 0.96, 1.0, 1.04, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 2.95]) 

389 nReplaced = filterBadValues(values) 

390 self.assertEqual(nReplaced, 4) 

391 

392 # test with bad values in first two positions 

393 values = np.array([2.0, 1.96, 1.0, 1.04, 0.95, 1.0, 1.05, 1.0, 1.05, 1.0, 0.95]) # median = 1.0 

394 nReplaced = filterBadValues(values) 

395 self.assertEqual(nReplaced, 2) 

396 

397 # test with bad values in first two positions and one in the middle 

398 values = np.array([2.0, 1.96, 1.0, 1.04, 0.95, 5.0, 1.04, 1.0, 1.05, 1.0, 0.95]) 

399 nReplaced = filterBadValues(values) 

400 self.assertEqual(nReplaced, 3) 

401 

402 # check that the last two good values are always used for correction, 

403 # including when there are more than three consecutive bad values. 

404 values = np.array([1.0, 0.96, 1.0, 1.02, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 2.95]) 

405 expected = np.array([1.0, 0.96, 1.0, 1.02, 1.01, 1.01, 1.01, 5.0, 1.05, 1.0, 1.025]) 

406 nReplaced = filterBadValues(values) 

407 residuals = np.abs(values - expected) 

408 self.assertEqual(nReplaced, 4) 

409 self.assertTrue(np.all(residuals < 1e-6)) 

410 

411 # check with one good point after an overflowing run of bad to make 

412 # sure the correction is always applied with good values, not the naive 

413 # average of the last two even if they might be bad 

414 values = np.array([1.0, 0.96, 1.0, 1.02, 2.95, 3.0, 4.05, 5.0, 1.05, 2.95, 1.0]) 

415 expected = np.array([1.0, 0.96, 1.0, 1.02, 1.01, 1.01, 1.01, 5.0, 1.05, 1.035, 1.0]) 

416 nReplaced = filterBadValues(values) 

417 residuals = np.abs(values - expected) 

418 self.assertEqual(nReplaced, 4) 

419 self.assertTrue(np.all(residuals < 1e-6)) 

420 

421 # check with non-default maxDelta 

422 values = np.array([1.0, 0.96, 1.0, 1.02, 2.95, 3.0, 4.05, 5.0, 1.05, 1.0, 2.95]) 

423 nReplaced = filterBadValues(values, maxDelta=10) 

424 self.assertEqual(nReplaced, 0) 

425 

426 values = np.array( 

427 [ 

428 1.0, 

429 1.0, 

430 1.0, 

431 1.1, 

432 1.0, 

433 1.0, 

434 1.0, 

435 1.0, 

436 1.0, 

437 1.0, 

438 ] 

439 ) 

440 nReplaced = filterBadValues(values, maxDelta=0.01) 

441 self.assertEqual(nReplaced, 1) 

442 

443 @vcr.use_cassette() 

444 def test_getEvent(self): 

445 # test the singular event getter, and what happens if the event doesn't 

446 # exist for the day 

447 eventMaker = self.tmaEventMaker 

448 events = eventMaker.getEvents(self.dayObs) 

449 nEvents = len(events) 

450 

451 event = eventMaker.getEvent(self.dayObs, 0) 

452 self.assertIsInstance(event, TMAEvent) 

453 self.assertEqual(event, events[0]) 

454 event = eventMaker.getEvent(self.dayObs, 100) 

455 self.assertIsInstance(event, TMAEvent) 

456 self.assertEqual(event, events[100]) 

457 

458 with self.assertLogs(level="WARNING") as cm: 

459 correctMsg = f"Event {nEvents + 1} not found for {self.dayObs}" 

460 event = eventMaker.getEvent(self.dayObs, nEvents + 1) 

461 msg = cm.output[0] 

462 self.assertIn(correctMsg, msg) 

463 

464 @vcr.use_cassette() 

465 def test_printing(self): 

466 eventMaker = self.tmaEventMaker 

467 events = eventMaker.getEvents(self.dayObs) 

468 

469 # test str(), repr(), and _ipython_display_() for an event 

470 print(str(events[0])) 

471 print(repr(events[0])) 

472 print(events[0]._ipython_display_()) 

473 

474 # spot-check both a slow and a track to print 

475 slews = [e for e in events if e.type == TMAState.SLEWING] 

476 tracks = [e for e in events if e.type == TMAState.TRACKING] 

477 eventMaker.printEventDetails(slews[0]) 

478 eventMaker.printEventDetails(tracks[0]) 

479 eventMaker.printEventDetails(events[-1]) 

480 

481 # check the full day trick works 

482 eventMaker.printFullDayStateEvolution(self.dayObs) 

483 

484 tma = TMAStateMachine() 

485 _initializeTma(tma) # the uninitialized state contains wrong types for printing 

486 eventMaker.printTmaDetailedState(tma) 

487 

488 @vcr.use_cassette() 

489 def test_getAxisData(self): 

490 eventMaker = self.tmaEventMaker 

491 events = eventMaker.getEvents(self.dayObs) 

492 

493 azData, elData = getAzimuthElevationDataForEvent(self.client, events[0]) 

494 self.assertIsInstance(azData, pd.DataFrame) 

495 self.assertIsInstance(elData, pd.DataFrame) 

496 

497 paddedAzData, paddedElData = getAzimuthElevationDataForEvent( 

498 self.client, events[0], prePadding=2, postPadding=1 

499 ) 

500 self.assertGreater(len(paddedAzData), len(azData)) 

501 self.assertGreater(len(paddedElData), len(elData)) 

502 

503 # just check this doesn't raise when called, and check we can pass the 

504 # data in 

505 plotEvent(self.client, events[0], azimuthData=azData, elevationData=elData) 

506 

507 @vcr.use_cassette() 

508 def test_plottingAndCommands(self): 

509 eventMaker = self.tmaEventMaker 

510 events = eventMaker.getEvents(self.dayObs) 

511 event = events[10] # this one has commands, and we'll check that later 

512 

513 # check we _can_ plot without a figure, and then stop doing that 

514 plotEvent(self.client, event) 

515 

516 fig = plt.figure(figsize=(10, 8)) 

517 # just check this doesn't raise when called 

518 plotEvent(self.client, event, fig=fig) 

519 plt.close(fig) 

520 

521 commandsToPlot = ["raDecTarget", "moveToTarget", "startTracking", "stopTracking"] 

522 commands = getCommandsDuringEvent(self.client, event, commandsToPlot, doLog=False) 

523 self.assertTrue(not all([time is None for time in commands.values()])) # at least one command 

524 

525 plotEvent(self.client, event, fig=fig, commands=commands) 

526 

527 del fig 

528 

529 @vcr.use_cassette() 

530 def test_findEvent(self): 

531 eventMaker = self.tmaEventMaker 

532 # addBlockInfo=True because it shouldn't affect the comparison, and 

533 # this also then ensures that the code is exercised too 

534 events = eventMaker.getEvents(self.dayObs, addBlockInfo=True) 

535 event = events[28] # this one has a contiguous event before it 

536 

537 time = event.begin 

538 found = eventMaker.findEvent(time) 

539 self.assertEqual(found, event) 

540 

541 dt = TimeDelta(0.01, format="sec") 

542 # must be just inside to get the same event back, because if a moment 

543 # is shared it gives the one which starts with the moment (whilst 

544 # logging info messages about it) 

545 time = event.end - dt 

546 found = eventMaker.findEvent(time) 

547 self.assertEqual(found, event) 

548 

549 # now check that if we're a hair after, we don't get the same event 

550 time = event.end + dt 

551 found = eventMaker.findEvent(time) 

552 self.assertNotEqual(found, event) 

553 

554 # Now check the cases which don't find an event at all. It would be 

555 # nice to check the log messages here, but it seems too fragile to be 

556 # worth it 

557 dt = TimeDelta(1, format="sec") 

558 tooEarlyOnDay = getDayObsStartTime(self.dayObs) + dt # 1 second after start of day 

559 found = eventMaker.findEvent(tooEarlyOnDay) 

560 self.assertIsNone(found) 

561 

562 # 1 second before end of day and this day does not end with an open 

563 # event 

564 tooLateOnDay = getDayObsStartTime(calcNextDay(self.dayObs)) - dt 

565 found = eventMaker.findEvent(tooLateOnDay) 

566 self.assertIsNone(found) 

567 

568 # going just inside the last event of the day should be fine 

569 lastEvent = events[-1] 

570 found = eventMaker.findEvent(lastEvent.end - dt) 

571 self.assertEqual(found, lastEvent) 

572 

573 # going at the very end of the last event of the day should actually 

574 # find nothing, because the last moment of an event isn't actually in 

575 # the event itself, because of how contiguous events are defined to 

576 # behave (being half-open intervals) 

577 found = eventMaker.findEvent(lastEvent.end) 

578 self.assertIsNone(found, lastEvent) 

579 

580 @vcr.use_cassette() 

581 def test_eventAssociatedWith(self): 

582 eventMaker = self.tmaEventMaker 

583 events = eventMaker.getEvents(self.dayObsWithBlockInfo) 

584 eventsWithBlockInfo = [e for e in events if e.blockInfos] 

585 eventsWithoutBlockInfo = [e for e in events if not e.blockInfos] 

586 self.assertEqual(len(events), 69) 

587 self.assertEqual(len(eventsWithBlockInfo), 65) 

588 self.assertEqual(len(eventsWithoutBlockInfo), 4) 

589 

590 self.assertIsNotNone(eventsWithoutBlockInfo[0].blockInfos) 

591 self.assertIsInstance(eventsWithoutBlockInfo[0].blockInfos, list) 

592 self.assertEqual(len(eventsWithoutBlockInfo[0].blockInfos), 0) 

593 

594 event = eventsWithBlockInfo[0] 

595 self.assertIsInstance(event, TMAEvent) 

596 self.assertTrue(event.associatedWith(ticket="SITCOM-906")) 

597 self.assertFalse(event.associatedWith(ticket="SITCOM-905")) 

598 

599 self.assertTrue(event.associatedWith(salIndex=100017)) 

600 self.assertFalse(event.associatedWith(salIndex=100018)) 

601 

602 self.assertTrue(event.associatedWith(block=6)) 

603 self.assertFalse(event.associatedWith(block=5)) 

604 

605 # check it works with any and all of the arguments 

606 self.assertTrue(event.associatedWith(block=6, salIndex=100017)) 

607 self.assertTrue(event.associatedWith(block=6, salIndex=100017, ticket="SITCOM-906")) 

608 

609 # check it's false if any are false 

610 self.assertFalse(event.associatedWith(block=7, salIndex=100017, ticket="SITCOM-906")) # 1 wrong 

611 self.assertFalse(event.associatedWith(block=6, salIndex=100018, ticket="SITCOM-906")) # 1 wrong 

612 self.assertFalse(event.associatedWith(block=6, salIndex=100017, ticket="SITCOM-907")) # 1 wrong 

613 self.assertFalse(event.associatedWith(block=1, salIndex=1, ticket="SITCOM-1")) # all wrong 

614 

615 # check with the blockSeqNum, with and without the other items 

616 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1)) 

617 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2)) 

618 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1, salIndex=100017)) 

619 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2, salIndex=100017)) 

620 self.assertTrue(event.associatedWith(block=6, blockSeqNum=1, salIndex=100017, ticket="SITCOM-906")) 

621 self.assertFalse(event.associatedWith(block=6, blockSeqNum=2, salIndex=100017, ticket="SITCOM-906")) 

622 

623 with self.assertRaises(ValueError): 

624 event.associatedWith() 

625 event.associatedWith(blockSeqNum=1) # nonsense to ask for a seqNum without a block number 

626 

627 

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

629 pass 

630 

631 

632def setup_module(module): 

633 lsst.utils.tests.init() 

634 

635 

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

637 lsst.utils.tests.init() 

638 unittest.main()