Coverage for event_normalizer/transformers.py: 100%

1"""Pipeline-specific extraction logic.""" 

2 

3import logging 

4from typing import Any 

5 

6from .models import EventLabels 

7 

8logger = logging.getLogger(__name__) 

9 

10 

11class Transformers: 

12 def __init__( 

13 self, 

14 group: str | None, 

15 pipeline: str | None, 

16 search: str | None, 

17 event: dict[str, Any], 

18 ): 

19 self.group = group 

20 self.pipeline = pipeline 

21 self.search = search 

22 self.event = event 

23 

24 @property 

25 def event_id(self) -> str: 

26 return str(self.event.get("graceid", "<unknown>")) 

27 

28 @property 

29 def extra_attributes(self) -> dict[str, Any]: 

30 extra = self.event.get("extra_attributes") 

31 

32 if extra is None: 

33 return {} 

34 

35 if not isinstance(extra, dict): 

36 logger.warning( 

37 "Invalid extra_attributes type for event %s: expected dict, got %s", 

38 self.event_id, 

39 type(extra).__name__, 

40 ) 

41 return {} 

42 

43 return extra 

44 

45 def _get_table(self, table_name: str) -> dict[str, Any] | None: 

46 table = self.extra_attributes.get(table_name) 

47 

48 if table is None: 

49 logger.debug( 

50 "Table %s is absent for event %s", 

51 table_name, 

52 self.event_id, 

53 ) 

54 return None 

55 

56 if not isinstance(table, dict): 

57 logger.warning( 

58 "Invalid %s table for event %s: expected dict, got %s", 

59 table_name, 

60 self.event_id, 

61 type(table).__name__, 

62 ) 

63 return None 

64 

65 return table 

66 

67 def transform_channels(self) -> dict[str, str]: 

68 """ 

69 Extract detector channel names into H1_channel, L1_channel, V1_channel. 

70 

71 Sources: 

72 - standard CBC: extra_attributes.SingleInspiral[*].channel 

73 - CWB Burst: extra_attributes.MultiBurst.hoft 

74 - MLy and aframe: extra_attributes.MLyBurst.channels 

75 """ 

76 channels: dict[str, str] = {} 

77 

78 # Standard CBC, excluding aframe. 

79 if self.group == "CBC" and self.pipeline != "aframe": 

80 tables = self.extra_attributes.get("SingleInspiral") 

81 

82 if tables is None: 

83 logger.debug( 

84 "SingleInspiral is absent for event %s", 

85 self.event_id, 

86 ) 

87 return channels 

88 

89 if not isinstance(tables, list): 

90 logger.warning( 

91 "Invalid SingleInspiral for event %s: expected list, got %s", 

92 self.event_id, 

93 type(tables).__name__, 

94 ) 

95 return channels 

96 

97 for table in tables: 

98 if not isinstance(table, dict): 

99 logger.warning( 

100 "Ignoring invalid SingleInspiral entry for event %s", 

101 self.event_id, 

102 ) 

103 continue 

104 

105 ifo = table.get("ifo") 

106 channel = table.get("channel") 

107 

108 if not ifo or not channel: 

109 logger.debug( 

110 "Incomplete SingleInspiral channel entry for event %s: %r", 

111 self.event_id, 

112 table, 

113 ) 

114 continue 

115 

116 ifo = str(ifo).strip() 

117 channel = str(channel).strip() 

118 

119 if not ifo or not channel: 

120 logger.debug( 

121 "Empty SingleInspiral channel entry for event %s: %r", 

122 self.event_id, 

123 table, 

124 ) 

125 continue 

126 

127 channels[f"{ifo}_channel"] = channel 

128 return channels 

129 

130 # CWB / standard Burst. 

131 if self.group == "Burst" and self.pipeline != "MLy": 

132 table = self._get_table("MultiBurst") 

133 if table is None: 

134 return channels 

135 

136 hoft = table.get("hoft") 

137 if not hoft: 

138 logger.debug( 

139 "MultiBurst.hoft is absent for event %s", 

140 self.event_id, 

141 ) 

142 return channels 

143 

144 raw_channels = hoft.split(",") if isinstance(hoft, str) else hoft 

145 

146 # MLy Burst and aframe CBC. 

147 elif self.pipeline in {"MLy", "aframe"}: 

148 table = self._get_table("MLyBurst") 

149 if table is None: 

150 return channels 

151 

152 raw_channels = table.get("channels", []) 

153 

154 else: 

155 return channels 

156 

157 if not isinstance(raw_channels, list): 

158 logger.warning( 

159 "Invalid channels value for event %s: expected list, got %s", 

160 self.event_id, 

161 type(raw_channels).__name__, 

162 ) 

163 return channels 

164 

165 for entry in raw_channels: 

166 if not isinstance(entry, str): 

167 logger.warning( 

168 "Ignoring non-string channel entry for event %s: %r", 

169 self.event_id, 

170 entry, 

171 ) 

172 continue 

173 

174 entry = entry.strip() 

175 

