Coverage for tests / test_tmaUtils.py: 17%

357 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-17 09:44 +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/>. 

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.dateTime import calcNextDay, getDayObsStartTime 

36from lsst.summit.utils.efdUtils import makeEfdClient 

37from lsst.summit.utils.enums import PowerState 

38from lsst.summit.utils.tmaUtils import ( 

39 AxisMotionState, 

40 TMAEvent, 

41 TMAEventMaker, 

42 TMAState, 

43 TMAStateMachine, 

44 _initializeTma, 

45 filterBadValues, 

46 getAxisAndType, 

47 getAzimuthElevationDataForEvent, 

48 getCommandsDuringEvent, 

49 getSlewsFromEventList, 

50 getTracksFromEventList, 

51 plotEvent, 

52) 

53 

54__all__ = [ 

55 "writeNewTmaEventTestTruthValues", 

56] 

57 

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

59vcr = getVcr() 

60 

61 

62def getTmaEventTestTruthValues(): 

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

64 

65 Returns 

66 ------- 

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

68 The sequence numbers of the events. 

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

70 The _startRow numbers of the events. 

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

72 The _endRow numbers of the events. 

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

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

75 ``event.type``. 

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

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

78 event's ``event.endReason``. 

79 """ 

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

81 

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

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

84 ) 

85 return seqNums, startRows, endRows, types, endReasons 

86 

87 

88def writeNewTmaEventTestTruthValues(): 

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

90 

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

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

93 git. 

94 

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

96 update the version number on the TMAEvent class. 

97 """ 

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

99 

100 eventMaker = TMAEventMaker() 

101 events = eventMaker.getEvents(dayObs) 

102 

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

104 

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

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

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

108 for event in events: 

109 line = ( 

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

111 f"{event.endReason.name}" 

112 ) 

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

114 

115 

116def makeValid(tma): 

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

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

119 if value == tma._UNINITIALIZED_VALUE: 

120 tma._parts[name] = 1 

121 

122 

123def _turnOn(tma): 

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

125 

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

127 sets values to turn the axes on. 

128 

129 Parameters 

130 ---------- 

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

132 The TMA state machine model to initialize. 

133 """ 

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

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

136 

137 

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

139 def test_tmaInit(self): 

140 tma = TMAStateMachine() 

141 self.assertFalse(tma._isValid) 

142 

143 # setting one axis should not make things valid 

144 tma._parts["azimuthMotionState"] = 1 

145 self.assertFalse(tma._isValid) 

146 

147 # setting all the other components should make things valid 

148 tma._parts["azimuthInPosition"] = 1 

149 tma._parts["azimuthSystemState"] = 1 

150 tma._parts["elevationInPosition"] = 1 

151 tma._parts["elevationMotionState"] = 1 

152 tma._parts["elevationSystemState"] = 1 

153 self.assertTrue(tma._isValid) 

154 

155 def test_tmaReferences(self): 

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

157 dict. 

158 """ 

159 tma = TMAStateMachine() 

160 

161 # setting one axis should not make things valid 

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

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

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

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

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

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

168 

169 def test_getAxisAndType(self): 

170 # check both the long and short form names work 

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

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

173 

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

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

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

177 

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

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

180 

181 def test_initStateLogic(self): 

182 tma = TMAStateMachine() 

183 self.assertFalse(tma._isValid) 

184 self.assertFalse(tma.isMoving) 

185 self.assertFalse(tma.canMove) 

186 self.assertFalse(tma.isTracking) 

187 self.assertFalse(tma.isSlewing) 

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

189 

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

191 self.assertTrue(tma._isValid) 

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

193 self.assertTrue(tma.canMove) 

194 self.assertTrue(tma.isNotMoving) 

195 self.assertFalse(tma.isMoving) 

196 self.assertFalse(tma.isTracking) 

197 self.assertFalse(tma.isSlewing) 

198 

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

200 self.assertTrue(tma._isValid) 

201 self.assertTrue(tma.canMove) 

202 self.assertTrue(tma.isNotMoving) 

203 self.assertFalse(tma.isMoving) 

204 self.assertFalse(tma.isTracking) 

205 self.assertFalse(tma.isSlewing) 

206 

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

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

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

210 # tma._axesInFault() 

211 # tma._axesOff() 

212 # tma._axesOn() 

213 # tma._axesInMotion() 

214 # tma._axesTRACKING() 

215 # tma._axesInPosition() 

216 

217 

218@vcr.use_cassette() 

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

220 @classmethod 

221 @vcr.use_cassette() 

222 def setUpClass(cls): 

223 try: 

224 cls.client = makeEfdClient(testing=True) 

225 except RuntimeError: 

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

227 

228 cls.dayObs = 20241210 

229 cls.dayObsWithBlockInfo = 20230615 

230 # get a sample expRecord here to test expRecordToTimespan 

