Coverage for python/lsst/summit/utils/blockUtils.py: 20%

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

21from __future__ import annotations 

22 

23import logging 

24import re 

25import time 

26from dataclasses import dataclass 

27from typing import TYPE_CHECKING 

28 

29import numpy as np 

30import pandas as pd 

31from astropy.time import Time 

32 

33from .efdUtils import efdTimestampToAstropy, getEfdData, makeEfdClient 

34from .enums import ScriptState 

35 

36if TYPE_CHECKING: 

37 from .tmaUtils import TMAEvent 

38 

39 try: 

40 from lsst_efd_client import EfdClient 

41 except ImportError: 

42 EfdClient = None # this is currently just for mypy 

43 

44__all__ = ("BlockParser", "BlockInfo", "ScriptStatePoint") 

45 

46 

47@dataclass(kw_only=True, frozen=True) 

48class BlockInfo: 

49 """Information about the execution of a "block". 

50 

51 Each BlockInfo instance contains information about a single block 

52 execution. This is identified by the block number and sequence number, 

53 which, when combined with the dayObs, define the block ID. 

54 

55 Each BlockInfo instance contains the following information: 

56 * The block ID - this is the primary identifier, as a string, for 

57 example "BL52_20230615_02", which is parsed into: 

58 * The block number, as an integer, for example 52, for "BLOCK-52". 

59 * The dayObs, as an integer, for example 20230615. 

60 * The seqNum - the execution number of that block on that day. 

61 * The begin and end times of the block execution, as astropy.time.Time 

62 * The SAL indices which were involved in the block execution, as a list 

63 * The SITCOM tickets which were involved in the block execution, as a 

64 list of strings, including the SITCOM- prefix. 

65 * The states of the script during the block execution, as a list of 

66 ``ScriptStatePoint`` instances. 

67 

68 Parameters 

69 ---------- 

70 blockNumber : `str` 

71 The block number, as a str - sometimes it'll be like 123 but others it 

72 will be like "T123" for test blocks. 

73 blockId : `str` 

74 The block ID, as a string. 

75 dayObs : `int` 

76 The dayObs the block was run on. 

77 seqNum : `int` 

78 The sequence number of the block. 

79 begin : `astropy.time.Time` 

80 The time the block execution began. 

81 end : `astropy.time.Time` 

82 The time the block execution ended. 

83 isTestCase : `bool` 

84 Whether this block is a test case type block, or a regular block. This 

85 is also reflected in the blockNumber. 

86 salIndices : `list` of `int` 

87 One or more SAL indices, relating to the block. 

88 tickets : `list` of `str` 

89 One or more SITCOM tickets, relating to the block. 

90 states : `list` of `lsst.summit.utils.blockUtils.ScriptStatePoint` 

91 The states of the script during the block. Each element is a 

92 ``ScriptStatePoint`` which contains: 

93 - the time, as an astropy.time.Time 

94 - the state, as a ``ScriptState`` enum 

95 - the reason for state change, as a string, if present 

96 """ 

97 

98 blockNumber: str 

99 blockId: str 

100 dayObs: int 

101 seqNum: int 

102 begin: Time 

103 end: Time 

104 salIndices: list 

105 tickets: list 

106 states: list 

107 isTestCase: bool 

108 

109 def __repr__(self) -> str: 

110 return ( 

111 f"BlockInfo(blockNumber={self.blockNumber}, blockId={self.blockId}, salIndices={self.salIndices}," 

112 f" tickets={self.tickets}, states={self.states!r}" 

113 ) 

114 

115 def _ipython_display_(self) -> None: 

116 """This is the function which runs when someone executes a cell in a 

117 notebook with just the class instance on its own, without calling 

118 print() or str() on it. 

119 """ 

120 print(self.__str__()) 

121 

122 def __str__(self) -> str: 

123 # no literal \n allowed inside {} portion of f-strings until python 