176 if not entry: 

177 logger.warning( 

178 "Ignoring empty channel entry for event %s", 

179 self.event_id, 

180 ) 

181 continue 

182 

183 # CWB hoft may contain calibration markers such as C00. 

184 if entry == "C00": 

185 continue 

186 

187 try: 

188 ifo, channel = (value.strip() for value in entry.split(":", 1)) 

189 except ValueError: 

190 logger.warning( 

191 "Invalid channel entry for event %s: %r", 

192 self.event_id, 

193 entry, 

194 ) 

195 continue 

196 

197 if not ifo or not channel: 

198 logger.warning( 

199 "Incomplete channel entry for event %s: %r", 

200 self.event_id, 

201 entry, 

202 ) 

203 continue 

204 

205 channels[f"{ifo}_channel"] = channel 

206 

207 return channels 

208 

209 def transform_coinc_table(self, keys: list[str]) -> dict[str, Any] | None: 

210 """Extract selected CoincInspiral values for CBC events.""" 

211 # aframe is a CBC event but does not use CoincInspiral. 

212 if self.group == "Burst" or self.pipeline == "aframe": 

213 return None 

214 

215 table = self._get_table("CoincInspiral") 

216 if table is None: 

217 return None 

218 

219 result: dict[str, Any] = {} 

220 

221 for key in keys: 

222 value = table.get(key) 

223 

224 if value is not None: 

225 result["coinc_" + key] = value 

226 else: 

227 logger.debug( 

228 "CoincInspiral.%s is absent for event %s", 

229 key, 

230 self.event_id, 

231 ) 

232 

233 return result or None 

234 

235 def transform_burst_table(self, keys: list[str]) -> dict[str, Any] | None: 

236 """ 

237 Extract selected Burst-like values from MultiBurst. 

238 """ 

239 # Standard CBC does not have a Burst table. 

240 # aframe is the explicit exception: it is CBC, but has MLyBurst attributes. 

241 if self.group == "CBC" and self.pipeline != "aframe": 

242 return None 

243 

244 # MLy uses MLyBurst, not MultiBurst 

245 if self.pipeline in {"MLy", "aframe"}: 

246 return None 

247 

248 table = self._get_table("MultiBurst") 

249 if table is None: 

250 return None 

251 

252 result: dict[str, Any] = {} 

253 

254 for key in keys: 

255 value = table.get(key) 

256 

257 if value is not None: 

258 result["burst_" + key] = value 

259 else: 

260 logger.debug( 

261 "MultiBurst.%s is absent for event %s", 

262 key, 

263 self.event_id, 

264 ) 

265 

266 return result or None 

267 

268 def transform_mly_table(self, keys: list[str]) -> dict[str, Any] | None: 

269 """ 

270 Extract selected MLy-specific values from MLyBurst. 

271 """ 

272 # MLy and aframe use MLyBurst 

273 if self.pipeline not in {"MLy", "aframe"}: 

274 return None 

275 

276 table = self._get_table("MLyBurst") 

277 if table is None: 

278 return None 

279 

280 result: dict[str, Any] = {} 

281 

282 for key in keys: 

283 # Handle nested scores dict 

284 if key.startswith("scores."): 

285 scores_key = key.split(".", 1)[1] 

286 scores = table.get("scores", {}) 

287 if isinstance(scores, dict): 

288 value = scores.get(scores_key) 

289 if value is not None: 

290 result[f"mly_scores_{scores_key}"] = value 

291 else: 

292 logger.debug( 

293 "MLyBurst.scores.%s is absent for event %s", 

294 scores_key, 

295 self.event_id, 

296 ) 

297 else: 

298 value = table.get(key) 

299 

300 if value is not None: 

301 result["mly_" + key] = value 

302 else: 

303 logger.debug( 

304 "MLyBurst.%s is absent for event %s", 

305 key, 

306 self.event_id, 

307 ) 

308 

309 return result or None 

310 

311 def transform_single_inspiral( 

312 self, 

313 keys: list[str], 

314 ) -> dict[str, dict[str, Any]]: 

315 """ 

316 Extract SingleInspiral technical fields grouped by interferometer. 

317 

318 The IFO is represented by the result dictionary key and later by the 

319 parent NormalizedEvent field: 

320 

321 - "H1" -> single_H1 

322 - "L1" -> single_L1 

323 - "V1" -> single_V1 

324 - "K1" -> single_K1 

325 

326 Channel values are intentionally excluded because they are normalized 

327 separately as H1_channel, L1_channel, V1_channel, and K1_channel. 

328 

329 Example result: 

330 { 

331 "H1": { 

332 "single_snr": 5.0, 

333 "single_mass1": 30.0, 

334 }, 

335 "L1": { 

336 "single_snr": 7.0, 

337 "single_mass1": 30.0, 

338 }, 

339 } 

340 """ 

341 # Standard CBC pipelines use SingleInspiral. 

342 # aframe is a CBC pipeline, but its technical attributes are in MLyBurst. 

