Coverage for gwcelery/tasks/orchestrator.py: 94%

1"""Tasks that comprise the alert orchestrator. 

2 

3The orchestrator is responsible for the vetting and annotation workflow to 

4produce preliminary, initial, and update alerts for gravitational-wave event 

5candidates. 

6""" 

7import json 

8import re 

9 

10from astropy.time import Time 

11from celery import chain, group 

12from ligo.rrt_chat import channel_creation 

13from rapidpe_rift_pipe import pastro as rpe_pastro 

14 

15from .. import app 

16from . import (alerts, bayestar, circulars, detchar, em_bright, 

17 external_skymaps, gcn, gracedb, igwn_alert, inference, p_astro, 

18 rrt_utils, skymaps, superevents) 

19from .core import get_first, identity 

20 

21 

22@igwn_alert.handler('superevent', 

23 'mdc_superevent', 

24 shared=False) 

25def handle_superevent(alert): 

26 """Schedule annotations for new superevents. 

27 

28 After waiting for a time specified by the 

29 :obj:`~gwcelery.conf.orchestrator_timeout` configuration variable for the 

30 choice of preferred event to settle down, this task performs data quality 

31 checks with :meth:`gwcelery.tasks.detchar.check_vectors` and calls 

32 :meth:`~gwcelery.tasks.orchestrator.earlywarning_preliminary_alert` to send 

33 a preliminary notice. 

34 """ 

35 superevent_id = alert['uid'] 

36 # launch PE and detchar based on new type superevents 

37 if alert['alert_type'] == 'new': 

38 # launching rapidpe 30s after merger. 

39 timeout = max( 

40 alert['object']['t_0'] - Time.now().gps + 

41 app.conf['rapidpe_timeout'], 0 

42 ) 

43 ( 

44 gracedb.get_superevent.si(superevent_id).set( 

45 countdown=timeout 

46 ) 

47 | 

48 _get_pe_far_and_event.s() 

49 | 

50 parameter_estimation.s(superevent_id, 'rapidpe') 

51 ).apply_async() 

52 

53 ( 

54 gracedb.get_superevent.si(superevent_id).set( 

55 countdown=app.conf['pe_timeout'] 

56 ) 

57 | 

58 _get_pe_far_and_event.s() 

59 | 

60 parameter_estimation.s(superevent_id, 'bilby') 

61 ).apply_async() 

62 

63 # run check_vectors. Create and upload omegascans 

64 group( 

65 detchar.omegascan.si(alert['object']['t_0'], superevent_id), 

66 

67 detchar.check_vectors.si( 

68 alert['object']['preferred_event_data'], 

69 superevent_id, 

70 alert['object']['t_start'], 

71 alert['object']['t_end'] 

72 ) 

73 ).delay() 

74 

75 elif alert['alert_type'] == 'label_added': 

76 label_name = alert['data']['name'] 

77 query = f'superevent: {superevent_id} group: CBC Burst' 

78 if alert['object']['category'] == 'MDC': 

79 query += ' MDC' 

80 elif alert['object']['category'] == 'Test': 

81 query += ' Test' 

82 

83 # launch less-significant preliminary alerts on LOW_SIGNIF_LOCKED 

84 if label_name == superevents.FROZEN_LABEL: 

85 # don't launch if EARLY_WARNING or SIGNIF_LOCKED is present 

86 skipping_labels = { 

87 superevents.SIGNIFICANT_LABEL, 

88 superevents.EARLY_WARNING_LABEL 

89 }.intersection(alert['object']['labels']) 

90 if skipping_labels: 

91 gracedb.upload.delay( 

92 None, None, superevent_id, 

93 "The superevent already has a significant/EW event, " 

94 "skipping launching less-significant alert" 

95 ) 

96 return 

97 ( 

98 gracedb.upload.s( 

99 None, 

100 None, 

101 superevent_id, 

102 "Automated DQ check before sending less-significant " 

103 "preliminary alert. New results supersede old results.", 

104 tags=['data_quality'] 

105 ) 

106 | 

107 detchar.check_vectors.si( 

108 alert['object']['preferred_event_data'], 

109 superevent_id, 

110 alert['object']['t_start'], 

111 alert['object']['t_end'] 

112 ) 

113 | 

114 earlywarning_preliminary_alert.s( 

115 alert, alert_type='less-significant') 

116 ).apply_async() 

117 

118 # launch significant alert on SIGNIF_LOCKED 

119 elif label_name == superevents.SIGNIFICANT_LABEL: 