231 cls.tmaEventMaker = TMAEventMaker(cls.client) 

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

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

234 

235 @vcr.use_cassette() 

236 def tearDown(self): 

237 loop = asyncio.get_event_loop() 

238 if self.client.influx_client is not None: 

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

240 

241 @vcr.use_cassette() 

242 def test_events(self): 

243 data = self.sampleData 

244 self.assertIsInstance(data, pd.DataFrame) 

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

246 

247 @vcr.use_cassette() 

248 def test_rowDataForValues(self): 

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

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

251 

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

253 # without being explicit about it here. 

254 correct = { 

255 "azimuthInPosition", 

256 "azimuthMotionState", 

257 "azimuthSystemState", 

258 "elevationInPosition", 

259 "elevationMotionState", 

260 "elevationSystemState", 

261 } 

262 self.assertSetEqual(rowsFor, correct) 

263 

264 @vcr.use_cassette() 

265 def test_monotonicTimeInDataframe(self): 

266 # ensure that each row is later than the previous 

267 times = self.sampleData["private_efdStamp"] 

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

269 

270 @vcr.use_cassette() 

271 def test_monotonicTimeApplicationOfRows(self): 

272 # ensure you can apply rows in the correct order 

273 tma = TMAStateMachine() 

274 row1 = self.sampleData.iloc[0] 

275 row2 = self.sampleData.iloc[1] 

276 

277 # just running this check it is OK 

278 tma.apply(row1) 

279 tma.apply(row2) 

280 

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

282 tma = TMAStateMachine() 

283 with self.assertRaises(ValueError): 

284 tma.apply(row2) 

285 tma.apply(row1) 

286 

287 @vcr.use_cassette() 

288 def test_fullDaySequence(self): 

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

290 # through the logic sieve 

291 for engineering in (True, False): 

292 tma = TMAStateMachine(engineeringMode=engineering) 

293 

294 _initializeTma(tma) 

295 

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

297 tma.apply(row) 

298 

299 @vcr.use_cassette() 

300 def test_endToEnd(self): 

301 eventMaker = self.tmaEventMaker 

302 events = eventMaker.getEvents(self.dayObs) 

303 self.assertIsInstance(events, list) 

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

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

306 

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

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

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

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

311 

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

313 for eventNum, event in enumerate(events): 

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

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

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

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

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

319 

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

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

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

323 

324 @vcr.use_cassette() 

325 def test_noDataBehaviour(self): 

326 eventMaker = self.tmaEventMaker 

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

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

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

330 events = eventMaker.getEvents(noDataDayObs) 

331 self.assertIsInstance(events, list) 

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

333 msg = cm.output[0] 

334 self.assertIn(correctMsg, msg) 

335 

336 @vcr.use_cassette() 

337 def test_helperFunctions(self): 

338 eventMaker = self.tmaEventMaker 

339 events = eventMaker.getEvents(self.dayObs) 

340 

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

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

343 foundSlews = getSlewsFromEventList(events) 

344 foundTracks = getTracksFromEventList(events) 

345 self.assertEqual(slews, foundSlews) 

346 self.assertEqual(tracks, foundTracks) 

347 

348 def test_filterBadValues(self): 

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

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

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

352 

353 # test no bad values 

354 # mean = median = 1.0 

355 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]) 

356 mean = np.mean(values) 

357 nReplaced = filterBadValues(values) 

358 self.assertEqual(nReplaced, 0) 

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

360 

361 # test with one bad values 

362 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]) 

363 nReplaced = filterBadValues(values) 

364 self.assertEqual(nReplaced, 1) 

365 

366 # test with two consecutive bad values 

367 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]) 

368 nReplaced = filterBadValues(values) 

369 self.assertEqual(nReplaced, 2) 

370 

371 # test with three consecutive bad values 

372 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]) 

373 nReplaced = filterBadValues(values) 

374 self.assertEqual(nReplaced, 3) 

375 

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

377 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]) 

378 nReplaced = filterBadValues(values) 

379 self.assertEqual(nReplaced, 4) 

380 

381 # test with more than three consecutive bad values 

382 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]) 

383 nReplaced = filterBadValues(values) 

384 self.assertEqual(nReplaced, 3) 

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

386 

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

388 # value at the end 

389 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]) 

390 nReplaced = filterBadValues(values) 

391 self.assertEqual(nReplaced, 4) 

392 

393 # test with bad values in first two positions 

394 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 

395 nReplaced = filterBadValues(values) 

396 self.assertEqual(nReplaced, 2) 

397 

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

399 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]) 

400 nReplaced = filterBadValues(values) 

401 self.assertEqual(nReplaced, 3) 

402 

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

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

405 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]) 

406 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]) 

407 nReplaced = filterBadValues(values) 

408 residuals = np.abs(values - expected) 

409 self.assertEqual(nReplaced, 4) 

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

411 

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

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

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

415 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]) 