124 # 3.12, but it can go in via a variable 

125 newline = " \n" 

126 return ( 

127 f"dayObs: {self.dayObs}\n" 

128 f"seqNum: {self.seqNum}\n" 

129 f"blockNumber: {self.blockNumber}\n" 

130 f"blockId: {self.blockId}\n" 

131 f"begin: {self.begin.isot}\n" 

132 f"end: {self.end.isot}\n" 

133 f"salIndices: {self.salIndices}\n" 

134 f"tickets: {self.tickets}\n" 

135 f"states: \n{newline.join([str(state) for state in self.states])}" 

136 ) 

137 

138 

139@dataclass(kw_only=True, frozen=True) 

140class ScriptStatePoint: 

141 """The execution state of a script at a point in time. 

142 

143 Parameters 

144 ---------- 

145 time : `astropy.time.Time` 

146 The time of the state change. 

147 state : `lsst.summit.utils.enums.ScriptState` 

148 The state of the script at this point in time. 

149 reason : `str` 

150 The reason for the state change, if given. 

151 """ 

152 

153 time: Time 

154 state: ScriptState 

155 reason: str 

156 

157 def __repr__(self) -> str: 

158 return f"ScriptStatePoint(time={self.time!r}, state={self.state!r}, reason={self.reason!r})" 

159 

160 def _ipython_display_(self) -> None: 

161 """This is the function which runs when someone executes a cell in a 

162 notebook with just the class instance on its own, without calling 

163 print() or str() on it. 

164 """ 

165 print(self.__str__()) 

166 

167 def __str__(self) -> str: 

168 reasonStr = f" - {self.reason}" if self.reason else "" 

169 return f"{self.state.name:>10} @ {self.time.isot}{reasonStr}" 

170 

171 

172class BlockParser: 

173 """A class to parse BLOCK data from the EFD. 

174 

175 Information on executed blocks is stored in the EFD (Electronic Facilities 

176 Database) in the ``lsst.sal.Script.logevent_state`` topic. This class 

177 parses that topic and provides methods to get information on the blocks 

178 which were run on a given dayObs. It also provides methods to get the 

179 events which occurred during a given block, and also to get the block in 

180 which a specified event occurred, if any. 

181 

182 Parameters 

183 ---------- 

184 dayObs : `int` 

185 The dayObs to get the block data for. 

186 client : `lsst_efd_client.efd_client.EfdClient`, optional 

187 The EFD client to use. If not specified, a new one is created. 

188 """ 

189 

190 def __init__(self, dayObs: int, client: EfdClient | None = None) -> None: 

191 self.log = logging.getLogger("lsst.summit.utils.blockUtils.BlockParser") 

192 self.dayObs = dayObs 

193 

194 self.client = client 

195 if client is None: 

196 self.client = makeEfdClient() 

197 

198 t0 = time.time() 

199 self.getDataForDayObs() 

200 self.log.debug(f"Getting data took {(time.time() - t0):.2f} seconds") 

201 t0 = time.time() 

202 self.augmentData() 

203 self.log.debug(f"Parsing data took {(time.time() - t0):.5f} seconds") 

204 

205 def getDataForDayObs(self) -> None: 

206 """Retrieve the data for the specified dayObs from the EFD.""" 

207 # Tiago thinks no individual block seqNums should take more than an 

208 # hour to run, so pad the dayObs by 1.5 hours to make sure we catch 

209 # any blocks which might span the end of the day. 

210 padding = 1.5 * 60 * 60 

211 data = getEfdData( 

212 self.client, 

213 "lsst.sal.Script.logevent_state", 

214 dayObs=self.dayObs, 

215 postPadding=padding, 

216 raiseIfTopicNotInSchema=False, 

217 ) 

218 self.data = data 

219 

220 def augmentData(self) -> None: 

221 """Parse each row in the data frame individually, pulling the 

222 information out into its own columns. 

223 """ 