120 # ensure superevent is locked before starting alert workflow 

121 r = chain() 

122 if superevents.FROZEN_LABEL not in alert['object']['labels']: 

123 r |= gracedb.create_label.si(superevents.FROZEN_LABEL, 

124 superevent_id) 

125 

126 r |= ( 

127 gracedb.get_events.si(query) 

128 | 

129 superevents.select_preferred_event.s() 

130 | 

131 _update_superevent_and_return_event_dict.s(superevent_id) 

132 | 

133 _leave_log_message_and_return_event_dict.s( 

134 superevent_id, 

135 "Superevent cleaned up after significant event. " 

136 ) 

137 | 

138 _leave_log_message_and_return_event_dict.s( 

139 superevent_id, 

140 "Automated DQ check before sending significant alert. " 

141 "New results supersede old results.", 

142 tags=['data_quality'] 

143 ) 

144 | 

145 detchar.check_vectors.s( 

146 superevent_id, 

147 alert['object']['t_start'], 

148 alert['object']['t_end'] 

149 ) 

150 | 

151 earlywarning_preliminary_alert.s( 

152 alert, alert_type='preliminary') 

153 ) 

154 r.apply_async() 

155 

156 # launch second preliminary on GCN_PRELIM_SENT 

157 elif label_name == 'GCN_PRELIM_SENT': 

158 ( 

159 identity.si().set( 

160 # https://git.ligo.org/emfollow/gwcelery/-/issues/478 

161 # FIXME: remove this task once https://github.com/celery/celery/issues/7851 is resolved # noqa: E501 

162 countdown=app.conf['superevent_clean_up_timeout'] 

163 ) 

164 | 

165 gracedb.get_events.si(query) 

166 | 

167 superevents.select_preferred_event.s() 

168 | 

169 _update_superevent_and_return_event_dict.s(superevent_id) 

170 | 

171 group( 

172 _leave_log_message_and_return_event_dict.s( 

173 superevent_id, 

174 "Superevent cleaned up after first preliminary alert" 

175 ), 

176 

177 gracedb.create_label.si('DQR_REQUEST', superevent_id) 

178 ) 

179 | 

180 get_first.s() 

181 | 

182 earlywarning_preliminary_alert.s( 

183 alert, alert_type='preliminary') 

184 ).apply_async() 

185 

186 # set pipeline preferred events 

187 # FIXME: Ideally this should combined with the previous canvas. 

188 # However, incorporating that group prevents canvas from executing 

189 # maybe related to https://github.com/celery/celery/issues/7851 

190 ( 

191 gracedb.get_events.si(query) 

192 | 

193 superevents.select_pipeline_preferred_event.s() 

194 | 

195 _set_pipeline_preferred_events.s(superevent_id) 

196 ).apply_async(countdown=app.conf['superevent_clean_up_timeout']) 

197 

198 elif label_name == 'LOW_SIGNIF_PRELIM_SENT': 

199 # similar workflow as the GCN_PRELIM_SENT 

200 # except block by condition evaluated at the end of timeout 

201 _revise_and_send_second_less_significant_alert.si( 

202 alert, query, superevent_id, 

203 ).apply_async(countdown=app.conf['superevent_clean_up_timeout']) 

204 

205 elif label_name == superevents.EARLY_WARNING_LABEL: 

206 if superevents.SIGNIFICANT_LABEL in alert['object']['labels']: 

207 # stop if full BW significant event already present 

208 gracedb.upload.delay( 

209 None, None, superevent_id, 

210 "Superevent superseded by full BW event, skipping EW." 

211 ) 

212 return 

213 # start the EW alert pipeline; is blocked by SIGNIF_LOCKED 

214 # ensure superevent is locked before starting pipeline 

215 r = chain() 

216 if superevents.FROZEN_LABEL not in alert['object']['labels']: 

217 r |= gracedb.create_label.si(superevents.FROZEN_LABEL, 

218 superevent_id) 

219 

220 r |= ( 

221 gracedb.get_events.si(query) 

222 | 

223 superevents.select_preferred_event.s() 

224 | 

225 _update_superevent_and_return_event_dict.s(superevent_id) 

226 | 

227 _leave_log_message_and_return_event_dict.s( 

228 superevent_id, 

229 "Superevent cleaned up before sending EW alert." 

230 ) 

231 | 

232 earlywarning_preliminary_alert.s( 

233 alert, alert_type='earlywarning') 

234 ) 

235 r.apply_async() 

236 