416 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]) 

417 nReplaced = filterBadValues(values) 

418 residuals = np.abs(values - expected) 

419 self.assertEqual(nReplaced, 4) 

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

421 

422 # check with non-default maxDelta 

423 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]) 

424 nReplaced = filterBadValues(values, maxDelta=10) 

425 self.assertEqual(nReplaced, 0) 

426 

427 values = np.array( 

428 [ 

429 1.0, 

430 1.0, 

431 1.0, 

432 1.1, 

433 1.0, 

434 1.0, 

435 1.0, 

436 1.0, 

437 1.0, 

438 1.0, 

439 ] 

440 ) 

441 nReplaced = filterBadValues(values, maxDelta=0.01) 

442 self.assertEqual(nReplaced, 1) 

443 

444 @vcr.use_cassette() 

445 def test_getEvent(self): 

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

447 # exist for the day 

448 eventMaker = self.tmaEventMaker 

449 events = eventMaker.getEvents(self.dayObs) 

450 nEvents = len(events) 

451 

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

453 self.assertIsInstance(event, TMAEvent) 

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

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

456 self.assertIsInstance(event, TMAEvent) 

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

458 

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

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

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

462 msg = cm.output[0] 

463 self.assertIn(correctMsg, msg) 

464 

465 @vcr.use_cassette() 

466 def test_printing(self): 

467 eventMaker = self.tmaEventMaker 

468 events = eventMaker.getEvents(self.dayObs) 

469 

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

471 print(str(events[0])) 

472 print(repr(events[0])) 

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

474 

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

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

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

478 eventMaker.printEventDetails(slews[0]) 

479 eventMaker.printEventDetails(tracks[0]) 

480 eventMaker.printEventDetails(events[-1]) 

481 

482 # check the full day trick works 

483 eventMaker.printFullDayStateEvolution(self.dayObs) 

484 

485 tma = TMAStateMachine() 

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

487 eventMaker.printTmaDetailedState(tma) 

488 

489 @vcr.use_cassette() 

490 def test_getAxisData(self): 

491 eventMaker = self.tmaEventMaker 

492 events = eventMaker.getEvents(self.dayObs) 

493 

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

495 self.assertIsInstance(azData, pd.DataFrame) 

496 self.assertIsInstance(elData, pd.DataFrame) 

497 

498 paddedAzData, paddedElData = getAzimuthElevationDataForEvent( 

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

500 ) 

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

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

503 

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

505 # data in 

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

507 

508 @vcr.use_cassette() 

509 def test_plottingAndCommands(self): 

510 eventMaker = self.tmaEventMaker 

511 events = eventMaker.getEvents(self.dayObs) 

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

513 

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

515 plotEvent(self.client, event) 

516 

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

518 # just check this doesn't raise when called 

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

520 plt.close(fig) 

521 

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

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

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

525 

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

527 

528 del fig 

529 

530 @vcr.use_cassette() 

531 def test_findEvent(self): 

532 eventMaker = self.tmaEventMaker 

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

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

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

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

537 

538 time = event.begin 

539 found = eventMaker.findEvent(time) 

540 self.assertEqual(found, event) 

541 

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

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

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

545 # logging info messages about it) 

546 time = event.end - dt 

547 found = eventMaker.findEvent(time) 

548 self.assertEqual(found, event) 

549 

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

551 time = event.end + dt 

552 found = eventMaker.findEvent(time) 

553 self.assertNotEqual(found, event) 

554 

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

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

557 # worth it 

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

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

560 found = eventMaker.findEvent(tooEarlyOnDay) 

561 self.assertIsNone(found) 

562 

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

564 # event 

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

566 found = eventMaker.findEvent(tooLateOnDay) 

567 self.assertIsNone(found) 

568 

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

570 lastEvent = events[-1] 

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

572 self.assertEqual(found, lastEvent) 

573 

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

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

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

577 # behave (being half-open intervals) 

578 found = eventMaker.findEvent(lastEvent.end) 

579 self.assertIsNone(found, lastEvent) 

580 

581 @vcr.use_cassette() 

582 def test_eventAssociatedWith(self): 

583 eventMaker = self.tmaEventMaker 

584 events = eventMaker.getEvents(self.dayObsWithBlockInfo) 

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

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

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

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

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

590 

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

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

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

594 

595 event = eventsWithBlockInfo[0] 

596 self.assertIsInstance(event, TMAEvent) 

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

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

599 

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

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

602 

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

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

605 

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

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

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

609 

610 # check it's false if any are false 

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

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

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

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

615 

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

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

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

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

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

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

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

623 

624 with self.assertRaises(ValueError): 

625 event.associatedWith() 

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

627 

628 

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

630 pass 

631 

632 

633def setup_module(module): 

634 lsst.utils.tests.init() 

635 

636 

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

638 lsst.utils.tests.init() 

639 unittest.main()