224 data = self.data 

225 blockPattern = r"BLOCK-(\d+)" 

226 blockIdPattern = r"B[LT]\d+(?:_\w+)+" 

227 

228 data["blockNum"] = pd.Series() 

229 data["blockDayObs"] = pd.Series() 

230 data["blockSeqNum"] = pd.Series() 

231 data["isTestCase"] = pd.Series() 

232 

233 if "lastCheckpoint" not in self.data.columns: 

234 nRows = len(self.data) 

235 self.log.warning( 

236 f"Found {nRows} rows of data and no 'lastCheckpoint' column was in the data," 

237 " so block data cannot be parsed." 

238 ) 

239 

240 # at some point the blockId column was added to the data, so if it's 

241 # present, we can use it to extract the blockNum and other information, 

242 # but before that we have to parse it out of the lastCheckpoint column 

243 if "blockId" in data.columns: 

244 blockNumberPattern = re.compile(r"[A-Z]*([0-9]+)(?=_)") 

245 

246 for index, row in data.iterrows(): 

247 # an example blockIdStr value is like BL365_O_20250420_000001 

248 blockIdStr = row["blockId"] 

249 match = blockNumberPattern.match(blockIdStr) 

250 if not match: # we've failed to get the 365 part in the example 

251 continue 

252 blockNum = match.group(1) 

253 

254 idStrSplit = blockIdStr.split("_") 

255 blockDayObs = int(idStrSplit[2]) 

256 if blockDayObs != self.dayObs: 

257 continue # we're in the padded region 

258 

259 isTestCase = False 

260 if idStrSplit[0].startswith("BT"): 

261 isTestCase = True 

262 

263 blockSeqNum = int(idStrSplit[3]) 

264 data.at[index, "blockNum"] = f"{'T' if isTestCase else ''}{blockNum}" 

265 data.at[index, "blockDayObs"] = int(blockDayObs) 

266 data.at[index, "blockSeqNum"] = int(blockSeqNum) 

267 data.at[index, "isTestCase"] = isTestCase 

268 

269 else: 

270 data["blockId"] = pd.Series() # add it, as it's not present 

271 for index, row in data.iterrows(): 

272 rowStr = row["lastCheckpoint"] 

273 

274 blockMatch = re.search(blockPattern, rowStr) 

275 blockNumber = int(blockMatch.group(1)) if blockMatch else None 

276 

277 blockIdMatch = re.search(blockIdPattern, rowStr) 

278 blockIdStr = blockIdMatch.group(0) if blockIdMatch else None 

279 if blockIdStr is None: 

280 continue 

281 data.at[index, "blockId"] = blockIdStr 

282 

283 idStrSplit = blockIdStr.split("_") 

284 blockDayObs = int(idStrSplit[2]) 

285 if blockDayObs != self.dayObs: 

286 continue # we're in the padded region 

287 

288 isTestCase = False 

289 if idStrSplit[0].startswith("BT"): 

290 isTestCase = True 

291 

292 data.at[index, "blockNum"] = f"{'T' if isTestCase else ''}{blockNumber}" 

293 data.at[index, "isTestCase"] = isTestCase 

294 blockDayObs = int(idStrSplit[2]) 

295 blockSeqNum = int(idStrSplit[3]) 

296 data.at[index, "blockDayObs"] = blockDayObs 

297 data.at[index, "blockSeqNum"] = blockSeqNum 

298 

299 def _listColumnValues(self, column: str, removeNone: bool = True) -> list[str]: 

300 """Get all the different values for the specified column, as a list. 

301 

302 Parameters 

303 ---------- 

304 column : `str` 

305 The column to get the values for. 

306 removeNone : `bool` 

307 Whether to remove None from the list of values. 

308 

309 Returns 

310 ------- 

311 values : `list` 

312 The values for the specified column. 

313 """ 

