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(ignore_result=True, 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_with_log( 

34 log_message='Superevent labeled ' 

35 '<span color="red">HIGH_PROFILE</span> ' 

36 'since event with lowest FAR is a Burst event.', 

37 label='HIGH_PROFILE', 

38 tags=['em_follow'], 

39 superevent_id=superevent_id) 

40 return 

41 

42 # annotation number condition 

43 preferred_event = superevent['preferred_event_data'] 

44 if preferred_event["search"] == "SSM": 

45 gracedb.create_label_with_log( 

46 log_message='Superevent labeled ' 

47 '<span color="red">HIGH_PROFILE</span> ' 

48 'since preferred event is from SSM search.', 

49 label='HIGH_PROFILE', 

50 tags=['em_follow'], 

51 superevent_id=superevent_id) 

52 return 

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

54 em_bright_dict = json.loads(em_bright) 

55 has_remnant = em_bright_dict['HasRemnant'] 

56 pastro_dict = json.loads(p_astro) 

57 p_bns = pastro_dict['BNS'] 

58 p_terr = pastro_dict['Terrestrial'] 

59 p_nsbh = pastro_dict['NSBH'] 

60 

61 with NamedTemporaryFile(content=skymap) as skymap_file: 

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

63 cl = 90 

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

65 sky_area = result.contour_areas[0] 

66 # This is commented out while we figure out the distance cutoff 

67 # is_far_away = not (gw_skymap.meta.get('distmean', np.nan) < 2000) 

68 if p_terr < 0.5: 

69 if p_bns > 0.1: 

70 gracedb.create_label_with_log( 

71 log_message='Superevent labeled ' 

72 '<span color="red">HIGH_PROFILE</span> since p_BNS > 10%.', # noqa: E501 

73 label='HIGH_PROFILE', 

74 tags=['em_follow'], 

75 superevent_id=superevent_id) 

76 return 

77 elif p_nsbh > 0.1: 

78 gracedb.create_label_with_log( 

79 log_message='Superevent labeled ' 

80 '<span color="red">HIGH_PROFILE</span> since p_NSBH > 10%.', # noqa: E501 

81 label='HIGH_PROFILE', 

82 tags=['em_follow'], 

83 superevent_id=superevent_id) 

84 return 

85 elif has_remnant > 0.1: 

86 gracedb.create_label_with_log( 

87 log_message='Superevent labeled ' 

88 '<span color="red">HIGH_PROFILE</span> since ' 

89 'p_HasRemnant > 10%.', 

90 label='HIGH_PROFILE', 

91 tags=['em_follow'], 

92 superevent_id=superevent_id) 

93 return 

94 elif sky_area < 100: 

95 gracedb.create_label_with_log( 

96 log_message='Superevent labeled ' 

97 '<span color="red">HIGH_PROFILE</span> since area of ' 

98 '90% confidence level in the skymap is < 100 sq.deg.', 

99 label='HIGH_PROFILE', 

100 tags=['em_follow'], 

101 superevent_id=superevent_id) 

102 return 

103 return "No conditions satisfied. Skipping"