343 if self.group == "Burst" or self.pipeline == "aframe": 

344 return {} 

345 

346 tables = self.extra_attributes.get("SingleInspiral") 

347 

348 if tables is None: 

349 logger.debug( 

350 "SingleInspiral is absent for event %s", 

351 self.event_id, 

352 ) 

353 return {} 

354 

355 if not isinstance(tables, list): 

356 logger.warning( 

357 "Invalid SingleInspiral for event %s: expected list, got %s", 

358 self.event_id, 

359 type(tables).__name__, 

360 ) 

361 return {} 

362 

363 result: dict[str, dict[str, Any]] = {} 

364 

365 for table in tables: 

366 if not isinstance(table, dict): 

367 logger.warning( 

368 "Ignoring invalid SingleInspiral entry for event %s", 

369 self.event_id, 

370 ) 

371 continue 

372 

373 ifo = table.get("ifo") 

374 

375 if not isinstance(ifo, str) or not ifo.strip(): 

376 logger.debug( 

377 "SingleInspiral entry without valid IFO for event %s: %r", 

378 self.event_id, 

379 table, 

380 ) 

381 continue 

382 

383 ifo = ifo.strip() 

384 

385 ifo_data: dict[str, Any] = {} 

386 

387 for key in keys: 

388 # The detector is represented by single_H1, single_L1, etc. 

389 # Channel values are handled separately by transform_channels(). 

390 if key in {"ifo", "channel"}: 

391 continue 

392 

393 value = table.get(key) 

394 

395 if value is not None: 

396 ifo_data[f"single_{key}"] = value 

397 else: 

398 logger.debug( 

399 "SingleInspiral[%s].%s is absent for event %s", 

400 ifo, 

401 key, 

402 self.event_id, 

403 ) 

404 

405 # Do not create empty single_H1/single_L1/etc. models. 

406 if not ifo_data: 

407 logger.debug( 

408 "SingleInspiral contains no selected technical fields " 

409 "for event %s, IFO %s", 

410 self.event_id, 

411 ifo, 

412 ) 

413 continue 

414 

415 if ifo in result: 

416 logger.warning( 

417 "Duplicate SingleInspiral entry for event %s, IFO %s; " 

418 "replacing previous values", 

419 self.event_id, 

420 ifo, 

421 ) 

422 

423 result[ifo] = ifo_data 

424 

425 return result 

426 

427 def transform_labels(self) -> dict[str, bool] | None: 

428 raw_labels = self.event.get("labels") 

429 

430 if not raw_labels: 

431 return None 

432 

433 if not isinstance(raw_labels, list): 

434 logger.warning( 

435 "Invalid labels for event %s: expected list, got %s", 

436 self.event_id, 

437 type(raw_labels).__name__, 

438 ) 

439 return None 

440 

441 result: dict[str, bool] = {} 

442 allowed = set(EventLabels.model_fields.keys()) 

443 

444 for label in raw_labels: 

445 if not isinstance(label, str): 

446 logger.warning( 

447 "Ignoring non-string label for event %s: %r", 

448 self.event_id, 

449 label, 

450 ) 

451 continue 

452 

453 field_name = f"label_{label}" 

454 

455 if field_name not in allowed: 

456 logger.warning( 

457 "Unknown label skipped for event %s: %s", 

458 self.event_id, 

459 label, 

460 ) 

461 continue 

462 

463 result[field_name] = True 

464 

465 return result or None 

466 

467 @staticmethod 

468 def transform_instruments(value: Any) -> list[str] | None: 

469 """ 

470 Normalize instrument values. 

471 

472 Examples: 

473 - "H1,L1,V1" -> ["H1", "L1", "V1"] 

474 - ["H1", "L1"] -> ["H1", "L1"] 

475 - ["H1,L1,V1"] -> ["H1", "L1", "V1"] 

476 - None -> None 

477 """ 

478 if value is None: 

479 return None 

480 

481 if isinstance(value, str): 

482 raw_values = [value] 

483 elif isinstance(value, list): 

484 raw_values = value 

485 else: 

486 logger.warning( 

487 "Invalid instruments value: expected str or list, got %s", 

488 type(value).__name__, 

489 ) 

490 return None 

491 

492 instruments: list[str] = [] 

493 

494 for item in raw_values: 

495 if not isinstance(item, str): 

496 logger.warning( 

497 "Ignoring non-string instrument value: %r", 

498 item, 

499 ) 

500 continue 

501 

502 instruments.extend( 

503 instrument.strip() 

504 for instrument in item.split(",") 

505 if instrument.strip() 

506 ) 

507 

508 return instruments or None 

509 

510 def transform_links(self) -> dict[str, str] | None: 

511 """Extract GraceDB URL links.""" 

512 links = self.event.get("links") 

513 

514 if not links: 

515 return None 

516 

517 if not isinstance(links, dict): 

518 logger.warning( 

519 "Invalid links for event %s: expected dict, got %s", 

520 self.event_id, 

521 type(links).__name__, 

522 ) 

523 return None 

524 

525 return links