314 values = set(self.data[column].dropna()) 

315 if None in values and removeNone: 

316 values.remove(None) 

317 return sorted(values) 

318 

319 def getBlockNums(self) -> list[str]: 

320 """Get the block numbers which were run on the specified dayObs. 

321 

322 Returns 

323 ------- 

324 blockNums : `list` of `int` 

325 The blocks which were run on the specified dayObs. 

326 """ 

327 return self._listColumnValues("blockNum") 

328 

329 def getSeqNums(self, block: int | str) -> list[int]: 

330 """Get the seqNums for the specified block. 

331 

332 Parameters 

333 ---------- 

334 block : `int` 

335 The block name or number to get the events for, e.g. 123 or T123. 

336 

337 Returns 

338 ------- 

339 seqNums : `list` of `int` 

340 The sequence numbers for the specified block. 

341 """ 

342 if isinstance(block, int): 

343 block = str(block) 

344 

345 seqNums = self.data[self.data["blockNum"] == block]["blockSeqNum"] 

346 # block header rows have no blockId or seqNum, but do have a blockNum 

347 # so appear here, so drop the nans as they don't relate to an actual 

348 # run of a block 

349 seqNums = seqNums.dropna() 

350 return sorted(set(seqNums)) 

351 

352 def getRows(self, block: str | int, seqNum: int | None = None) -> pd.DataFrame: 

353 """Get all rows of data which relate to the specified block. 

354 

355 If the seqNum is specified, only the rows for that sequence number are 

356 returned, otherwise all the rows relating to any block execution that 

357 day are returned. If the specified seqNum doesn't occur on the current 

358 day, an empty dataframe is returned. 

359 

360 Parameters 

361 ---------- 

362 block : `int` 

363 The block number to get the events for. 

364 seqNum : `int`, optional 

365 The sequence number, if specified, to get the row data for. If not 

366 specified, all data for the specified block is returned. 

367 

368 Returns 

369 ------- 

370 data : `pandas.DataFrame` 

371 The row data. 

372 """ 

373 if isinstance(block, int): 

374 block = str(block) 

375 

376 # Because we query for a whole dayObs, but BLOCKs can overlap the day 

377 # start/end, it's possible for the block's blockDayObs not to be the 

378 # same as self.dayObs around the beginning or end of the day, so filter 

379 # with an extra `& (self.data['blockDayObs'] == self.dayObs` when 

380 # getting the relevant rows. 

381 rowsForBlock = self.data[ 

382 np.logical_and(self.data["blockNum"] == block, self.data["blockDayObs"] == self.dayObs) 

383 ] 

384 if rowsForBlock.empty: 

385 self.log.warning(f"No rows found for {block=} on dayObs={self.dayObs}") 

386 if seqNum is None: 

387 return rowsForBlock 

388 return rowsForBlock[rowsForBlock["blockSeqNum"] == seqNum] 

389 

390 def printBlockEvolution(self, block: str | int, seqNum: int | None = None) -> None: 

391 """Display the evolution of the specified block. 

392 

393 If the seqNum is specified, the evolution of that specific block 

394 exection is displayed, otherwise all executions of that block are 

395 printed. 

396 

397 Parameters 

398 ---------- 

399 block : `int` 

400 The block number to get the events for. 

401 seqNum : `int`, optional 

402 The sequence number, if specified, to print the evolution of. If 

403 not specified, all sequence numbers for the block are printed. 

404 """ 

405 if isinstance(block, int): 

406 block = str(block) 

407 

408 if seqNum is None: 

409 seqNums = self.getSeqNums(block) 

410 else: 

411 seqNums = [seqNum] 

412 print(f"Evolution of BLOCK {block} for dayObs={self.dayObs} {seqNum=}:") 

413 for seqNum in seqNums: 

414 blockInfo = self.getBlockInfo(block, seqNum) 

415 print(blockInfo, "\n") 

416 