237 # launch initial/retraction alert on ADVOK/ADVNO 

238 elif label_name == 'ADVOK': 

239 initial_alert((None, None, None), alert) 

240 elif label_name == 'ADVNO': 

241 retraction_alert(alert) 

242 elif label_name == 'ADVREQ': 

243 if app.conf['create_mattermost_channel']: 

244 _create_mattermost_channel.si(superevent_id).delay() 

245 

246 # check DQV label on superevent, run check_vectors if required 

247 elif alert['alert_type'] == 'event_added': 

248 # FIXME Check if this should be changed to 'update' alert_types instead 

249 # of 'event_added'. 'event_added' seems like it's just a new event in 

250 # the superevent window, not necessarily an event that should be 

251 # promoted to preferred event 

252 start = alert['data']['t_start'] 

253 end = alert['data']['t_end'] 

254 

255 if 'DQV' in gracedb.get_labels(superevent_id): 

256 ( 

257 detchar.check_vectors.s( 

258 alert['object']['preferred_event_data'], 

259 superevent_id, 

260 start, 

261 end 

262 ) 

263 | 

264 _update_if_dqok.s(superevent_id) 

265 ).apply_async() 

266 

267 

268@igwn_alert.handler('cbc_gstlal', 

269 'cbc_spiir', 

270 'cbc_pycbc', 

271 'cbc_mbta', 

272 shared=False) 

273def handle_cbc_event(alert): 

274 """Perform annotations for CBC events that depend on pipeline-specific 

275 matched-filter parameter estimates. 

276 

277 Notes 

278 ----- 

279 This IGWN alert message handler is triggered by a new upload or by updates 

280 that include the file ``pipeline.p_astro.json``. If also generates 

281 pipeline.p_astro.json information for pipelines that do not provide 

282 such information. 

283 

284 The table below lists which files are created as a result of a new upload, 

285 and which tasks generate them. 

286 

287 ============================== ================================================== 

288 File Task 

289 ============================== ================================================== 

290 ``bayestar.multiorder.fits`` :meth:`gwcelery.tasks.bayestar.localize` 

291 ``em_bright.json`` :meth:`gwcelery.tasks.em_bright.source_properties` 

292 ``pipeline.p_astro.json`` :meth:`gwcelery.tasks.p_astro.compute_p_astro` 

293 ============================== ================================================== 

294 

295 """ # noqa: E501 

296 graceid = alert['uid'] 

297 pipeline = alert['object']['pipeline'].lower() 

298 search = alert['object']['search'].lower() 

299 

300 # no annotations for events used in VT analysis 

301 if search == superevents.VT_SEARCH_NAME.lower(): 

302 return 

303 

304 priority = 0 if superevents.should_publish(alert['object']) else 1 

305 

306 # Pipelines that use the GWCelery p-astro method 

307 # - spiir (all searches) 

308 # - pycbc for EarlyWarning search 

309 # - periodic MDC generated by first-two-years (based on gstlal) 

310 # FIXME: Remove this once all pipelines compute their own p-astro 

311 pipelines_stock_p_astro = {('spiir', 'earlywarning'), 

312 ('pycbc', 'earlywarning'), 

313 ('gstlal', 'mdc')} 

314 

315 # em_bright and p_astro calculation 

316 if alert['alert_type'] == 'new': 

317 instruments = superevents.get_instruments_in_ranking_statistic( 

318 alert['object']) 

319 extra_attributes = alert['object']['extra_attributes'] 

320 snr = superevents.get_snr(alert['object']) 

321 far = alert['object']['far'] 

322 mass1 = extra_attributes['SingleInspiral'][0]['mass1'] 

323 mass2 = extra_attributes['SingleInspiral'][0]['mass2'] 

324 chi1 = extra_attributes['SingleInspiral'][0]['spin1z'] 

325 chi2 = extra_attributes['SingleInspiral'][0]['spin2z'] 

326 

327 ( 

328 em_bright.source_properties.si(mass1, mass2, chi1, chi2, snr, 

329 pipeline=pipeline, search=search) 

330 | 

331 gracedb.upload.s( 

332 'em_bright.json', graceid, 

333 'em bright complete', ['em_bright', 'public'] 

334 ) 

335 | 

336 gracedb.create_label.si('EMBRIGHT_READY', graceid) 

337 ).apply_async(priority=priority) 

338 

339 # p_astro calculation for pipelines that does not provide a 

340 # stock p_astro (upload pipeline.p_astro.json) 

