Coverage for gwcelery/tasks/raven.py: 99%
194 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-25 18:01 +0000
1"""Search for GRB-GW coincidences with ligo-raven."""
2import ligo.raven.search
3from celery import group
4from celery.utils.log import get_task_logger
6from .. import app
7from . import external_skymaps, gracedb
8from .core import identity
10log = get_task_logger(__name__)
13@app.task(shared=False)
14def calculate_coincidence_far(superevent, exttrig, tl, th,
15 use_superevent_skymap=None):
16 """Compute coincidence FAR for external trigger and superevent coincidence
17 by calling ligo.raven.search.calc_signif_gracedb, using sky map info if
18 available.
20 Parameters
21 ----------
22 superevent : dict
23 Superevent dictionary
24 exttrig : dict
25 External event dictionary
26 tl : int
27 Lower bound of search window in seconds
28 th : int
29 Upper bound of search window in seconds
30 use_superevent_skymap : bool
31 If True/False, use/don't use skymap info from superevent.
32 Else if None, check SKYMAP_READY label in external event.
34 Returns
35 -------
36 joint_far : dict
37 Dictionary containing joint false alarm rate, including sky map info
38 if available
40 """
41 superevent_id = superevent['superevent_id']
42 exttrig_id = exttrig['graceid']
43 far_grb = exttrig['far']
45 # Don't compute coinc FAR for SNEWS coincidences
46 if exttrig['pipeline'] == 'SNEWS':
47 return {}
49 # Define max far thresholds for targeted subthreshold search
50 if exttrig['search'] == 'SubGRBTargeted':
51 far_thresholds = app.conf['raven_targeted_far_thresholds']
52 far_gw_thresh = far_thresholds['GW'][exttrig['pipeline']]
53 far_grb_thresh = far_thresholds['GRB'][exttrig['pipeline']]
54 else:
55 far_gw_thresh = None
56 far_grb_thresh = None
58 # Get rate for expected number of astrophysical external triggers if needed
59 if exttrig['search'] in {'GRB', 'SubGRB', 'MDC'}:
60 ext_rate = app.conf['raven_ext_rates'][exttrig['search']]
61 else:
62 ext_rate = None
64 if ({'EXT_SKYMAP_READY', 'SKYMAP_READY'}.issubset(exttrig['labels']) or
65 {'EXT_SKYMAP_READY', 'EM_READY'}.issubset(exttrig['labels'])):
66 # if both sky maps available, calculate spatial coinc far
67 use_preferred_event_skymap = (
68 not use_superevent_skymap
69 if use_superevent_skymap is not None else
70 'SKYMAP_READY' not in exttrig['labels'])
71 se_skymap = external_skymaps.get_skymap_filename(
72 (superevent['preferred_event'] if use_preferred_event_skymap
73 else superevent_id), is_gw=True)
74 ext_skymap = external_skymaps.get_skymap_filename(
75 exttrig_id, is_gw=False)
76 ext_moc = '.multiorder.fits' in ext_skymap
78 return ligo.raven.search.calc_signif_gracedb(
79 superevent_id, exttrig_id, tl, th,
80 se_dict=superevent, ext_dict=exttrig,
81 grb_search=exttrig['search'],
82 se_fitsfile=se_skymap, ext_fitsfile=ext_skymap,
83 se_moc=True, ext_moc=ext_moc,
84 incl_sky=True, gracedb=gracedb.client,
85 em_rate=ext_rate,
86 far_grb=far_grb,
87 far_gw_thresh=far_gw_thresh,
88 far_grb_thresh=far_grb_thresh,
89 use_preferred_event_skymap=use_preferred_event_skymap)
90 else:
91 return ligo.raven.search.calc_signif_gracedb(
92 superevent_id, exttrig_id, tl, th,
93 se_dict=superevent, ext_dict=exttrig,
94 grb_search=exttrig['search'],
95 incl_sky=False, gracedb=gracedb.client,
96 em_rate=ext_rate,
97 far_grb=far_grb,
98 far_gw_thresh=far_gw_thresh,
99 far_grb_thresh=far_grb_thresh)
102@app.task(shared=False)
103def coincidence_search(gracedb_id, alert_object, group=None, pipelines=[],
104 searches=[], se_searches=[]):
105 """Perform ligo-raven search for coincidences. Determines time window to
106 use. If events found, launches RAVEN pipeline.
108 Parameters
109 ----------
110 gracedb_id : str
111 GraceDB ID of the trigger that launched RAVEN
112 alert_object : dict
113 Alert dictionary
114 group : str
115 Burst or CBC
116 pipelines : list
117 List of external trigger pipeline names
118 searches : list
119 List of external trigger searches
120 se_searches : list
121 List of superevent searches
123 """
124 tl, th = _time_window(gracedb_id, group, pipelines, searches)
126 (
127 search.si(gracedb_id, alert_object, tl, th, group, pipelines,
128 searches, se_searches)
129 |
130 raven_pipeline.s(gracedb_id, alert_object, tl, th, group)
131 ).delay()
134def _time_window(gracedb_id, group, pipelines, searches):
135 """Determine the time window to use given the parameters of the search.
137 Parameters
138 ----------
139 gracedb_id : str
140 GraceDB ID of the trigger that launched RAVEN
141 group : str
142 Burst or CBC
143 pipelines : list
144 List of external trigger pipeline names
145 searches : list
146 List of external trigger searches
147 se_searches : list
148 List of superevent searches
150 Returns
151 -------
152 tl, th : tuple
153 Tuple of lower bound and upper bound of search window
155 """
156 tl_cbc, th_cbc = app.conf['raven_coincidence_windows']['GRB_CBC']
157 tl_subfermi, th_subfermi = \
158 app.conf['raven_coincidence_windows']['GRB_CBC_SubFermi']
159 tl_subswift, th_subswift = \
160 app.conf['raven_coincidence_windows']['GRB_CBC_SubSwift']
161 tl_burst, th_burst = app.conf['raven_coincidence_windows']['GRB_Burst']
162 tl_snews, th_snews = app.conf['raven_coincidence_windows']['SNEWS']
164 if 'SNEWS' in pipelines:
165 tl, th = tl_snews, th_snews
166 # Use Targeted search window if CBC or Burst
167 elif not {'SubGRB', 'SubGRBTargeted'}.isdisjoint(searches):
168 if 'Fermi' in pipelines:
169 tl, th = tl_subfermi, th_subfermi
170 elif 'Swift' in pipelines:
171 tl, th = tl_subswift, th_subswift
172 else:
173 raise ValueError('Specify Fermi or Swift as pipeline when ' +
174 'launching subthreshold search')
175 elif group == 'CBC':
176 tl, th = tl_cbc, th_cbc
177 elif group == 'Burst':
178 tl, th = tl_burst, th_burst
179 else:
180 raise ValueError('Invalid RAVEN search request for {0}'.format(
181 gracedb_id))
182 if 'S' in gracedb_id:
183 # If triggering on a superevent, need to reverse the time window
184 tl, th = -th, -tl
186 return tl, th
189@app.task(shared=False)
190def search(gracedb_id, alert_object, tl=-5, th=5, group=None,
191 pipelines=[], searches=[], se_searches=[]):
192 """Perform ligo-raven search to look for coincidences. This function
193 does a query of GraceDB and uploads a log message of the result(s).
195 Parameters
196 ----------
197 gracedb_id : str
198 GraceDB ID of the trigger that launched RAVEN
199 alert_object : dict
200 Alert dictionary
201 tl : int
202 Lower bound of search window in seconds
203 th : int
204 Upper bound of search window in seconds
205 group : str
206 Burst or CBC
207 pipelines : list
208 List of external trigger pipelines for performing coincidence search
209 against
210 searches : list
211 List of external trigger searches
212 se_searches : list
213 List of superevent searches
215 Returns
216 -------
217 results : list
218 List with the dictionaries of related GraceDB events
220 """
221 return ligo.raven.search.search(gracedb_id, tl, th,
222 event_dict=alert_object,
223 gracedb=gracedb.client,
224 group=group, pipelines=pipelines,
225 searches=searches,
226 se_searches=se_searches)
229@app.task(shared=False)
230def raven_pipeline(raven_search_results, gracedb_id, alert_object, tl, th,
231 gw_group, use_superevent_skymap=None):
232 """Executes the full RAVEN pipeline, including adding
233 the external trigger to the superevent, calculating the
234 coincidence false alarm rate, applying 'EM_COINC' to the
235 appropriate events, and checking whether the candidate(s) pass all
236 publishing conditions.
238 Parameters
239 ----------
240 raven_search_results : list
241 List of dictionaries of each related gracedb trigger
242 gracedb_id : str
243 GraceDB ID of the trigger that launched RAVEN
244 alert_object : dict
245 Alert dictionary, either a superevent or an external event
246 tl : int
247 Lower bound of search window in seconds
248 th : int
249 Upper bound of search window in seconds
250 gw_group : str
251 Burst or CBC
252 use_superevent_skymap : bool
253 If True/False, use/don't use skymap info from superevent.
254 Else if None, checks SKYMAP_READY label in external event.
256 """
257 if not raven_search_results:
258 return
259 if 'S' not in gracedb_id:
260 raven_search_results = preferred_superevent(raven_search_results)
261 for result in raven_search_results:
262 if 'S' in gracedb_id:
263 superevent_id = gracedb_id
264 exttrig_id = result['graceid']
265 superevent = alert_object
266 ext_event = result
267 else:
268 superevent_id = result['superevent_id']
269 exttrig_id = gracedb_id
270 superevent = result
271 ext_event = alert_object
272 # Don't continue if it is a different superevent than previous one.
273 if ext_event['superevent'] is not None \
274 and ext_event['superevent'] != superevent['superevent_id']:
275 return
277 canvas = (
278 gracedb.add_event_to_superevent.si(superevent_id, exttrig_id)
279 |
280 calculate_coincidence_far.si(
281 superevent, ext_event, tl, th,
282 use_superevent_skymap=use_superevent_skymap
283 )
284 |
285 group(gracedb.create_label.si('EM_COINC', superevent_id),
286 gracedb.create_label.si('EM_COINC', exttrig_id),
287 trigger_raven_alert.s(superevent, gracedb_id,
288 ext_event, gw_group))
289 )
290 canvas.delay()
293@app.task(shared=False)
294def preferred_superevent(raven_search_results):
295 """Chooses the superevent with the lowest FAR for an external
296 event to be added to. This is to prevent errors from trying to
297 add one external event to multiple superevents.
299 Parameters
300 ----------
301 raven_search_results : list
302 List of dictionaries of each related gracedb trigger
304 Returns
305 -------
306 superevent : list
307 List containing single chosen superevent
309 """
310 minfar, idx = min((result['far'], idx) for (idx, result) in
311 enumerate(raven_search_results))
312 return [raven_search_results[idx]]
315@app.task(shared=False)
316def update_coinc_far(coinc_far_dict, superevent, ext_event):
317 """Update joint info in superevent based on the current preferred
318 coincidence. In order of priority, the determing conditions are the
319 following:
321 * A SNEWS coincidence is preferred over GRB.
322 * Likely astrophysical external candidates are preferred over likely
323 non-astrophysical candidates.
324 * Candidates that pass publishing thresholds are preferred over those
325 that do not.
326 * A spacetime joint FAR over a time-only joint FAR.
327 * Lower joint FARs are preferred over higher joint FARs.
329 Parameters
330 ----------
331 coinc_far_dict : dict
332 Dictionary containing coincidence false alarm rate results from
333 RAVEN
334 superevent : dict
335 Superevent dictionary
336 ext_event: dict
337 External event dictionary
339 Returns
340 -------
341 joint_far : dict
342 Dictionary containing joint false alarm rate passed to the function
343 as an initial argument
345 """
346 # Get graceids
347 superevent_id = superevent['superevent_id']
348 ext_id = ext_event['graceid']
350 # Get the latest info to prevent race condition
351 superevent_latest = gracedb.get_superevent(superevent_id)
353 # Joint FAR isn't computed for SNEWS coincidence
354 # Choose SNEWS coincidence over any other type of coincidence
355 if ext_event['pipeline'] == 'SNEWS':
356 gracedb.update_superevent(superevent_id, em_type=ext_id,
357 time_coinc_far=None,
358 space_coinc_far=None)
359 return coinc_far_dict
361 # Load needed variables
362 infty = float('inf')
363 new_time_far = coinc_far_dict['temporal_coinc_far']
364 new_space_far = coinc_far_dict['spatiotemporal_coinc_far']
365 # Map None to infinity to make logic easier
366 new_space_far_f = new_space_far if new_space_far else infty
367 old_time_far = superevent_latest['time_coinc_far']
368 old_time_far_f = old_time_far if old_time_far else infty
369 old_space_far = superevent_latest['space_coinc_far']
370 old_space_far_f = old_space_far if old_space_far else infty
371 is_far_improved = (new_space_far_f < old_space_far_f or
372 (new_time_far < old_time_far_f and
373 old_space_far_f == infty))
375 if superevent_latest['em_type']:
376 # If previous preferred external event, load to compare
377 emtype_event = gracedb.get_event(superevent_latest['em_type'])
378 # Don't overwrite SNEWS with GRB event
379 snews_to_grb = \
380 (emtype_event['pipeline'] == 'SNEWS' and
381 ext_event['pipeline'] != 'SNEWS')
382 # Determine which events are likely real or not
383 is_old_grb_real, is_new_grb_real = \
384 ('NOT_GRB' not in emtype_event['labels'],
385 'NOT_GRB' not in ext_event['labels'])
386 is_old_raven_alert, is_new_raven_alert = \
387 ('RAVEN_ALERT' in emtype_event['labels'],
388 'RAVEN_ALERT' in ext_event['labels'])
389 # Use new event only if it is better old event information
390 is_event_improved = ((is_new_grb_real and not is_old_grb_real) or
391 (is_new_raven_alert and not is_old_raven_alert))
392 # if both real or both not, use FAR to differentiate
393 if is_old_grb_real == is_new_grb_real \
394 and is_old_raven_alert == is_new_raven_alert:
395 is_event_improved = is_far_improved
396 else:
397 snews_to_grb = False
398 is_event_improved = is_far_improved
400 if is_event_improved and not snews_to_grb:
401 gracedb.update_superevent(superevent_id, em_type=ext_id,
402 time_coinc_far=new_time_far,
403 space_coinc_far=new_space_far)
404 return coinc_far_dict
407@app.task(shared=False)
408def trigger_raven_alert(coinc_far_dict, superevent, gracedb_id,
409 ext_event, gw_group):
410 """Determine whether an event should be published as a preliminary alert.
411 If yes, then triggers an alert by applying `RAVEN_ALERT` to the preferred
412 event.
414 All of the following conditions must be true to either trigger an alert or
415 include coincidence info into the next alert include:
417 * The external event must be a threshold GRB or SNEWS event.
418 * If triggered on a SNEWS event, the GW false alarm rate must pass
419 :obj:`~gwcelery.conf.snews_gw_far_threshold`.
420 * The event's RAVEN coincidence false alarm rate, weighted by the
421 group-specific trials factor as specified by the
422 :obj:`~gwcelery.conf.preliminary_alert_trials_factor` configuration
423 setting, is less than or equal to
424 :obj:`~gwcelery.conf.preliminary_alert_far_threshold`. This FAR also
425 must not be negative.
426 * If the coincidence involves a GRB, then both sky maps must be present.
428 Parameters
429 ----------
430 coinc_far_dict : dict
431 Dictionary containing coincidence false alarm rate results from
432 RAVEN
433 superevent : dict
434 Superevent dictionary
435 gracedb_id : str
436 GraceDB ID of the trigger that launched RAVEN
437 ext_event : dict
438 External event dictionary
439 gw_group : str
440 Burst or CBC
442 """
443 preferred_gwevent_id = superevent['preferred_event']
444 superevent_id = superevent['superevent_id']
445 ext_id = ext_event['graceid']
446 # Specify group is not given, currently missing for subthreshold searches
447 gw_group = gw_group or superevent['preferred_event_data']['group']
448 gw_group = gw_group.lower()
449 gw_search = superevent['preferred_event_data']['search'].lower()
450 pipeline = ext_event['pipeline']
451 if gw_search in app.conf['significant_alert_trials_factor'][gw_group]:
452 trials_factor = \
453 app.conf['significant_alert_trials_factor'][gw_group][gw_search]
454 else:
455 trials_factor = 1
456 missing_skymap = True
457 comments = []
458 messages = []
460 # Since the significance of SNEWS triggers is so high, we will publish
461 # any trigger coincident with a decently significant GW candidate
462 if 'SNEWS' == pipeline:
463 gw_far = superevent['far']
464 far_type = 'gw'
465 far_threshold = app.conf['snews_gw_far_threshold']
466 pass_far_threshold = gw_far * trials_factor < far_threshold
467 is_far_negative = gw_far < 0
468 is_ext_subthreshold = False
469 missing_skymap = False
470 # Set coinc FAR to gw FAR only for the sake of a message below
471 time_coinc_far = space_coinc_far = coinc_far = None
472 coinc_far_f = gw_far
474 # The GBM team requested we not send automatic alerts from subthreshold
475 # GRBs. This checks that at least one threshold GRB present as well as
476 # the coinc far
477 else:
478 # check whether the GRB is threshold or sub-thresholds
479 is_ext_subthreshold = 'SubGRB' == ext_event['search']
481 # Use spatial FAR if available, otherwise use temporal
482 time_coinc_far = coinc_far_dict['temporal_coinc_far']
483 space_coinc_far = coinc_far_dict['spatiotemporal_coinc_far']
484 if space_coinc_far is not None:
485 coinc_far = space_coinc_far
486 missing_skymap = False
487 else:
488 coinc_far = time_coinc_far
490 far_type = 'joint'
491 if gw_search in app.conf['significant_alert_far_threshold'][gw_group]:
492 far_threshold = (
493 app.conf['significant_alert_far_threshold'][gw_group]
494 [gw_search]
495 )
496 else:
497 # Fallback in case an event is uploaded to an unlisted search
498 far_threshold = -1 * float('inf')
499 coinc_far_f = coinc_far * trials_factor * (trials_factor - 1.)
500 pass_far_threshold = coinc_far_f <= far_threshold
501 is_far_negative = coinc_far_f < 0
503 # Get most recent labels to prevent race conditions
504 ext_labels = gracedb.get_labels(ext_id)
505 no_previous_alert = {'RAVEN_ALERT'}.isdisjoint(ext_labels)
506 likely_real_ext_event = {'NOT_GRB'}.isdisjoint(ext_labels)
507 is_test_event = (superevent['preferred_event_data']['group'] == 'Test' or
508 ext_event['group'] == 'Test')
510 # If publishable, trigger an alert by applying `RAVEN_ALERT` label to
511 # preferred event
512 if pass_far_threshold and not is_ext_subthreshold and \
513 likely_real_ext_event and not missing_skymap and \
514 not is_test_event and no_previous_alert and \
515 not is_far_negative:
516 comments.append(('RAVEN: publishing criteria met for {0}-{1}. '
517 'Triggering RAVEN alert'.format(
518 preferred_gwevent_id, ext_id)))
519 # Add label to local dictionary and to event on GraceDB server
520 # NOTE: We may prefer to apply the superevent label first and the grab
521 # labels to refresh in the future
522 superevent['labels'] += 'RAVEN_ALERT'
523 # Add RAVEN_ALERT to preferred event last to avoid race conditions
524 # where superevent is expected to have it once alert is issued
525 alert_canvas = (
526 gracedb.create_label.si('RAVEN_ALERT', superevent_id)
527 |
528 gracedb.create_label.si('HIGH_PROFILE', superevent_id)
529 |
530 gracedb.create_label.si('RAVEN_ALERT', ext_id)
531 |
532 gracedb.create_label.si('RAVEN_ALERT', preferred_gwevent_id)
533 )
534 else:
535 alert_canvas = identity.si()
536 if not pass_far_threshold:
537 comments.append(('RAVEN: publishing criteria not met for {0}-{1},'
538 ' {2} FAR (w/ trials) too large '
539 '({3:.4g} > {4:.4g})'.format(
540 preferred_gwevent_id, ext_id, far_type,
541 coinc_far_f, far_threshold)))
542 if is_ext_subthreshold:
543 comments.append(('RAVEN: publishing criteria not met for {0}-{1},'
544 ' {1} is subthreshold'.format(preferred_gwevent_id,
545 ext_id)))
546 if not likely_real_ext_event:
547 ext_far = ext_event['far']
548 grb_far_threshold = \
549 app.conf['raven_targeted_far_thresholds']['GRB'][pipeline]
550 extra_sentence = ''
551 if ext_far is not None and grb_far_threshold < ext_far:
552 extra_sentence = (' This due to the GRB FAR being too high '
553 '({0:.4g} > {1:.4g})'.format(
554 ext_far, grb_far_threshold))
555 comments.append(('RAVEN: publishing criteria not met for {0}-{1},'
556 ' {1} is likely non-astrophysical.{2}'.format(
557 preferred_gwevent_id, ext_id, extra_sentence)))
558 if is_test_event:
559 comments.append('RAVEN: {0}-{1} is non-astrophysical, '
560 'at least one event is a Test event'.format(
561 preferred_gwevent_id, ext_id))
562 if missing_skymap:
563 comments.append('RAVEN: Will only publish GRB coincidence '
564 'if spatial-temporal FAR is present. '
565 'Waiting for both sky maps to be available '
566 'first.')
567 if is_far_negative:
568 comments.append(('RAVEN: publishing criteria not met for {0}-{1},'
569 ' {2} FAR is negative ({3:.4g})'.format(
570 preferred_gwevent_id, ext_id, far_type,
571 coinc_far_f)))
572 for comment in comments:
573 messages.append(gracedb.upload.si(None, None, superevent_id, comment,
574 tags=['ext_coinc']))
576 # Update coincidence FAR with latest info, including the application of
577 # RAVEN_ALERT, then issue alert
578 (
579 update_coinc_far.si(coinc_far_dict, superevent, ext_event)
580 |
581 group(
582 alert_canvas,
583 external_skymaps.plot_overlap_integral.s(superevent, ext_event),
584 *messages
585 )
586 ).delay()
589@app.task(shared=False)
590def sog_paper_pipeline(ext_event, superevent):
591 """Determine whether an a speed of gravity measurment manuscript should be
592 created for a given coincidence. This is denoted by applying the
593 ``SOG_READY`` label to a superevent.
595 All of the following conditions must be true for a SoG paper:
597 * The coincidence is significant and FARs more significant than in
598 :obj:`~sog_paper_far_threshold`.
599 * The external event is a high-significance GRB and from an MOU partner.
600 * The GW event is a CBC candidate.
602 Parameters
603 ----------
604 superevent : dict
605 Superevent dictionary
606 ext_event : dict
607 External event dictionary
609 """
610 gw_far = superevent['far']
611 coinc_far = superevent['space_coinc_far']
612 gw_far_threshold = app.conf['sog_paper_far_threshold']['gw']
613 joint_far_threshold = app.conf['sog_paper_far_threshold']['joint']
615 # Check publishing conditions
616 pass_gw_far_threshold = gw_far <= gw_far_threshold
617 pass_joint_far_threshold = coinc_far <= joint_far_threshold
618 is_grb = ext_event['search'] in ['GRB', 'MDC']
619 is_mou_partner = ext_event['pipeline'] in ['Fermi', 'Swift']
620 is_cbc = superevent['preferred_event_data']['group'] == 'CBC'
622 if is_grb and is_cbc and is_mou_partner and \
623 pass_gw_far_threshold and pass_joint_far_threshold:
624 # Trigger SOG_READY label alert to alert SOG analysts
625 gracedb.create_label.si('SOG_READY',
626 superevent['superevent_id']).delay()