Coverage for gwcelery/tasks/rrt_utils.py: 100%

1import json 

2 

3from ligo.skymap.io import read_sky_map 

4from ligo.skymap.postprocess.crossmatch import crossmatch 

5 

6from .. import app 

7from ..util.tempfile import NamedTemporaryFile 

8from . import gracedb 

9 

10 

11@app.task(shared=False) 

12def check_high_profile(skymap, em_bright, 

13 p_astro, superevent): 

14 superevent_id = superevent['superevent_id'] 

15 # conditions are defined in L2100046 

16 # RAVEN_ALERT HIGH_PROFILE is implemented in raven.py 

17 # Checking if the label is applied beforehand 

18 if 'HIGH_PROFILE' in superevent['labels']: 

19 return "HIGH_PROFILE already applied" 

20 

21 # low-far unmodelled burst condition 

22 far_list = [] 

23 gw_events = superevent["gw_events"] 

24 for event in gw_events: 

25 events_dict = gracedb.get_event(event) 

26 far_list.append({"group": events_dict["group"], 

27 "search": events_dict["search"], 

28 "far": events_dict["far"]}) 

29 far_list_sorted = sorted(far_list, key=lambda k: k["far"]) 

30 

31 if far_list_sorted[0]["group"] == "Burst" and \ 

32 far_list_sorted[0]["search"] != "BBH": 

33 gracedb.create_label.si( 

34 'HIGH_PROFILE', superevent_id).delay() 

35 return "Event with the lowest FAR is a Burst event. Applying label" 

36 

37 # annotation number condition 

38 preferred_event = superevent['preferred_event_data'] 

39 if preferred_event["group"] == "CBC": 

40 em_bright_dict = json.loads(em_bright) 

41 has_remnant = em_bright_dict['HasRemnant'] 

42 

43 pastro_dict = json.loads(p_astro) 

44 p_bns = pastro_dict['BNS'] 

45 p_terr = pastro_dict['Terrestrial'] 

46 p_nsbh = pastro_dict['NSBH'] 

47 

48 with NamedTemporaryFile(content=skymap) as skymap_file: 

49 gw_skymap = read_sky_map(skymap_file.name, moc=True) 

50 cl = 90 

51 result = crossmatch(gw_skymap, contours=[cl / 100]) 

52 sky_area = result.contour_areas[0] 

53 

54 if p_terr < 0.5: 

55 if (p_bns > 0.1 or p_nsbh > 0.1 or has_remnant > 0.1 or sky_area < 100): # noqa: E501 

56 gracedb.create_label.si( 

57 'HIGH_PROFILE', superevent_id).delay() 

58 return "Annotations condition satisfied. Applying label" 

59 return "No conditions satisfied. Skipping"