341 if (pipeline, search) in pipelines_stock_p_astro: 

342 ( 

343 p_astro.compute_p_astro.s(snr, 

344 far, 

345 mass1, 

346 mass2, 

347 pipeline, 

348 instruments) 

349 | 

350 gracedb.upload.s( 

351 f'{pipeline}.p_astro.json', graceid, 

352 'p_astro computation complete', ['p_astro', 'public'] 

353 ) 

354 | 

355 gracedb.create_label.si('PASTRO_READY', graceid) 

356 ).apply_async(priority=priority) 

357 

358 # Start BAYESTAR for all CBC pipelines. 

359 ( 

360 gracedb.download.s('coinc.xml', graceid) 

361 | 

362 bayestar.localize.s(graceid) 

363 | 

364 gracedb.upload.s( 

365 'bayestar.multiorder.fits', graceid, 

366 'sky localization complete', ['sky_loc', 'public'] 

367 ) 

368 | 

369 gracedb.create_label.si('SKYMAP_READY', graceid) 

370 ).apply_async(priority=priority) 

371 

372 

373@igwn_alert.handler('burst_olib', 

374 'burst_cwb', 

375 'burst_mly', 

376 shared=False) 

377def handle_burst_event(alert): 

378 """Perform annotations for burst events that depend on pipeline-specific 

379 """ # noqa: E501 

380 graceid = alert['uid'] 

381 pipeline = alert['object']['pipeline'].lower() 

382 search = alert['object']['search'].lower() 

383 priority = 0 if superevents.should_publish(alert['object']) else 1 

384 

385 # em_bright calculation for Burst-cWB-BBH 

386 if alert['alert_type'] == 'new': 

387 if (pipeline, search) in [('cwb', 'bbh')]: 

388 extra_attributes = alert['object']['extra_attributes'] 

389 multiburst = extra_attributes['MultiBurst'] 

390 snr = multiburst.get('snr') 

391 mchirp = multiburst.get('mchirp', 0.0) 

392 # FIXME once ingestion of mchip is finalised 

393 # In case mchirp is not there or zero 

394 # produce em_brigth with all zero 

395 if mchirp == 0.0: 

396 m12 = 30.0 

397 else: 

398 m12 = 2**(0.2) * mchirp 

399 ( 

400 em_bright.source_properties.si(m12, m12, 0.0, 0.0, snr) 

401 | 

402 gracedb.upload.s( 

403 'em_bright.json', graceid, 

404 'em bright complete', ['em_bright', 'public'] 

405 ) 

406 | 

407 gracedb.create_label.si('EMBRIGHT_READY', graceid) 

408 ).apply_async(priority=priority) 

409 

410 if alert['alert_type'] != 'log': 

411 return 

412 

413 filename = alert['data']['filename'] 

414 

415 # Pipeline is uploading a flat resultion skymap file 

416 # Converting to a multiirder ones with proper name. 

417 # FIXME: Remove block when CWB starts to upload skymaps 

418 # in multiorder format 

419 if filename.endswith('.fits.gz'): 

420 new_filename = filename.replace('.fits.gz', '.multiorder.fits') 

421 flatten_msg = ( 

422 'Multi-resolution FITS file created from ' 

423 '<a href="/api/events/{graceid}/files/' 

424 '{filename}">{filename}</a>').format( 

425 graceid=graceid, filename=filename) 

426 tags = ['sky_loc', 'lvem', 'public'] 

427 ( 

428 gracedb.download.si(filename, graceid) 

429 | 

430 skymaps.unflatten.s(new_filename) 

431 | 

432 gracedb.upload.s( 

433 new_filename, graceid, flatten_msg, tags) 

434 | 

435 gracedb.create_label.si('SKYMAP_READY', graceid) 

436 ).apply_async(priority=priority) 

437 

438 

439@igwn_alert.handler('superevent', 

440 'mdc_superevent', 

441 shared=False) 

442def handle_posterior_samples(alert): 

443 """Generate multi-resolution and flat-resolution FITS files and skymaps 

444 from an uploaded HDF5 file containing posterior samples. 

445 """ 

446 if alert['alert_type'] != 'log' or \ 

447 not alert['data']['filename'].endswith('.posterior_samples.hdf5'): 

448 return 

449 superevent_id = alert['uid'] 

450 filename = alert['data']['filename'] 

451 info = '{} {}'.format(alert['data']['comment'], filename) 

452 prefix, _ = filename.rsplit('.posterior_samples.') 

453 skymap_filename = f'{prefix}.multiorder.fits' 