417 def getBlockInfo(self, block: int | str, seqNum: int) -> BlockInfo | None: 

418 """Get the block info for the specified block. 

419 

420 Parses the rows relating to this block execution, and returns 

421 the information as a ``BlockInfo`` instance. 

422 

423 Parameters 

424 ---------- 

425 block : `int` 

426 The block number. 

427 seqNum : `int` 

428 The sequence number. 

429 

430 Returns 

431 ------- 

432 blockInfo : `lsst.summit.utils.blockUtils.BlockInfo` 

433 The block info. 

434 """ 

435 if isinstance(block, int): 

436 block = str(block) 

437 

438 rows = self.getRows(block, seqNum=seqNum) 

439 if rows.empty: 

440 print(f"No {seqNum=} on dayObs={self.dayObs} for {block=}") 

441 return None 

442 

443 blockIds = set() 

444 testCases = set() 

445 tickets = set() 

446 salIndices = set() 

447 statePoints = [] 

448 sitcomPattern = r"SITCOM-(\d+)" 

449 

450 for index, row in rows.iterrows(): 

451 salIndices.add(row["salIndex"]) 

452 blockIds.add(row["blockId"]) 

453 testCases.add(row["isTestCase"]) 

454 

455 lastCheckpoint = row["lastCheckpoint"] 

456 sitcomMatches = re.findall(sitcomPattern, lastCheckpoint) 

457 tickets.update(sitcomMatches) 

458 

459 time = efdTimestampToAstropy(row["private_efdStamp"]) 

460 state = ScriptState(row["state"]) 

461 reason = row["reason"] 

462 statePoint = ScriptStatePoint(time=time, state=state, reason=reason) 

463 statePoints.append(statePoint) 

464 

465 # check that blockIds, blockNames, and testCases are all length == 1 

466 if any(len(s) != 1 for s in [blockIds, testCases]): 

467 raise RuntimeError( 

468 f"Expected exactly one unique value for blockIds and testCases, " 

469 f"but found: blockIds={blockIds}, testCases={testCases} " 

470 f"for {seqNum=}" 

471 ) 

472 

473 blockId = blockIds.pop() 

474 isTestCase = testCases.pop() 

475 

476 blockInfo = BlockInfo( 

477 blockNumber=block, 

478 blockId=blockId, 

479 dayObs=self.dayObs, 

480 seqNum=seqNum, 

481 begin=efdTimestampToAstropy(rows.iloc[0]["private_efdStamp"]), 

482 end=efdTimestampToAstropy(rows.iloc[-1]["private_efdStamp"]), 

483 salIndices=sorted(salIndices), 

484 tickets=[f"SITCOM-{ticket}" for ticket in sorted(tickets)], 

485 states=statePoints, 

486 isTestCase=isTestCase, 

487 ) 

488 

489 return blockInfo 

490 

491 def getEventsForBlock(self, events: list[TMAEvent], block: str, seqNum: int) -> list[TMAEvent]: 

492 """Get the events which occurred during the specified block. 

493 

494 Parameters 

495 ---------- 

496 events : `list` of `lsst.summit.utils.tmaUtils.TMAEvent` 

497 The list of candidate events. 

498 block : `int` 

499 The block number to get the events for. 

500 seqNum : `int` 

501 The sequence number to get the events for. 

502 

503 Returns 

504 ------- 

505 events : `list` of `lsst.summit.utils.tmaUtils.TMAEvent` 

506 The events. 

507 """ 

508 blockInfo = self.getBlockInfo(block, seqNum) 

509 if blockInfo is None: 

510 return [] 

511 begin = blockInfo.begin 

512 end = blockInfo.end 

513 

514 # each event's end being past the begin time and their 

515 # starts being before the end time means we get all the 

516 # events in the window and also those that overlap the 

517 # start/end too 

518 return [e for e in events if e.end >= begin and e.begin <= end]