454 labels = ['pe', 'sky_loc'] 

455 ifos = superevents.get_instruments(alert['object']['preferred_event_data']) 

456 

457 ( 

458 gracedb.download.si(filename, superevent_id) 

459 | 

460 skymaps.skymap_from_samples.s(superevent_id, ifos) 

461 | 

462 group( 

463 skymaps.annotate_fits.s( 

464 skymap_filename, superevent_id, labels 

465 ), 

466 

467 gracedb.upload.s( 

468 skymap_filename, superevent_id, 

469 'Multiresolution FITS file generated from "{}"'.format(info), 

470 labels 

471 ) 

472 ) 

473 ).delay() 

474 

475 # em_bright from LALInference posterior samples 

476 ( 

477 gracedb.download.si(filename, superevent_id) 

478 | 

479 em_bright.em_bright_posterior_samples.s() 

480 | 

481 gracedb.upload.s( 

482 '{}.em_bright.json'.format(prefix), superevent_id, 

483 'em-bright computed from "{}"'.format(info), 

484 'pe' 

485 ) 

486 ).delay() 

487 

488 

489@app.task(bind=True, shared=False) 

490def _create_mattermost_channel(self, superevent_id): 

491 """ 

492 Creates a mattermost channel when ADVREQ label is applied and 

493 posts a cooresponding gracedb link of that event in the channel 

494 

495 Channel name : O4 RRT {superevent_id} 

496 

497 Parameters: 

498 ------------ 

499 superevent_id: str 

500 The superevent id 

501 """ 

502 gracedb_url = self.app.conf['gracedb_host'] 

503 channel_creation.rrt_channel_creation( 

504 superevent_id, gracedb_url) 

505 

506 

507@app.task(shared=False) 

508def _set_pipeline_preferred_events(pipeline_event, superevent_id): 

509 """Return group for setting pipeline preferred event using 

510 :meth:`gracedb.add_pipeline_preferred_event`. 

511 

512 Parameters 

513 ---------- 

514 pipeline_event: dict 

515 {pipeline: event_dict} key value pairs, returned by 

516 :meth:`superevents.select_pipeline_preferred_event`. 

517 

518 superevent_id: str 

519 The superevent id 

520 """ 

521 return group( 

522 gracedb.add_pipeline_preferred_event(superevent_id, 

523 event['graceid']) 

524 for event in pipeline_event.values() 

525 ) 

526 

527 

528@app.task(shared=False, ignore_result=True) 

529def _update_if_dqok(event, superevent_id): 

530 """Update `preferred_event` of `superevent_id` to `event_id` if `DQOK` 

531 label has been applied. 

532 """ 

533 if 'DQOK' in gracedb.get_labels(superevent_id): 

534 event_id = event['graceid'] 

535 gracedb.update_superevent(superevent_id, 

536 preferred_event=event_id, 

537 t_0=event["gpstime"]) 

538 gracedb.upload.delay( 

539 None, None, superevent_id, 

540 comment=f'DQOK applied based on new event {event_id}') 

541 

542 

543@gracedb.task(shared=False) 

544def _create_voevent(classification, *args, **kwargs): 

545 r"""Create a VOEvent record from an EM bright JSON file. 

546 

547 Parameters 

548 ---------- 

549 classification : tuple, None 

550 A collection of JSON strings, generated by 

551 :meth:`gwcelery.tasks.em_bright.source_properties` and 

552 :meth:`gwcelery.tasks.p_astro.compute_p_astro` or 

553 content of ``{gstlal,mbta}.p_astro.json`` uploaded 

554 by {gstlal,mbta} respectively; or None 

555 \*args 

556 Additional positional arguments passed to 

557 :meth:`gwcelery.tasks.gracedb.create_voevent`. 

558 \*\*kwargs 

559 Additional keyword arguments passed to 

560 :meth:`gwcelery.tasks.gracedb.create_voevent`. 

561 

562 Returns 

563 ------- 

564 str 

565 The filename of the newly created VOEvent. 

566 

567 """ 

568 kwargs = dict(kwargs) 

569 

570 if classification is not None: 

571 # Merge source classification and source properties into kwargs. 

572 for text in classification: 

573 # Ignore filenames, only load dict in bytes form 

574 if text is not None: 

575 kwargs.update(json.loads(text)) 

576 

577 # FIXME: These keys have differ between em_bright.json 

578 # and the GraceDB REST API. 

579 try: 

580 kwargs['ProbHasNS'] = kwargs.pop('HasNS') 

581 except KeyError: 

582 pass 

583 

584 try: 

585 kwargs['ProbHasRemnant'] = kwargs.pop('HasRemnant') 

586 except KeyError: 

587 pass 

588 

589 skymap_filename = kwargs.get('skymap_filename') 

590 if skymap_filename is not None: 

591 skymap_type = re.sub( 

592 r'(\.multiorder)?\.fits(\..+)?(,[0-9]+)?$', '', skymap_filename) 

593 kwargs.setdefault('skymap_type', skymap_type) 

594 

595 # FIXME: remove ._orig_run when this bug is fixed: 

596 # https://github.com/getsentry/sentry-python/issues/370 

597 return gracedb.create_voevent._orig_run(*args, **kwargs) 

598 

599 

600@app.task(shared=False) 

601def _create_label_and_return_filename(filename, label, graceid): 

602 gracedb.create_label.delay(label, graceid) 

603 return filename 

604 

605 

606@app.task(shared=False) 

607def _leave_log_message_and_return_event_dict(event, superevent_id, 

608 message, **kwargs): 

609 """Wrapper around :meth:`gracedb.upload` 

610 that returns the event dictionary. 

611 """ 

612 gracedb.upload.delay(None, None, superevent_id, message, **kwargs) 

613 return event 

614 

615 

616@gracedb.task(shared=False) 

617def _update_superevent_and_return_event_dict(event, superevent_id): 

618 """Wrapper around :meth:`gracedb.update_superevent` 

619 that returns the event dictionary. 

620 """ 

621 gracedb.update_superevent(superevent_id, 

622 preferred_event=event['graceid'], 

623 t_0=event['gpstime']) 

624 return event 

625 

626 

627@gracedb.task(shared=False) 

628def _proceed_if_not_blocked_by(files, superevent_id, block_by): 

629 """Return files in case the superevent does not have labels `block_by` 

630 

631 Parameters 

632 ---------- 

633 files : tuple 

634 List of files 

635 superevent_id : str 

636 The superevent id corresponding to files 

637 block_by : set 

638 Set of blocking labels. E.g. `{'ADVOK', 'ADVNO'}` 

639 """ 

640 superevent_labels = gracedb.get_labels(superevent_id) 

641 blocking_labels = block_by.intersection(superevent_labels) 

642 if blocking_labels: 

643 gracedb.upload.delay( 

644 None, None, superevent_id, 

645 f"Blocking automated notice due to labels {blocking_labels}" 

646 ) 

647 return None 

648 else: 

649 return files 

650 

651 

652@gracedb.task(shared=False) 

653def _revise_and_send_second_less_significant_alert(alert, query, 

654 superevent_id): 

655 superevent_labels = gracedb.get_labels(superevent_id) 

656 blocking_labels = { 

657 'ADVREQ', 'ADVOK', 'ADVNO', 

658 superevents.SIGNIFICANT_LABEL, 

659 superevents.EARLY_WARNING_LABEL, 

660 } 

661 if blocking_labels.intersection(superevent_labels): 

662 return 

663 

664 ( 

665 gracedb.get_events.si(query) 

666 | 

667 superevents.select_preferred_event.s() 

668 | 

669 _update_superevent_and_return_event_dict.s(superevent_id) 

670 | 

671 _leave_log_message_and_return_event_dict.s( 

672 superevent_id, 

673 "Superevent cleaned up before second less-significant alert" 

674 ) 

675 | 

676 earlywarning_preliminary_alert.s( 

677 alert, alert_type='less-significant') 

678 ).delay() 

679 

680 # set pipeline preferred events 

681 # FIXME: Ideally this should combined with the previous canvas. 

682 # However, incorporating that group prevents canvas from executing 

683 # maybe related to https://github.com/celery/celery/issues/7851 

684 ( 

685 gracedb.get_events.si(query) 

686 | 

687 superevents.select_pipeline_preferred_event.s() 

688 | 

689 _set_pipeline_preferred_events.s(superevent_id) 

690 ).delay() 

691 

692 

693@app.task(shared=False) 

694def _annotate_fits_and_return_input(input_list, superevent_id): 

695 """Unpack the output of the skymap, embright, p-astro download group in the 

696 beginning of the 

697 :meth:`~gwcelery.tasks.orchestartor.earlywarning_preliminary_alert` canvas 

698 and call :meth:`~gwcelery.tasks.skymaps.annotate_fits`. 

699 

700 

701 Parameters 

702 ---------- 

703 input_list : list 

704 The output of the group that downloads the skymap, embright, and 

705 p-astro files. This list is in the form [skymap, skymap_filename], 

706 [em_bright, em_bright_filename], [p_astro_dict, p_astro_filename], 

707 though the em-bright and p-astro lists can be populated by Nones 

708 superevent_id : str 

709 A list of the sky map, em_bright, and p_astro filenames. 

710 """ 

711 

712 skymaps.annotate_fits_tuple( 

713 input_list[0], 

714 superevent_id, 

715 ['sky_loc', 'public'] 

716 ) 

717 

718 return input_list 

719 

720 

721@gracedb.task(shared=False) 

722def _unpack_args_and_send_earlywarning_preliminary_alert(input_list, alert, 

723 alert_type): 

724 """Unpack the output of the skymap, embright, p-astro download group in the 

725 beginning of the 

726 :meth:`~gwcelery.tasks.orchestartor.earlywarning_preliminary_alert` canvas 

727 and call 

728 :meth:`gwcelery.tasks.orchestrator.earlywarning_preliminary_initial_update_alert`. 

729 

730 

731 Parameters 

732 ---------- 

733 input_list : list 

734 The output of the group that downloads the skymap, embright, and 

735 p-astro files. This list is in the form [skymap, skymap_filename], 

736 [em_bright, em_bright_filename], [p_astro_dict, p_astro_filename], 

737 though the em-bright and p-astro lists can be populated by Nones 

738 alert : dict 

739 IGWN-Alert dictionary 

740 alert_type : str 

741 alert_type passed to 

742 :meth:`earlywarning_preliminary_initial_update_alert` 

743 """ 

744 if input_list is None: # alert is blocked by blocking labels 

745 return 

746 

747 [skymap, skymap_filename], [em_bright, em_bright_filename], \ 

748 [p_astro_dict, p_astro_filename] = input_list 

749 

750 # Update to latest state after downloading files 

751 superevent = gracedb.get_superevent(alert['object']['superevent_id']) 

752 

753 earlywarning_preliminary_initial_update_alert.delay( 

754 [skymap_filename, em_bright_filename, p_astro_filename], 

755 superevent, alert_type, 

756 filecontents=[skymap, em_bright, p_astro_dict] 

757 ) 

758 

759 

760@app.task(ignore_result=True, shared=False) 

761def earlywarning_preliminary_alert(event, alert, alert_type='preliminary', 

762 initiate_voevent=True): 

763 """Produce a preliminary alert by copying any sky maps. 

764 

765 This consists of the following steps: 

766 

767 1. Copy any sky maps and source classification from the preferred event 

768 to the superevent. 

769 2. Create standard annotations for sky maps including all-sky plots by 

770 calling :meth:`gwcelery.tasks.skymaps.annotate_fits`. 

771 3. Create a preliminary VOEvent. 

772 4. Send the VOEvent to GCN and notices to SCiMMA and GCN. 

773 5. Apply the GCN_PRELIM_SENT or LOW_SIGNIF_PRELIM_SENT 

774 depending on the significant or less-significant alert 

775 respectively. 

776 6. Create and upload a GCN Circular draft. 

777 """ 

778 priority = 0 if superevents.should_publish(event) else 1 

779 preferred_event_id = event['graceid'] 

780 superevent_id = alert['uid'] 

781 

782 # Define alert payloads depending on group-pipeline-search 

783 # cbc-*-* : p_astro/em_bright 

784 # burst.cwb-bbh : p_astro/em_bright 

785 # burst-*-* : NO p_astro/em_bright 

786 alert_group = event['group'].lower() 

787 alert_pipeline = event['pipeline'].lower() 

788 alert_search = event['search'].lower() 

789 if alert_pipeline == 'cwb' and alert_search == 'bbh': 

790 skymap_filename = alert_pipeline + '.multiorder.fits' 

791 p_astro_filename = alert_pipeline + '.p_astro.json' 

792 em_bright_filename = 'em_bright.json' 

793 elif alert_group == 'cbc': 

794 skymap_filename = 'bayestar.multiorder.fits' 

795 p_astro_filename = alert_pipeline + '.p_astro.json' 

796 em_bright_filename = 'em_bright.json' 

797 elif alert_group == 'burst': 

798 skymap_filename = event['pipeline'].lower() + '.multiorder.fits' 

799 p_astro_filename = None 

800 em_bright_filename = None 

801 else: 

802 raise NotImplementedError( 

803 'Valid skymap required for preliminary alert' 

804 ) 

805 

806 # Determine if the event should be made public. 

807 is_publishable = (superevents.should_publish( 

808 event, significant=alert_type != 'less-significant') and 

809 {'DQV', 'INJ'}.isdisjoint( 

810 event['labels'])) 

811 

812 # Download files from events and upload to superevent with relevant 

813 # annotations. Pass file contents down the chain so alerts task doesn't 

814 # need to download them again. 

815 # Note: this is explicitly made a chain to fix an issue described in #464. 

816 canvas = chain( 

817 group( 

818 gracedb.download.si(skymap_filename, preferred_event_id) 

819 | 

820 group( 

821 identity.s(), 

822 

823 gracedb.upload.s( 

824 skymap_filename, 

825 superevent_id, 

826 message='Localization copied from {}'.format( 

827 preferred_event_id), 

828 tags=['sky_loc', 'public'] 

829 ) 

830 | 

831 _create_label_and_return_filename.s('SKYMAP_READY', 

832 superevent_id) 

833 ), 

834 

835 ( 

836 gracedb.download.si(em_bright_filename, preferred_event_id) 

837 | 

838 group( 

839 identity.s(), 

840 

841 gracedb.upload.s( 

842 em_bright_filename, 

843 superevent_id, 

844 message='Source properties copied from {}'.format( 

845 preferred_event_id), 

846 tags=['em_bright', 'public'] 

847 ) 

848 | 

849 _create_label_and_return_filename.s('EMBRIGHT_READY', 

850 superevent_id) 

851 ) 

852 ) if em_bright_filename is not None else 

853 identity.s([None, None]), 

854 

855 ( 

856 gracedb.download.si(p_astro_filename, preferred_event_id) 

857 | 

858 group( 

859 identity.s(), 

860 

861 gracedb.upload.s( 

862 p_astro_filename, 

863 superevent_id, 

864 message='Source classification copied from {}'.format( 

865 preferred_event_id), 

866 tags=['p_astro', 'public'] 

867 ) 

868 | 

869 _create_label_and_return_filename.s('PASTRO_READY', 

870 superevent_id) 

871 ) 

872 ) if p_astro_filename is not None else 

873 identity.s([None, None]) 

874 ) 

875 | 

876 # Need annotate skymap task in body of chord instead of header because 

877 # this task simply calls another task, which is to be avoided in chord 

878 # headers. Note that any group that chains to a task is automatically 

879 # upgraded to a chord. 

880 _annotate_fits_and_return_input.s(superevent_id) 

881 ) 

882 

883 # Switch for disabling all but MDC alerts. 

884 if app.conf['only_alert_for_mdc']: 

885 if event.get('search') != 'MDC': 

886 canvas |= gracedb.upload.si( 

887 None, None, superevent_id, 

888 ("Skipping alert because gwcelery has been configured to only" 

889 " send alerts for MDC events.")) 

890 canvas.apply_async(priority=priority) 

891 return 

892 

893 # Send notice and upload GCN circular draft for online events. 

894 if is_publishable and initiate_voevent: 

895 # presence of advocate action blocks significant prelim alert 

896 # presence of adv action or significant event blocks EW alert 

897 # presence of adv action or significant event or EW event blocks 

898 # less significant alert 

899 if alert_type == 'earlywarning': 

900 # Launch DQR for significant early warning events after a timeout. 

901 # If a full BW significant trigger arrives before the end of the 

902 # timeout, the latter will apply the label instead, and this call 

903 # is a noop. 

904 gracedb.create_label.si( 

905 'DQR_REQUEST', 

906 superevent_id 

907 ).apply_async(countdown=600) 

908 blocking_labels = ( 

909 {'ADVOK', 'ADVNO'} if alert_type == 'preliminary' 

910 else 

911 {superevents.SIGNIFICANT_LABEL, 'ADVOK', 'ADVNO'} 

912 if alert_type == 'earlywarning' 

913 else 

914 {superevents.EARLY_WARNING_LABEL, superevents.SIGNIFICANT_LABEL, 

915 'ADVOK', 'ADVNO'} 

916 if alert_type == 'less-significant' 

917 else 

918 set() 

919 ) 

920 canvas |= ( 

921 _proceed_if_not_blocked_by.s(superevent_id, blocking_labels) 

922 | 

923 _unpack_args_and_send_earlywarning_preliminary_alert.s( 

924 alert, alert_type 

925 ) 

926 ) 

927 

928 canvas.apply_async(priority=priority)